30 changed files with 3087 additions and 211 deletions
+48 -57
View File
@@ -7,34 +7,34 @@
业务按键或业务 NFC 线程。
产测固件复用 SDK 底层接口、ES8311 配置、DP1312EA 驱动和 Type A 读卡实现。
测试结果通过 USB CDC 和调试串口输出。上位机只需解析以 `FT1 ` 开头的行,
测试结果通过 DBG UART 输出。上位机只需解析以 `FT1 ` 开头的行,
其他 SDK 调试信息可以忽略。
默认通信参数:
- USB CDC,115200 8N1;
- 同一条协议也通过 `printf` 输出到调试日志;
- DBG UART(GPIO29/30),115200 8N1;
- 协议与 SDK 调试日志共用 `printf` 输出;
- 每条协议消息为一行 UTF-8/ASCII JSON,以 `\r\n` 结束;
- 协议前缀固定为 `FT1 `,协议版本为 `1`。
- 协议前缀固定为 `FT1 `,协议版本为 `2`。
## 2. 测试流程
```text
长按电源键开机
-> 立即在后台打开 GNSS,持续累计 NMEA、卫星数和定位状态
-> 等待音频子系统就绪
-> ES8311 ID/初始化,设置所有播放音量为 100%
-> 播放开机提示音
-> 电源 ADC
-> 按住对话键录音,松开后回放,最多录 5 秒
-> 按住 GPIO84(原三击测试按键)录音,松开后回放,最多录 5 秒
-> 工人判断音质
正常:按另一个按键三次
异常:不再操作,60 秒后 KEY_TRIPLE 失败
-> eSIM READY 和 ICCID
-> 4G 注册、PDP 激活、获取 IPv4
-> Ping 业务域名
-> 前面全部通过后启用 NFC 测试
-> 读取 NTAG213 UID
-> 播放“全部测试通过”
-> 播放一次 GNSS 测试提示音并读取后台累计结果
-> 未达到门槛时保持呼吸灯并持续等待
-> GNSS 通过后结束搜索
-> 播放关机提示音
-> 输出最终报告
-> 正常关机
@@ -49,16 +49,19 @@
|---:|---|---|---|---:|
| 0 | `BOOT_AUDIO` | 读取 ES8311 ID,初始化外置 Codec,播放 `U:/factory_boot_zh.mp3` | Codec ID 为 `0x83`,文件存在且播放结束 | 15 秒播放超时 |
| 1 | `POWER_ADC` | 读取 6 次 VBAT,去掉最大值和最小值后平均 | `3500mV <= 平均值 <= 4200mV` | - |
| 2 | `AUDIO_LOOPBACK` | 按住电源/对话键录音,松开后按原音量回放 | 录音和回放 API 成功且至少有一帧 PCM | 等待按键 60 秒,录音最长 5 秒 |
| 3 | `KEY_TRIPLE` | GPIO84 按下并释放三次,每次输出进度 PASS | 60 秒内累计三次有效按键 | 60 秒 |
| 4 | `ESIM` | 查询 SIM 状态并读取 ICCID | `READY`,ICCID 为 18-22 位十进制数字 | 30 秒 |
| 5 | `DATA_CALL` | 等待注册、激活 CID 1 IPv4 PDP | 返回非 `0.0.0.0` 的 IPv4 | 注册等待 120 秒 |
| 6 | `PING` | Ping `banban.api.talkingq.com` 3 次 | 至少收到 2 次回复 | 单包 4 秒,总等待 20 秒 |
| 7 | `NFC` | 检查 DP1312EA 版本并轮询 NTAG213 | 版本 `0x82`,成功读取 UID | 60 秒 |
| 2 | `AUDIO_LOOPBACK` | 按住 GPIO84 录音,松开后按原音量回放 | 录音和回放 API 成功且至少有一帧 PCM | 等待按键 60 秒,录音最长 5 秒 |
| 3 | `ESIM` | 查询 SIM 状态并读取 ICCID | `READY`,且能读到 18-22 位 ICCID | 30 秒 |
| 4 | `DATA_CALL` | 等待注册、激活 CID 1 IPv4 PDP | 返回非 `0.0.0.0` 的 IPv4 | 注册等待 120 秒 |
| 5 | `NFC` | 检查 DP1312EA 版本并轮询 NTAG213 | 版本 `0x82`,成功读取 UID | 60 秒 |
| 6 | `GNSS` | 开机后后台搜索,到节点后循环读取累计值 | `open_ret=0` 且 `nmea_count>=5`;卫星数和定位状态只记录 | 无限等待直到通过;打开失败时每 2 秒重试 |
说明:`AUDIO_LOOPBACK` 的软件通过仅代表数据链路完成。最终声学判断由工人完成;
如果声音异常,工人不按三击键,随后 `KEY_TRIPLE` 超时,整机产测失败。这避免了
上位机额外的人工确认按钮。
说明:`AUDIO_LOOPBACK` 使用 GPIO84 按键,软件通过仅代表录放数据链路完成。
最终声学判断由工人完成。
GPIO70 蓝色状态灯在普通流程中长亮,Loopback 使用 250ms 慢闪,NFC 使用 100ms
快闪,GNSS 使用 PWM1 做约 1.4 秒一轮的呼吸效果。如果 PWM 配置或使能失败,
GNSS 会自动降级为两次 80ms 短闪后熄灭 520ms,不影响测试判定。GNSS 未达到
基础硬件门槛时不会报失败或退出节点,指示灯持续呼吸;通过后才恢复常亮并进入关机流程。
## 4. 音频和文件系统接口
@@ -66,7 +69,9 @@
```text
U:/factory_boot_zh.mp3
U:/factory_pass_zh.mp3
U:/factory_audio_loopback_zh.mp3
U:/factory_nfc_test_zh.mp3
U:/factory_gnss_test_zh.mp3
U:/factory_shutdown_zh.mp3
```
@@ -77,8 +82,8 @@ audio_prompts/factory_test/customer_fs/
```
资源为 16kHz、单声道、48kbps MP3。代码不会下载提示音,也不会在运行时回退到
业务网络资源。缺少开机提示音会返回 `1002`;最终提示音缺失或播放失败会返回
`1801`,并禁止自动关机。
业务网络资源。节点提示音缺失或播放失败会返回 `1002`/`1003`;关机提示音播放
失败会返回 `1801`,并禁止自动关机。本固件没有“全部测试通过”音频。
本分支不生成、不覆盖 `ql-config/config/EG800AKCN_11LG/customer_fs.bin`。
Windows 打包方法见 [factory_test_windows_fs.md](factory_test_windows_fs.md)。
@@ -89,7 +94,7 @@ Windows 打包方法见 [factory_test_windows_fs.md](factory_test_windows_fs.md)
| 字段 | 说明 |
|---|---|
| `v` | 协议版本,当前为 1 |
| `v` | 协议版本,当前为 2 |
| `type` | `session_start`、`node_start`、`node_progress`、`node_result`、`session_result` 或 `power_off` |
| `seq` | 本次开机内单调递增的消息序号 |
| `session` | RTC 有效时使用时间生成,否则为 `BOOT-<uptime>` |
@@ -104,7 +109,7 @@ Windows 打包方法见 [factory_test_windows_fs.md](factory_test_windows_fs.md)
### 5.2 会话开始
```text
FT1 {"v":1,"type":"session_start","seq":1,"session":"20260923-163010","uptime_ms":2050,"imei":"860123456789012","iccid":"","nfc_uid":"","fw":"FACTORY_TEST_1.0.0","ping_host":"banban.api.talkingq.com","start_time":"2026-09-23T16:30:10"}
FT1 {"v":2,"type":"session_start","seq":1,"session":"20260923-163010","uptime_ms":2050,"imei":"860123456789012","iccid":"","nfc_uid":"","fw":"FACTORY_TEST_1.1.1","start_time":"2026-09-23T16:30:10"}
```
RTC 无效时 `start_time` 为 `null`。RTC 无效不会导致产测失败;测试耗时始终以
@@ -113,51 +118,38 @@ RTC 无效时 `start_time` 为 `null`。RTC 无效不会导致产测失败;测
### 5.3 节点开始
```text
FT1 {"v":1,"type":"node_start","seq":2,"session":"20260923-163010","uptime_ms":2100,"imei":"860123456789012","iccid":"","nfc_uid":"","node":1,"name":"POWER_ADC"}
FT1 {"v":2,"type":"node_start","seq":2,"session":"20260923-163010","uptime_ms":2100,"imei":"860123456789012","iccid":"","nfc_uid":"","node":1,"name":"POWER_ADC"}
```
### 5.4 ADC 通过
```text
FT1 {"v":1,"type":"node_result","seq":3,"session":"20260923-163010","uptime_ms":2450,"imei":"860123456789012","iccid":"","nfc_uid":"","node":1,"name":"POWER_ADC","result":"PASS","code":0,"elapsed_ms":350,"data":{"voltage_mv":3865,"sample_min_mv":3858,"sample_max_mv":3873,"limit_min_mv":3500,"limit_max_mv":4200}}
FT1 {"v":2,"type":"node_result","seq":3,"session":"20260923-163010","uptime_ms":2450,"imei":"860123456789012","iccid":"","nfc_uid":"","node":1,"name":"POWER_ADC","result":"PASS","code":0,"elapsed_ms":350,"data":{"voltage_mv":3865,"sample_min_mv":3858,"sample_max_mv":3873,"limit_min_mv":3500,"limit_max_mv":4200}}
```
### 5.5 三击按键进度
每次完整按下并释放都输出一条 `node_progress`:
### 5.5 数据连接
```text
FT1 {"v":1,"type":"node_progress","seq":8,"session":"20260923-163010","uptime_ms":12300,"imei":"860123456789012","iccid":"","nfc_uid":"","node":3,"name":"KEY_TRIPLE","result":"PASS","data":{"current":1,"required":3}}
FT1 {"v":2,"type":"node_result","seq":14,"session":"20260923-163010","uptime_ms":28000,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"","node":4,"name":"DATA_CALL","result":"PASS","code":0,"elapsed_ms":8200,"data":{"ip":"10.23.45.67","ip_type":"IPv4"}}
```
第三次松开后再输出正式的 `node_result=PASS`。
### 5.6 网络和 Ping
### 5.6 NFC、GNSS 和最终报告
```text
FT1 {"v":1,"type":"node_result","seq":14,"session":"20260923-163010","uptime_ms":28000,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"","node":5,"name":"DATA_CALL","result":"PASS","code":0,"elapsed_ms":8200,"data":{"ip":"10.23.45.67","ip_type":"IPv4"}}
```
```text
FT1 {"v":1,"type":"node_result","seq":16,"session":"20260923-163010","uptime_ms":31500,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"","node":6,"name":"PING","result":"PASS","code":0,"elapsed_ms":3400,"data":{"host":"banban.api.talkingq.com","sent":3,"received":3,"lost":0,"avg_rtt_ms":86}}
```
### 5.7 NFC 和最终报告
```text
FT1 {"v":1,"type":"node_result","seq":18,"session":"20260923-163010","uptime_ms":34000,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","node":7,"name":"NFC","result":"PASS","code":0,"elapsed_ms":1900,"data":{"chip_version":"0x82","uid":"04A1B2C3D4E5F6"}}
FT1 {"v":2,"type":"node_result","seq":18,"session":"20260923-163010","uptime_ms":34000,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","node":5,"name":"NFC","result":"PASS","code":0,"elapsed_ms":1900,"data":{"chip_version":"0x82","uid":"04A1B2C3D4E5F6"}}
FT1 {"v":2,"type":"node_result","seq":20,"session":"20260923-163010","uptime_ms":35200,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","node":6,"name":"GNSS","result":"PASS","code":0,"elapsed_ms":950,"data":{"open_ret":0,"nmea_count":65,"sat_count":0,"fix_status":0,"search_ms":35100,"required_nmea_count":5,"required_sat_count":0}}
```
最终报告包含 IMEI、ICCID、NFC UID、失败码、开始时间、总耗时和全部节点状态:
```text
FT1 {"v":1,"type":"session_result","seq":19,"session":"20260923-163010","uptime_ms":40120,"result":"PASS","fail_code":0,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","start_time":"2026-09-23T16:30:10","duration_ms":38070,"nodes":{"BOOT_AUDIO":"PASS","POWER_ADC":"PASS","AUDIO_LOOPBACK":"PASS","KEY_TRIPLE":"PASS","ESIM":"PASS","DATA_CALL":"PASS","PING":"PASS","NFC":"PASS"}}
FT1 {"v":2,"type":"session_result","seq":21,"session":"20260923-163010","uptime_ms":40120,"result":"PASS","fail_code":0,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","start_time":"2026-09-23T16:30:10","duration_ms":38070,"nodes":{"BOOT_AUDIO":"PASS","POWER_ADC":"PASS","AUDIO_LOOPBACK":"PASS","ESIM":"PASS","DATA_CALL":"PASS","NFC":"PASS","GNSS":"PASS"}}
```
上位机收到 `power_off` 后应立即落盘本次记录:
```text
FT1 {"v":1,"type":"power_off","seq":20,"session":"20260923-163010","uptime_ms":40150,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","code":0}
FT1 {"v":2,"type":"power_off","seq":22,"session":"20260923-163010","uptime_ms":40150,"imei":"860123456789012","iccid":"89860xxxxxxxxxxxxxxx","nfc_uid":"04A1B2C3D4E5F6","code":0}
```
## 6. 失败码
@@ -166,25 +158,24 @@ FT1 {"v":1,"type":"power_off","seq":20,"session":"20260923-163010","uptime_ms":4
|---:|---|
| 0 | 成功 |
| 1001 | ES8311 初始化或设备 ID 检查失败 |
| 1002 | 开机提示音文件不存在 |
| 1003 | 开机提示音播放失败或超时 |
| 1002 | 当前节点提示音文件不存在 |
| 1003 | 当前节点提示音播放失败或超时 |
| 1101 | ADC 初始化或读取失败 |
| 1102 | 电压低于 3500mV |
| 1103 | 电压高于 4200mV |
| 1201 | 录音句柄或 160KB 缓冲区创建失败 |
| 1202 | 没有录到一帧有效 PCM |
| 1203 | PCM 回放失败 |
| 1204 | 60 秒内未按下对话键 |
| 1301 | 音质异常未继续,或 60 秒内未完成三击 |
| 1204 | 60 秒内未按下 Loopback 按键 |
| 1401 | eSIM 未进入 READY |
| 1402 | ICCID 读取失败 |
| 1403 | ICCID 长度或字符格式错误 |
| 1403 | ICCID 长度不在 18-22 位范围内 |
| 1501 | 4G 网络注册超时 |
| 1502 | IPv4 PDP 激活失败 |
| 1503 | 未获取有效 IPv4 地址 |
| 1601 | Ping 上下文或同步对象创建失败 |
| 1602 | Ping 启动或等待回调失败 |
| 1603 | 3 包中收到少于 2 包 |
| 1601 | GNSS 打开失败 |
| 1602 | GNSS 打开成功但未收到 NMEA |
| 1603 | GNSS 已收到 NMEA,但未发现卫星且没有有效定位 |
| 1701 | 前置节点未全部通过,NFC 锁定 |
| 1702 | DP1312EA 版本寄存器不是 `0x82` |
| 1703 | 60 秒内没有成功刷卡 |
@@ -196,18 +187,18 @@ FT1 {"v":1,"type":"power_off","seq":20,"session":"20260923-163010","uptime_ms":4
- `interface/factory_test/factory_test.c`:顺序状态机和具体测试;
- `interface/factory_test/factory_audio.c`:ES8311、100% 音量、MP3、录音回放;
- `interface/factory_test/factory_gnss.c`:开机后台搜星、NMEA/GSV/GGA 累计;
- `interface/factory_test/factory_protocol.c`:FT1 JSON 日志;
- `interface/factory_test/factory_test.h`:阈值、超时、业务域名、GPIO 和文件路径;
- `interface/factory_test/Makefile`:独立产测编译单元。
修改 Ping 地址时只需修改 `FT_PING_HOST`。ADC 阈值、按键超时和 NFC 超时也集中在
`factory_test.h`。GPIO84 是第二个按键,GPIO68 是功放控制;产测代码不会把 GPIO68
ADC 阈值、按键超时和 NFC 超时集中在 `factory_test.h`。GPIO84 是第二个按键,GPIO68 是功放控制;产测代码不会把 GPIO68
当作 LED,也不会把 GPIO86 当作红色 LED,避免和 NFC 复位冲突。
## 8. 上板前检查
1. 确认 PCB 上对话键确实连接模组 PWRKEY,第二按键连接 GPIO84。
2. 确认 GPIO68 为 NS4150B `PA_CTL`,GPIO86 为 NFC Reset。
3. 将三段 MP3 打入 U 盘文件系统,不能只更新 APP。
4. 首板先观察 ES8311 ID、ADC 原始值、IP 和 DP1312EA 版本日志。
3. 将五段 MP3 打入 U 盘文件系统,不能只更新 APP。
4. 首板先观察 ES8311 ID、ADC 原始值、IP、DP1312EA 版本和 GNSS 累计值。
5. 上位机只把 `FT1 ` 行作为协议消息,SDK 的普通 `printf` 不应进入数据库。
+7 -1
View File
@@ -2,10 +2,13 @@
## 1. 说明
产测固件从 U 盘文件系统根目录读取以下三个文件:
产测音频包包含以下六个文件:
```text
factory_boot_zh.mp3
factory_audio_loopback_zh.mp3
factory_nfc_test_zh.mp3
factory_gnss_test_zh.mp3
factory_pass_zh.mp3
factory_shutdown_zh.mp3
```
@@ -44,6 +47,9 @@ dir audio_prompts\factory_test\customer_fs
```text
factory_boot_zh.mp3
factory_audio_loopback_zh.mp3
factory_nfc_test_zh.mp3
factory_gnss_test_zh.mp3
factory_pass_zh.mp3
factory_shutdown_zh.mp3
```
+6
View File
@@ -0,0 +1,6 @@
__pycache__/
*.pyc
build/
dist/
dist-*/
vendor/
+106
View File
@@ -0,0 +1,106 @@
# EG800 FT1 工厂烧录与产测上位机
本目录包含上位机源码、默认配置、测试、构建脚本和发布文档。最终交付物统一生成到
`factory_host/dist/`,工厂不需要接触项目源码。
## 工程结构
```text
factory_host/
├─ src/ # Python 上位机源码
│ ├─ factory_host.py
│ └─ ft1_protocol.py
├─ config/ # 开发和发布使用的默认配置
│ └─ factory_host_config.json
├─ scripts/ # 音频、固件和 release 构建脚本
│ ├─ build_factory_audio_package.py
│ ├─ build_factory_firmware.ps1
│ └─ build_release.ps1
├─ tests/ # 自动测试
│ └─ test_factory_host.py
├─ docs/
│ ├─ USAGE.md # 工厂现场操作说明
│ └─ DEVELOPMENT.md # 开发、换包和发布说明
├─ firmware/ # 默认产测与正式固件
├─ vendor/ # 本地 Aboot 厂商工具,不提交
├─ build/ # 中间产物和开发日志,不提交
├─ dist/ # 最终交付目录,每次发布会重建
├─ requirements.txt
└─ README.md
```
`build/` 和 `dist/` 都被 Git 忽略。发布脚本只会清理 `dist/`,不会修改 `src/`、
`config/`、`docs/`、`tests/` 或 `build/`。
## 功能
- 产测固件与正式固件独立保存并醒目标识。
- 持续检测 ASR AROM 下载口,按物理 USB 设备去重,支持多工位自动等待设备上电后烧录。
- 同一设备烧录成功后锁定,端口断开并重新出现后才允许烧录下一台。
- 自动烧录忽略同一设备枚举出的 DBG、AT、DIAG 和运行态串口,不会为它们创建伪工位。
- 自动识别 DBG UART,解析设备每 500 ms 广播的 FT1 状态。
- 显示七个测试节点、IMEI、ICCID、NFC UID、失败码和实时日志。
- 自动保存汇总 CSV 与单机 JSON,烧录和扫描过程不弹出控制台黑框。
## 快速验证
在项目根目录执行:
```powershell
python -m pip install -r .\factory_host\requirements.txt
python -B -m unittest discover -s .\factory_host\tests -p "test_*.py" -v
python -B .\factory_host\src\factory_host.py
```
源码模式会读取 `factory_host/config/factory_host_config.json`,运行日志写入
`factory_host/build/logs/`,固件和下载器从现有 `factory_host/dist/` 解析。
## 生成 Release
默认使用项目中的以下输入:
```text
factory_host/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip
factory_host/firmware/official/EG800AKCN_11LG-official-fac.zip
factory_host/vendor/aboot-tools-2024.08.08-win-x64/
```
执行:
```powershell
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_release.ps1
```
生成:
```text
factory_host/dist/
├─ START-EG800-FACTORY-TOOL.exe
├─ firmware/
│ ├─ EG800AKCN_11LG-FT1-factory-test.zip
│ └─ EG800AKCN_11LG-official-fac.zip
├─ .runtime/ # Windows 隐藏目录
│ ├─ burn_tool/
│ ├─ docs/USAGE.md
│ ├─ factory_host_config.json
│ ├─ release_manifest.json
│ └─ logs/
└─ EG800-Factory-Tool-Full-Package.zip
```
交付工厂时只发送 `EG800-Factory-Tool-Full-Package.zip`。解压后双击醒目的
`START-EG800-FACTORY-TOOL.exe`;不要单独移动 EXE、固件或 `.runtime`。
默认配置使用相对于发布目录的路径:
```text
firmware/EG800AKCN_11LG-FT1-factory-test.zip
firmware/EG800AKCN_11LG-official-fac.zip
.runtime/burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe
```
## 文档
- 工厂操作:[docs/USAGE.md](docs/USAGE.md)
- 开发、更新固件与发布:[docs/DEVELOPMENT.md](docs/DEVELOPMENT.md)
- 固件侧 FT1 协议:[../docs/factory_test.md](../docs/factory_test.md)
@@ -0,0 +1,39 @@
{
"flash": {
"aboot_path": ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/aboot.exe",
"downloader_path": ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe",
"args": [
"-u",
"-a",
"-s",
"{baudrate}",
"-q",
"{firmware}"
],
"auto_args": [
"-p",
"{port}",
"-a",
"-s",
"{baudrate}",
"-q",
"{firmware}"
],
"timeout_seconds": 900,
"workers": 2
},
"firmwares": {
"production": "firmware/EG800AKCN_11LG-FT1-factory-test.zip",
"official": "firmware/EG800AKCN_11LG-official-fac.zip"
},
"serial": {
"baudrate": 115200,
"encoding": "utf-8",
"line_ending": "\\r\\n"
},
"protocol": {
"prefix": "FT1 ",
"version": 2,
"auto_save": true
}
}
+150
View File
@@ -0,0 +1,150 @@
# EG800 产测上位机开发与发布
## 1. 目录职责
| 目录 | 用途 | 是否提交 |
| --- | --- | --- |
| `src/` | 上位机与 FT1 协议源码 | 是 |
| `config/` | 默认发布配置 | 是 |
| `scripts/` | 音频、固件、上位机发布脚本 | 是 |
| `tests/` | 单元测试 | 是 |
| `docs/` | 工厂与开发文档 | 是 |
| `firmware/` | 默认产测与正式固件 | 是,二进制由 Git LFS 管理 |
| `vendor/` | 完整 Aboot 厂商工具 | 否,本地依赖 |
| `build/` | 中间产物、备份和开发日志 | 否 |
| `dist/` | 最终 release 目录和 ZIP | 否 |
不要在 `dist/` 保存需要长期保留的文件。`build_release.ps1` 每次发布前都会校验目标位于
`factory_host` 内,然后清空并重建目标目录。
## 2. 环境准备
在项目根目录执行:
```powershell
python -m pip install -r .\factory_host\requirements.txt
python -m pip install pyinstaller
```
完整固件构建还需要当前 EG800 SDK 的编译环境能够正常执行 `build.bat app` 和
`build.bat firmware`。
发布还要求本机存在完整厂商工具目录:
```text
factory_host/vendor/aboot-tools-2024.08.08-win-x64/
```
`vendor/` 被 Git 忽略,不能只放一个 `adownload.exe`;请保留原工具的 DLL、配置、驱动和
平台文件。发布脚本缺少该目录时会直接报错,不会生成残缺包。
## 3. 开发与测试
运行源码:
```powershell
python -B .\factory_host\src\factory_host.py
```
运行全部测试:
```powershell
python -B -m unittest discover -s .\factory_host\tests -p "test_*.py" -v
```
源码模式读取 `factory_host/config/factory_host_config.json`,日志写入
`factory_host/build/logs/`。默认相对固件路径会解析到 `factory_host/dist/firmware/`。
## 4. 更新产测音频和固件
音频源位于:
```text
audio_prompts/factory_test/customer_fs/
```
生成并验证音频文件系统:
```powershell
python -B .\factory_host\scripts\build_factory_audio_package.py
```
输出位于 `factory_host/build/audio/`。脚本会检查五个固定 MP3 的名称与格式,生成
262144 字节 `customer_fs.bin`,反向解包并逐文件核对 SHA-256。
生成产测完整固件:
```powershell
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_factory_firmware.ps1
```
输出位于 `factory_host/build/firmware/`。脚本会临时替换 SDK 的业务
`customer_fs.bin`,完成后恢复原文件并核对哈希。恢复失败时不得继续发布。
确认新固件后,将其更新为默认产测固件:
```text
factory_host/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip
```
也可以不覆盖默认文件,在发布时通过 `-FactoryFirmware` 显式选择。
## 5. 生成交付包
使用默认两份固件:
```powershell
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_release.ps1
```
使用指定固件,路径相对于项目根目录,也支持绝对路径:
```powershell
powershell.exe -NoProfile -ExecutionPolicy Bypass `
-File .\factory_host\scripts\build_release.ps1 `
-FactoryFirmware "factory_host\build\firmware\EG800AKCN_11LG-FT1-factory-test.zip" `
-OfficialFirmware "factory_host\firmware\official\EG800AKCN_11LG-official-fac.zip"
```
发布脚本会:
1. 清空并重建 `factory_host/dist/`。
2. 从 `src/` 生成无控制台窗口的单文件 EXE。
3. 复制完整 Aboot 工具目录,不能只复制 `adownload.exe`。
4. 复制产测与正式固件到 `dist/firmware/`。
5. 把配置、使用说明、下载器和日志目录放入 `.runtime/`。
6. 生成包含 Git revision、文件大小和 SHA-256 的 `release_manifest.json`。
7. 生成 `EG800-Factory-Tool-Full-Package.zip`。
## 6. 发布检查
发布前至少执行:
```powershell
python -B -m unittest discover -s .\factory_host\tests -p "test_*.py" -v
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_release.ps1
tar.exe -tf .\factory_host\dist\EG800-Factory-Tool-Full-Package.zip
```
ZIP 顶层只能包含:
```text
START-EG800-FACTORY-TOOL.exe
firmware/
.runtime/
```
还应在一台真实设备上验证:自动等待上电烧录、断开后才允许下一台、DBG UART 自动识别、
七项 FT1 状态、GNSS 累计数据、JSON/CSV 保存,以及产测固件和正式固件切换。
自动烧录端口枚举必须保留 USB 元数据,只接受 ASR AROM 的 `VID_2ECC/PID_3004`,并用
`location`(其次为序列号、HWID、COM 名)标识物理设备。不要把运行态的 DBG、AT、DIAG
接口交给 `adownload.exe`,否则一台设备会被误建成多个工位。
## 7. 固件替换规则
- 产测与正式固件都必须是 Aboot 可直接烧录的内层 ZIP,不要选择外层 `*-target.zip`。
- 默认文件名保持不变时无需修改配置。
- 修改文件名时同步更新 `config/factory_host_config.json`。
- 每次发布核对 `.runtime/release_manifest.json` 中的大小和 SHA-256。
- 不要把源码、`build/`、SDK 编译目录或历史日志交付给工厂。
+188
View File
@@ -0,0 +1,188 @@
# EG800 产测上位机使用说明
## 1. 发布目录
现场使用时至少保留以下内容,并保持相对目录不变:
```text
factory_host/dist/
├─ START-EG800-FACTORY-TOOL.exe
├─ firmware/
│ ├─ EG800AKCN_11LG-FT1-factory-test.zip
│ └─ EG800AKCN_11LG-official-fac.zip
└─ .runtime/
├─ factory_host_config.json
├─ burn_tool/
│ └─ aboot-tools-2024.08.08-win-x64/
│ ├─ adownload.exe
│ └─ 其他随工具提供的 DLL、配置和驱动文件
├─ docs/USAGE.md
├─ release_manifest.json
└─ logs/
```
`adownload.exe` 不能脱离原来的工具目录单独复制,否则可能缺少 DLL、配置或驱动。
工厂现场只需接收并解压 `EG800-Factory-Tool-Full-Package.zip`,然后双击根目录的
`START-EG800-FACTORY-TOOL.exe`。不要把 `factory_host` 源码目录交付给现场;`.runtime`
用于保存下载器、配置和日志,程序启动时会自动将其设为 Windows 隐藏目录,不应移动或删除。
## 2. 首次配置
上位机读取 `.runtime/factory_host_config.json`。推荐配置:
```json
{
"flash": {
"downloader_path": ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe",
"args": ["-u", "-a", "-s", "{baudrate}", "-q", "{firmware}"],
"timeout_seconds": 900,
"workers": 2
},
"firmwares": {
"production": "firmware/EG800AKCN_11LG-FT1-factory-test.zip",
"official": "firmware/EG800AKCN_11LG-official-fac.zip"
},
"serial": {
"baudrate": 115200,
"encoding": "utf-8",
"line_ending": "\\r\\n"
},
"protocol": {
"prefix": "FT1 ",
"version": 2,
"auto_save": true
}
}
```
- `production`:产测固件,包含 FT1 测试程序以及开机、Loopback、NFC、GNSS、关机五段音频。
- `official`:正式业务固件,供产测完成后换烧。
- 固件路径可以使用相对于 EXE 所在发布目录的路径,也可以填写绝对路径。
- 必须选择可直接烧录的内层 FT1 ZIP 或 `*-fac.zip`,不要选择外层 `*-target.zip` 归档。
- 修改配置后重新启动上位机,或使用界面中的重新加载配置功能。
## 3. 烧录产测固件
1. 启动 `START-EG800-FACTORY-TOOL.exe`,固件类型选择“产测固件”或“正式固件”,确认顶部完整路径。
2. 连接一个或多个烧录器,设备保持关机。
3. 点击醒目的“启动自动烧录”,工具每 0.5 秒持续检测 ASR AROM 下载口,并为每台物理设备建立一个工位。
4. 工位显示“烧录中·等待上电”后给对应设备开机,设备进入 AROM 后自动开始烧录。
5. 工位显示“烧录成功”后先断开当前设备;工具检测到端口消失后才会允许下一次烧录,避免同一设备反复烧录。
6. 换上下一台关机设备并上电,端口重新出现后自动开始下一次烧录。
7. 全部生产完成后点击“停止烧录”;不要在实际写入过程中拔线或关闭程序。
默认参数使用 `adownload.exe -u -a` 自动等待 AROM 下载设备,因此烧录口和产测 UART 是两个不同角色。
自动烧录只接受 `VID_2ECC/PID_3004` 的 AROM 下载口,并按 USB 物理位置去重;同一设备的
DBG、AT、DIAG 及运行态串口不会创建额外工位。启动监听前已经存在的端口作为基线,按标准流程
先启动自动烧录,再给待烧录设备上电。
## 4. 自动识别并开始产测
烧录完成、设备正常启动并连接调试 UART 后:
1. 点击“扫描串口”可只识别设备并查看工位;直接点击“开始产测”也会先自动扫描。
2. 上位机被动监听所有候选 COM 口,不向设备发送 `IDENTIFY` 或 `START` 命令。
3. 只有能持续返回合法 FT1 状态广播的 UART 才会自动添加为工位。当前固件使用 `QL_DEBUG_UART_PORT`(GPIO29/30),不使用 USB CDC;AROM、主 UART、DIAG 和 AT 口不会被添加。
4. 每发现一台设备,上位机自动创建或复用一个空工位,并在工位表显示对应 COM 口。
5. 设备上电后自动开始测试;上位机只接收并解析每 500ms 一次的完整状态广播,不需要发送启动命令。
6. 未手工选中工位时,“开始产测”会启动所有已识别工位;选中工位时只启动选中的设备。
每个工位会实时显示:
- 当前测试节点和通过数量;
- 整机通过或失败状态;
- IMEI、ICCID、NFC UID;
- 失败码、结构化测试数据和实时串口日志。
产测固件开机后立即在后台启动 GNSS 搜索,同时依次执行:开机音频、电源 ADC、音频 Loopback、eSIM、数据连接和 NFC。最后的 GNSS 节点只播放一次提示音,然后循环读取此前累计的 NMEA、卫星数和定位状态,直到达到基础硬件判定门槛。
GNSS 的基础硬件判定门槛为:GNSS 成功打开并累计收到至少 5 条 NMEA。卫星数、定位状态、经纬度、定位精度、TTFF 和 1PPS 只用于日志观察,不参与产线判定,因此可以在室内完成基础数据链路测试;天线和实际搜星能力需要另行室外抽检。
指示灯状态:普通流程长亮,Loopback 慢闪,NFC 快闪,GNSS 为呼吸灯。如果设备运行时无法启用 PWM,GNSS 会自动改为“双短闪 + 长灭”。GNSS 未通过时会无限等待并持续呼吸;通过后才播放关机音并自动关机。
## 5. 测试结果与日志
默认自动保存位置:
```text
factory_host/dist/.runtime/logs/factory_results.csv
factory_host/dist/.runtime/logs/sessions/*.json
```
- `factory_results.csv`:每台设备一行,适合生产汇总。
- `sessions/*.json`:保存完整协议消息、节点耗时、失败码和各节点数据,适合追查单台设备。
- 普通 SDK 日志会显示在实时日志区,但只有 `FT1 ` 开头的合法协议消息会参与结果判定。
- 最终判定以设备发送的 `session_result` 为准,上位机不会根据普通文本日志猜测通过或失败。
## 6. 替换固件包
### 方法一:修改配置文件
把新固件复制到稳定目录,然后修改:
```json
"firmwares": {
"production": "新产测固件.zip",
"official": "新正式固件-fac.zip"
}
```
推荐每次替换后记录文件名、版本、文件大小和 SHA-256,避免现场混用旧包。
### 方法二:在界面选择
选择“产测固件”或“正式固件”类型,再使用界面的固件选择按钮指定对应 ZIP。选择后检查界面顶部显示的完整路径。
正式固件通常不包含 FT1 产测协议。换烧正式固件后,上位机不能再自动识别为 FT1 产测工位,这是正常现象。
## 7. 更新产测音频并生成固件
音频源目录:
```text
audio_prompts/factory_test/customer_fs/
├─ factory_boot_zh.mp3
├─ factory_audio_loopback_zh.mp3
├─ factory_nfc_test_zh.mp3
├─ factory_gnss_test_zh.mp3
└─ factory_shutdown_zh.mp3
```
文件必须保持脚本要求的名称和音频格式。更新后在项目根目录执行:
```powershell
python -B .\factory_host\scripts\build_factory_audio_package.py
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_factory_firmware.ps1
```
生成物:
```text
factory_host/build/audio/customer_fs.bin
factory_host/build/audio/EG800AKCN_11LG-FT1-factory-audio.zip
factory_host/build/firmware/EG800AKCN_11LG-FT1-factory-test.zip
factory_host/build/firmware/firmware_manifest.json
```
固件脚本会临时替换构建目录中的 `customer_fs.bin`,完成后自动恢复业务镜像并核对哈希。若脚本报告恢复失败,不要继续发布固件,应先检查备份和当前业务镜像。
## 8. 重新打包上位机
完整的工程结构、固件更新和发布参数见 `factory_host/docs/DEVELOPMENT.md`。
开发环境先安装依赖:
```powershell
python -m pip install -r .\factory_host\requirements.txt
python -m pip install pyinstaller
```
运行测试并打包:
```powershell
python -B -m unittest discover -s .\factory_host\tests -p "test_*.py" -v
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\factory_host\scripts\build_release.ps1
```
发布前至少使用一台设备验证:产测固件烧录、调试 UART 自动识别、七项测试结果、GNSS 室外或窗边搜星、JSON/CSV 保存,以及正式固件换烧。
+3
View File
@@ -0,0 +1,3 @@
# Optional: improves COM enumeration and serial I/O. The application has a
# Windows file-handle fallback when pyserial is unavailable.
pyserial>=3.5
@@ -0,0 +1,202 @@
"""Validate and package the fixed FT1 audio prompts."""
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
import subprocess
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path
EXPECTED_FILES = (
"factory_boot_zh.mp3",
"factory_audio_loopback_zh.mp3",
"factory_nfc_test_zh.mp3",
"factory_gnss_test_zh.mp3",
"factory_shutdown_zh.mp3",
)
IMAGE_SIZE = 262144
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest().upper()
def _skip_id3(data: bytes) -> int:
if not data.startswith(b"ID3") or len(data) < 10:
return 0
size = 0
for value in data[6:10]:
if value & 0x80:
raise ValueError("ID3 使用了无效的 synchsafe size")
size = (size << 7) | value
return 10 + size
def inspect_mp3(path: Path) -> dict:
data = path.read_bytes()
offset = _skip_id3(data)
while offset + 4 <= len(data) and not (data[offset] == 0xFF and data[offset + 1] & 0xE0 == 0xE0):
offset += 1
if offset + 4 > len(data):
raise ValueError(f"{path.name}: 找不到 MPEG 音频帧")
header = int.from_bytes(data[offset : offset + 4], "big")
version_bits = (header >> 19) & 0x3
layer_bits = (header >> 17) & 0x3
bitrate_index = (header >> 12) & 0xF
sample_index = (header >> 10) & 0x3
channel_mode = (header >> 6) & 0x3
if version_bits == 1 or layer_bits != 1 or bitrate_index in (0, 15) or sample_index == 3:
raise ValueError(f"{path.name}: MP3 帧头不受支持")
version = {3: "MPEG1", 2: "MPEG2", 0: "MPEG2.5"}[version_bits]
sample_rates = {
3: (44100, 48000, 32000),
2: (22050, 24000, 16000),
0: (11025, 12000, 8000),
}
mpeg1_bitrates = (0, 32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320)
mpeg2_bitrates = (0, 8, 16, 24, 32, 40, 48, 56, 64, 80, 96, 112, 128, 144, 160)
bitrate = (mpeg1_bitrates if version_bits == 3 else mpeg2_bitrates)[bitrate_index]
sample_rate = sample_rates[version_bits][sample_index]
channels = 1 if channel_mode == 3 else 2
if (sample_rate, channels, bitrate) != (16000, 1, 48):
raise ValueError(
f"{path.name}: 需要 16000Hz/单声道/48kbps,实际为 {sample_rate}Hz/{channels}ch/{bitrate}kbps"
)
duration_ms = round(len(data) * 8 / (bitrate * 1000) * 1000)
return {
"name": path.name,
"bytes": len(data),
"sha256": sha256(path),
"mpeg": version,
"sample_rate_hz": sample_rate,
"channels": channels,
"bitrate_kbps": bitrate,
"estimated_duration_ms": duration_ms,
}
def build_package(root: Path, output: Path) -> tuple[Path, Path, dict]:
source = root / "audio_prompts" / "factory_test" / "customer_fs"
tool = root / "ql-cross-tool" / "win32" / "mklfs-v2.exe"
if not source.is_dir():
raise FileNotFoundError(f"音频目录不存在: {source}")
if not tool.is_file():
raise FileNotFoundError(f"mklfs-v2.exe 不存在: {tool}")
actual = sorted(path.name for path in source.iterdir() if path.is_file())
if actual != sorted(EXPECTED_FILES):
raise ValueError(f"音频目录必须且只能包含 {EXPECTED_FILES},实际为 {actual}")
output.mkdir(parents=True, exist_ok=True)
image = output / "customer_fs.bin"
if image.exists():
image.unlink()
audio = [inspect_mp3(source / name) for name in EXPECTED_FILES]
result = subprocess.run(
[str(tool), "-m", "-c", str(source), "-s", str(IMAGE_SIZE), "-i", str(image)],
cwd=str(root),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
if result.returncode != 0:
raise RuntimeError(f"mklfs-v2 失败 ({result.returncode}): {result.stdout}\n{result.stderr}")
if not image.is_file() or image.stat().st_size != IMAGE_SIZE:
raise RuntimeError(f"customer_fs.bin 大小错误: {image.stat().st_size if image.exists() else '不存在'}")
verified_files = []
with tempfile.TemporaryDirectory(prefix="verify-", dir=output) as verify_dir:
verify_path = Path(verify_dir)
verify = subprocess.run(
[str(tool), "-o", "-c", str(verify_path), "-s", str(IMAGE_SIZE), "-i", str(image)],
cwd=str(root),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
if verify.returncode != 0:
raise RuntimeError(f"customer_fs 解包验证失败 ({verify.returncode}): {verify.stdout}\n{verify.stderr}")
extracted = sorted(path.name for path in verify_path.iterdir() if path.is_file())
if extracted != sorted(EXPECTED_FILES):
raise RuntimeError(f"customer_fs 文件清单错误: {extracted}")
for name in EXPECTED_FILES:
source_hash = sha256(source / name)
extracted_hash = sha256(verify_path / name)
if source_hash != extracted_hash:
raise RuntimeError(f"customer_fs 解包内容不一致: {name}")
verified_files.append({"name": name, "sha256": extracted_hash})
try:
revision = subprocess.run(
["git", "rev-parse", "HEAD"], cwd=str(root), capture_output=True, text=True, check=True
).stdout.strip()
except (OSError, subprocess.SubprocessError):
revision = ""
manifest = {
"format": "banban-factory-audio-v1",
"generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"git_revision": revision,
"filesystem": {
"name": "customer_fs.bin",
"bytes": image.stat().st_size,
"sha256": sha256(image),
"extract_verified": True,
"verified_files": verified_files,
},
"mount_paths": [f"U:/{name}" for name in EXPECTED_FILES],
"audio": audio,
}
manifest_path = output / "manifest.json"
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8-sig")
readme_path = output / "README.txt"
readme_path.write_text(
"班班 FT1 产测音频包\n\n"
"customer_fs.bin 固定为 262144 字节,用于 EG800AKCN_11LG 的 customer_fs 分区。\n"
"五个 MP3 必须位于 U:/ 根目录;不要将此镜像混入正式业务固件。\n"
"烧录后以 FT1 BOOT_AUDIO 和最终 session_result 作为设备侧验证。\n",
encoding="utf-8-sig",
)
package = output / "EG800AKCN_11LG-FT1-factory-audio.zip"
if package.exists():
package.unlink()
with zipfile.ZipFile(package, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
archive.write(image, "customer_fs.bin")
archive.write(manifest_path, "manifest.json")
archive.write(readme_path, "README.txt")
for name in EXPECTED_FILES:
archive.write(source / name, f"customer_fs/{name}")
manifest["package"] = {"name": package.name, "bytes": package.stat().st_size, "sha256": sha256(package)}
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8-sig")
return image, package, manifest
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[2])
parser.add_argument("--output", type=Path, default=Path(__file__).resolve().parents[1] / "build" / "audio")
args = parser.parse_args()
image, package, manifest = build_package(args.root.resolve(), args.output.resolve())
print(f"customer_fs={image}")
print(f"package={package}")
print(f"customer_fs_sha256={manifest['filesystem']['sha256']}")
print(f"package_sha256={manifest['package']['sha256']}")
if __name__ == "__main__":
main()
@@ -0,0 +1,61 @@
param(
[string]$Output = "build\firmware"
)
$ErrorActionPreference = "Stop"
$scriptRoot = Split-Path -Parent $MyInvocation.MyCommand.Path
$factoryHostRoot = Split-Path -Parent $scriptRoot
$projectRoot = Split-Path -Parent $factoryHostRoot
$audioImage = Join-Path $factoryHostRoot "build\audio\customer_fs.bin"
$configuredImage = Join-Path $projectRoot "ql-config\config\EG800AKCN_11LG\customer_fs.bin"
$flashPackage = Join-Path $projectRoot "target\EG800AKCN_11LG\EG800AKCN_11LG.zip"
$outDir = if ([System.IO.Path]::IsPathRooted($Output)) { $Output } else { Join-Path $factoryHostRoot $Output }
$backup = Join-Path $outDir "customer_fs.business-backup.bin"
$export = Join-Path $outDir "EG800AKCN_11LG-FT1-factory-test.zip"
if (-not (Test-Path -LiteralPath $audioImage)) {
throw "Factory audio image is missing. Run build_factory_audio_package.py first."
}
if ((Get-Item -LiteralPath $audioImage).Length -ne 262144) {
throw "Factory customer_fs.bin is not 262144 bytes."
}
New-Item -ItemType Directory -Force -Path $outDir | Out-Null
Copy-Item -LiteralPath $configuredImage -Destination $backup -Force
$originalHash = (Get-FileHash -LiteralPath $configuredImage -Algorithm SHA256).Hash
$audioHash = (Get-FileHash -LiteralPath $audioImage -Algorithm SHA256).Hash
try {
Copy-Item -LiteralPath $audioImage -Destination $configuredImage -Force
if ((Get-FileHash -LiteralPath $configuredImage -Algorithm SHA256).Hash -ne $audioHash) {
throw "Temporary customer_fs.bin hash verification failed."
}
& cmd.exe /d /c "build.bat app"
if ($LASTEXITCODE -ne 0) { throw "build.bat app failed with exit code $LASTEXITCODE" }
& cmd.exe /d /c "build.bat firmware"
if ($LASTEXITCODE -ne 0) { throw "build.bat firmware failed with exit code $LASTEXITCODE" }
if (-not (Test-Path -LiteralPath $flashPackage)) {
throw "Flashable inner package was not generated: $flashPackage"
}
Copy-Item -LiteralPath $flashPackage -Destination $export -Force
$exportHash = (Get-FileHash -LiteralPath $export -Algorithm SHA256).Hash
@{
package = (Split-Path -Leaf $export)
bytes = (Get-Item -LiteralPath $export).Length
sha256 = $exportHash
customer_fs_sha256 = $audioHash
git_revision = (& git -C $projectRoot rev-parse HEAD).Trim()
generated_at = (Get-Date).ToString("o")
} | ConvertTo-Json | Set-Content -LiteralPath (Join-Path $outDir "firmware_manifest.json") -Encoding UTF8
Write-Host "Factory firmware: $export"
Write-Host "SHA256: $exportHash"
}
finally {
Copy-Item -LiteralPath $backup -Destination $configuredImage -Force
$restoredHash = (Get-FileHash -LiteralPath $configuredImage -Algorithm SHA256).Hash
if ($restoredHash -ne $originalHash) {
throw "Business customer_fs.bin restore verification failed. Backup: $backup"
}
Write-Host "Business customer_fs.bin restored: $restoredHash"
}
+141
View File
@@ -0,0 +1,141 @@
param(
[string]$Output = "dist",
[string]$FactoryFirmware = "",
[string]$OfficialFirmware = ""
)
$ErrorActionPreference = "Stop"
$scriptRoot = Split-Path -Parent $MyInvocation.MyCommand.Path
$factoryHostRoot = Split-Path -Parent $scriptRoot
$projectRoot = Split-Path -Parent $factoryHostRoot
$buildRoot = Join-Path $factoryHostRoot "build"
function Resolve-ProjectInput([string]$Value, [string]$DefaultRelativePath) {
$selected = if ([string]::IsNullOrWhiteSpace($Value)) { $DefaultRelativePath } else { $Value }
if ([System.IO.Path]::IsPathRooted($selected)) {
return [System.IO.Path]::GetFullPath($selected)
}
return [System.IO.Path]::GetFullPath((Join-Path $projectRoot $selected))
}
$outputCandidate = if ([System.IO.Path]::IsPathRooted($Output)) {
$Output
} else {
Join-Path $factoryHostRoot $Output
}
$outDir = [System.IO.Path]::GetFullPath($outputCandidate)
$factoryHostPrefix = [System.IO.Path]::GetFullPath($factoryHostRoot).TrimEnd("\") + "\"
if (-not $outDir.StartsWith($factoryHostPrefix, [System.StringComparison]::OrdinalIgnoreCase)) {
throw "Output 必须位于 factory_host 目录内: $outDir"
}
$pyinstallerDist = Join-Path $buildRoot "pyinstaller-dist"
$pyinstallerWork = Join-Path $buildRoot "pyinstaller-work"
$pyinstallerSpec = Join-Path $buildRoot "pyinstaller-spec"
$buildExe = Join-Path $pyinstallerDist "EG800FactoryHost.exe"
$releaseExe = Join-Path $outDir "START-EG800-FACTORY-TOOL.exe"
$runtimeDir = Join-Path $outDir ".runtime"
$firmwareDir = Join-Path $outDir "firmware"
$toolSource = Join-Path $factoryHostRoot "vendor\aboot-tools-2024.08.08-win-x64"
$toolDestination = Join-Path $runtimeDir "burn_tool\aboot-tools-2024.08.08-win-x64"
$factoryFirmwareSource = Resolve-ProjectInput $FactoryFirmware "factory_host\firmware\test\EG800AKCN_11LG-FT1-factory-test.zip"
$officialFirmwareSource = Resolve-ProjectInput $OfficialFirmware "factory_host\firmware\official\EG800AKCN_11LG-official-fac.zip"
$factoryFirmwareDestination = Join-Path $firmwareDir "EG800AKCN_11LG-FT1-factory-test.zip"
$officialFirmwareDestination = Join-Path $firmwareDir "EG800AKCN_11LG-official-fac.zip"
$releaseZip = Join-Path $outDir "EG800-Factory-Tool-Full-Package.zip"
python -c "import PyInstaller" 2>$null
if ($LASTEXITCODE -ne 0) {
throw "未找到 PyInstaller。先执行: python -m pip install pyinstaller pyserial"
}
foreach ($required in @($toolSource, $factoryFirmwareSource, $officialFirmwareSource)) {
if (-not (Test-Path -LiteralPath $required)) {
throw "发布所需文件不存在: $required"
}
}
# dist is disposable. Source and intermediate artifacts live elsewhere.
New-Item -ItemType Directory -Force -Path $outDir, $buildRoot | Out-Null
Get-ChildItem -LiteralPath $outDir -Force | Remove-Item -Recurse -Force
if (Test-Path -LiteralPath $pyinstallerDist) {
Remove-Item -LiteralPath $pyinstallerDist -Recurse -Force
}
python -m PyInstaller --noconfirm --clean --onefile --windowed `
--name EG800FactoryHost `
--distpath $pyinstallerDist `
--workpath $pyinstallerWork `
--specpath $pyinstallerSpec `
(Join-Path $factoryHostRoot "src\factory_host.py")
if ($LASTEXITCODE -ne 0 -or -not (Test-Path -LiteralPath $buildExe)) {
throw "PyInstaller 构建失败,退出码: $LASTEXITCODE"
}
New-Item -ItemType Directory -Force -Path $runtimeDir, $firmwareDir | Out-Null
Move-Item -LiteralPath $buildExe -Destination $releaseExe -Force
Copy-Item -LiteralPath (Join-Path $factoryHostRoot "config\factory_host_config.json") `
-Destination (Join-Path $runtimeDir "factory_host_config.json") -Force
Copy-Item -LiteralPath $toolSource -Destination $toolDestination -Recurse -Force
Copy-Item -LiteralPath $factoryFirmwareSource -Destination $factoryFirmwareDestination -Force
Copy-Item -LiteralPath $officialFirmwareSource -Destination $officialFirmwareDestination -Force
New-Item -ItemType Directory -Force -Path `
(Join-Path $runtimeDir "docs"), (Join-Path $runtimeDir "logs") | Out-Null
Copy-Item -LiteralPath (Join-Path $factoryHostRoot "docs\USAGE.md") `
-Destination (Join-Path $runtimeDir "docs\USAGE.md") -Force
$revision = ""
try {
$revisionOutput = & git -C $projectRoot rev-parse HEAD 2>$null
if ($LASTEXITCODE -eq 0 -and $revisionOutput) {
$revision = ($revisionOutput | Select-Object -First 1).Trim()
}
} catch {
$revision = ""
}
$manifest = [ordered]@{
format = "eg800-factory-host-release-v1"
generated_at = (Get-Date).ToUniversalTime().ToString("o")
git_revision = $revision
files = [ordered]@{
executable = [ordered]@{
path = "START-EG800-FACTORY-TOOL.exe"
bytes = (Get-Item -LiteralPath $releaseExe).Length
sha256 = (Get-FileHash -LiteralPath $releaseExe -Algorithm SHA256).Hash
}
factory_firmware = [ordered]@{
path = "firmware/EG800AKCN_11LG-FT1-factory-test.zip"
bytes = (Get-Item -LiteralPath $factoryFirmwareDestination).Length
sha256 = (Get-FileHash -LiteralPath $factoryFirmwareDestination -Algorithm SHA256).Hash
}
official_firmware = [ordered]@{
path = "firmware/EG800AKCN_11LG-official-fac.zip"
bytes = (Get-Item -LiteralPath $officialFirmwareDestination).Length
sha256 = (Get-FileHash -LiteralPath $officialFirmwareDestination -Algorithm SHA256).Hash
}
downloader = [ordered]@{
path = ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe"
bytes = (Get-Item -LiteralPath (Join-Path $toolDestination "adownload.exe")).Length
sha256 = (Get-FileHash -LiteralPath (Join-Path $toolDestination "adownload.exe") -Algorithm SHA256).Hash
}
}
}
$manifest | ConvertTo-Json -Depth 5 | Set-Content `
-LiteralPath (Join-Path $runtimeDir "release_manifest.json") -Encoding UTF8
$runtimeItem = Get-Item -LiteralPath $runtimeDir -Force
$runtimeItem.Attributes = $runtimeItem.Attributes -bor [System.IO.FileAttributes]::Hidden
Push-Location $outDir
try {
& tar.exe -a -c -f $releaseZip "START-EG800-FACTORY-TOOL.exe" "firmware" ".runtime"
if ($LASTEXITCODE -ne 0) {
throw "tar.exe 创建完整交付包失败,退出码: $LASTEXITCODE"
}
} finally {
Pop-Location
}
Write-Host "发布目录: $outDir"
Write-Host "上位机: $releaseExe"
Write-Host "完整交付包: $releaseZip"
Write-Host "SHA256: $((Get-FileHash -LiteralPath $releaseZip -Algorithm SHA256).Hash)"
File diff suppressed because it is too large Load Diff
+285
View File
@@ -0,0 +1,285 @@
"""Parser and state model for the Banban FT1 factory-test protocol."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any
PREFIX = "FT1 "
VERSION = 2
MESSAGE_TYPES = {
"session_start",
"node_start",
"node_progress",
"node_result",
"session_result",
"power_off",
"device_ready",
"command_ack",
"status_snapshot",
}
NODE_NAMES = (
"BOOT_AUDIO",
"POWER_ADC",
"AUDIO_LOOPBACK",
"ESIM",
"DATA_CALL",
"NFC",
"GNSS",
)
ERROR_MESSAGES = {
0: "成功",
1001: "ES8311 初始化或设备 ID 检查失败",
1002: "当前节点提示音文件不存在",
1003: "当前节点提示音播放失败或超时",
1101: "ADC 初始化或读取失败",
1102: "电压低于 3500mV",
1103: "电压高于 4200mV",
1201: "录音句柄或缓冲区创建失败",
1202: "没有录到有效 PCM",
1203: "PCM 回放失败",
1204: "60 秒内未按下 Loopback 按键",
1401: "eSIM 未进入 READY",
1402: "ICCID 读取失败",
1403: "ICCID 格式错误",
1501: "4G 网络注册超时",
1502: "IPv4 PDP 激活失败",
1503: "未获取有效 IPv4 地址",
1601: "GNSS 打开失败",
1602: "GNSS 未收到 NMEA",
1603: "GNSS 已收到 NMEA,但未发现卫星",
1701: "前置节点未全部通过,NFC 锁定",
1702: "DP1312EA 版本不是 0x82",
1703: "60 秒内未成功刷卡",
1704: "NFC UID 无效",
1801: "最终提示音播放失败",
1802: "正常关机接口异常返回",
}
class ProtocolError(ValueError):
pass
def error_text(code: Any) -> str:
try:
number = int(code)
except (TypeError, ValueError):
return f"未知失败码 {code}"
return ERROR_MESSAGES.get(number, f"未知失败码 {number}")
def _require_int(message: dict, key: str, minimum: int = 0) -> int:
value = message.get(key)
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise ProtocolError(f"字段 {key} 必须是大于等于 {minimum} 的整数")
return value
def parse_ft1_line(line: str) -> dict | None:
"""Parse one serial line. Non-FT1 lines return None and are not records."""
line = line.strip("\r\n")
if not line.startswith(PREFIX):
return None
try:
message = json.loads(line[len(PREFIX) :])
except json.JSONDecodeError as exc:
raise ProtocolError(f"JSON 无效: {exc.msg}") from exc
if not isinstance(message, dict):
raise ProtocolError("协议载荷必须是 JSON 对象")
if message.get("v") != VERSION:
raise ProtocolError(f"不支持的协议版本: {message.get('v')}")
msg_type = message.get("type")
if msg_type not in MESSAGE_TYPES:
raise ProtocolError(f"不支持的消息类型: {msg_type}")
_require_int(message, "seq", 1)
_require_int(message, "uptime_ms", 0)
if not isinstance(message.get("session"), str) or not message["session"]:
raise ProtocolError("缺少 session")
if msg_type in {"node_start", "node_progress", "node_result"}:
node = _require_int(message, "node", 0)
if node >= len(NODE_NAMES):
raise ProtocolError(f"节点编号越界: {node}")
if message.get("name") != NODE_NAMES[node]:
raise ProtocolError(f"节点名称不匹配: {node}/{message.get('name')}")
if msg_type == "node_progress":
data = message.get("data")
if not isinstance(data, dict):
raise ProtocolError("node_progress 缺少 data")
current = _require_int(data, "current", 0)
required = _require_int(data, "required", 1)
if current > required:
raise ProtocolError("节点进度超过 required")
if msg_type == "node_result":
if message.get("result") not in {"PASS", "FAIL"}:
raise ProtocolError("node_result 的 result 必须为 PASS 或 FAIL")
_require_int(message, "code", 0)
_require_int(message, "elapsed_ms", 0)
if not isinstance(message.get("data"), dict):
raise ProtocolError("node_result 的 data 必须是对象")
if msg_type == "session_result":
if message.get("result") not in {"PASS", "FAIL"}:
raise ProtocolError("session_result 的 result 必须为 PASS 或 FAIL")
_require_int(message, "fail_code", 0)
_require_int(message, "duration_ms", 0)
nodes = message.get("nodes")
if not isinstance(nodes, dict):
raise ProtocolError("session_result 缺少 nodes")
missing = [name for name in NODE_NAMES if nodes.get(name) not in {"WAIT", "PASS", "FAIL"}]
if missing:
raise ProtocolError("节点汇总无效: " + ", ".join(missing))
if msg_type == "power_off":
_require_int(message, "code", 0)
if msg_type == "device_ready":
if not isinstance(message.get("fw"), str):
raise ProtocolError("device_ready 缺少 fw")
if msg_type == "command_ack":
if message.get("command") != "START" or not isinstance(message.get("accepted"), bool):
raise ProtocolError("command_ack 字段无效")
if msg_type == "status_snapshot":
active_node = message.get("active_node")
if isinstance(active_node, bool) or not isinstance(active_node, int) or active_node < -1 or active_node >= len(NODE_NAMES):
raise ProtocolError("status_snapshot active_node 无效")
nodes = message.get("nodes")
if not isinstance(nodes, dict):
raise ProtocolError("status_snapshot 缺少 nodes")
missing = [name for name in NODE_NAMES if nodes.get(name) not in {"WAIT", "PASS", "FAIL"}]
if missing:
raise ProtocolError("status_snapshot 节点状态缺失: " + ", ".join(missing))
return message
@dataclass
class FT1Session:
session: str = ""
last_seq: int = 0
firmware: str = ""
start_time: str | None = None
imei: str = ""
iccid: str = ""
nfc_uid: str = ""
result: str = ""
fail_code: int = 0
duration_ms: int = 0
current_node: str = ""
session_start_seen: bool = False
session_result_seen: bool = False
power_off_seen: bool = False
device_ready_seen: bool = False
start_ack_seen: bool = False
nodes: dict[str, str] = field(default_factory=lambda: {name: "WAIT" for name in NODE_NAMES})
node_details: dict[str, dict] = field(default_factory=dict)
messages: list[dict] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
def reset(self, session: str = "") -> None:
fresh = type(self)(session=session)
self.__dict__.update(fresh.__dict__)
def apply(self, message: dict) -> str:
msg_type = message["type"]
incoming_session = message["session"]
if msg_type == "session_start" and incoming_session != self.session:
self.reset(incoming_session)
elif not self.session:
self.reset(incoming_session)
self.warnings.append("未收到 session_start,记录可能不完整")
elif incoming_session != self.session:
raise ProtocolError(f"会话突变: {self.session} -> {incoming_session}")
seq = message["seq"]
if seq <= self.last_seq:
raise ProtocolError(f"seq 未递增: {seq} <= {self.last_seq}")
if self.last_seq and seq != self.last_seq + 1:
self.warnings.append(f"消息序号缺口: {self.last_seq} -> {seq}")
self.last_seq = seq
self.messages.append(message)
for key in ("imei", "iccid", "nfc_uid"):
value = message.get(key)
if isinstance(value, str) and value:
setattr(self, key, value)
if msg_type == "session_start":
self.session_start_seen = True
self.firmware = str(message.get("fw", ""))
self.start_time = message.get("start_time")
return f"会话开始 {self.firmware or self.session}"
if msg_type == "device_ready":
self.device_ready_seen = True
self.firmware = str(message.get("fw", ""))
return f"设备已识别 {self.firmware}"
if msg_type == "command_ack":
self.start_ack_seen = bool(message.get("accepted"))
return "START 已确认" if self.start_ack_seen else "START 被拒绝"
if msg_type == "status_snapshot":
self.nodes.update(message["nodes"])
active_node = message["active_node"]
self.current_node = NODE_NAMES[active_node] if active_node >= 0 else ""
if active_node >= 0:
self.nodes[self.current_node] = "RUNNING"
return "状态广播 " + ", ".join(f"{name}={self.nodes[name]}" for name in NODE_NAMES)
if msg_type == "node_start":
name = message["name"]
self.current_node = name
self.nodes[name] = "RUNNING"
return f"开始 {name}"
if msg_type == "node_progress":
name = message["name"]
data = message["data"]
self.current_node = name
self.nodes[name] = f"{data['current']}/{data['required']}"
self.node_details[name] = {"data": data}
return f"{name} {data['current']}/{data['required']}"
if msg_type == "node_result":
name = message["name"]
self.nodes[name] = message["result"]
self.node_details[name] = {
"code": message["code"],
"elapsed_ms": message["elapsed_ms"],
"data": message["data"],
}
return f"{name} {message['result']} ({message['code']})"
if msg_type == "session_result":
self.session_result_seen = True
self.result = message["result"]
self.fail_code = message["fail_code"]
self.duration_ms = message["duration_ms"]
self.start_time = message.get("start_time")
self.nodes.update(message["nodes"])
return f"整机 {self.result}: {error_text(self.fail_code)}"
self.power_off_seen = True
return "设备准备关机"
@property
def complete(self) -> bool:
return self.session_start_seen and self.session_result_seen
@property
def passed_count(self) -> int:
return sum(state == "PASS" for state in self.nodes.values())
def summary(self) -> dict:
return {
"protocol_version": VERSION,
"session": self.session,
"firmware": self.firmware,
"start_time": self.start_time,
"imei": self.imei,
"iccid": self.iccid,
"nfc_uid": self.nfc_uid,
"result": self.result or "INCOMPLETE",
"fail_code": self.fail_code,
"fail_reason": error_text(self.fail_code) if self.result else "会话未完成",
"duration_ms": self.duration_ms,
"nodes": dict(self.nodes),
"node_details": dict(self.node_details),
"session_start_seen": self.session_start_seen,
"session_result_seen": self.session_result_seen,
"power_off_seen": self.power_off_seen,
"protocol_complete": self.complete,
"last_seq": self.last_seq,
"warnings": list(self.warnings),
}
+219
View File
@@ -0,0 +1,219 @@
import json
import sys
import tempfile
import unittest
from pathlib import Path
FACTORY_HOST_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(FACTORY_HOST_DIR / "src"))
import factory_host
from ft1_protocol import FT1Session, ProtocolError, parse_ft1_line
class FactoryHostTests(unittest.TestCase):
def test_expand_args(self):
args = factory_host.expand_args(
["--port", "{port}", "{firmware}", "--station", "{station}"],
aboot=Path(r"D:\burn_tool\aboot.exe"),
firmware=Path(r"D:\fw\prod.zip"),
port="COM14",
station="2",
baudrate=115200,
)
self.assertEqual(args[1], "COM14")
self.assertEqual(args[2], r"D:\fw\prod.zip")
self.assertEqual(args[-1], "2")
def test_merge_and_config_round_trip(self):
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "config.json"
config = factory_host.load_config(path)
config["flash"]["workers"] = 4
factory_host.save_config(config, path)
loaded = factory_host.load_config(path)
self.assertEqual(loaded["flash"]["workers"], 4)
self.assertIn("protocol", loaded)
def test_parse_command_template(self):
self.assertEqual(factory_host.parse_command_template('"C:\\Program Files\\aboot.exe" --port COM14'), [r"C:\Program Files\aboot.exe", "--port", "COM14"])
def test_default_flash_waits_for_arom_usb(self):
args = factory_host.DEFAULT_CONFIG["flash"]["args"]
self.assertIn("-u", args)
self.assertIn("-a", args)
self.assertNotIn("{port}", args)
def test_auto_flash_binds_each_detected_port(self):
args = factory_host.DEFAULT_CONFIG["flash"]["auto_args"]
self.assertIn("-p", args)
self.assertIn("-a", args)
self.assertIn("{port}", args)
def test_auto_flash_accepts_only_asr_arom_port(self):
arom = factory_host.SerialPortInfo(
device="COM8", hwid=r"USB\VID_2ECC&PID_3004\1234", vid=0x2ECC, pid=0x3004)
arom_from_hwid = factory_host.SerialPortInfo(
device="COM22", hwid=r"USB\VID_2ECC&PID_3004\5678")
runtime = factory_host.SerialPortInfo(
device="COM13", hwid=r"USB\VID_2ECC&PID_3017&MI_00\1234", vid=0x2ECC, pid=0x3017,
description="ASR Serial Download Device")
unrelated = factory_host.SerialPortInfo(
device="COM3", hwid=r"USB\VID_1A86&PID_7523\5678", vid=0x1A86, pid=0x7523)
misleading = factory_host.SerialPortInfo(
device="COM4", description="Generic BootROM Serial Device")
self.assertTrue(factory_host.is_arom_download_port(arom))
self.assertTrue(factory_host.is_arom_download_port(arom_from_hwid))
self.assertFalse(factory_host.is_arom_download_port(runtime))
self.assertFalse(factory_host.is_arom_download_port(unrelated))
self.assertFalse(factory_host.is_arom_download_port(misleading))
def test_auto_flash_ignores_eg800_runtime_interfaces(self):
roles = (
("COM13", 0x03, "Quectel USB AT Port"),
("COM14", 0x05, "USB Serial Device"),
("COM15", 0x04, "Quectel USB Modem"),
("COM16", 0x02, "Quectel USB DIAG Port"),
)
for device, interface, description in roles:
info = factory_host.SerialPortInfo(
device=device,
description=description,
hwid=rf"USB\VID_2C7C&PID_6002&MI_{interface:02X}\1234",
vid=0x2C7C,
pid=0x6002,
)
with self.subTest(device=device):
self.assertFalse(factory_host.is_arom_download_port(info))
def test_auto_flash_groups_interfaces_by_physical_location(self):
first = factory_host.SerialPortInfo(
device="COM8", vid=0x2ECC, pid=0x3004, location="1-4:1.0")
second = factory_host.SerialPortInfo(
device="COM9", vid=0x2ECC, pid=0x3004, location="1-4:1.1")
ports = factory_host.arom_download_ports([second, first])
self.assertEqual(len(ports), 1)
self.assertEqual(next(iter(ports.values())).device, "COM8")
def test_auto_flash_port_identity_survives_com_number_change(self):
before = factory_host.SerialPortInfo(
device="COM8", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001")
after = factory_host.SerialPortInfo(
device="COM21", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001")
self.assertEqual(
factory_host.download_device_key(before),
factory_host.download_device_key(after),
)
def test_project_root_and_relative_path(self):
self.assertTrue((factory_host.FACTORY_HOST_DIR / "vendor" / "aboot-tools-2024.08.08-win-x64").is_dir())
self.assertTrue((factory_host.FACTORY_HOST_DIR / "firmware").is_dir())
self.assertEqual(factory_host.FACTORY_HOST_DIR, FACTORY_HOST_DIR)
self.assertEqual(factory_host.CONFIG_PATH, FACTORY_HOST_DIR / "config" / "factory_host_config.json")
self.assertEqual(factory_host.LOG_DIR, FACTORY_HOST_DIR / "build" / "logs")
config = factory_host.load_config(factory_host.CONFIG_PATH)
self.assertEqual(
Path(config["firmwares"]["production"]).name,
"EG800AKCN_11LG-FT1-factory-test.zip",
)
self.assertTrue(
factory_host.resolve_project_path(config["firmwares"]["production"]).is_file()
)
self.assertTrue(
factory_host.resolve_project_path(config["firmwares"]["official"]).is_file()
)
self.assertTrue(config["firmwares"]["production"].startswith("firmware/"))
self.assertTrue(config["firmwares"]["official"].startswith("firmware/"))
self.assertTrue(config["flash"]["aboot_path"].startswith(".runtime/"))
self.assertTrue(config["flash"]["downloader_path"].startswith(".runtime/"))
def test_ft1_ignores_sdk_log_and_parses_node(self):
self.assertIsNone(parse_ft1_line("DP1312EA is OK!"))
message = parse_ft1_line(
'FT1 {"v":2,"type":"node_result","seq":3,"session":"BOOT-1",'
'"uptime_ms":2450,"imei":"860123456789012","iccid":"","nfc_uid":"",'
'"node":1,"name":"POWER_ADC","result":"PASS","code":0,"elapsed_ms":350,'
'"data":{"voltage_mv":3865}}'
)
self.assertEqual(message["data"]["voltage_mv"], 3865)
def test_ft1_session_result_and_sequence_guard(self):
session = FT1Session()
start = parse_ft1_line(
'FT1 {"v":2,"type":"session_start","seq":1,"session":"BOOT-1",'
'"uptime_ms":2050,"imei":"860123456789012","iccid":"","nfc_uid":"",'
'"fw":"FACTORY_TEST_1.1.1","start_time":null}'
)
session.apply(start)
result = parse_ft1_line(
'FT1 {"v":2,"type":"session_result","seq":2,"session":"BOOT-1",'
'"uptime_ms":40120,"result":"FAIL","fail_code":1203,"imei":"860123456789012",'
'"iccid":"","nfc_uid":"","start_time":null,"duration_ms":38070,"nodes":'
'{"BOOT_AUDIO":"PASS","POWER_ADC":"PASS","AUDIO_LOOPBACK":"FAIL",'
'"ESIM":"WAIT","DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}'
)
session.apply(result)
self.assertEqual(session.result, "FAIL")
self.assertEqual(session.nodes["AUDIO_LOOPBACK"], "FAIL")
self.assertTrue(session.complete)
with self.assertRaises(ProtocolError):
session.apply(result)
def test_ft1_rejects_node_name_mismatch(self):
with self.assertRaises(ProtocolError):
parse_ft1_line(
'FT1 {"v":2,"type":"node_start","seq":2,"session":"BOOT-1",'
'"uptime_ms":2100,"node":1,"name":"NFC"}'
)
def test_ft1_control_messages(self):
ready = parse_ft1_line(
'FT1 {"v":2,"type":"device_ready","seq":1,"session":"BOOT-1",'
'"uptime_ms":100,"fw":"FACTORY_TEST_1.1.1"}'
)
ack = parse_ft1_line(
'FT1 {"v":2,"type":"command_ack","seq":2,"session":"BOOT-1",'
'"uptime_ms":120,"command":"START","accepted":true}'
)
session = FT1Session()
self.assertEqual(session.apply(ready), "设备已识别 FACTORY_TEST_1.1.1")
self.assertEqual(session.apply(ack), "START 已确认")
self.assertTrue(session.device_ready_seen)
self.assertTrue(session.start_ack_seen)
def test_ft1_status_snapshot_keeps_passed_nodes(self):
first = parse_ft1_line(
'FT1 {"v":2,"type":"status_snapshot","seq":1,"session":"BOOT-1",'
'"uptime_ms":100,"active_node":0,"nodes":{"BOOT_AUDIO":"WAIT",'
'"POWER_ADC":"WAIT","AUDIO_LOOPBACK":"WAIT","ESIM":"WAIT",'
'"DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}'
)
second = parse_ft1_line(
'FT1 {"v":2,"type":"status_snapshot","seq":2,"session":"BOOT-1",'
'"uptime_ms":600,"active_node":1,"nodes":{"BOOT_AUDIO":"PASS",'
'"POWER_ADC":"WAIT","AUDIO_LOOPBACK":"WAIT","ESIM":"WAIT",'
'"DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}'
)
session = FT1Session()
session.apply(first)
session.apply(second)
self.assertEqual(session.nodes["BOOT_AUDIO"], "PASS")
self.assertEqual(session.current_node, "POWER_ADC")
def test_ft1_parses_low_threshold_gnss_result(self):
message = parse_ft1_line(
'FT1 {"v":2,"type":"node_result","seq":9,"session":"BOOT-1",'
'"uptime_ms":30600,"imei":"860123456789012","iccid":"8986000000000000000",'
'"nfc_uid":"04A1B2C3D4E5F6","node":6,"name":"GNSS","result":"PASS",'
'"code":0,"elapsed_ms":800,"data":{"open_ret":0,"nmea_count":5,'
'"sat_count":0,"fix_status":0,"search_ms":28500,'
'"required_nmea_count":5,"required_sat_count":0}}'
)
session = FT1Session()
session.apply(message)
self.assertEqual(session.nodes["GNSS"], "PASS")
self.assertEqual(session.node_details["GNSS"]["data"]["sat_count"], 0)
if __name__ == "__main__":
unittest.main()
@@ -11,6 +11,7 @@ vpath %.c ../custom_app ../custom_app/lib/libnfc
SRC_FILES+= \
factory_test.c \
factory_protocol.c \
factory_gnss.c \
factory_audio.c \
hal_i2c.c \
DP1312EA.c \
@@ -5,7 +5,6 @@
#include "ql_audio.h"
#include "ql_fs.h"
#include "ql_gpio.h"
#include "ql_power.h"
#include "ql_rtos.h"
#include "factory_audio.h"
@@ -16,7 +15,9 @@
#define ES8311_DEVICE_ID 0x83
#define ES8311_VOLUME_100 0x8E
#define FT_PCM_CHUNK_BYTES 640
#define FT_MP3_TIMEOUT_TICKS (15000 / 5)
#define FT_PCM_CHUNK_MS 20
#define FT_PCM_WRITE_RETRIES 40
#define FT_MP3_TIMEOUT_MS 15000
extern void sw_i2c_init(void);
extern int8 sw_i2c_start(void);
@@ -157,7 +158,7 @@ int factory_audio_play_prompt(const char *path)
return -1;
}
ret = ql_rtos_semaphore_wait(g_mp3_sem, FT_MP3_TIMEOUT_TICKS);
ret = ql_rtos_semaphore_wait(g_mp3_sem, FT_MP3_TIMEOUT_MS);
if (ret != 0 || g_mp3_result != 2) {
ql_stop_mp3_play(g_mp3_handle, 0);
factory_pa_control(0);
@@ -167,15 +168,20 @@ int factory_audio_play_prompt(const char *path)
return 0;
}
static int wait_pwrkey_level(unsigned char expected, u32 timeout_ms)
static int wait_loopback_key_level(PIN_LEVEL_E expected, u32 timeout_ms)
{
PIN_LEVEL_E level = PIN_LEVEL_HIGH;
u32 start = ft_now_ms();
while (ql_get_pwrkey_status() != expected) {
if (ft_now_ms() - start >= timeout_ms) return -1;
do {
ql_gpio_get_level(FT_LOOPBACK_KEY_PIN, &level);
if (level == expected) {
ql_rtos_task_sleep_ms(40);
ql_gpio_get_level(FT_LOOPBACK_KEY_PIN, &level);
if (level == expected) return 0;
}
ql_rtos_task_sleep_ms(10);
}
ql_rtos_task_sleep_ms(40);
return ql_get_pwrkey_status() == expected ? 0 : -1;
} while (ft_now_ms() - start < timeout_ms);
return -1;
}
int factory_audio_wait_and_loopback(u32 wait_timeout_ms, u32 *recorded_ms)
@@ -187,21 +193,29 @@ int factory_audio_wait_and_loopback(u32 wait_timeout_ms, u32 *recorded_ms)
audio_record_handle record_handle = 0;
audio_track_handle track_handle = 0;
int ret = -1;
u32 release_wait_start = 0;
if (recorded_ms) *recorded_ms = 0;
if (wait_pwrkey_level(0, 10000) != 0 ||
wait_pwrkey_level(1, wait_timeout_ms) != 0) {
ql_gpio_init(FT_LOOPBACK_KEY_PIN, PIN_DIRECTION_IN, PIN_PULL_PU, PIN_LEVEL_HIGH);
if (wait_loopback_key_level(PIN_LEVEL_HIGH, 5000) != 0 ||
wait_loopback_key_level(PIN_LEVEL_LOW, wait_timeout_ms) != 0) {
return -4;
}
buffer = (unsigned char *)malloc(FT_AUDIO_BUFFER_BYTES);
if (!buffer) return -1;
if (!buffer) {
ret = -1;
goto cleanup;
}
record_handle = ql_audio_record_open(FT_AUDIO_RATE_HZ, FT_AUDIO_CHANNELS,
0, factory_record_callback);
if (!record_handle) goto cleanup;
while (ql_get_pwrkey_status() && total < FT_AUDIO_BUFFER_BYTES) {
while (total < FT_AUDIO_BUFFER_BYTES) {
PIN_LEVEL_E level = PIN_LEVEL_HIGH;
ql_gpio_get_level(FT_LOOPBACK_KEY_PIN, &level);
if (level != PIN_LEVEL_LOW) break;
request = FT_PCM_CHUNK_BYTES;
if (request > FT_AUDIO_BUFFER_BYTES - total) {
request = FT_AUDIO_BUFFER_BYTES - total;
@@ -215,10 +229,21 @@ int factory_audio_wait_and_loopback(u32 wait_timeout_ms, u32 *recorded_ms)
ql_audio_record_close(record_handle, 0);
record_handle = 0;
if (total == FT_AUDIO_BUFFER_BYTES && ql_get_pwrkey_status()) {
if (wait_pwrkey_level(0, 30000) != 0) goto cleanup;
} else if (wait_pwrkey_level(0, 3000) != 0) {
goto cleanup;
{
PIN_LEVEL_E level = PIN_LEVEL_HIGH;
ql_gpio_get_level(FT_LOOPBACK_KEY_PIN, &level);
if (total == FT_AUDIO_BUFFER_BYTES && level == PIN_LEVEL_LOW) {
release_wait_start = ft_now_ms();
while (1) {
ql_gpio_get_level(FT_LOOPBACK_KEY_PIN, &level);
if (level != PIN_LEVEL_LOW) break;
if (ft_now_ms() - release_wait_start >= 30000) goto cleanup;
ql_rtos_task_sleep_ms(10);
}
ql_rtos_task_sleep_ms(40);
} else if (wait_loopback_key_level(PIN_LEVEL_HIGH, 3000) != 0) {
goto cleanup;
}
}
if (total < FT_PCM_CHUNK_BYTES) {
@@ -233,8 +258,9 @@ int factory_audio_wait_and_loopback(u32 wait_timeout_ms, u32 *recorded_ms)
es8311_write(0x32, ES8311_VOLUME_100);
ql_set_volume(AUDIOHAL_SPK_VOL_11);
factory_pa_control(1);
/* Match the SDK PCM path and feed the asynchronous track in real time. */
track_handle = ql_audio_track_open(FT_AUDIO_RATE_HZ, FT_AUDIO_CHANNELS,
0, factory_track_callback);
0x100, factory_track_callback);
if (!track_handle) {
ret = -3;
goto cleanup;
@@ -242,11 +268,16 @@ int factory_audio_wait_and_loopback(u32 wait_timeout_ms, u32 *recorded_ms)
for (offset = 0; offset < total; offset += FT_PCM_CHUNK_BYTES) {
unsigned int size = total - offset;
unsigned int retries = 0;
if (size > FT_PCM_CHUNK_BYTES) size = FT_PCM_CHUNK_BYTES;
if (ql_audio_track_write(track_handle, (char *)(buffer + offset), size) != 0) {
ret = -3;
goto cleanup;
while (ql_audio_track_write(track_handle, (char *)(buffer + offset), size) != 0) {
if (++retries > FT_PCM_WRITE_RETRIES) {
ret = -3;
goto cleanup;
}
ql_rtos_task_sleep_ms(5);
}
ql_rtos_task_sleep_ms(FT_PCM_CHUNK_MS);
}
ql_rtos_task_sleep_ms(total * 1000U /
(FT_AUDIO_RATE_HZ * 2U * FT_AUDIO_CHANNELS) + 300U);
@@ -0,0 +1,118 @@
#include <stdio.h>
#include <string.h>
#include "ql_gnss.h"
#include "ql_rtos.h"
#include "factory_gnss.h"
#define FT_GNSS_NMEA_GGA_GSV ((1 << 0) | (1 << 2))
typedef struct {
volatile int open_ret;
volatile u32 nmea_count;
volatile int sat_count;
volatile int fix_status;
volatile int started;
u32 start_ms;
} factory_gnss_context_t;
static factory_gnss_context_t g_factory_gnss;
static u32 factory_gnss_now_ms(void)
{
return ql_rtos_get_systicks() * 5U;
}
static int nmea_field_int(const char *line, int line_len, int wanted_field,
int *value)
{
int field = 0;
int i;
int parsed = 0;
int number = 0;
for (i = 0; i < line_len; ++i) {
char ch = line[i];
if (ch == ',') {
if (field == wanted_field) break;
++field;
continue;
}
if (field == wanted_field) {
if (ch < '0' || ch > '9') break;
number = number * 10 + (ch - '0');
parsed = 1;
}
}
if (!parsed) return 0;
*value = number;
return 1;
}
static void factory_gnss_callback(char *data, int len)
{
int i;
if (!data || len <= 0) return;
for (i = 0; i + 5 < len; ++i) {
int line_end;
int value;
if (data[i] != '$') continue;
++g_factory_gnss.nmea_count;
for (line_end = i; line_end < len; ++line_end) {
if (data[line_end] == '\r' || data[line_end] == '\n') break;
}
if (data[i + 3] == 'G' && data[i + 4] == 'S' && data[i + 5] == 'V' &&
nmea_field_int(data + i, line_end - i, 3, &value) &&
value > g_factory_gnss.sat_count) {
g_factory_gnss.sat_count = value;
}
if (data[i + 3] == 'G' && data[i + 4] == 'G' && data[i + 5] == 'A' &&
nmea_field_int(data + i, line_end - i, 6, &value) &&
value > g_factory_gnss.fix_status) {
g_factory_gnss.fix_status = value;
}
i = line_end;
}
}
void factory_gnss_start(void)
{
int cfgsys_ret;
int cfgmsg_ret;
int mode_ret;
memset(&g_factory_gnss, 0, sizeof(g_factory_gnss));
g_factory_gnss.open_ret = -1;
g_factory_gnss.start_ms = factory_gnss_now_ms();
cfgsys_ret = ql_gnss_cfgsys_set(1);
cfgmsg_ret = ql_gnss_cfgmsg_set(FT_GNSS_NMEA_GGA_GSV);
mode_ret = ql_set_gnss_nmea_mode(PACKET_MODE);
g_factory_gnss.open_ret = ql_gnss_open(factory_gnss_callback);
g_factory_gnss.started = (g_factory_gnss.open_ret == 0);
printf("[FT_GNSS] start open_ret=%d cfgsys_ret=%d cfgmsg_ret=%d mode_ret=%d\r\n",
g_factory_gnss.open_ret, cfgsys_ret, cfgmsg_ret, mode_ret);
}
void factory_gnss_snapshot(factory_gnss_result_t *result)
{
if (!result) return;
result->open_ret = g_factory_gnss.open_ret;
result->nmea_count = g_factory_gnss.nmea_count;
result->sat_count = g_factory_gnss.sat_count;
result->fix_status = g_factory_gnss.fix_status;
result->search_ms = factory_gnss_now_ms() - g_factory_gnss.start_ms;
}
void factory_gnss_stop(void)
{
int ret;
if (!g_factory_gnss.started) return;
ret = ql_gnss_close();
g_factory_gnss.started = 0;
printf("[FT_GNSS] stop ret=%d nmea=%lu sat=%d fix=%d\r\n", ret,
(unsigned long)g_factory_gnss.nmea_count,
g_factory_gnss.sat_count, g_factory_gnss.fix_status);
}
@@ -0,0 +1,18 @@
#ifndef BANBAN_FACTORY_GNSS_H
#define BANBAN_FACTORY_GNSS_H
#include "ql_type.h"
typedef struct {
int open_ret;
u32 nmea_count;
int sat_count;
int fix_status;
u32 search_ms;
} factory_gnss_result_t;
void factory_gnss_start(void);
void factory_gnss_snapshot(factory_gnss_result_t *result);
void factory_gnss_stop(void);
#endif
@@ -9,6 +9,7 @@
#include "factory_protocol.h"
#define FT_PROTOCOL_LINE_MAX 1024
#define FT_PROTOCOL_UART_PORT QL_DEBUG_UART_PORT
typedef struct {
char session[32];
@@ -20,9 +21,17 @@ typedef struct {
u32 seq;
ft_state_t nodes[FT_NODE_COUNT];
int cdc_open;
ql_mutex_t tx_mutex;
ql_task_t status_task;
volatile int status_broadcast;
int active_node;
} ft_protocol_context_t;
static ft_protocol_context_t g_protocol;
static void ft_emit(const char *fmt, ...);
u32 ft_now_ms(void);
static void ft_status_task(void *arg);
u32 ft_now_ms(void)
{
@@ -32,8 +41,8 @@ u32 ft_now_ms(void)
const char *ft_node_name(ft_node_t node)
{
static const char *names[FT_NODE_COUNT] = {
"BOOT_AUDIO", "POWER_ADC", "AUDIO_LOOPBACK", "KEY_TRIPLE",
"ESIM", "DATA_CALL", "PING", "NFC"
"BOOT_AUDIO", "POWER_ADC", "AUDIO_LOOPBACK",
"ESIM", "DATA_CALL", "NFC", "GNSS"
};
if (node < 0 || node >= FT_NODE_COUNT) {
@@ -64,10 +73,16 @@ static void ft_emit(const char *fmt, ...)
if (len < 0) return;
if (len >= (int)sizeof(line)) len = sizeof(line) - 1;
printf("%s", line);
if (g_protocol.cdc_open) {
ql_uart_write(QL_USB_CDC_PORT, (unsigned char *)line, len);
if (g_protocol.tx_mutex) ql_rtos_mutex_lock(g_protocol.tx_mutex, QL_WAIT_FOREVER);
if (FT_PROTOCOL_UART_PORT == QL_DEBUG_UART_PORT) {
/* DBG UART is the printf console on this board. Use the same
transport as the visible boot/application logs so FT1 frames are
actually received by the host terminal. */
printf("%s", line);
} else if (g_protocol.cdc_open) {
ql_uart_write(FT_PROTOCOL_UART_PORT, (unsigned char *)line, len);
}
if (g_protocol.tx_mutex) ql_rtos_mutex_unlock(g_protocol.tx_mutex);
}
static void ft_make_session(void)
@@ -94,10 +109,50 @@ static void ft_make_session(void)
void ft_protocol_init(void)
{
int uart_ret;
memset(&g_protocol, 0, sizeof(g_protocol));
g_protocol.cdc_open =
(ql_uart_open(QL_USB_CDC_PORT, QL_UART_BAUD_115200, QL_FC_NONE) == 0);
g_protocol.active_node = -1;
ql_rtos_mutex_create(&g_protocol.tx_mutex);
uart_ret = ql_uart_open(FT_PROTOCOL_UART_PORT, QL_UART_BAUD_115200, QL_FC_NONE);
g_protocol.cdc_open = (uart_ret == 0 || uart_ret == QL_UART_ALREAD_OPEN);
ft_make_session();
printf("[FT1] control UART port=%d open_ret=%d ready=%d\r\n",
FT_PROTOCOL_UART_PORT, uart_ret, g_protocol.cdc_open);
}
static void ft_status_task(void *arg)
{
(void)arg;
while (g_protocol.status_broadcast) {
int i;
char payload[FT_PROTOCOL_LINE_MAX - 16];
int offset = snprintf(payload, sizeof(payload),
"{\"v\":%d,\"type\":\"status_snapshot\",\"seq\":%lu,\"session\":\"%s\",\"uptime_ms\":%lu,\"active_node\":%d,\"nodes\":{",
FT_PROTOCOL_VERSION, (unsigned long)++g_protocol.seq,
g_protocol.session, (unsigned long)ft_now_ms(),
g_protocol.active_node);
for (i = 0; i < FT_NODE_COUNT && offset > 0 && offset < (int)sizeof(payload); ++i) {
offset += snprintf(payload + offset, sizeof(payload) - offset,
"%s\"%s\":\"%s\"",
i ? "," : "", ft_node_name((ft_node_t)i),
ft_state_name(g_protocol.nodes[i]));
}
if (offset > 0 && offset < (int)sizeof(payload)) {
snprintf(payload + offset, sizeof(payload) - offset, "}}");
ft_emit("%s", payload);
}
ql_rtos_task_sleep_ms(500);
}
}
void ft_protocol_start_status_broadcast(void)
{
if (g_protocol.status_broadcast) return;
g_protocol.status_broadcast = 1;
/* The snapshot formatter keeps a near-1 KiB JSON buffer on the task
stack; leave headroom for libc formatting and UART calls. */
ql_rtos_task_create(&g_protocol.status_task, 8192, 15,
"ft_status", ft_status_task, NULL);
}
static void ft_copy(char *dst, unsigned int dst_size, const char *src)
@@ -131,16 +186,17 @@ void ft_protocol_session_start(void)
ft_emit("{\"v\":%d,\"type\":\"session_start\",\"seq\":%lu,"
"\"session\":\"%s\",\"uptime_ms\":%lu,\"imei\":\"%s\","
"\"iccid\":\"%s\",\"nfc_uid\":\"%s\",\"fw\":\"%s\","
"\"ping_host\":\"%s\",\"start_time\":%s%s%s}",
"\"start_time\":%s%s%s}",
FT_PROTOCOL_VERSION, (unsigned long)++g_protocol.seq,
g_protocol.session, (unsigned long)ft_now_ms(), g_protocol.imei,
g_protocol.iccid, g_protocol.nfc_uid, FT_FIRMWARE_VERSION,
FT_PING_HOST, g_protocol.start_time[0] ? "\"" : "null",
g_protocol.start_time[0] ? "\"" : "null",
g_protocol.start_time, g_protocol.start_time[0] ? "\"" : "");
}
void ft_protocol_node_start(ft_node_t node)
{
if (node >= 0 && node < FT_NODE_COUNT) g_protocol.active_node = node;
ft_emit("{\"v\":%d,\"type\":\"node_start\",\"seq\":%lu,"
"\"session\":\"%s\",\"uptime_ms\":%lu,\"imei\":\"%s\","
"\"iccid\":\"%s\",\"nfc_uid\":\"%s\",\"node\":%d,"
@@ -169,6 +225,7 @@ void ft_protocol_node_result(ft_node_t node, int passed, ft_error_t code,
const char *result = passed ? "PASS" : "FAIL";
if (node >= 0 && node < FT_NODE_COUNT) {
g_protocol.nodes[node] = passed ? FT_STATE_PASS : FT_STATE_FAIL;
g_protocol.active_node = -1;
}
if (!data_json || !data_json[0]) data_json = "{}";
@@ -191,9 +248,8 @@ void ft_protocol_session_result(int passed, ft_error_t code)
"\"nfc_uid\":\"%s\",\"start_time\":%s%s%s,"
"\"duration_ms\":%lu,\"nodes\":{"
"\"BOOT_AUDIO\":\"%s\",\"POWER_ADC\":\"%s\","
"\"AUDIO_LOOPBACK\":\"%s\",\"KEY_TRIPLE\":\"%s\","
"\"ESIM\":\"%s\",\"DATA_CALL\":\"%s\","
"\"PING\":\"%s\",\"NFC\":\"%s\"}}",
"\"AUDIO_LOOPBACK\":\"%s\",\"ESIM\":\"%s\","
"\"DATA_CALL\":\"%s\",\"NFC\":\"%s\",\"GNSS\":\"%s\"}}",
FT_PROTOCOL_VERSION, (unsigned long)++g_protocol.seq,
g_protocol.session, (unsigned long)ft_now_ms(), passed ? "PASS" : "FAIL",
code, g_protocol.imei, g_protocol.iccid, g_protocol.nfc_uid,
@@ -203,7 +259,7 @@ void ft_protocol_session_result(int passed, ft_error_t code)
ft_state_name(g_protocol.nodes[0]), ft_state_name(g_protocol.nodes[1]),
ft_state_name(g_protocol.nodes[2]), ft_state_name(g_protocol.nodes[3]),
ft_state_name(g_protocol.nodes[4]), ft_state_name(g_protocol.nodes[5]),
ft_state_name(g_protocol.nodes[6]), ft_state_name(g_protocol.nodes[7]));
ft_state_name(g_protocol.nodes[6]));
}
void ft_protocol_power_off(void)
@@ -10,6 +10,7 @@ typedef enum {
} ft_state_t;
void ft_protocol_init(void);
void ft_protocol_start_status_broadcast(void);
void ft_protocol_set_imei(const char *imei);
void ft_protocol_set_iccid(const char *iccid);
void ft_protocol_set_nfc_uid(const char *uid);
@@ -8,23 +8,20 @@
#include "ql_gpio.h"
#include "ql_iic.h"
#include "ql_nw.h"
#include "ql_ping.h"
#include "ql_power.h"
#include "ql_pwm.h"
#include "ql_rtos.h"
#include "ql_sim.h"
#include "sockets.h"
#include "DP1312EA.h"
#include "factory_audio.h"
#include "factory_gnss.h"
#include "factory_protocol.h"
#include "factory_test.h"
#include "nfc.h"
#include "reg.h"
#define FT_PING_COUNT 3
#define FT_PING_REQUIRED 2
#define FT_PING_TIMEOUT_MS 4000
#define FT_PING_WAIT_TICKS (20000 / 5)
#define FT_NFC_VERSION_EXPECTED 0x82
int g_testmode = 1;
@@ -35,22 +32,115 @@ extern s8 Read_NTAG213(u8 *uid_buf, u8 *data_buf, u16 *data_len);
void handle_nfc_bind(const char *uid) { (void)uid; }
void handle_nfc_trigger(const char *uid) { (void)uid; }
static ql_sem_t g_ping_sem;
static ql_ping_stat_t g_ping_stats;
static volatile int g_ping_finished;
static char g_ip_address[64];
static volatile int g_led_mode;
static ql_task_t g_led_task;
enum {
FT_LED_OFF = 0,
FT_LED_ON,
FT_LED_LOOPBACK,
FT_LED_NFC,
FT_LED_GNSS
};
static int factory_led_pwm_set(unsigned short duty, int *enabled)
{
QL_PWM_CONFIG_EX config;
config.pwm_psc = 1;
config.high_one_cycle_duration = duty;
config.total_one_cycle_duration = 320;
config.clk_src = QL_PWM_CLK_SRC_32K;
if (ql_pwm_config_plus_ex(FT_STATUS_LED_PIN, &config) != 0) return -1;
if (!*enabled) {
if (ql_pwm_enable(FT_STATUS_LED_PIN) != 0) {
ql_pwm_disable(FT_STATUS_LED_PIN);
return -1;
}
*enabled = 1;
}
return 0;
}
static void factory_led_pwm_stop(int *enabled)
{
if (*enabled) ql_pwm_disable(FT_STATUS_LED_PIN);
*enabled = 0;
}
static void factory_led_task(void *arg)
{
int phase = 0;
int last_mode = -1;
int pwm_enabled = 0;
int pwm_failed = 0;
int breath_duty = 16;
int breath_step = 16;
(void)arg;
ql_gpio_init(FT_STATUS_LED_PIN, PIN_DIRECTION_OUT, PIN_PULL_DISABLE,
PIN_LEVEL_HIGH);
while (1) {
int mode = g_led_mode;
if (mode != last_mode) {
phase = 0;
breath_duty = 16;
breath_step = 16;
pwm_failed = 0;
last_mode = mode;
}
if (mode == FT_LED_GNSS && !pwm_failed) {
if (factory_led_pwm_set((unsigned short)breath_duty,
&pwm_enabled) == 0) {
breath_duty += breath_step;
if (breath_duty >= 304 || breath_duty <= 16) {
breath_step = -breath_step;
}
ql_rtos_task_sleep_ms(40);
continue;
}
pwm_failed = 1;
factory_led_pwm_stop(&pwm_enabled);
printf("[FT_LED] GPIO70 PWM unavailable, use GNSS double-flash\r\n");
}
if (pwm_enabled) {
factory_led_pwm_stop(&pwm_enabled);
ql_gpio_init(FT_STATUS_LED_PIN, PIN_DIRECTION_OUT,
PIN_PULL_DISABLE, PIN_LEVEL_HIGH);
}
if (mode == FT_LED_OFF) {
ql_gpio_set_level(FT_STATUS_LED_PIN, PIN_LEVEL_LOW);
ql_rtos_task_sleep_ms(100);
} else if (mode == FT_LED_ON) {
ql_gpio_set_level(FT_STATUS_LED_PIN, PIN_LEVEL_HIGH);
ql_rtos_task_sleep_ms(100);
} else if (mode == FT_LED_GNSS) {
static const int levels[] = {1, 0, 1, 0};
static const int delays[] = {80, 80, 80, 520};
ql_gpio_set_level(FT_STATUS_LED_PIN,
levels[phase] ? PIN_LEVEL_HIGH : PIN_LEVEL_LOW);
ql_rtos_task_sleep_ms(delays[phase]);
phase = (phase + 1) % 4;
} else {
phase = !phase;
ql_gpio_set_level(FT_STATUS_LED_PIN,
phase ? PIN_LEVEL_HIGH : PIN_LEVEL_LOW);
ql_rtos_task_sleep_ms(mode == FT_LED_LOOPBACK ? 250 : 100);
}
}
}
static void factory_led_mode(int mode)
{
g_led_mode = mode;
}
static int iccid_valid(const char *iccid)
{
unsigned int i;
unsigned int len;
if (!iccid) return 0;
len = strlen(iccid);
if (len < 18 || len > 22) return 0;
for (i = 0; i < len; ++i) {
if (iccid[i] < '0' || iccid[i] > '9') return 0;
}
return 1;
return len >= 18 && len <= 22;
}
static ft_error_t run_boot_audio(char *data, unsigned int data_size)
@@ -100,7 +190,16 @@ static ft_error_t run_power_adc(char *data, unsigned int data_size)
static ft_error_t run_audio_loopback(char *data, unsigned int data_size)
{
u32 recorded_ms = 0;
int ret = factory_audio_wait_and_loopback(FT_DIALOG_KEY_TIMEOUT_MS, &recorded_ms);
int ret = factory_audio_play_prompt(FT_AUDIO_LOOPBACK_PATH);
if (ret == -2) {
return FT_ERR_AUDIO_FILE_MISSING;
}
if (ret != 0) {
return FT_ERR_AUDIO_PLAY;
}
factory_led_mode(FT_LED_LOOPBACK);
ret = factory_audio_wait_and_loopback(FT_DIALOG_KEY_TIMEOUT_MS, &recorded_ms);
factory_led_mode(FT_LED_ON);
snprintf(data, data_size,
"{\"recorded_ms\":%lu,\"max_ms\":%d,\"volume_percent\":100}",
(unsigned long)recorded_ms, FT_AUDIO_MAX_MS);
@@ -111,43 +210,6 @@ static ft_error_t run_audio_loopback(char *data, unsigned int data_size)
return FT_OK;
}
static int wait_gpio_level(unsigned int pin, PIN_LEVEL_E expected, u32 timeout_ms)
{
PIN_LEVEL_E level = PIN_LEVEL_HIGH;
u32 start = ft_now_ms();
do {
ql_gpio_get_level(pin, &level);
if (level == expected) {
ql_rtos_task_sleep_ms(40);
ql_gpio_get_level(pin, &level);
if (level == expected) return 0;
}
ql_rtos_task_sleep_ms(10);
} while (ft_now_ms() - start < timeout_ms);
return -1;
}
static ft_error_t run_key_triple(char *data, unsigned int data_size)
{
int count = 0;
u32 start = ft_now_ms();
ql_gpio_init(FT_TRIPLE_KEY_PIN, PIN_DIRECTION_IN, PIN_PULL_PU, PIN_LEVEL_HIGH);
if (wait_gpio_level(FT_TRIPLE_KEY_PIN, PIN_LEVEL_HIGH, 5000) != 0) {
return FT_ERR_TRIPLE_KEY_TIMEOUT;
}
while (count < 3 && ft_now_ms() - start < FT_TRIPLE_KEY_TIMEOUT_MS) {
u32 remaining = FT_TRIPLE_KEY_TIMEOUT_MS - (ft_now_ms() - start);
if (wait_gpio_level(FT_TRIPLE_KEY_PIN, PIN_LEVEL_LOW, remaining) != 0) break;
++count;
ft_protocol_node_progress(FT_NODE_KEY_TRIPLE, count, 3);
remaining = FT_TRIPLE_KEY_TIMEOUT_MS - (ft_now_ms() - start);
if (wait_gpio_level(FT_TRIPLE_KEY_PIN, PIN_LEVEL_HIGH, remaining) != 0) break;
}
snprintf(data, data_size, "{\"press_count\":%d,\"required\":3}", count);
return count == 3 ? FT_OK : FT_ERR_TRIPLE_KEY_TIMEOUT;
}
static ft_error_t run_esim(char *data, unsigned int data_size)
{
int status = QL_SIM_STATUS_UNKNOWN;
@@ -207,66 +269,28 @@ static ft_error_t run_data_call(char *data, unsigned int data_size)
return FT_OK;
}
static void factory_ping_callback(ql_ping_state_e state, int err,
ql_ping_stat_t *stats)
{
(void)err;
if (stats) memcpy(&g_ping_stats, stats, sizeof(g_ping_stats));
if (state == QL_PING_FINISHED) {
g_ping_finished = 1;
if (g_ping_sem) ql_rtos_semaphore_release(g_ping_sem);
}
}
static ft_error_t run_ping(char *data, unsigned int data_size)
{
ql_ping_ctx_t *ctx;
ql_ping_cfg_t cfg = {
.pdpCid = 1,
.host = FT_PING_HOST,
.timeout = FT_PING_TIMEOUT_MS,
.dataSize = 32,
.pingNum = FT_PING_COUNT
};
int ret;
memset(&g_ping_stats, 0, sizeof(g_ping_stats));
g_ping_finished = 0;
if (ql_rtos_semaphore_create(&g_ping_sem, 0) != 0) return FT_ERR_PING_INIT;
ctx = ql_ping_init(cfg, factory_ping_callback);
if (!ctx) {
ql_rtos_semaphore_delete(g_ping_sem);
g_ping_sem = NULL;
return FT_ERR_PING_INIT;
}
ret = ql_ping(ctx);
if (ret == 0) ret = ql_rtos_semaphore_wait(g_ping_sem, FT_PING_WAIT_TICKS);
ql_ping_deinit(ctx);
ql_rtos_semaphore_delete(g_ping_sem);
g_ping_sem = NULL;
snprintf(data, data_size,
"{\"host\":\"%s\",\"sent\":%u,\"received\":%u,"
"\"lost\":%u,\"avg_rtt_ms\":%lu}",
FT_PING_HOST, g_ping_stats.sentNum, g_ping_stats.rcvdNum,
g_ping_stats.lostNum, (unsigned long)g_ping_stats.avgRtt);
if (ret != 0 || !g_ping_finished) return FT_ERR_PING_START;
if (g_ping_stats.rcvdNum < FT_PING_REQUIRED) return FT_ERR_PING_LOSS;
return FT_OK;
}
static ft_error_t run_nfc(char *data, unsigned int data_size)
{
u8 uid[7] = {0};
u8 ntag_data[180] = {0};
u16 ntag_len = 0;
u8 version;
int audio_ret;
u32 start = ft_now_ms();
audio_ret = factory_audio_play_prompt(FT_AUDIO_NFC_PATH);
if (audio_ret == -2) {
return FT_ERR_AUDIO_FILE_MISSING;
}
if (audio_ret != 0) {
return FT_ERR_AUDIO_PLAY;
}
factory_led_mode(FT_LED_NFC);
DP1312EA_Reset();
version = DP1312EA_ReadRawRC(VersionReg);
if (version != FT_NFC_VERSION_EXPECTED) {
snprintf(data, data_size, "{\"chip_version\":\"0x%02X\"}", version);
factory_led_mode(FT_LED_ON);
return FT_ERR_NFC_CHIP;
}
@@ -277,14 +301,55 @@ static ft_error_t run_nfc(char *data, unsigned int data_size)
snprintf(data, data_size,
"{\"chip_version\":\"0x%02X\",\"uid\":\"%s\"}",
version, (char *)output_uuid);
factory_led_mode(FT_LED_ON);
return FT_OK;
}
ql_rtos_task_sleep_ms(200);
} while (ft_now_ms() - start < FT_NFC_TIMEOUT_MS);
factory_led_mode(FT_LED_ON);
return FT_ERR_NFC_TIMEOUT;
}
static ft_error_t run_gnss(char *data, unsigned int data_size)
{
factory_gnss_result_t result;
int audio_ret;
factory_led_mode(FT_LED_GNSS);
audio_ret = factory_audio_play_prompt(FT_AUDIO_GNSS_PATH);
if (audio_ret != 0) {
factory_gnss_stop();
factory_led_mode(FT_LED_ON);
return audio_ret == -2 ? FT_ERR_AUDIO_FILE_MISSING : FT_ERR_AUDIO_PLAY;
}
while (1) {
factory_gnss_snapshot(&result);
if (result.open_ret == 0 &&
result.nmea_count >= FT_GNSS_MIN_NMEA_COUNT) {
snprintf(data, data_size,
"{\"open_ret\":%d,\"nmea_count\":%lu,\"sat_count\":%d,"
"\"fix_status\":%d,\"search_ms\":%lu,"
"\"required_nmea_count\":%d,\"required_sat_count\":0}",
result.open_ret, (unsigned long)result.nmea_count,
result.sat_count, result.fix_status,
(unsigned long)result.search_ms,
FT_GNSS_MIN_NMEA_COUNT);
factory_gnss_stop();
factory_led_mode(FT_LED_ON);
return FT_OK;
}
if (result.open_ret != 0) {
ql_rtos_task_sleep_s(2);
factory_gnss_start();
} else {
ql_rtos_task_sleep_ms(500);
}
}
}
static ft_error_t run_node(ft_node_t node, char *data, unsigned int data_size)
{
data[0] = '\0';
@@ -292,11 +357,10 @@ static ft_error_t run_node(ft_node_t node, char *data, unsigned int data_size)
case FT_NODE_BOOT_AUDIO: return run_boot_audio(data, data_size);
case FT_NODE_POWER_ADC: return run_power_adc(data, data_size);
case FT_NODE_AUDIO_LOOPBACK: return run_audio_loopback(data, data_size);
case FT_NODE_KEY_TRIPLE: return run_key_triple(data, data_size);
case FT_NODE_ESIM: return run_esim(data, data_size);
case FT_NODE_DATA_CALL: return run_data_call(data, data_size);
case FT_NODE_PING: return run_ping(data, data_size);
case FT_NODE_NFC: return run_nfc(data, data_size);
case FT_NODE_GNSS: return run_gnss(data, data_size);
default: return FT_ERR_NFC_LOCKED;
}
}
@@ -311,12 +375,16 @@ static void factory_test_task(void *arg)
int audio_ret;
(void)arg;
factory_gnss_start();
ql_rtos_task_sleep_s(2);
ft_protocol_init();
ql_rtos_task_create(&g_led_task, 2048, 15, "ft_led", factory_led_task, NULL);
factory_led_mode(FT_LED_ON);
if (ql_dev_get_imei(imei, sizeof(imei)) == 0) {
ft_protocol_set_imei(imei);
}
ft_protocol_session_start();
ft_protocol_start_status_broadcast();
for (node = FT_NODE_BOOT_AUDIO; node < FT_NODE_COUNT; ++node) {
ft_protocol_node_start(node);
@@ -330,11 +398,7 @@ static void factory_test_task(void *arg)
}
}
audio_ret = factory_audio_play_prompt(FT_AUDIO_PASS_PATH);
if (audio_ret == 0) {
ql_rtos_task_sleep_ms(300);
audio_ret = factory_audio_play_prompt(FT_AUDIO_SHUTDOWN_PATH);
}
audio_ret = factory_audio_play_prompt(FT_AUDIO_SHUTDOWN_PATH);
if (audio_ret != 0) {
ft_protocol_session_result(0, FT_ERR_FINAL_AUDIO);
while (1) ql_rtos_task_sleep_s(60);
@@ -4,9 +4,8 @@
#include "ql_type.h"
#include "ql_gpio.h"
#define FT_PROTOCOL_VERSION 1
#define FT_FIRMWARE_VERSION "FACTORY_TEST_1.0.0"
#define FT_PING_HOST "banban.api.talkingq.com"
#define FT_PROTOCOL_VERSION 2
#define FT_FIRMWARE_VERSION "FACTORY_TEST_1.1.1"
#define FT_ADC_MIN_MV 3500
#define FT_ADC_MAX_MV 4200
@@ -18,26 +17,28 @@
#define FT_AUDIO_BUFFER_BYTES (FT_AUDIO_RATE_HZ * 2 * FT_AUDIO_MAX_MS / 1000)
#define FT_DIALOG_KEY_TIMEOUT_MS 60000
#define FT_TRIPLE_KEY_TIMEOUT_MS 60000
#define FT_SIM_TIMEOUT_MS 30000
#define FT_NFC_TIMEOUT_MS 60000
#define FT_GNSS_MIN_NMEA_COUNT 5
#define FT_TRIPLE_KEY_PIN GPIO_PIN_NO_84
#define FT_LOOPBACK_KEY_PIN GPIO_PIN_NO_84
#define FT_STATUS_LED_PIN GPIO_PIN_NO_70
#define FT_PA_PIN GPIO_PIN_NO_68
#define FT_AUDIO_BOOT_PATH "U:/factory_boot_zh.mp3"
#define FT_AUDIO_PASS_PATH "U:/factory_pass_zh.mp3"
#define FT_AUDIO_LOOPBACK_PATH "U:/factory_audio_loopback_zh.mp3"
#define FT_AUDIO_NFC_PATH "U:/factory_nfc_test_zh.mp3"
#define FT_AUDIO_GNSS_PATH "U:/factory_gnss_test_zh.mp3"
#define FT_AUDIO_SHUTDOWN_PATH "U:/factory_shutdown_zh.mp3"
typedef enum {
FT_NODE_BOOT_AUDIO = 0,
FT_NODE_POWER_ADC,
FT_NODE_AUDIO_LOOPBACK,
FT_NODE_KEY_TRIPLE,
FT_NODE_ESIM,
FT_NODE_DATA_CALL,
FT_NODE_PING,
FT_NODE_NFC,
FT_NODE_GNSS,
FT_NODE_COUNT
} ft_node_t;
@@ -53,16 +54,15 @@ typedef enum {
FT_ERR_RECORD_EMPTY = 1202,
FT_ERR_LOOPBACK_PLAY = 1203,
FT_ERR_DIALOG_KEY_TIMEOUT = 1204,
FT_ERR_TRIPLE_KEY_TIMEOUT = 1301,
FT_ERR_SIM_NOT_READY = 1401,
FT_ERR_ICCID_READ = 1402,
FT_ERR_ICCID_FORMAT = 1403,
FT_ERR_NETWORK_REGISTER = 1501,
FT_ERR_PDP_ACTIVATE = 1502,
FT_ERR_IP_INVALID = 1503,
FT_ERR_PING_INIT = 1601,
FT_ERR_PING_START = 1602,
FT_ERR_PING_LOSS = 1603,
FT_ERR_GNSS_OPEN = 1601,
FT_ERR_GNSS_NO_NMEA = 1602,
FT_ERR_GNSS_NO_SATELLITE = 1603,
FT_ERR_NFC_LOCKED = 1701,
FT_ERR_NFC_CHIP = 1702,
FT_ERR_NFC_TIMEOUT = 1703,