zhangyisong/test_led_strip
1
0
Fork 0
forked from EmbeddedTeam/app_photomagnetic
Code Pull Requests Activity

Compare commits

base: zhangyisong:main
Branches Tags
zhangyisong:main
zhangyisong:feat/add_mcuboot
zhangyisong:feat/multi_inf_sensors
zhangyisong:feat/b_version
zhangyisong:Light-Initialization
zhangyisong:feat_coro
zhangyisong:v1.0.0
zhangyisong:v0.3.2
zhangyisong:v0.3.1
zhangyisong:v0.3.0
..
compare: zhangyisong:Light-Initialization
Branches Tags
zhangyisong:feat/add_mcuboot
zhangyisong:main
zhangyisong:feat/multi_inf_sensors
zhangyisong:feat/b_version
zhangyisong:Light-Initialization
zhangyisong:feat_coro
zhangyisong:v1.0.0
zhangyisong:v0.3.2
zhangyisong:v0.3.1
zhangyisong:v0.3.0

1 Commits
main ... Light-Init

Author SHA1 Message Date
zhangjing
141c3b8aee 🐞 fix(app_photomagnetic):1.main.cpp 添加初始化rgb(红绿蓝闪完，蓝灯常亮) 2026-03-27 22:39:59 +08:00
25 changed files with 120 additions and 2306 deletions
Show Stats Expand all files Collapse all files

19
CMakeLists.txt
View File

@ -3,24 +3,7 @@ cmake_minimum_required(VERSION 3.20.0)
find_package(Zephyr REQUIRED HINTS $ENV{ZEPHYR_BASE})
project(app_photomagnetic)
target_include_directories(app PRIVATE include)
target_sources(app PRIVATE src/main.cpp)
if(CONFIG_SAMPLE_HELLOWORLD)
message("CONFIG_SAMPLE_HELLOWORLD is enabled")
target_sources(app PRIVATE src/sample_hello_world.cpp)
elseif(CONFIG_SAMPLE_NTC)
message("CONFIG_SAMPLE_NTC is enabled")
target_sources(app PRIVATE src/sample_ntc.cpp)
else()
message("build main app")
target_sources(app PRIVATE
src/main.cpp
src/led.cpp
src/temp.cpp
src/watdog.cpp
src/com.cpp
)
endif()

8
Kconfig
View File

@ -1,8 +0,0 @@
menu "app thermotherapy"
source "Kconfig.zephyr"
endmenu
config SAMPLE_HELLOWORLD
bool "Hello World sample"
config SAMPLE_NTC
bool "NTC sample"

18
boards/dr2501a_g070rb.overlay Normal file
View File

@ -0,0 +1,18 @@
&indicator {
led-strip = <&led_strip>;
compatible = "led-strip-indicator";
en-gpios = <&gpiob 13 GPIO_ACTIVE_HIGH>;
standby {
rgb = <0 0 255>; // Blue
};
running {
rgb = <0 255 0>; // Green
};
pause {
rgb = <254 254 0>;
// interval-ms = <500>;
};
error {
rgb = <255 0 0>; //Red
};
};

71
boards/use_ms.overlay
View File

@ -1,71 +0,0 @@
&ch9438_uart0 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor0 {
mode = "wrist";
};
&ch9438_uart1 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor1 {
mode = "wrist";
};
&ch9438_uart2 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor2 {
mode = "wrist";
};
&ch9438_uart3 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor3 {
mode = "wrist";
};
&ch9438_uart4 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor4 {
mode = "wrist";
};
&ch9438_uart5 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor5 {
mode = "wrist";
};
&ch9438_uart6 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor6 {
mode = "wrist";
};
&ch9438_uart7 {
status = "okay";
current-speed = <9600>;
};
&gd_sensor7 {
mode = "wrist";
};

317
doc/光磁头软件架构文档.md
View File

@ -1,317 +0,0 @@
# 光磁头软件架构文档
> 本文档为光磁头（小板 / 温度传感端）固件的**设计规范**，源码实现应遵循本文档的模块划分、数据流与设计约束。
> 配套协议：[`通讯协议.md`](./通讯协议.md)
> 硬件平台：STM32G0B0（板级 `dr2501a_g0b0ce`）
---
## 1. 系统定位
光磁头是光磁热疗系统的**温度传感端**，通过一根 UART 线与主板（控制端）通信，向主板提供 9 路皮肤温度反馈，供主板 PID 温控闭环使用。
| 职责 | 说明 |
|---|---|
| 温度采集 | 9 路 godtek 温度传感器（8 路经 CH9438 扩展 + 1 路片内直连） |
| 温度上报 | 每 20ms 主动上报 9 路**最大值**（温度闭环反馈） |
| 命令响应 | 响应主板查询（ID / 温度）与运行状态设置 |
| 状态指示 | 按主板运行状态点亮红外灯与灯带 |
```mermaid
flowchart LR
subgraph 光磁头[光磁头 传感端]
S[9 路 godtek 传感器] -->|UART 触发| C[采集聚合 infrared]
C -->|20ms 周期| T[温度上报 temp]
T --> P[协议 com<br/>usart1 0x7EE7]
P -->|运行状态| L[LED 指示 led]
end
P <-->|UART 115200| M[主板 控制端]
```
---
## 2. 硬件资源
| 外设 | 用途 | 关键配置 |
|---|---|---|
| `spi2` + CH9438（INT# PB13） | SPI→8×UART 扩展，传感器 ch0~7 | SPI 500kHz，8 口均 9600 8N1（overlay `use_ms.overlay`） |
| `usart2`（PA2/PA3） | 片内第 9 路传感器（ch8） | 9600 8N1 |
| `usart1`（PA9/PA10） | 与主板通信协议口 | 115200，帧头 0x7EE7 + CRC16 |
| `usart3`（PA5/PB0） | 调试 console | 115200 |
| `spi1` + WS2812（40 灯） | 灯带指示（standby/running/pause/error） | SPI 4MHz |
| `led_inf`（PB1） | 红外加热指示 LED | 运行中点亮 |
| `led_mcu_state`（PA15） | 心跳灯 | 1s 周期闪烁 |
| IWDG | 系统看门狗 | 20ms 喂狗 |
```mermaid
graph TD
subgraph MCU[STM32G0B0]
U1[usart1 协议口<br/>115200] -->|0x7EE7 帧| MB[主板]
U2[usart2 第9路<br/>9600] --> G8[godtek ch8]
U3[usart3 console]
SP2[spi2 500kHz] --> CH[CH9438<br/>INT# PB13]
CH --> G0[godtek ch0]
CH --> G1[godtek ch1]
CH --> G7[godtek ch7]
SP1[spi1 4MHz] --> WS[WS2812 灯带 ×40]
LED1[led_inf PB1]
LED2[led_mcu_state PA15]
WDT[IWDG]
end
```
> 传感器工作模式：8 路外置为 `wrist`（0xAC 测量命令），片内第 9 路默认 `surface`（0xAA）。
---
## 3. 软件架构
```mermaid
graph TD
subgraph 应用层
MAIN[main 主线程]
TEMP[temp 采集调度<br/>20ms 上报]
COM[com 协议处理<br/>回调表/上电发ID]
LED[led 状态指示]
WDG[watdog 喂狗]
end
subgraph 聚合层
IR[infrared 9路聚合<br/>触发注册/缓存/最大值]
end
subgraph 驱动层
UC[uart_com 帧协议<br/>0x7EE7+CRC16]
GD[godtek_temp 传感器驱动<br/>组帧/校验/触发]
CH8[ch9438 SPI→8×UART]
IND[led_strip_indicator 灯带]
end
TEMP --> IR
TEMP --> COM
COM --> UC
IR --> GD
GD --> CH8
GD --> U2[usart2]
LED --> COM
LED --> IND
MAIN -.后台驻留.-> TEMP
```
### 模块职责
| 模块 | 职责 | 关键约束 |
|---|---|---|
| `infrared`（聚合层） | 9 路传感器统一抽象：设备列表、触发注册、通道缓存、取最大值 | 缓存须 ISR 安全；通道编号固定 ch0~7=CH9438、ch8=片内 |
| `temp`（应用层） | 采集触发注册 + 20ms 周期上报最大值 | 触发注册必须在传感器驱动初始化之后 |
| `com`（应用层） | 协议帧收发、命令回调、运行状态信号广播 | 回调在系统 workqueue 线程执行；上电主动发 ID |
| `led`（应用层） | 运行状态 → 红外灯 + 灯带；心跳灯 | 状态变化由 com 信号驱动，不轮询 |
| `watdog`（应用层） | IWDG 喂狗 | 20ms 周期；配置失败不得阻塞启动 |
| `main`（应用层） | 入口 + 后台驻留 | 各模块由 SYS_INIT 初始化 |
---
## 4. 数据流
### 4.1 温度采集链（触发驱动，无轮询）
9 路传感器**全部由 UART 数据触发驱动**，不占用 CPU 轮询：
```mermaid
flowchart LR
S[godtek 传感器自发上报] --> CH[CH9438 INT# 中断<br/>或 usart2 RX 中断]
CH --> WQ[系统 workqueue<br/>或 ISR]
WQ --> GD[godtek 驱动<br/>组帧 + 校验]
GD -->|有效帧| TRIG[触发回调<br/>更新第 I 路缓存]
TRIG --> CACHE[volatile 通道缓存]
```
- **帧校验**：`+`/`-` 符号 + 6 位十进制数字 + 温度范围（-40.0 ~ +125.0°C），坏帧直接丢弃，不上报
- **上下文**：CH9438 端口回调在系统 workqueue（SPI 安全）；片内 usart2 回调在 ISR —— 回调内不得做任何阻塞操作
### 4.2 温度上报链（20ms 周期）
```mermaid
flowchart LR
T[temp 周期任务<br/>20ms] --> M[取 9 路缓存最大值]
M --> C[组帧上报]
C -->|0x7EE7 帧 00 02 整数 小数 CRC| U[usart1 → 主板]
```
温度值格式：×10 整数（如 `345` = 34.5°C），上报数据 2 字节 `[整数][小数]`。
### 4.3 命令响应链（RX ISR → workqueue）
```mermaid
flowchart LR
RX[usart1 RX ISR<br/>逐字节帧状态机] -->|完整帧| P[拷贝到 pending 缓冲<br/>提交 work]
P --> WQ2[系统 workqueue<br/>执行命令回调]
WQ2 --> H[ID / 温度查询 / 运行状态]
H --> LED2[运行状态 → LED 指示]
H --> R[ID / 温度查询 → 回帧]
```
> 命令回调必须在 workqueue 线程上下文执行（ISR 只做轻量拷贝），可安全使用 `printk` / `uart_poll_out`，避免 ISR 内阻塞导致温度上报停止。
---
## 5. 通信协议
帧结构（详见 `通讯协议.md`）：
```
| 7E | E7 | CMD | LEN | DATA[0..LEN-1] | CRC_L | CRC_H |
|----- 帧头 -----| | |-- CRC16 Modbus (lsb-msb) --|
```
| 指令 | 方向 | 内容 | 响应行为 |
|---|---|---|---|
| `0x00` 温度 | 小板→主板（主动） | 每 20ms 上报最大值，2 字节 `[整数][小数]` | — |
| `0x00` 温度 | 主板请求→小板回复 | 请求时回复当前最大值 | 查询回复 |
| `0x01` ID | 小板→主板（主动） | **上电主动发一次**，3 字节（hwinfo 前 3 字节） | — |
| `0x01` ID | 主板请求→小板回复 | 请求时回复 | 查询回复 |
| `0x02` 运行状态 | 主板→小板 | 1 字节：0x00 待机 / 0x01 运行 / 0x02 暂停 / 0x03 故障 | 更新状态并驱动 LED |
```mermaid
sequenceDiagram
participant S as 小板 光磁头
participant M as 主板 控制端
Note over S: 系统上电
S->>M: 【主动】ID 帧 (0x01)
loop 每 20ms
S->>M: 【主动】温度最大值帧 (0x00)
end
Note over M: 运行状态变化
M->>S: 运行状态帧 (0x02) → LED 指示
opt 主板查询
M->>S: 请求 ID (0x01)
S-->>M: 回复 ID
M->>S: 请求温度 (0x00)
S-->>M: 回复温度最大值
end
```
---
## 6. 初始化顺序
各模块通过 SYS_INIT 按以下顺序初始化（同优先级内不得存在跨模块依赖）：
| 阶段 | 优先级 | 模块 | 说明 |
|---|---|---|---|
| POST_KERNEL | 驱动默认 | ch9438 | SPI 初始化、**波特率配置（9600，含 FCR 时序处理）**、中断使能 |
| POST_KERNEL | 传感器默认 | godtek | UART 回调注册、500ms 周期测量命令 |
| APPLICATION | 40 | led | GPIO / 灯带初始化 + 注册状态槽 |
| APPLICATION | 45 | temp | **触发注册**（必须在 godtek 之后）+ 启动 20ms 上报 |
| APPLICATION | 50 | com | 协议回调表注册 + **上电主动发 ID** |
| APPLICATION | 60 | watdog | IWDG 安装 + 启动喂狗 |
| APPLICATION | 90 | ch9438 flush | POR 残留 FIFO 清空（边沿中断补偿） |
| — | — | main | 主循环（后台线程） |
```mermaid
sequenceDiagram
participant B as 启动
participant D as 驱动层
participant A as 应用层
B->>D: ch9438 初始化（波特率 9600 时序）
B->>D: godtek 初始化（回调 + 500ms 测量命令）
B->>A: led 初始化（APPLICATION 40）
B->>A: temp 初始化（APPLICATION 45）触发注册
B->>A: com 初始化（APPLICATION 50）上电发 ID
B->>A: watdog 初始化（APPLICATION 60）
B->>A: ch9438 flush（APPLICATION 90）
A->>D: 传感器数据触发 → 缓存 → 20ms 上报
```
---
## 7. 关键设计决策
### 7.1 波特率 POR 时序（重点坑）
**现象**：烧录后波特率正常（9600），断电重启后回退为 115200，部分端口只收到 `0x00`。
**根因**：CH9438 的 FCR 复位位（RFIFORST/TFIFORST）会**异步重置波特率分频器**。POR 冷启动时 FCR 复位持续较久，把紧随其后的 9600 配置（LCR/DLL/DLM）清回芯片默认值 115200；烧录时芯片未断电、复位瞬间完成，因此正常。
**修复**（初始化顺序）：
```
写 LCR(DLAB=1) → 写 DLL/DLM(9600) → 验证
→ 引脚配置 → FCR=0x07（复位+使能，触发发生器重锁存）→ 1ms 延时
→ 重写 LCR(DLAB=1) → 重写 DLL/DLM(9600) → 回读验证（连 LCR 一起检查）→ LCR(DLAB=0)
```
**回读验证假象**：DLAB 未锁存时，读 "DLL" 实际读到 RBR（uart7 短接回环的 0x20）、"DLM" 读到 IER（0x4E）——自洽的谎言。**必须连 LCR 一起回读比对**，确认 DLAB 真正置位；再加 8 次 ×20ms 重试兜底（POR 后 UART 模块可能晚就绪，写入被静默丢弃）。
### 7.2 触发驱动采集
9 路全部由 UART 数据触发更新缓存，无轮询，响应实时；发送侧固定 20ms 周期取缓存最大值，采集与上报解耦。
### 7.3 缓存线程安全
volatile 字段 + 编译期通道索引，触发回调（ISR / workqueue 双上下文）与查询无锁，避免死锁与中断延迟。
### 7.4 回调上下文
协议命令回调全部在系统 workqueue 线程执行，ISR 只做数据拷贝 —— 避免 ISR 内 `uart_poll_out` / `printk` 阻塞导致温度上报停止。
### 7.5 通道编号稳定
`DT_FOREACH` 按设备树节点顺序展开，若 usart2 排在 spi2 前会导致通道错位（ch0 变成片内传感器）。通过按父节点 compatible 分两段构建设备列表，保证 ch0~7 = CH9438、ch8 = 片内。
### 7.6 坏帧丢弃
帧校验含符号、6 位数字与温度范围三重检查，字节丢失/错位产生的坏帧直接丢弃，不污染缓存。
---
## 8. 调试与诊断
### 8.1 打印规范
所有诊断打印由 `APP_DEBUG_PRINT` 宏控制（各模块默认 0），默认只保留启动确认与错误打印：
| 宏 | 位置 | 打印内容 |
|---|---|---|
| `APP_DEBUG_PRINT=1` | ch9438 | 波特率回读、work 计数、FIFO 扫描、运行期诊断 |
| `APP_DEBUG_PRINT=1` | godtek | 原始字节 dump、命令追踪 |
| `APP_DEBUG_PRINT=1` | temp | 每秒 9 路通道状态（含有效掩码） |
| `APP_DEBUG_PRINT=1` | com | 协议错误统计 |
启动确认打印（应保留的最小集）：
```
[ch9438] v1.0 port=0 ok ← CH9438 8 端口初始化成功
[godtek] mode = 0xac: baudrate = 9600 ← 传感器模式与波特率确认
[app_led] Init: ... ← LED 初始化
[app_temp]temp init ← 温度模块初始化
[infra] ch0 ready=1 ... ch8 ready=1 ← 9 路设备就绪
[com] init: rx table=3, id sent ← 协议就绪 + 上电 ID
[watchdog] init ← 看门狗就绪
```
### 8.2 常见问题排查
| 现象 | 排查方向 |
|---|---|
| 重启后波特率异常（0x00 / 乱码） | 确认 ch9438 初始化含 FCR 复位后重写波特率 + 回读验证（§7.1） |
| 温度恒为 0.0 / 无数据 | 检查传感器 UART 波特率（9600）、测量命令周期（500ms）、帧校验是否丢帧 |
| 灯带无指示 | 检查 `led_strip_indicator` 节点、WS2812 SPI 时序 |
| 主板收不到上报 | 检查 usart1 115200、帧头 0x7EE7、CRC16 lsb-msb 与主板一致 |
### 8.3 回环自测
短接 CH9438 UART7 的 TX/RX（PIN25 ↔ PIN26），验证中断接收链路（ISR → workqueue → 回调 → FIFO 读取），用于驱动回归测试。
---
## 9. 构建与烧录
```bash
# 构建（board 定义在 boards/dr2501a，overlay 配置传感器波特率/模式）
west build -p auto -b dr2501a_g0b0ce/stm32g0b0xx app/app_photomagnetic \
-d build -DOVERLAY_CONFIG=boards/use_ms.overlay
# 烧录
west flash -d build
```
> 板级传感器配置（波特率、模式）统一在 `app/app_photomagnetic/boards/use_ms.overlay` 中，不修改板级 dts。

14
dts/bindings/heating-pad.yaml
View File

@ -1,14 +0,0 @@
description: |
Heating pad using PWM
compatible: "heating-pad-pwm"
include: base.yaml
properties:
pwms:
type: phandle-array
description: PWM device to use for heating pad
required: true
temp-sensor:
type: phandle
description: Temperature sensor to use for heating pad
required: true

66
include/com.hpp
View File

@ -1,66 +0,0 @@
#ifndef __THER_COM_HPP__
#define __THER_COM_HPP__
#include "led.hpp"
#include <etl/delegate.h>
#include <etl/signal.h>
#include <led_strip_indicator/led_strip_indicator.hpp>
#include <uart_com/simple_protocal.hpp>
#include <zephyr/drivers/hwinfo.h>
#include <zephyr/drivers/sensor.h>
#include <zpp/driver.hpp>
#include <zpp/error.hpp>
namespace ther {
class Com {
public:
enum HostStatus : uint8_t { STANDBY, RUNNING, PAUSE, ERROR };
using RunningStateSignal = etl::signal<void(HostStatus), 10>;
static auto Init() -> zpp::error {
s_proto->SetRxCallbackTable(kRxCallbackTable);
auto size = hwinfo_get_device_id(buff, sizeof(buff));
s_id_buff = etl::span<uint8_t>(buff, size);
return zpp::ok();
}
static auto SendTemp(sensor_value val) -> void {
const uint8_t temp_data[] = {(uint8_t)val.val1, (uint8_t)val.val2};
s_proto->Send(R_TEMP, temp_data);
}
static auto AddRunningState(RunningStateSignal::slot_type slot) -> bool {
return s_running_state_sig.connect(slot);
}
private:
enum Addr : uint8_t {
R_TEMP = 0,
R_GET_ID,
W_RUNNING_STATE,
};
inline static RunningStateSignal s_running_state_sig{};
static auto RunningState(uart_com::DataType data) -> void {
if (data.size() != 1) {
return;
}
s_running_state_sig(HostStatus(data[0]));
}
static auto GetId(uart_com::DataType data) -> void {
printk("handle get id");
s_proto->Send(R_GET_ID, uart_com::DataType(s_id_buff));
};
constexpr static std::pair<const uint8_t,
uart_com::SimpleProtocal::CallbackType>
kRxCallbackTable[] = {
{R_GET_ID, GetId},
{W_RUNNING_STATE, RunningState},
};
inline static auto s_proto = (uart_com::SimpleProtocal *)DEVICE_DT_GET(
DT_COMPAT_GET_ANY_STATUS_OKAY(uart_com_simple_protocal));
inline static uint8_t buff[20];
inline static auto s_heating_state = false;
inline static etl::span<uint8_t> s_id_buff;
};
} // namespace ther
#endif

144
include/heating.hpp
View File

@ -1,144 +0,0 @@
#ifndef __THER_HEATING_HPP__
#define __THER_HEATING_HPP__
#include <algorithm>
#include <cmath>
#include <zephyr/device.h>
#include <zephyr/drivers/pwm.h>
#include <zephyr/drivers/sensor.h>
#include <zephyr/kernel.h>
#include <zephyr/sys/printk.h>
namespace ther {
class HeatingPad {
public:
static auto Start(sensor_value target_temp) -> void {
s_target = sensor_value_to_double(&target_temp);
s_integral = 0.0;
s_last_error = 0.0;
s_last_time = k_uptime_get();
printk("[heat] start target=%.1f C\n", s_target);
if (!device_is_ready(s_pwm_spec.dev)) {
printk("[heat] ERR: PWM not ready\n");
return;
}
printk("[heat] PWM OK\n");
k_work_init(&s_work, WorkHandler);
k_timer_init(&s_timer, TimerTick, nullptr);
k_timer_start(&s_timer, K_NO_WAIT, K_MSEC(kUpdatePeriodMs));
s_active = true;
}
/// Start PWM with raw duty cycle (0.0 ~ 1.0).
static auto Start(float duty_cycle) -> void {
duty_cycle = std::clamp(duty_cycle, 0.0f, 1.0f);
const uint32_t pulse =
static_cast<uint32_t>(s_pwm_spec.period * duty_cycle);
pwm_set_pulse_dt(&s_pwm_spec, pulse);
s_active = true;
}
static auto Stop() -> void {
s_active = false;
k_timer_stop(&s_timer);
pwm_set_pulse_dt(&s_pwm_spec, 0);
}
static auto CurrentTemp() -> sensor_value {
if (sensor_sample_fetch(s_temp_sensor) == 0) {
sensor_value temp;
sensor_channel_get(s_temp_sensor, SENSOR_CHAN_AMBIENT_TEMP, &temp);
s_current_temp = temp;
}
return s_current_temp;
}
private:
static constexpr uint32_t kPeriodNs = PWM_KHZ(5);
static constexpr uint32_t kUpdatePeriodMs = 100;
static constexpr double kOutMin = 0.0;
static constexpr double kOutMax = 1.0;
static constexpr double kKp = 2.0;
static constexpr double kKi = 0.02;
inline static bool s_active{false};
inline static double s_target{0.0};
inline static double s_integral{0.0};
inline static double s_last_error{0.0};
inline static int64_t s_last_time{0};
inline static pwm_dt_spec s_pwm_spec = {
.dev = DEVICE_DT_GET(DT_NODELABEL(pwm1)),
.channel = 1,
.period = PWM_KHZ(5),
.flags = PWM_POLARITY_NORMAL,
};
inline static const device *s_temp_sensor =
DEVICE_DT_GET(DT_NODELABEL(heating_pad_ntc));
inline static k_timer s_timer;
inline static k_work s_work;
inline static bool s_work_pending{false};
/// Timer fires → submit work. Skips if previous work hasn't finished.
static auto TimerTick(k_timer * /*timer*/) -> void {
if (s_work_pending) {
return;
}
s_work_pending = true;
k_work_submit(&s_work);
}
/// PID update running in system workqueue context.
static auto WorkHandler(k_work * /*work*/) -> void {
double current_temp;
if (ReadTemperature(current_temp) != 0) {
s_work_pending = false;
return;
}
printk("[heat] cur=%.1f C target=%.1f C\n", current_temp, s_target);
const int64_t now = k_uptime_get();
const double dt = static_cast<double>(now - s_last_time) / 1000.0;
s_last_time = now;
const double error = s_target - current_temp;
s_integral += kKi * error * dt;
s_integral = std::clamp(s_integral, kOutMin, kOutMax);
double output = kKp * error + s_integral;
output = std::clamp(output, kOutMin, kOutMax);
const uint32_t pulse = static_cast<uint32_t>(output * kPeriodNs);
pwm_set_pulse_dt(&s_pwm_spec, pulse);
s_work_pending = false;
}
static auto ReadTemperature(double &out_temp) -> int {
if (0 != sensor_sample_fetch(s_temp_sensor)) {
printk("[heat] ERR: sensor fetch failed\n");
return -EIO;
}
sensor_value val{};
if (0 !=
sensor_channel_get(s_temp_sensor, SENSOR_CHAN_AMBIENT_TEMP, &val)) {
printk("[heat] ERR: channel_get failed\n");
return -EIO;
}
out_temp = sensor_value_to_double(&val);
s_current_temp = val;
return 0;
}
inline static sensor_value s_current_temp{};
};
} // namespace ther
#endif

162
include/infrared.hpp
View File

@ -1,162 +0,0 @@
#ifndef __THER_INFRARED_HPP__
#define __THER_INFRARED_HPP__
#include <etl/algorithm.h>
#include <etl/array.h>
#include <etl/delegate.h>
#include <etl/tuple.h>
#include <utility>
#include <zephyr/device.h>
#include <zephyr/devicetree.h>
#include <zephyr/drivers/sensor.h>
#include <zpp/result.hpp>
#include <zpp/value.hpp>
#include <zpp/work_queue.hpp>
namespace ther {
class Infrared {
public:
static auto Init() -> zpp::error {
return InitImpl(std::make_index_sequence<s_dev.size()>{});
}
/* Register a callback invoked every scan period (20 ms) with the
* hottest sensor value. Runs in the system workqueue thread context -
* keep the callback lightweight (no blocking, no SPI). */
static auto
AddCallbackWhenSensorValueReady(etl::delegate<void(sensor_value)> cb)
-> zpp::error {
if (s_cb.is_valid()) {
return zpp::error_code::k_busy;
}
s_cb = cb;
return zpp::ok();
}
/* Hottest sensor among the latest cached samples (no polling: the
* trigger callback keeps the cache fresh). */
static auto GetMaxSensorValue() -> zpp::result<std::pair<int, sensor_value>> {
bool found = false;
int max_id = 0;
sensor_value max_val{};
for (size_t id = 0; id < s_dev.size(); ++id) {
if (!s_latest_valid[id]) {
continue;
}
const sensor_value val{s_latest_val1[id], s_latest_val2[id]};
if (!found || val.val1 > max_val.val1 ||
(val.val1 == max_val.val1 && val.val2 > max_val.val2)) {
found = true;
max_id = static_cast<int>(id);
max_val = val;
}
}
if (!found) {
return zpp::error_code::k_nodata;
}
return std::make_pair(max_id, max_val);
}
/* Snapshot of all cached sensor values (pure memory reads, safe in
* trigger context). Channels without a sample yet read as 0.
* auto return: the body is parsed in complete-class context where the
* trailing members (s_dev, caches) are visible. */
static auto GetAllSensorValues() {
etl::array<sensor_value, s_dev.size()> out{};
for (size_t id = 0; id < s_dev.size(); ++id) {
if (s_latest_valid[id]) {
out[id] = sensor_value{s_latest_val1[id], s_latest_val2[id]};
}
}
return out;
}
/* Bitmask of channels that have received at least one sample (diag). */
static auto GetChannelValidMask() -> uint32_t {
uint32_t mask = 0;
for (size_t id = 0; id < s_dev.size(); ++id) {
if (s_latest_valid[id]) {
mask |= (1u << id);
}
}
return mask;
}
private:
/* Per-sensor trigger handler; the channel index is a compile-time
* constant. Runs in the UART trigger context (workqueue thread for
* CH9438 ports, ISR context for on-chip UARTs) - keep it lock-free. */
template <size_t I>
static void OnSensorDataReady(const struct device *dev,
const struct sensor_trigger *trig) {
sensor_value val;
if (0 != sensor_channel_get(dev, SENSOR_CHAN_AMBIENT_TEMP, &val)) {
printk("[infra] ch%u channel_get failed\n", (unsigned)I);
return;
}
s_latest_val1[I] = val.val1;
s_latest_val2[I] = val.val2;
s_latest_valid[I] = true;
if (s_cb.is_valid()) {
s_cb(MaxCachedValue()); /* notify with the hottest sensor value */
}
}
/* Latest cached max value; pure memory reads, safe in trigger context. */
static auto MaxCachedValue() -> sensor_value {
bool found = false;
sensor_value max_val{};
for (size_t id = 0; id < s_dev.size(); ++id) {
if (!s_latest_valid[id]) {
continue;
}
const sensor_value val{s_latest_val1[id], s_latest_val2[id]};
if (!found || val.val1 > max_val.val1 ||
(val.val1 == max_val.val1 && val.val2 > max_val.val2)) {
found = true;
max_val = val;
}
}
return max_val;
}
template <size_t... Is>
static auto InitImpl(std::index_sequence<Is...>) -> zpp::error {
static sensor_trigger tri{.type = SENSOR_TRIG_DATA_READY,
.chan = SENSOR_CHAN_AMBIENT_TEMP};
int ret = 0;
((printk("[infra] ch%u ready=%d\n", (unsigned)Is,
device_is_ready(s_dev[Is])),
ret |= sensor_trigger_set(s_dev[Is], &tri, &OnSensorDataReady<Is>)),
...);
return ret != 0 ? zpp::error{-ENODEV} : zpp::ok();
}
/* Channel order: ch0~7 = CH9438 SPI-UART ports, ch8 = on-chip usart2 inner
* sensor. DT_FOREACH follows .dtsi soc node order (usart2 before spi2), so
* split by parent compatible instead of relying on node order. */
#define INFRARED_DEV(node) \
COND_CODE_1(DT_NODE_HAS_COMPAT(DT_PARENT(node), wch_ch9438_uart), \
(DEVICE_DT_GET(node), ), ())
#define INFRARED_INNER_DEV(node) \
COND_CODE_1(DT_NODE_HAS_COMPAT(DT_PARENT(node), st_stm32_usart), \
(DEVICE_DT_GET(node), ), ())
inline static etl::array s_dev{
DT_FOREACH_STATUS_OKAY(godtek_temp_uart, INFRARED_DEV)
DT_FOREACH_STATUS_OKAY(godtek_temp_uart, INFRARED_INNER_DEV)};
#undef INFRARED_DEV
#undef INFRARED_INNER_DEV
/* Field-level volatile caches: written by the trigger handlers, read by
* GetMaxSensorValue. sensor_value itself cannot be volatile (no volatile
* copy/assign), so the two 32-bit fields are cached separately. A torn
* cross-field read is possible but negligible for slowly changing temps. */
inline static volatile int32_t s_latest_val1[s_dev.size()]{};
inline static volatile int32_t s_latest_val2[s_dev.size()]{};
inline static volatile bool s_latest_valid[s_dev.size()]{};
inline static etl::delegate<void(sensor_value)> s_cb;
};
} // namespace ther
#endif

33
include/led.hpp
View File

@ -1,33 +0,0 @@
#ifndef __THER_LED_HPP__
#define __THER_LED_HPP__
#include <etl/algorithm.h>
#include <etl/array.h>
#include <zephyr/drivers/led.h>
#include <zpp/ct_string.hpp>
#include <zpp/work_queue.hpp>
namespace ther {
template <led_dt_spec kSpec> class Led {
public:
static auto On() -> zpp::error { return led_on_dt(&kSpec); }
static auto Off() -> zpp::error { return led_off_dt(&kSpec); }
template <typename TRep, typename TPeriod>
static auto Flash(std::chrono::duration<TRep, TPeriod> period) {
s_work.submit(period);
}
private:
static auto Flash() -> void {
static bool flag = false;
if (flag) {
led_off_dt(&kSpec);
} else {
led_on_dt(&kSpec);
}
flag = !flag;
}
static inline zpp::periodic_work<> s_work{Flash};
};
} // namespace ther
#endif

76
include/ntc.hpp
View File

@ -1,76 +0,0 @@
#ifndef THER_NTC_HPP
#define THER_NTC_HPP
#include <array>
#include <cmath>
#include <zephyr/device.h>
#include <zephyr/drivers/sensor.h>
#include <zephyr/sys/printk.h>
#include <zpp/result.hpp>
namespace ther {
class NtcGroup {
public:
/// Read all NTC sensors, return the maximum temperature value.
/// Returns error if every sensor read fails.
static auto GetSensorValue() -> zpp::result<sensor_value> {
bool found = false;
sensor_value max_val{};
for (size_t i = 0; i < NTC_COUNT; ++i) {
const auto *dev = s_devices[i];
if (0 != sensor_sample_fetch(dev)) {
printk("[ntc] #%zu ERR fetch\n", i + 1);
continue;
}
sensor_value val{};
if (0 != sensor_channel_get(dev, SENSOR_CHAN_AMBIENT_TEMP, &val)) {
printk("[ntc] #%zu ERR channel_get\n", i + 1);
continue;
}
// Always cache the latest reading for individual query
s_current_temp[i] = val;
if (!found || val.val1 > max_val.val1 ||
(val.val1 == max_val.val1 && val.val2 > max_val.val2)) {
max_val = val;
found = true;
}
}
if (found) {
return max_val;
}
return zpp::error_code::k_io;
}
/// Return the cached value for a single NTC sensor.
static auto GetSensorValue(uint8_t id) -> sensor_value {
if (id >= NTC_COUNT) {
return sensor_value{};
}
return s_current_temp[id];
}
private:
static constexpr size_t NTC_COUNT = 8;
inline static std::array<sensor_value, NTC_COUNT> s_current_temp{};
inline static const device *s_devices[] = {
DEVICE_DT_GET(DT_NODELABEL(pole_ntc1)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc2)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc3)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc4)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc5)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc6)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc7)),
DEVICE_DT_GET(DT_NODELABEL(pole_ntc8)),
};
};
} // namespace ther
#endif // THER_NTC_HPP

6
include/watchdog.hpp
View File

@ -3,7 +3,7 @@
#include <zephyr/drivers/watchdog.h>
#include <zpp/assert.hpp>
#include <zpp/timer.hpp>
namespace ther {
namespace app {
class WatchDogConfig {
public:
WatchDogConfig() {
@ -28,6 +28,6 @@ private:
int chan_id;
};
} // namespace ther
} // namespace app
#endif
#endif

24
prj.conf
View File

@ -1,30 +1,14 @@
CONFIG_STDOUT_CONSOLE=y
CONFIG_CBPRINTF_FP_SUPPORT=y
CONFIG_STD_CPP23=y
CONFIG_STD_CPP20=y
CONFIG_CPP=y
CONFIG_REQUIRES_FULL_LIBCPP=y
CONFIG_CONSOLE=y
CONFIG_SERIAL=y
CONFIG_UART_INTERRUPT_DRIVEN=y
CONFIG_PMC_COM=y
CONFIG_HWINFO=y
CONFIG_LED=y
# CONFIG_PRINTK=n
CONFIG_ADC=y
CONFIG_SPI=y
CONFIG_SPI_STM32=y
# SPI 中断模式：传输带 1s 超时，多路高频 SPI 下不会永久挂死（与 ch9438 测试对齐）
CONFIG_SPI_STM32_INTERRUPT=y
CONFIG_SENSOR=y
CONFIG_ADC_MCP320X_ACQUISITION_THREAD_STACK_SIZE=2048
CONFIG_REBOOT=y
CONFIG_WATCHDOG=y
CONFIG_PWM=y
# # Debug logging
# CONFIG_LOG=y
# CONFIG_LOG_MODE_IMMEDIATE=y
# CONFIG_ADC_LOG_LEVEL_DBG=y
# CONFIG_SENSOR_LOG_LEVEL_DBG=y
# CONFIG_SPI_LOG_LEVEL_DBG=y
CONFIG_WATCHDOG=y

343
scripts/test.py
View File

@ -1,343 +0,0 @@
#!/usr/bin/env python3
"""
Test script for app_photomagnetic communication protocol.
Frame format (SimpleProtocal, per DTS config):
H0(0x7E) H1(0xE7) CMD(1B) LEN(1B) DATA[0..N] CRC_LO CRC_HI
CRC = CRC-16 Modbus over CMD+LEN+DATA, LSB-MSB order
Commands:
0x00 TEMP D→H (2 bytes: val1 val2)
0x01 GET_ID H→D / D→H (20 bytes HWID)
0x02 RUNNING_STATE H→D (1 byte: 0=Standby 1=Running)
0x03 W_HEATING_STATE H→D (1 byte: 0=stop, >0=target °C)
0x04 R_HEATING_STATE H→D / D→H (3 bytes: state val1 val2)
0x64 R_TEMP_POLE_NTC H→D (1 byte index) / D→H (2 bytes: val1 val2)
0x65 R_HEATING_PAD_NTC H→D (0 bytes) / D→H (2 bytes: val1 val2)
Usage:
python test.py /dev/ttyUSB0 --monitor
python test.py /dev/ttyUSB0 --heating 42
python test.py /dev/ttyUSB0 --read-pole-ntc 3
python test.py /dev/ttyUSB0 --read-heating-ntc
"""
import argparse
import sys
import time
try:
import serial
except ImportError:
print("Please install pyserial: pip install pyserial", file=sys.stderr)
sys.exit(1)
# ── Protocol constants ────────────────────────────────────────
HEADER = b"\x7e\xe7"
HEADER_SIZE = len(HEADER)
CMD_TEMP = 0x00
CMD_GET_ID = 0x01
CMD_RUNNING_STATE = 0x02
CMD_W_HEATING = 0x03
CMD_R_HEATING = 0x04
CMD_READ_POLE_NTC = 0x64
CMD_READ_HEATING_NTC = 0x65
HEATING_STATE_NAMES = {0: "OFF", 1: "ON"}
# ── CRC-16 Modbus ─────────────────────────────────────────────
def crc16_modbus(data: bytes) -> int:
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 1:
crc = (crc >> 1) ^ 0xA001
else:
crc >>= 1
return crc & 0xFFFF
# ── Frame helpers ─────────────────────────────────────────────
def build_frame(cmd: int, payload: bytes = b"") -> bytes:
body = bytes([cmd, len(payload)]) + payload
crc = crc16_modbus(body)
return HEADER + body + bytes([crc & 0xFF, (crc >> 8) & 0xFF])
def parse_frame(frame: bytes) -> tuple[int, bytes] | None:
if len(frame) < HEADER_SIZE + 2 + 2:
return None
if frame[:HEADER_SIZE] != HEADER:
return None
cmd = frame[HEADER_SIZE]
length = frame[HEADER_SIZE + 1]
expected = HEADER_SIZE + 2 + length + 2
if len(frame) != expected:
return None
data = frame[HEADER_SIZE + 2 : HEADER_SIZE + 2 + length]
crc_body = frame[HEADER_SIZE : HEADER_SIZE + 2 + length]
actual_crc = crc16_modbus(crc_body)
wire_crc = (frame[expected - 1] << 8) | frame[expected - 2]
if actual_crc != wire_crc:
return None
return cmd, data
# ── Serial reader ─────────────────────────────────────────────
class ProtocolReader:
def __init__(self, ser: serial.Serial):
self._ser = ser
self._buf = bytearray()
def read_frame(self, timeout: float = 1.0) -> bytes | None:
deadline = time.time() + timeout
while time.time() < deadline:
if len(self._buf) >= 2:
idx = self._buf.find(HEADER)
if idx > 0:
del self._buf[:idx]
elif idx < 0:
self._buf.clear()
if len(self._buf) >= HEADER_SIZE and self._buf[:HEADER_SIZE] == HEADER:
if len(self._buf) >= HEADER_SIZE + 2 + 2:
length = self._buf[HEADER_SIZE + 1]
total = HEADER_SIZE + 2 + length + 2
if len(self._buf) >= total:
frame = bytes(self._buf[:total])
del self._buf[:total]
return frame
elif len(self._buf) >= HEADER_SIZE:
del self._buf[0]
continue
try:
chunk = self._ser.read(self._ser.in_waiting or 1)
if chunk:
self._buf.extend(chunk)
except (serial.SerialTimeoutException, serial.SerialException):
pass
return None
def _read_reply(ser: serial.Serial, timeout: float = 2.0) -> tuple[int, bytes] | None:
reader = ProtocolReader(ser)
frame = reader.read_frame(timeout)
if frame is None:
print("✗ No response (timeout)")
return None
result = parse_frame(frame)
if result is None:
print(f"✗ Invalid frame: {frame.hex()}")
return None
return result
def _temp_from_data(data: bytes) -> float:
return data[0] + data[1] / 100.0 if len(data) >= 2 else 0.0
# ── Command handlers ──────────────────────────────────────────
def cmd_get_id(ser: serial.Serial, timeout: float = 2.0):
print("→ Sending GET_ID...")
ser.write(build_frame(CMD_GET_ID))
r = _read_reply(ser, timeout)
if r is None:
return
cmd, data = r
if cmd != CMD_GET_ID:
print(f"✗ Unexpected cmd=0x{cmd:02X}")
return
print(f"✓ Device ID ({len(data)} bytes): {data.hex(' ')}")
def cmd_set_state(ser: serial.Serial, state: int):
names = {0: "Standby", 1: "Running", 2: "Pause", 3: "Error"}
name = names.get(state, "Unknown")
print(f"→ Setting state: {name} (0x{state:02X})")
ser.write(build_frame(CMD_RUNNING_STATE, bytes([state])))
print("✓ Sent")
def cmd_write_heating(ser: serial.Serial, target: int):
if target == 0:
print("→ Stopping heating")
else:
print(f"→ Starting heating to {target}°C")
ser.write(build_frame(CMD_W_HEATING, bytes([target])))
print("✓ Sent")
def cmd_read_heating(ser: serial.Serial, timeout: float = 2.0):
print("→ Querying heating state...")
ser.write(build_frame(CMD_R_HEATING))
r = _read_reply(ser, timeout)
if r is None:
return
cmd, data = r
if cmd != CMD_R_HEATING:
print(f"✗ Unexpected cmd=0x{cmd:02X}")
return
if len(data) < 3:
print(f"✗ Short data: {data.hex()}")
return
state = HEATING_STATE_NAMES.get(data[0], f"Unknown({data[0]})")
print(f"✓ Heating: {state}, temp={_temp_from_data(data[1:3]):.2f}°C")
def cmd_read_pole_ntc(ser: serial.Serial, index: int, timeout: float = 2.0):
print(f"→ Reading pole NTC #{index}...")
ser.write(build_frame(CMD_READ_POLE_NTC, bytes([index])))
r = _read_reply(ser, timeout)
if r is None:
return
cmd, data = r
if cmd != CMD_READ_POLE_NTC:
print(f"✗ Unexpected cmd=0x{cmd:02X}")
return
if len(data) < 2:
print(f"✗ Short data: {data.hex()}")
return
print(f"✓ Pole NTC #{index}: {_temp_from_data(data):.2f}°C")
def cmd_read_heating_ntc(ser: serial.Serial, timeout: float = 2.0):
print("→ Reading heating pad NTC...")
ser.write(build_frame(CMD_READ_HEATING_NTC))
r = _read_reply(ser, timeout)
if r is None:
return
cmd, data = r
if cmd != CMD_READ_HEATING_NTC:
print(f"✗ Unexpected cmd=0x{cmd:02X}")
return
if len(data) < 2:
print(f"✗ Short data: {data.hex()}")
return
print(f"✓ Heating pad NTC: {_temp_from_data(data):.2f}°C")
def cmd_monitor(ser: serial.Serial):
print("Monitoring... (Ctrl+C to stop)")
reader = ProtocolReader(ser)
try:
while True:
frame = reader.read_frame(timeout=0.5)
if frame is None:
continue
result = parse_frame(frame)
if result is None:
print(f"⚠ Bad frame: {frame.hex()}")
continue
cmd, data = result
if cmd == CMD_TEMP:
t = _temp_from_data(data)
print(f"🌡 Temp: {t:.2f}°C (raw: {data.hex(' ')})")
elif cmd == CMD_GET_ID:
print(f"🆔 Device ID: {data.hex(' ')}")
elif cmd == CMD_RUNNING_STATE:
print(f"🔁 Running state: 0x{data.hex()}")
elif cmd == CMD_R_HEATING:
if len(data) >= 3:
state = HEATING_STATE_NAMES.get(data[0], f"?{data[0]}")
t = _temp_from_data(data[1:3])
print(f"🔥 Heating: {state}, temp={t:.2f}°C")
else:
print(f"🔥 Heating: {data.hex()}")
elif cmd == CMD_READ_POLE_NTC:
t = _temp_from_data(data)
print(f"📡 Pole NTC reply: {t:.2f}°C")
elif cmd == CMD_READ_HEATING_NTC:
t = _temp_from_data(data)
print(f"🔥 Heating pad NTC reply: {t:.2f}°C")
else:
print(f"📦 cmd=0x{cmd:02X} data={data.hex()}")
except KeyboardInterrupt:
print("\nDone.")
def cmd_loop(ser: serial.Serial, count: int = 0):
i = 0
print("Looping GET_ID... (Ctrl+C to stop)")
try:
while count == 0 or i < count:
cmd_get_id(ser, timeout=1.0)
i += 1
time.sleep(0.5)
except KeyboardInterrupt:
print("\nDone.")
# ── Main ──────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Test script for app_photomagnetic comm protocol"
)
parser.add_argument("port", help="Serial port")
parser.add_argument("--baud", type=int, default=115200)
parser.add_argument("--get-id", action="store_true")
parser.add_argument(
"--state", type=lambda x: int(x, 0), metavar="N",
help="Set running state")
parser.add_argument(
"--heating", type=int, metavar="TEMP",
help="Write heating target (0=stop)")
parser.add_argument(
"--read-heating", action="store_true",
help="Read heating state + temp")
parser.add_argument(
"--read-pole-ntc", type=int, metavar="IDX",
help="Read pole NTC by index (0..7)")
parser.add_argument(
"--read-heating-ntc", action="store_true",
help="Read heating pad NTC temperature")
parser.add_argument("--monitor", action="store_true")
parser.add_argument(
"--loop", type=int, nargs="?", const=0, metavar="N",
help="Loop GET_ID N times")
parser.add_argument(
"--raw", type=lambda x: bytes.fromhex(x), metavar="HEX",
help="Send raw CMD+DATA")
args = parser.parse_args()
ser = serial.Serial(args.port, args.baud, timeout=0.1)
print(f"Connected to {args.port} @ {args.baud} baud")
try:
if args.get_id:
cmd_get_id(ser)
elif args.state is not None:
cmd_set_state(ser, args.state)
elif args.heating is not None:
cmd_write_heating(ser, args.heating)
elif args.read_heating:
cmd_read_heating(ser)
elif args.read_pole_ntc is not None:
cmd_read_pole_ntc(ser, args.read_pole_ntc)
elif args.read_heating_ntc:
cmd_read_heating_ntc(ser)
elif args.monitor:
cmd_monitor(ser)
elif args.loop is not None:
cmd_loop(ser, args.loop)
elif args.raw is not None:
payload = args.raw
c = payload[0]
d = payload[1:] if len(payload) > 1 else b""
print(f"→ Sending CMD=0x{c:02X} data={d.hex()}")
ser.write(build_frame(c, d))
print("✓ Sent")
else:
print("Use --get-id, --state, --heating, --read-heating, "
"--read-pole-ntc, --read-heating-ntc, --monitor, --loop, --raw")
finally:
ser.close()
if __name__ == "__main__":
main()

606
scripts/test_gui.py
View File

@ -1,606 +0,0 @@
#!/usr/bin/env python3
"""
GUI test tool for app_photomagnetic communication protocol.
Built with Python native tkinter + threading.
"""
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import threading
import time
import sys
from enum import IntEnum
try:
import serial
import serial.tools.list_ports
except ImportError:
print("Please install pyserial: pip install pyserial", file=sys.stderr)
sys.exit(1)
# ── Protocol constants ────────────────────────────────────────
HEADER = b"\x7e\xe7"
HEADER_SIZE = len(HEADER)
class Cmd(IntEnum):
TEMP = 0x00
GET_ID = 0x01
RUNNING_STATE = 0x02
W_HEATING = 0x03
R_HEATING = 0x04
READ_POLE_NTC = 0x64
READ_HEATING_NTC = 0x65
HEATING_STATE_NAMES = {0: "OFF", 1: "ON"}
STATE_NAMES = {0: "Standby", 1: "Running", 2: "Pause", 3: "Error"}
# ── Protocol helpers (from test.py) ───────────────────────────
def crc16_modbus(data: bytes) -> int:
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
return crc & 0xFFFF
def build_frame(cmd: int, payload: bytes = b"") -> bytes:
body = bytes([cmd, len(payload)]) + payload
crc = crc16_modbus(body)
return HEADER + body + bytes([crc & 0xFF, (crc >> 8) & 0xFF])
def parse_frame(frame: bytes) -> tuple[int, bytes] | None:
if len(frame) < HEADER_SIZE + 2 + 2:
return None
if frame[:HEADER_SIZE] != HEADER:
return None
cmd = frame[HEADER_SIZE]
length = frame[HEADER_SIZE + 1]
expected = HEADER_SIZE + 2 + length + 2
if len(frame) != expected:
return None
data = frame[HEADER_SIZE + 2: HEADER_SIZE + 2 + length]
crc_body = frame[HEADER_SIZE: HEADER_SIZE + 2 + length]
actual_crc = crc16_modbus(crc_body)
wire_crc = (frame[expected - 1] << 8) | frame[expected - 2]
if actual_crc != wire_crc:
return None
return cmd, data
def temp_from_data(data: bytes) -> float:
return data[0] + data[1] / 100.0 if len(data) >= 2 else 0.0
# ── Serial I/O thread ─────────────────────────────────────────
class SerialWorker:
def __init__(self, gui_callback):
self.ser: serial.Serial | None = None
self._lock = threading.Lock()
self._running = False
self._monitoring = False
self._reader_thread: threading.Thread | None = None
self._gui = gui_callback # on_frame(cmd, data) called from reader thread
@property
def is_open(self) -> bool:
return self.ser is not None and self.ser.is_open
def open(self, port: str, baud: int) -> str | None:
"""Open serial port. Returns error string or None on success."""
try:
ser = serial.Serial(port, baud, timeout=0.05)
with self._lock:
self.ser = ser
self._running = True
self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True)
self._reader_thread.start()
return None
except serial.SerialException as e:
return str(e)
def close(self):
with self._lock:
self._running = False
self._monitoring = False
if self.ser:
try:
self.ser.close()
except Exception:
pass
self.ser = None
if self._reader_thread:
self._reader_thread.join(timeout=2)
self._reader_thread = None
def send(self, cmd: int, payload: bytes = b"") -> bool:
"""Send a frame. Returns True if sent."""
with self._lock:
if not self.ser or not self.ser.is_open:
return False
try:
self.ser.write(build_frame(cmd, payload))
return True
except serial.SerialException:
return False
def send_raw(self, data: bytes) -> bool:
"""Send raw bytes directly."""
with self._lock:
if not self.ser or not self.ser.is_open:
return False
try:
self.ser.write(data)
return True
except serial.SerialException:
return False
@property
def monitoring(self) -> bool:
return self._monitoring
@monitoring.setter
def monitoring(self, value: bool):
self._monitoring = value
def _reader_loop(self):
buf = bytearray()
while self._running:
# Read data
with self._lock:
if not self.ser or not self.ser.is_open:
time.sleep(0.05)
continue
try:
chunk = self.ser.read(self.ser.in_waiting or 1)
except serial.SerialException:
chunk = b""
if not chunk:
time.sleep(0.01)
continue
buf.extend(chunk)
# Try to extract frames
while len(buf) >= HEADER_SIZE + 2 + 2:
# Find header
idx = buf.find(HEADER)
if idx > 0:
del buf[:idx]
continue
if idx < 0:
buf.clear()
break
length = buf[HEADER_SIZE + 1]
total = HEADER_SIZE + 2 + length + 2
if len(buf) < total:
break # wait for more
frame = bytes(buf[:total])
del buf[:total]
result = parse_frame(frame)
if result:
self._gui.on_frame_received(result[0], result[1])
else:
self._gui.on_log(f"⚠ Bad frame: {frame.hex()}")
# ── GUI Application ───────────────────────────────────────────
class PhotomagneticGUI:
def __init__(self):
self.root = tk.Tk()
self.root.title("Photomagnetic Communication Tool")
self.root.geometry("850x700")
self.root.minsize(700, 550)
self.worker = SerialWorker(self)
self._monitor_after_id = None
self._last_temp_time = 0.0
self._temp_interval = 0.0
self._build_ui()
# ── UI Build ───────────────────────────────────────────
def _build_ui(self):
# ── Top: Serial connection ──
conn_frame = ttk.LabelFrame(self.root, text="Serial Connection", padding=8)
conn_frame.pack(fill=tk.X, padx=8, pady=4)
ttk.Label(conn_frame, text="Port:").grid(row=0, column=0, sticky=tk.W)
self._port_var = tk.StringVar()
self._port_combo = ttk.Combobox(conn_frame, textvariable=self._port_var, width=25)
self._port_combo.grid(row=0, column=1, padx=4, sticky=tk.W)
ttk.Button(conn_frame, text="↻", width=3,
command=self._scan_ports).grid(row=0, column=2, padx=2)
ttk.Label(conn_frame, text="Baud:").grid(row=0, column=3, padx=(12, 0), sticky=tk.W)
self._baud_var = tk.StringVar(value="115200")
baud_combo = ttk.Combobox(conn_frame, textvariable=self._baud_var,
values=("9600", "19200", "38400", "57600",
"115200", "230400", "460800"),
width=10)
baud_combo.grid(row=0, column=4, padx=4, sticky=tk.W)
self._connect_btn = ttk.Button(conn_frame, text="Connect", command=self._toggle_connect)
self._connect_btn.grid(row=0, column=5, padx=(12, 0))
self._status_lbl = ttk.Label(conn_frame, text="Disconnected", foreground="gray")
self._status_lbl.grid(row=0, column=6, padx=8, sticky=tk.W)
conn_frame.columnconfigure(6, weight=1)
# ── Main content: left (controls) + right (display) ──
main_frame = ttk.Frame(self.root)
main_frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=2)
left = ttk.Frame(main_frame)
left.pack(side=tk.LEFT, fill=tk.BOTH, expand=False)
right = ttk.Frame(main_frame)
right.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=(6, 0))
# ── Left: Command panel ──
cmd_frame = ttk.LabelFrame(left, text="Commands", padding=8)
cmd_frame.pack(fill=tk.X)
# Row 1: GET_ID + State
r1 = ttk.Frame(cmd_frame)
r1.pack(fill=tk.X, pady=2)
ttk.Button(r1, text="Get Device ID", width=14,
command=self._cmd_get_id).pack(side=tk.LEFT)
ttk.Label(r1, text="State:").pack(side=tk.LEFT, padx=(10, 2))
self._state_combo = ttk.Combobox(r1, values=["Standby", "Running", "Pause", "Error"],
state="readonly", width=10)
self._state_combo.current(0)
self._state_combo.pack(side=tk.LEFT)
ttk.Button(r1, text="Set", width=4,
command=self._cmd_set_state).pack(side=tk.LEFT, padx=4)
# Row 2: Heating
r2 = ttk.Frame(cmd_frame)
r2.pack(fill=tk.X, pady=2)
ttk.Label(r2, text="Heating:").pack(side=tk.LEFT)
self._heating_var = tk.IntVar(value=0)
self._heating_spin = tk.Spinbox(r2, from_=0, to=60, textvariable=self._heating_var,
width=5, state="readonly")
self._heating_spin.pack(side=tk.LEFT, padx=2)
ttk.Label(r2, text="°C").pack(side=tk.LEFT)
ttk.Button(r2, text="Set / Stop", width=10,
command=self._cmd_set_heating).pack(side=tk.LEFT, padx=6)
ttk.Button(r2, text="Read State", width=10,
command=self._cmd_read_heating).pack(side=tk.LEFT)
# Row 3: NTC
r3 = ttk.Frame(cmd_frame)
r3.pack(fill=tk.X, pady=2)
ttk.Label(r3, text="Pole NTC #").pack(side=tk.LEFT)
self._ntc_idx_var = tk.IntVar(value=0)
ntc_spin = tk.Spinbox(r3, from_=0, to=7, textvariable=self._ntc_idx_var,
width=3, state="readonly")
ntc_spin.pack(side=tk.LEFT, padx=2)
ttk.Button(r3, text="Read Pole NTC", width=12,
command=self._cmd_read_pole_ntc).pack(side=tk.LEFT, padx=6)
ttk.Button(r3, text="Read Pad NTC", width=12,
command=self._cmd_read_heating_ntc).pack(side=tk.LEFT)
# Row 4: Monitor + Raw
r4 = ttk.Frame(cmd_frame)
r4.pack(fill=tk.X, pady=2)
self._monitor_btn = ttk.Button(r4, text="▶ Monitor", width=12,
command=self._toggle_monitor)
self._monitor_btn.pack(side=tk.LEFT)
ttk.Label(r4, text="Raw Hex:").pack(side=tk.LEFT, padx=(10, 2))
self._raw_var = tk.StringVar()
ttk.Entry(r4, textvariable=self._raw_var, width=18).pack(side=tk.LEFT, padx=2)
ttk.Button(r4, text="Send", width=5,
command=self._cmd_raw).pack(side=tk.LEFT, padx=2)
# Row 5: Loop
r5 = ttk.Frame(cmd_frame)
r5.pack(fill=tk.X, pady=2)
self._loop_btn = ttk.Button(r5, text="▶ Loop GET_ID", width=14,
command=self._toggle_loop)
self._loop_btn.pack(side=tk.LEFT)
# ── Right top: Live display ──
disp_frame = ttk.LabelFrame(right, text="Live Values", padding=6)
disp_frame.pack(fill=tk.X)
self._disp_labels = {}
entries = [
("Temperature", "--- °C", "temp"),
("Sample Rate", "---", "sample_rate"),
("Heating State", "---", "heat_state"),
("Heating Temp.", "--- °C", "heat_temp"),
("Device ID", "---", "dev_id"),
("Running State", "---", "run_state"),
]
for i, (label, default, key) in enumerate(entries):
ttk.Label(disp_frame, text=label + ":").grid(row=i, column=0, sticky=tk.W, padx=4)
lbl = ttk.Label(disp_frame, text=default, font=("Consolas", 11, "bold"),
foreground="#333")
lbl.grid(row=i, column=1, sticky=tk.W, padx=4)
self._disp_labels[key] = lbl
# Pole NTC sub-frame
self._ntc_labels = []
ntc_header = ttk.Label(disp_frame, text="Pole NTCs:")
ntc_header.grid(row=len(entries), column=0, sticky=tk.W, padx=4, pady=(4, 0))
ntc_row = ttk.Frame(disp_frame)
ntc_row.grid(row=len(entries), column=1, sticky=tk.W, padx=4, pady=(4, 0))
for i in range(8):
lbl = ttk.Label(ntc_row, text=f"{i}:---", font=("Consolas", 10), width=8)
lbl.pack(side=tk.LEFT)
self._ntc_labels.append(lbl)
disp_frame.columnconfigure(1, weight=1)
# ── Bottom: Log ──
log_frame = ttk.LabelFrame(right, text="Log", padding=4)
log_frame.pack(fill=tk.BOTH, expand=True, pady=(4, 0))
log_toolbar = ttk.Frame(log_frame)
log_toolbar.pack(fill=tk.X, pady=(0, 2))
ttk.Button(log_toolbar, text="Clear", width=6,
command=self._clear_log).pack(side=tk.RIGHT)
self._log_text = scrolledtext.ScrolledText(log_frame, height=12, font=("Consolas", 9),
wrap=tk.WORD, state=tk.DISABLED)
self._log_text.pack(fill=tk.BOTH, expand=True)
self._scan_ports()
# ── Serial connection ────────────────────────────────
def _scan_ports(self):
ports = [p.device for p in serial.tools.list_ports.comports()]
self._port_combo["values"] = ports
if ports and not self._port_var.get():
self._port_var.set(ports[0])
def _toggle_connect(self):
if self.worker.is_open:
self._disconnect()
else:
self._connect()
def _connect(self):
port = self._port_var.get().strip()
if not port:
messagebox.showerror("Error", "Select a serial port")
return
try:
baud = int(self._baud_var.get())
except ValueError:
messagebox.showerror("Error", "Invalid baud rate")
return
err = self.worker.open(port, baud)
if err:
messagebox.showerror("Connection Error", err)
return
self._connect_btn.configure(text="Disconnect")
self._status_lbl.configure(text=f"Connected @ {baud}", foreground="green")
self.on_log(f"Connected to {port} @ {baud} baud")
def _disconnect(self):
self._stop_monitor()
self._stop_loop()
self.worker.close()
self._connect_btn.configure(text="Connect")
self._status_lbl.configure(text="Disconnected", foreground="gray")
self.on_log("Disconnected")
def _check_connected(self) -> bool:
if not self.worker.is_open:
self.on_log("✗ Not connected")
return False
return True
# ── Commands ─────────────────────────────────────────
def _cmd_get_id(self):
if not self._check_connected():
return
self.worker.send(Cmd.GET_ID)
self.on_log("→ GET_ID sent")
def _cmd_set_state(self):
if not self._check_connected():
return
idx = self._state_combo.current()
self.worker.send(Cmd.RUNNING_STATE, bytes([idx]))
self.on_log(f"→ Set state: {self._state_combo.get()} ({idx})")
def _cmd_set_heating(self):
if not self._check_connected():
return
target = self._heating_var.get()
self.worker.send(Cmd.W_HEATING, bytes([target]))
if target == 0:
self.on_log("→ Stop heating")
else:
self.on_log(f"→ Set heating to {target}°C")
def _cmd_read_heating(self):
if not self._check_connected():
return
self.worker.send(Cmd.R_HEATING)
self.on_log("→ Read heating state")
def _cmd_read_pole_ntc(self):
if not self._check_connected():
return
idx = self._ntc_idx_var.get()
self.worker.send(Cmd.READ_POLE_NTC, bytes([idx]))
self.on_log(f"→ Read pole NTC #{idx}")
def _cmd_read_heating_ntc(self):
if not self._check_connected():
return
self.worker.send(Cmd.READ_HEATING_NTC)
self.on_log("→ Read heating pad NTC")
def _cmd_raw(self):
if not self._check_connected():
return
hex_str = self._raw_var.get().strip()
if not hex_str:
return
try:
payload = bytes.fromhex(hex_str.replace(" ", ""))
except ValueError:
self.on_log(f"✗ Invalid hex: {hex_str}")
return
cmd = payload[0]
data = payload[1:] if len(payload) > 1 else b""
self.worker.send(cmd, data)
self.on_log(f"→ Raw: CMD=0x{cmd:02X} data={data.hex()}")
self._raw_var.set("")
# ── Monitor ──────────────────────────────────────────
def _toggle_monitor(self):
if self.worker.monitoring:
self._stop_monitor()
else:
self._start_monitor()
def _start_monitor(self):
if not self._check_connected():
return
self.worker.monitoring = True
self._monitor_btn.configure(text="■ Stop Monitor")
self.on_log("▶ Monitoring started")
def _stop_monitor(self):
self.worker.monitoring = False
self._monitor_btn.configure(text="▶ Monitor")
if self._monitor_after_id:
self.root.after_cancel(self._monitor_after_id)
self._monitor_after_id = None
# ── Loop ─────────────────────────────────────────────
def _toggle_loop(self):
if self._loop_running:
self._stop_loop()
else:
self._start_loop()
def _start_loop(self):
if not self._check_connected():
return
self._loop_running = True
self._loop_btn.configure(text="■ Stop Loop")
self.on_log("▶ Loop GET_ID started")
self._loop_tick()
def _stop_loop(self):
self._loop_running = False
self._loop_btn.configure(text="▶ Loop GET_ID")
if hasattr(self, '_loop_after_id') and self._loop_after_id:
self.root.after_cancel(self._loop_after_id)
self._loop_after_id = None
def _loop_tick(self):
if not self._loop_running or not self.worker.is_open:
self._stop_loop()
return
self.worker.send(Cmd.GET_ID)
self._loop_after_id = self.root.after(1000, self._loop_tick)
# ── Callbacks from SerialWorker (called in reader thread) ──
def on_frame_received(self, cmd: int, data: bytes):
"""Called from reader thread. Schedule UI update."""
self.root.after(0, self._handle_frame, cmd, data)
def on_log(self, msg: str):
"""Thread-safe log write."""
self.root.after(0, self._append_log, msg)
# ── Frame handling (runs in main thread via after) ───
def _handle_frame(self, cmd: int, data: bytes):
raw_hex = data.hex(" ")
# ── 0x00 TEMP: 设备主动推送, 只更新 live value, 不输出 log ──
if cmd == Cmd.TEMP:
t = temp_from_data(data)
now = time.time()
if self._last_temp_time > 0:
self._temp_interval = now - self._last_temp_time
hz = 1.0 / self._temp_interval if self._temp_interval > 0 else 0
self._disp_labels["sample_rate"].configure(
text=f"{self._temp_interval*1000:.0f}ms ({hz:.1f}Hz)")
self._last_temp_time = now
self._disp_labels["temp"].configure(text=f"{t:.2f} °C")
elif cmd == Cmd.GET_ID:
id_str = data.hex(" ")
id_show = id_str[:47] + "..." if len(id_str) > 50 else id_str
self._disp_labels["dev_id"].configure(text=id_show)
self._append_log(f"🆔 Device ID ({len(data)}B): {raw_hex}")
elif cmd == Cmd.RUNNING_STATE:
state = data[0] if data else -1
name = STATE_NAMES.get(state, f"Unknown({state})")
self._disp_labels["run_state"].configure(text=name)
self._append_log(f"🔁 State: {name}")
elif cmd == Cmd.W_HEATING:
target = data[0] if data else 0
if target == 0:
self._append_log(f"✓ Heating stopped (ACK)")
else:
self._append_log(f"✓ Heating set to {target}°C (ACK)")
elif cmd == Cmd.R_HEATING:
if len(data) >= 3:
s = HEATING_STATE_NAMES.get(data[0], f"?{data[0]}")
t = temp_from_data(data[1:3])
self._disp_labels["heat_state"].configure(text=s)
self._disp_labels["heat_temp"].configure(text=f"{t:.2f} °C")
self._append_log(f"🔥 Heating: {s}, temp={t:.2f}°C")
else:
self._append_log(f"🔥 Heating: {raw_hex}")
elif cmd == Cmd.READ_POLE_NTC:
if len(data) >= 2:
t = temp_from_data(data)
self._append_log(f"📡 Pole NTC reply: {t:.2f}°C ({raw_hex})")
else:
self._append_log(f"📡 Pole NTC: {raw_hex}")
elif cmd == Cmd.READ_HEATING_NTC:
t = temp_from_data(data)
self._append_log(f"🔥 Pad NTC: {t:.2f}°C ({raw_hex})")
else:
self._append_log(f"📦 CMD=0x{cmd:02X} data={raw_hex}")
def _set_ntc_display(self, idx: int, temp: float):
if 0 <= idx < len(self._ntc_labels):
self._ntc_labels[idx].configure(text=f"{idx}:{temp:.1f}")
def _clear_log(self):
self._log_text.configure(state=tk.NORMAL)
self._log_text.delete(1.0, tk.END)
self._log_text.configure(state=tk.DISABLED)
def _append_log(self, msg: str):
self._log_text.configure(state=tk.NORMAL)
self._log_text.insert(tk.END, msg + "\n")
self._log_text.see(tk.END)
self._log_text.configure(state=tk.DISABLED)
# ── Run ──────────────────────────────────────────────
def run(self):
self.root.mainloop()
self._disconnect()
if __name__ == "__main__":
app = PhotomagneticGUI()
app.run()

6
src/com.cpp
View File

@ -1,6 +0,0 @@
#include <com.hpp>
#include <zephyr/init.h>
static auto Init() -> int { return static_cast<int>(ther::Com::Init()); }
inline SYS_INIT(Init, APPLICATION, 50);

29
src/led.cpp
View File

@ -1,29 +0,0 @@
#include <com.hpp>
#include <etl/utility.h>
#include <led.hpp>
#include <led_strip_indicator/led_strip_indicator.hpp>
#include <zephyr/init.h>
using namespace ther;
namespace {
using McuState = Led<LED_DT_SPEC_GET(DT_NODELABEL(led_mcu_state))>;
using Inf = Led<LED_DT_SPEC_GET(DT_NODELABEL(led_inf))>;
auto indicator = ZPP_DRV_GET_P(ledstrip::Indicator, DT_NODELABEL(indicator));
constexpr const char *MODULE = "app_led";
} // namespace
static auto Init() -> int {
Com::AddRunningState([](Com::HostStatus state) {
indicator->Status(static_cast<uint8_t>(state));
if (state == Com::HostStatus::RUNNING) {
Inf::On();
} else {
Inf::Off();
}
});
using namespace std::chrono_literals;
McuState::Flash(1s);
printk("[%s] Init: MCU state LED flashing at 1s interval\n", MODULE);
return 0;
}
SYS_INIT(Init, APPLICATION, 50);

101
src/main.cpp
View File

@ -1,10 +1,97 @@
#include <ctime>
#include <etl/vector.h>
#include <zephyr/drivers/hwinfo.h>
#include <zephyr/drivers/sensor.h>
#include <zephyr/device.h>
#include <zephyr/kernel.h>
auto main(void) -> int {
printk("[main] init on %s\n", CONFIG_BOARD);
while (1) {
k_sleep(K_SECONDS(1));
}
#include <led_strip_indicator/led_strip_indicator.hpp>
#include <uart_com/protocal.hpp>
#include <zpp/device.hpp>
#include "watchdog.hpp"
#include "zephyr/drivers/gpio.h"
#include "zephyr/kernel.h"
#include "zpp/fmt.hpp"
#include "zpp/timer.hpp"
#include <zephyr/drivers/led_strip.h>
#include <zephyr/sys/printk.h>
namespace {
auto sensor = DEVICE_DT_GET(DT_NODELABEL(godtek));
auto &pmc = ZPP_DRV_GET(uart_com::Protocal, DT_NODELABEL(pm_protocal));
auto &led_strip = ZPP_DRV_GET(ledstrip::Indicator, DT_NODELABEL(indicator));
enum Command { kTemp = 0x00, kGetId, kRunningState };
enum RunningState {};
constexpr auto kDeviceIdSize = 20;
enum StatusId : uint8_t { kStandby, kRunning, kPause, kError};
volatile bool flag{false};
gpio_dt_spec led_g = GPIO_DT_SPEC_GET(DT_NODELABEL(led_g), gpios);
k_timer led_timer;
} // namespace
const led_rgb red{255, 0, 0};
const led_rgb green{0, 255, 0};
const led_rgb blue{0, 0, 255};
auto rgb_init(void)->int{
led_strip.TurnOn(red);
k_msleep(10);
led_strip.TurnOff();
led_strip.TurnOn(green);
k_msleep(10);
led_strip.TurnOff();
led_strip.TurnOn(blue); // 常亮
// k_msleep(10);
// led_strip.TurnOff();
return 0;
}
auto main(void) -> int {
gpio_pin_configure_dt(&led_g, GPIO_OUTPUT_ACTIVE);
k_timer_init(&led_timer, [](k_timer* tim) {
gpio_pin_toggle_dt(&led_g);
}, [](k_timer* tim) {});
k_timer_start(&led_timer, K_NO_WAIT, K_SECONDS(1));
rgb_init();
pmc.SetRxCallback(kGetId, [](uart_com::DataType data) -> void {
uint8_t buffer[kDeviceIdSize];
auto size = hwinfo_get_device_id(buffer, sizeof(buffer));
pmc.Send(kGetId, uart_com::DataType(buffer, size));
});
pmc.SetRxCallback(kRunningState, [](uart_com::DataType data) -> void {
auto state = static_cast<StatusId>(data[0]);
pmc.Send(
kRunningState,
uart_com::DataType(reinterpret_cast<uint8_t *>(&state), sizeof(state)));
led_strip.Status(state).on_error([](zpp::error_code code) {});
});
auto wdt = app::WatchDogConfig{};
sensor_trigger tri{.type = SENSOR_TRIG_DATA_READY,
.chan = SENSOR_CHAN_AMBIENT_TEMP};
if (auto r = sensor_trigger_set(
sensor, &tri,
[](const device *dev, const sensor_trigger *trig) { flag = true; });
r != 0) {
return -ENODEV;
};
while (1) {
if (flag) {
sensor_value val{0, 0};
if (sensor_sample_fetch_chan(sensor, SENSOR_CHAN_AMBIENT_TEMP) == 0) {
sensor_channel_get(sensor, SENSOR_CHAN_AMBIENT_TEMP, &val);
const uint8_t data[2] = {static_cast<uint8_t>(val.val1),
static_cast<uint8_t>(val.val2)};
pmc.Send("temp", data);
}
flag = false;
}
wdt.Feed();
}
return 0;
}

18
src/sample_hello_world.cpp
View File

@ -1,18 +0,0 @@
#include <zephyr/drivers/led.h>
#include <zephyr/kernel.h>
led_dt_spec status = LED_DT_SPEC_GET(DT_NODELABEL(status_led));
auto main() -> int {
printk("hello world %s\n", CONFIG_BOARD);
printk("hello world %s\n", CONFIG_BOARD);
printk("hello world %s\n", CONFIG_BOARD);
printk("hello world %s\n", CONFIG_BOARD);
while (true) {
printk("hello world\n");
led_on_dt(&status);
k_sleep(K_MSEC(1000));
led_off_dt(&status);
k_sleep(K_MSEC(1000));
}
return 0;
}

18
src/sample_ntc.cpp
View File

@ -1,18 +0,0 @@
#include <zephyr/device.h>
#include <zephyr/drivers/sensor.h>
#include <zephyr/kernel.h>
#define NTC_INS(node) DEVICE_DT_GET(node),
const device *ntc = DEVICE_DT_GET(DT_NODELABEL(pole_ntc1));
auto main() -> int {
sensor_value v;
while (1) {
if (0 == sensor_sample_fetch(ntc)) {
sensor_channel_get(ntc, SENSOR_CHAN_AMBIENT_TEMP, &v);
printk("temp: %d.%d\n", v.val1, v.val2);
}
k_sleep(K_SECONDS(1));
}
return 0;
}

93
src/temp.cpp
View File

@ -1,93 +0,0 @@
#include <com.hpp>
#include <etl/utility.h>
#include <infrared.hpp>
#include <zephyr/drivers/gpio.h>
#include <zephyr/init.h>
#include <zephyr/kernel.h>
/* 1 = enable verbose diagnostics (per-second channel dump). */
#ifndef APP_DEBUG_PRINT
#define APP_DEBUG_PRINT 0
#endif
using namespace ther;
/* 调试打印开关：
* 1 = 每次触发打印全部 9 路 + 最大值
* 0 / 注释 = 只打印最大值（生产模式）
*/
#ifndef APP_TEMP_PRINT_ALL_CHANNELS
#define APP_TEMP_PRINT_ALL_CHANNELS 0
#endif
namespace {
using namespace std::string_literals;
constexpr const char *MODULE = "app_temp";
constexpr auto kPeriodSend = std::chrono::milliseconds(20);
/* 触发回调：只做调试打印（采集/缓存由 Infrared 内部完成），
* 发送由 20ms 周期的 OnSendTick 负责。 */
constexpr auto OnSensorValueReady =
etl::delegate<void(sensor_value)>::create(+[](sensor_value val) {
#if APP_TEMP_PRINT_ALL_CHANNELS
const auto all = Infrared::GetAllSensorValues();
for (size_t i = 0; i < all.size(); ++i) {
printk("[%s]ch%u: %d.%d\n", MODULE, (unsigned)i, all[i].val1,
all[i].val2);
}
printk("[%s]max: %d.%d (%lld ms)\n", MODULE, val.val1, val.val2,
(long long)k_uptime_get());
#endif
});
static void OnSendTick(struct k_work *work);
/* 每 kPeriodSend(20ms) 发送一次最新缓存的最大值。 */
static K_WORK_DELAYABLE_DEFINE(s_send_work, OnSendTick);
static void OnSendTick(struct k_work *work) {
if (auto r = Infrared::GetMaxSensorValue(); r.ok()) {
Com::SendTemp(r.value().second);
}
#if APP_DEBUG_PRINT
static uint32_t s_send_tick; /* diag: send counter, ~1s cadence */
if (++s_send_tick % 50 == 0) { /* every ~1 s */
const auto all = Infrared::GetAllSensorValues();
printk("[%s]diag tick=%u mask=0x%02X val:", MODULE, (unsigned)s_send_tick,
(unsigned)Infrared::GetChannelValidMask());
for (size_t i = 0; i < all.size(); ++i) {
printk(" %u=%d.%d", (unsigned)i, all[i].val1, all[i].val2);
}
printk("\n");
}
#endif /* APP_DEBUG_PRINT */
k_work_schedule(&s_send_work, K_MSEC(kPeriodSend.count()));
}
} // namespace
#define ZEPHYR_USER_NODE DT_PATH(zephyr_user)
auto InitEnblePins() {
const gpio_dt_spec en_pin = GPIO_DT_SPEC_GET(ZEPHYR_USER_NODE, en_ir_gpios);
gpio_pin_configure_dt(&en_pin, GPIO_OUTPUT_ACTIVE);
}
static auto Init() -> int {
InitEnblePins();
printk("[%s]temp init\n", MODULE);
if (auto r = Infrared::Init(); r.code_id() != zpp::error_code::k_ok) {
printk("[%s] Infrared::Init failed: %d\n", MODULE,
static_cast<int>(r.code_id()));
return 0;
}
if (auto r = Infrared::AddCallbackWhenSensorValueReady(OnSensorValueReady);
r.code_id() != zpp::error_code::k_ok) {
printk("[%s] AddCallback failed: %d\n", MODULE,
static_cast<int>(r.code_id()));
}
/* 启动 20ms 周期发送 */
k_work_schedule(&s_send_work, K_MSEC(kPeriodSend.count()));
return 0;
}
/* Register AFTER the godtek/ch9438 drivers (POST_KERNEL): registering the
* sensor trigger earlier gets wiped by the sensor driver's own Init,
* which resets data_ready_handler to nullptr. */
SYS_INIT(Init, APPLICATION, 50);

16
src/watdog.cpp
View File

@ -1,16 +0,0 @@
#include <watchdog.hpp>
#include <zpp/work_queue.hpp>
using namespace ther;
namespace {
WatchDogConfig wdt;
} // namespace
static auto Init() -> int {
printk("[watchdog] init\n");
using namespace std::chrono_literals;
static zpp::periodic_work<> work{[]() { wdt.Feed(); }};
work.submit(50ms);
return 0;
}
SYS_INIT(Init, POST_KERNEL, 50);

15
testcase.yaml
View File

@ -1,15 +0,0 @@
common:
platform:
- native_sim/native/64
- nucleo_f429zi
tests:
sample.helloworld:
extra_configs:
- CONFIG_SAMPLE_HELLOWORLD=y
sample.ntc:
extra_configs:
- CONFIG_SAMPLE_NTC=y
sample.ms:
extra_args:
- DTC_OVERLAY_FILE=boards/use_ms.overlay
extra_configs:

51
west.yml
View File

@ -1,51 +0,0 @@
manifest:
projects:
- import:
name-allowlist:
- cmsis_6
- hal_stm32
- hal_ti
path-prefix: extern
name: zephyr
path: zephyr
remote: zephyr
revision: v4.4.0
- name: etl
path: modules/etl
remote: robotstorm
revision: master
- name: uart_com
path: modules/uart_com
remote: robotstorm
revision: main
- name: zpp
path: modules/zpp
remote: robotstorm
revision: dev
- name: dr2501a
path: boards/dr2501a
remote: robotstorm
revision: main
- name: led_strip_indicator
path: modules/led_strip_indicator
remote: robotstorm
revision: main
- name: godtek_temp
path: modules/godtek_temp
remote: robotstorm
revision: main
- name: ch9438
path: modules/ch9438
remote: robotstorm
revision: main
# - name: heading_pad
# path: modules/heading_pad
# revision: main
# url: undefined
remotes:
- name: zephyr
url-base: https://gitcode.com/gh_mirrors/ze
- name: robotstorm
url-base: https://git.robotstorm.tech/EmbeddedTeam
self:
west-commands: scripts/west-commands.yml

172
zbuild.py
View File

@ -1,172 +0,0 @@
#!/usr/bin/env python3
"""
Zephyr build helper for app_i_core_link.
Usage:
python zbuild.py dr2501a_g070rb
python zbuild.py dr2501a_g070rb -p auto
python zbuild.py -p always (auto-detect board from west.yml)
python zbuild.py dr2501a_g070rb -T sample.helloworld
Options:
board Board name (optional, auto-detected from west.yml if omitted)
-p PURSE Pristine option: auto / always / never
-t TARGET CMake target to run after build, e.g. test / flash
-T TEST Test name/sample identifier (e.g. sample.helloworld)
Steps:
1. west topdir → find Zephyr workspace root
2. west config --local manifest.file <this_script_dir>/west.yml
3. west build [-p PURSE] -b BOARD [-t TARGET] [-T TEST] <this_script_dir>
"""
import argparse
import os
import re
import subprocess
import sys
def get_script_dir() -> str:
"""Absolute path to the directory containing this script."""
return os.path.dirname(os.path.abspath(__file__))
def run(cmd: list[str], cwd: str | None = None) -> str:
"""Run a command and return stdout. Exit on failure."""
proc = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
if proc.returncode != 0:
print(f" ✗ {' '.join(cmd)}", file=sys.stderr)
print(f" {proc.stderr.strip()}", file=sys.stderr)
sys.exit(proc.returncode)
return proc.stdout.strip()
def discover_boards(manifest_path: str) -> list[str]:
"""Parse west.yml and return board names whose path starts with 'boards/'."""
if not os.path.exists(manifest_path):
return []
with open(manifest_path) as f:
content = f.read()
# Match entries like:
# - name: <board_name>
# path: boards/<something>
boards = re.findall(r"name:\s*(\S+)\s*\n\s+path:\s*boards/\S+", content)
return boards
def main():
parser = argparse.ArgumentParser(
description="Zephyr build helper for app_i_core_link"
)
parser.add_argument(
"board",
nargs="?",
default=None,
help="Board name (optional, auto-detected from west.yml if omitted)",
)
parser.add_argument(
"-p",
"--pristine",
default=None,
choices=["auto", "always", "never"],
help="Pristine build option",
)
parser.add_argument(
"-t",
"--target",
default=None,
help="CMake target to run after build (e.g. test, flash)",
)
parser.add_argument(
"-T",
"--test",
default=None,
dest="test_id",
help="Test name / sample identifier (e.g. sample.helloworld)",
)
parser.add_argument(
"--update",
action="store_true",
help="Run west update after setting manifest",
)
args = parser.parse_args()
# ── 0. Auto-detect board from west.yml when not specified ───
if args.board is None:
script_dir = get_script_dir()
manifest = os.path.join(script_dir, "west.yml")
boards = discover_boards(manifest)
if len(boards) == 0:
print(
" ✗ No board specified and no board found in west.yml "
"(no entry with path: boards/...)",
file=sys.stderr,
)
sys.exit(1)
elif len(boards) == 1:
args.board = boards[0]
print(f" → Auto-detected board: {args.board}")
else:
print(
" ✗ No board specified. Multiple boards found in west.yml:",
", ".join(boards),
"\n Please specify one explicitly.",
file=sys.stderr,
)
sys.exit(1)
script_dir = get_script_dir()
manifest = os.path.join(script_dir, "west.yml")
print(f" script dir : {script_dir}")
print(f" board : {args.board}")
print(f" pristine : {args.pristine or '(none)'}")
print(f" target : {args.target or '(none)'}")
print(f" test : {args.test_id or '(none)'}")
print(f" update : {'yes' if args.update else 'no'}")
# ── 1. Find workspace root ───────────────────────────────
print("\n [1/3] Finding west workspace root...")
topdir = run(["west", "topdir"])
print(f" → {topdir}")
# ── 2. Set local manifest ────────────────────────────────
print("\n [2/3] Setting local manifest...")
if not os.path.exists(manifest):
print(f" ✗ manifest not found: {manifest}", file=sys.stderr)
sys.exit(1)
run(["west", "config", "--local", "manifest.file", manifest], cwd=topdir)
print(f" → manifest.file = {manifest}")
# ── 2.5. Run west update (optional) ───────────────────────
if args.update:
print("\n [2.5/3] Running west update...")
run(["west", "update", "--fetch", "smart"], cwd=topdir)
print(" → update done")
# ── 3. Run west build ─────────────────────────────────────
print("\n [3/3] Running west build...")
build_cmd = ["west", "build"]
if args.pristine:
build_cmd += ["-p", args.pristine]
build_cmd += ["-b", args.board]
if args.target:
build_cmd += ["-t", args.target]
if args.test_id:
build_cmd += ["-T", args.test_id]
build_cmd += [script_dir]
print(f" $ {' '.join(build_cmd)}")
proc = subprocess.run(build_cmd, cwd=topdir)
if proc.returncode != 0:
print(f"\n ✗ Build failed (exit code {proc.returncode})")
sys.exit(proc.returncode)
print("\n ✓ Build succeeded")
if __name__ == "__main__":
main()