diff --git a/docs/factory_test.md b/docs/factory_test.md index 64d6ffb..6ccae49 100644 --- a/docs/factory_test.md +++ b/docs/factory_test.md @@ -25,13 +25,10 @@ -> ES8311 ID/初始化，设置所有播放音量为 100% -> 播放开机提示音 -> 电源 ADC - -> 按住对话键录音，松开后回放，最多录 5 秒 + -> 按住 GPIO84（原三击测试按键）录音，松开后回放，最多录 5 秒 -> 工人判断音质 - 正常：按另一个按键三次 - 异常：不再操作，60 秒后 KEY_TRIPLE 失败 -> eSIM READY 和 ICCID -> 4G 注册、PDP 激活、获取 IPv4 - -> Ping 业务域名 -> 前面全部通过后启用 NFC 测试 -> 读取 NTAG213 UID -> 播放“全部测试通过” @@ -49,16 +46,13 @@ |---:|---|---|---|---:| | 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`，ICCID 为 18-22 位十进制数字 | 30 秒 | +| 4 | `DATA_CALL` | 等待注册、激活 CID 1 IPv4 PDP | 返回非 `0.0.0.0` 的 IPv4 | 注册等待 120 秒 | +| 5 | `NFC` | 检查 DP1312EA 版本并轮询 NTAG213 | 版本 `0x82`，成功读取 UID | 60 秒 | -说明：`AUDIO_LOOPBACK` 的软件通过仅代表数据链路完成。最终声学判断由工人完成； -如果声音异常，工人不按三击键，随后 `KEY_TRIPLE` 超时，整机产测失败。这避免了 -上位机额外的人工确认按钮。 +说明：`AUDIO_LOOPBACK` 使用 GPIO84 按键，软件通过仅代表录放数据链路完成。 +最终声学判断由工人完成。 ## 4. 音频和文件系统接口 @@ -104,7 +98,6 @@ 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"} ``` RTC 无效时 `start_time` 为 `null`。RTC 无效不会导致产测失败；测试耗时始终以 @@ -122,36 +115,25 @@ FT1 {"v":1,"type":"node_start","seq":2,"session":"20260923-163010","uptime_ms":2 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}} ``` -### 5.5 三击按键进度 - -每次完整按下并释放都输出一条 `node_progress`： - -```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}} -``` - -第三次松开后再输出正式的 `node_result=PASS`。 - -### 5.6 网络和 Ping +### 5.5 NFC 和最终报告 ```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 和最终报告 +### 5.6 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":1,"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"}} ``` 最终报告包含 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":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","ESIM":"PASS","DATA_CALL":"PASS","NFC":"PASS"}} ``` 上位机收到 `power_off` 后应立即落盘本次记录： @@ -174,17 +156,13 @@ FT1 {"v":1,"type":"power_off","seq":20,"session":"20260923-163010","uptime_ms":4 | 1201 | 录音句柄或 160KB 缓冲区创建失败 | | 1202 | 没有录到一帧有效 PCM | | 1203 | PCM 回放失败 | -| 1204 | 60 秒内未按下对话键 | -| 1301 | 音质异常未继续，或 60 秒内未完成三击 | +| 1204 | 60 秒内未按下 Loopback 按键 | | 1401 | eSIM 未进入 READY | | 1402 | ICCID 读取失败 | | 1403 | ICCID 长度或字符格式错误 | | 1501 | 4G 网络注册超时 | | 1502 | IPv4 PDP 激活失败 | | 1503 | 未获取有效 IPv4 地址 | -| 1601 | Ping 上下文或同步对象创建失败 | -| 1602 | Ping 启动或等待回调失败 | -| 1603 | 3 包中收到少于 2 包 | | 1701 | 前置节点未全部通过，NFC 锁定 | | 1702 | DP1312EA 版本寄存器不是 `0x82` | | 1703 | 60 秒内没有成功刷卡 | @@ -200,8 +178,7 @@ FT1 {"v":1,"type":"power_off","seq":20,"session":"20260923-163010","uptime_ms":4 - `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. 上板前检查 diff --git a/factory_host/.gitignore b/factory_host/.gitignore new file mode 100644 index 0000000..f1520a4 --- /dev/null +++ b/factory_host/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +logs/ +dist/ +.pyinstaller-work/ +.pyinstaller-spec/ diff --git a/factory_host/README.md b/factory_host/README.md new file mode 100644 index 0000000..e9664bc --- /dev/null +++ b/factory_host/README.md @@ -0,0 +1,86 @@ +# EG800 FT1 工厂烧录与产测上位机 + +本工具配套 `test/factory-test` 分支的独立产测固件，支持批量烧录、USB CDC +协议采集、6 个硬件节点判定和结果落盘。 + +## 主要功能 + +- 产测固件和正式固件分开保存，每个工位可独立选择。 +- 烧录串口和产测 CDC 串口分开配置，避免把 DIAG/AT/CDC 当成同一个端口。 +- 使用原厂 `adownload.exe -u -a`，先进入等待状态，再由关机设备上电触发烧录。 +- 提供“烧录选中”单工位入口；USB 自动等待的批量任务按工位顺序执行，避免抢设备。 +- 只解析以 `FT1 ` 开头的协议行；普通 SDK 日志只显示，不进入产测数据库。 +- 严格校验协议版本、消息类型、节点编号/名称、会话 ID 和递增序号。 +- 显示 `BOOT_AUDIO`、`POWER_ADC`、`AUDIO_LOOPBACK`、`ESIM`、 + `DATA_CALL`、`NFC` 的实时状态与结构化数据。 +- 收到 `session_result` 或 `power_off` 时自动保存完整 JSON 和汇总 CSV。 +- 保存 IMEI、ICCID、NFC UID、失败码、固件版本、节点耗时和总耗时。 + +## 目录和启动 + +完整烧录工具位于： + +```text +burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe +``` + +发布程序： + +```text +factory_host/dist/EG800FactoryHost.exe +``` + +默认“产测固件”指向已包含三个产测 MP3 的 FT1 整包；默认“正式固件”指向 +`version` 目录中的正式内层 `-fac.zip`。两者都可在外置 +`factory_host_config.json` 中替换，烧录时不要选择 `-target.zip` 外层归档。 + +双击表格中的工位可以分别设置烧录口和产测 CDC 口。默认 USB 自动等待模式不需要 + +填写烧录口；产测 CDC 口仍需填写。“烧录选中”要求只选择一个工位；“批量烧录” +在未选中行时作用于全部工位，选中后只作用于选中工位。 + +单工位烧录时保持设备关机并连接 USB，选择工位后点击“烧录选中”。工位状态显示 +“等待上电”后再给设备开机。此时 `adownload` 会捕获 AROM USB 下载口并开始烧录， +操作顺序与 Aboot 的 `usbOnly + usbAutoEnable` 默认设置一致。 + +串口和协议记录位于： + +```text +factory_host/dist/logs/factory_results.csv +factory_host/dist/logs/sessions/*.json +``` + +## FT1 结果规则 + +产测结果完全采用固件的 `session_result`，上位机不会用模糊正则猜测硬件是否通过。 +非零失败码会显示 `docs/factory_test.md` 中对应的中文含义。收到 `power_off` 后会再次 +刷新同一 JSON 文件，但不会在 CSV 中重复增加记录。 + +`AUDIO_LOOPBACK=PASS` 表示 GPIO84（原三击测试按键）的录放数据链路成功。 +Loopback 仍然是按住录音、松开回放，最长录音 5 秒。 + +## 产测音频包 + +重新生成音频包： + +```powershell +python -B .\factory_host\build_factory_audio_package.py +``` + +脚本会执行以下检查： + +1. 源目录必须且只能包含三个固定文件名。 +2. 每个文件必须为 16 kHz、单声道、48 kbps MP3。 +3. 使用 SDK `mklfs-v2.exe -m` 生成 262144 字节 `customer_fs.bin`。 +4. 立即反向解包镜像并逐文件比对 SHA-256。 +5. 输出 ZIP、清单及校验值到 `factory_host/dist/factory_audio/`。 + +不要直接覆盖正式业务 `customer_fs.bin`。需要构建产测整包时，按照 +`docs/factory_test_windows_fs.md` 先备份业务镜像，再临时放入产测镜像。 + +## 开发验证 + +```powershell +python -B -m unittest discover -s factory_host -p "test_*.py" -v +powershell.exe -NoProfile -ExecutionPolicy Bypass -File factory_host\build_release.ps1 +``` diff --git a/factory_host/build_factory_audio_package.py b/factory_host/build_factory_audio_package.py new file mode 100644 index 0000000..5ef8d3d --- /dev/null +++ b/factory_host/build_factory_audio_package.py @@ -0,0 +1,200 @@ +"""Validate and package the three 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_pass_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().parent.parent) + parser.add_argument("--output", type=Path, default=Path(__file__).resolve().parent / "dist" / "factory_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() diff --git a/factory_host/build_factory_firmware.ps1 b/factory_host/build_factory_firmware.ps1 new file mode 100644 index 0000000..6189c7e --- /dev/null +++ b/factory_host/build_factory_firmware.ps1 @@ -0,0 +1,60 @@ +param( + [string]$Output = "dist\factory_firmware" +) + +$ErrorActionPreference = "Stop" +$toolRoot = Split-Path -Parent $MyInvocation.MyCommand.Path +$projectRoot = Split-Path -Parent $toolRoot +$audioImage = Join-Path $toolRoot "dist\factory_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 = Join-Path $toolRoot $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" +} diff --git a/factory_host/build_release.ps1 b/factory_host/build_release.ps1 new file mode 100644 index 0000000..267eef2 --- /dev/null +++ b/factory_host/build_release.ps1 @@ -0,0 +1,21 @@ +param( + [string]$Output = "dist" +) + +$ErrorActionPreference = "Stop" +$root = Split-Path -Parent $MyInvocation.MyCommand.Path +python -c "import PyInstaller" 2>$null +if ($LASTEXITCODE -ne 0) { + throw "未找到 PyInstaller。先执行: python -m pip install pyinstaller pyserial" +} + +$outDir = Join-Path $root $Output +New-Item -ItemType Directory -Force -Path $outDir | Out-Null +python -m PyInstaller --noconfirm --clean --onefile --windowed ` + --name EG800FactoryHost ` + --distpath $outDir ` + --workpath (Join-Path $root ".pyinstaller-work") ` + --specpath (Join-Path $root ".pyinstaller-spec") ` + (Join-Path $root "factory_host.py") +Copy-Item (Join-Path $root "factory_host_config.json") $outDir -Force +Write-Host "输出: $(Join-Path $outDir 'EG800FactoryHost.exe')" diff --git a/factory_host/factory_host.py b/factory_host/factory_host.py new file mode 100644 index 0000000..4cd9eb0 --- /dev/null +++ b/factory_host/factory_host.py @@ -0,0 +1,779 @@ +# -*- coding: utf-8 -*- +"""EG800 factory host: batch flashing, serial production tests, and reports. + +The aboot command line is intentionally configuration-driven. A firmware +package must be selected explicitly and the configured argument list may use +{aboot}, {firmware}, {port}, and {station} placeholders. +""" + +from __future__ import annotations + +import csv +import json +import os +import queue +import re +import shlex +import subprocess +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from tkinter import END, BOTH, LEFT, RIGHT, X, Y, StringVar, Tk, Toplevel, filedialog, messagebox +from tkinter import ttk + +from ft1_protocol import FT1Session, NODE_NAMES, ProtocolError, error_text, parse_ft1_line + + +APP_DIR = Path(sys.executable).resolve().parent if getattr(sys, "frozen", False) else Path(__file__).resolve().parent + + +def discover_project_root(start: Path) -> Path: + """Find the checkout root from either the source or the dist directory.""" + for candidate in (start, *start.parents): + if (candidate / "burn_tool").exists() and ((candidate / "version").exists() or (candidate / "target").exists()): + return candidate + return start.parent + + +ROOT_DIR = discover_project_root(APP_DIR) +CONFIG_PATH = APP_DIR / "factory_host_config.json" +LOG_DIR = APP_DIR / "logs" +SESSION_DIR = LOG_DIR / "sessions" + + +DEFAULT_CONFIG = { + "flash": { + "aboot_path": "burn_tool/aboot.exe", + "downloader_path": "burn_tool/adownload.exe", + "args": ["-u", "-a", "-s", "{baudrate}", "-q", "{firmware}"], + "timeout_seconds": 900, + "workers": 2, + }, + "firmwares": {"production": "", "official": ""}, + "serial": {"baudrate": 115200, "encoding": "utf-8", "line_ending": "\\r\\n"}, + "protocol": {"prefix": "FT1 ", "version": 1, "auto_save": True}, +} + + +def load_config(path: Path = CONFIG_PATH) -> dict: + if not path.exists(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(DEFAULT_CONFIG, ensure_ascii=False, indent=2), encoding="utf-8-sig") + return json.loads(json.dumps(DEFAULT_CONFIG)) + try: + raw = json.loads(path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError) as exc: + raise RuntimeError(f"配置文件读取失败: {exc}") from exc + merged = json.loads(json.dumps(DEFAULT_CONFIG)) + merge_dict(merged, raw) + return merged + + +def merge_dict(dst: dict, src: dict) -> None: + for key, value in src.items(): + if isinstance(value, dict) and isinstance(dst.get(key), dict): + merge_dict(dst[key], value) + else: + dst[key] = value + + +def save_config(config: dict, path: Path = CONFIG_PATH) -> None: + path.write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8-sig") + + +def resolve_project_path(value: str) -> Path: + raw = Path(value) + if raw.is_absolute(): + return raw + candidates = [ROOT_DIR / raw, APP_DIR / raw, Path.cwd() / raw] + for candidate in candidates: + if candidate.exists(): + return candidate.resolve() + return candidates[0].resolve() + + +def find_downloader(configured: str, aboot_configured: str = "") -> Path | None: + candidates = [] + if configured: + raw = Path(configured) + candidates.extend([raw, ROOT_DIR / raw, APP_DIR / raw]) + if aboot_configured: + raw = Path(aboot_configured) + candidates.extend([raw.parent / "adownload.exe", ROOT_DIR / raw.parent / "adownload.exe", APP_DIR / raw.parent / "adownload.exe"]) + candidates.extend([ROOT_DIR / "burn_tool" / "adownload.exe", APP_DIR / "burn_tool" / "adownload.exe"]) + for candidate in candidates: + try: + if candidate.is_file(): + return candidate.resolve() + except OSError: + continue + for root in (ROOT_DIR / "burn_tool", APP_DIR / "burn_tool"): + try: + if root.is_dir(): + found = next(root.rglob("adownload.exe"), None) + if found and found.is_file(): + return found.resolve() + except OSError: + continue + return None + + +def expand_args(args: list[str], *, aboot: Path, firmware: Path, port: str, station: str, baudrate: int = 115200) -> list[str]: + values = {"aboot": str(aboot), "firmware": str(firmware), "port": port, "station": station, "baudrate": str(baudrate)} + return [str(arg).format_map(values) for arg in args] + + +def parse_command_template(template: str) -> list[str]: + """Parse a user-entered Windows command line while retaining quoted paths.""" + if not template.strip(): + return [] + return [part.strip('"') for part in shlex.split(template, posix=False)] + + +def list_serial_ports() -> list[str]: + """Return present COM ports without requiring a third-party module.""" + ports = [] + try: + import serial.tools.list_ports # type: ignore + + ports = [item.device for item in serial.tools.list_ports.comports()] + except ImportError: + pass + if not ports and os.name == "nt": + try: + result = subprocess.run( + ["powershell.exe", "-NoProfile", "-Command", "Get-CimInstance Win32_SerialPort | Select-Object -ExpandProperty DeviceID"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + ports = [line.strip() for line in result.stdout.splitlines() if re.fullmatch(r"COM\d+", line.strip(), re.I)] + except (OSError, subprocess.SubprocessError): + pass + return sorted(set(ports), key=lambda item: int(item[3:]) if item[3:].isdigit() else item) + + +class SerialReader: + """Small Windows serial reader using pyserial when available, file handles otherwise.""" + + def __init__(self, port: str, baudrate: int, encoding: str, on_line): + self.port = port + self.baudrate = baudrate + self.encoding = encoding or "utf-8" + self.on_line = on_line + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._stream = None + self._pyserial = None + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._run, name=f"serial-{self.port}", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + try: + if self._stream is not None: + self._stream.close() + except OSError: + pass + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=1.0) + + def send(self, text: str, line_ending: str = "\r\n") -> None: + payload = (text + line_ending).encode(self.encoding, errors="replace") + if self._pyserial is not None: + self._pyserial.write(payload) + elif self._stream is not None: + self._stream.write(payload) + + def _run(self) -> None: + try: + try: + import serial # type: ignore + + self._pyserial = serial.Serial(self.port, self.baudrate, timeout=0.25) + self._stream = self._pyserial + except ImportError: + if os.name == "nt": + subprocess.run(["mode", self.port, f"BAUD={self.baudrate}", "PARITY=n", "DATA=8", "STOP=1"], capture_output=True, check=False) + device = self.port if self.port.startswith("\\\\.\\") else "\\\\.\\" + self.port + self._stream = open(device, "r+b", buffering=0) + buffer = bytearray() + while not self._stop.is_set(): + chunk = self._stream.read(256) + if not chunk: + continue + buffer.extend(chunk) + while b"\n" in buffer: + raw, _, remainder = buffer.partition(b"\n") + buffer = bytearray(remainder) + self.on_line(raw.rstrip(b"\r").decode(self.encoding, errors="replace")) + if buffer: + self.on_line(buffer.decode(self.encoding, errors="replace")) + except Exception as exc: # serial backends surface OS-specific errors + self.on_line(f"[SERIAL_ERROR] {exc}") + finally: + try: + if self._stream is not None: + self._stream.close() + except OSError: + pass + self._stream = None + self._pyserial = None + + +@dataclass +class Station: + number: str + flash_port: str = "" + test_port: str = "" + status: str = "待机" + firmware_kind: str = "production" + message: str = "" + reader: SerialReader | None = None + protocol: FT1Session = field(default_factory=FT1Session) + report_path: Path | None = None + report_saved: bool = False + + +class FactoryHost(Tk): + def __init__(self): + super().__init__() + self.title("EG800 工厂烧录与产测上位机") + self.geometry("1440x850") + self.minsize(1100, 680) + self.config_data = load_config() + self.stations: dict[str, Station] = {} + self.events: queue.Queue = queue.Queue() + self.executor: ThreadPoolExecutor | None = None + self.flash_cancel = threading.Event() + self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S") + LOG_DIR.mkdir(parents=True, exist_ok=True) + SESSION_DIR.mkdir(parents=True, exist_ok=True) + self.report_path = LOG_DIR / f"factory_{self.session_id}.csv" + self._build_ui() + self._load_default_rows() + self.protocol("WM_DELETE_WINDOW", self._on_close) + self.after(100, self._drain_events) + + def _build_ui(self) -> None: + top = ttk.Frame(self, padding=8) + top.pack(fill=X) + ttk.Label(top, text="固件:").pack(side=LEFT) + self.firmware_kind = StringVar(value="production") + ttk.Radiobutton(top, text="产测固件", variable=self.firmware_kind, value="production").pack(side=LEFT, padx=3) + ttk.Radiobutton(top, text="正式固件", variable=self.firmware_kind, value="official").pack(side=LEFT, padx=3) + self.firmware_label = StringVar() + ttk.Label(top, textvariable=self.firmware_label, width=55).pack(side=LEFT, padx=12) + ttk.Button(top, text="选择固件", command=self.choose_firmware).pack(side=LEFT) + ttk.Button(top, text="扫描串口", command=self.scan_ports).pack(side=LEFT, padx=4) + + status = ttk.Frame(self, padding=(8, 0, 8, 6)) + status.pack(fill=X) + self.tool_status = StringVar() + ttk.Label(status, textvariable=self.tool_status).pack(side=LEFT) + ttk.Label(status, text=f"配置: {CONFIG_PATH}").pack(side=RIGHT) + + controls = ttk.Frame(self, padding=(8, 0, 8, 6)) + controls.pack(fill=X) + ttk.Button(controls, text="添加工位", command=self.add_station).pack(side=LEFT) + ttk.Button(controls, text="删除选中", command=self.remove_station).pack(side=LEFT, padx=4) + ttk.Button(controls, text="烧录选中", command=self.start_selected_flash).pack(side=LEFT, padx=(16, 4)) + ttk.Button(controls, text="批量烧录", command=self.start_flash).pack(side=LEFT) + ttk.Button(controls, text="停止烧录", command=self.stop_flash).pack(side=LEFT) + ttk.Button(controls, text="开始串口产测", command=self.start_tests).pack(side=LEFT, padx=16) + ttk.Button(controls, text="停止串口", command=self.stop_serial).pack(side=LEFT) + ttk.Button(controls, text="打开配置", command=self.open_config).pack(side=LEFT, padx=16) + ttk.Button(controls, text="重新加载", command=self.reload_config).pack(side=LEFT) + ttk.Button(controls, text="导出报表", command=self.export_report).pack(side=RIGHT) + + table_frame = ttk.Frame(self, padding=(8, 0, 8, 4)) + table_frame.pack(fill=BOTH, expand=True) + columns = ("station", "flash_port", "test_port", "status", "session", "imei", "progress", "firmware", "message") + self.table = ttk.Treeview(table_frame, columns=columns, show="headings", selectmode="extended") + headings = { + "station": "工位", "flash_port": "烧录口", "test_port": "产测口", "status": "状态", + "session": "会话", "imei": "IMEI", "progress": "节点", "firmware": "固件", "message": "最近信息", + } + widths = { + "station": 55, "flash_port": 75, "test_port": 75, "status": 105, "session": 145, + "imei": 125, "progress": 70, "firmware": 80, "message": 440, + } + for column in columns: + self.table.heading(column, text=headings[column]) + self.table.column(column, width=widths[column], anchor="w") + self.table.pack(side=LEFT, fill=BOTH, expand=True) + scroll = ttk.Scrollbar(table_frame, orient="vertical", command=self.table.yview) + scroll.pack(side=RIGHT, fill=Y) + hscroll = ttk.Scrollbar(table_frame, orient="horizontal", command=self.table.xview) + hscroll.pack(side="bottom", fill=X) + self.table.configure(yscrollcommand=scroll.set, xscrollcommand=hscroll.set) + self.table.bind("", self.edit_station) + self.table.bind("<>", self._refresh_node_table) + + node_frame = ttk.LabelFrame(self, text="FT1 节点状态", padding=6) + node_frame.pack(fill=X, padx=8, pady=(0, 4)) + node_columns = ("id", "name", "state", "code", "elapsed", "data") + self.node_table = ttk.Treeview(node_frame, columns=node_columns, show="headings", height=8) + for column, label, width in ( + ("id", "ID", 40), ("name", "节点", 150), ("state", "状态", 80), + ("code", "代码", 60), ("elapsed", "耗时(ms)", 85), ("data", "数据", 700), + ): + self.node_table.heading(column, text=label) + self.node_table.column(column, width=width, anchor="w") + self.node_table.pack(fill=X) + + bottom = ttk.LabelFrame(self, text="实时日志", padding=6) + bottom.pack(fill=BOTH, expand=False, padx=8, pady=(0, 8)) + self.log_text = ttk.Treeview(bottom, columns=("time", "station", "text"), show="headings", height=8) + for column, label, width in (("time", "时间", 90), ("station", "工位", 70), ("text", "日志", 850)): + self.log_text.heading(column, text=label) + self.log_text.column(column, width=width, anchor="w") + self.log_text.pack(fill=BOTH, expand=True) + self._refresh_tool_status() + + def _load_default_rows(self) -> None: + ports = list_serial_ports() + for index, port in enumerate(ports, 1): + self.add_station(number=str(index), flash_port=port, test_port=port) + if not ports: + self.add_station(number="1") + + def _refresh_tool_status(self) -> None: + downloader = find_downloader(self.config_data["flash"].get("downloader_path", ""), self.config_data["flash"].get("aboot_path", "")) + firmware = resolve_project_path(self.config_data["firmwares"].get(self.firmware_kind.get(), "")) + tool_text = f"adownload: {downloader}" if downloader else "adownload: 未找到（请把 burn_tool 完整目录放在根目录或在配置中指定）" + package_text = f"固件: {firmware}" if firmware.is_file() else "固件: 未选择" + self.tool_status.set(tool_text + " | " + package_text) + self.firmware_label.set(str(firmware) if firmware.is_file() else "未选择当前固件") + + def choose_firmware(self) -> None: + path = filedialog.askopenfilename(title="选择固件包", filetypes=(("固件包", "*.zip *.bin *.img *.json"), ("所有文件", "*.*"))) + if path: + self.config_data["firmwares"][self.firmware_kind.get()] = path + save_config(self.config_data) + self._refresh_tool_status() + + def scan_ports(self) -> None: + ports = list_serial_ports() + self._log("系统", f"检测到串口: {', '.join(ports) if ports else '无'}") + for station in self.stations.values(): + if not station.flash_port and ports: + station.flash_port = ports.pop(0) + if not station.test_port: + station.test_port = station.flash_port + self._update_row(station) + + def open_config(self) -> None: + try: + if os.name == "nt": + os.startfile(str(CONFIG_PATH)) # type: ignore[attr-defined] + else: + subprocess.Popen(["xdg-open", str(CONFIG_PATH)]) + except OSError as exc: + messagebox.showerror("打开配置失败", str(exc)) + + def reload_config(self) -> None: + try: + self.config_data = load_config() + self._refresh_tool_status() + self._log("系统", "配置已重新加载") + except RuntimeError as exc: + messagebox.showerror("配置错误", str(exc)) + + def add_station(self, number: str | None = None, flash_port: str = "", test_port: str = "") -> None: + if number is None: + number = str(len(self.stations) + 1) + while number in self.stations: + number = str(int(number) + 1) if number.isdigit() else f"{number}_new" + station = Station(number=number, flash_port=flash_port, test_port=test_port, firmware_kind=self.firmware_kind.get()) + self.stations[number] = station + self.table.insert("", END, iid=number, values=(number, flash_port, test_port, station.status, "", "", "0/6", "产测固件", "")) + + def remove_station(self) -> None: + for iid in self.table.selection(): + station = self.stations.pop(iid, None) + if station and station.reader: + station.reader.stop() + self.table.delete(iid) + + def edit_station(self, _event=None) -> None: + selected = self.table.selection() + if not selected: + return + station = self.stations[selected[0]] + dialog = Toplevel(self) + dialog.title(f"编辑工位 {station.number}") + dialog.transient(self) + ttk.Label(dialog, text="烧录串口（自动 USB 模式可空）:").grid(row=0, column=0, padx=8, pady=8) + flash_port_var = StringVar(value=station.flash_port) + ttk.Entry(dialog, textvariable=flash_port_var, width=20).grid(row=0, column=1, padx=8, pady=8) + ttk.Label(dialog, text="产测 CDC 串口:").grid(row=1, column=0, padx=8, pady=8) + test_port_var = StringVar(value=station.test_port) + ttk.Entry(dialog, textvariable=test_port_var, width=20).grid(row=1, column=1, padx=8, pady=8) + ttk.Label(dialog, text="固件:").grid(row=2, column=0, padx=8, pady=8) + kind_var = StringVar(value=station.firmware_kind) + ttk.Combobox(dialog, textvariable=kind_var, values=("production", "official"), state="readonly", width=17).grid(row=2, column=1, padx=8, pady=8) + + def apply(): + station.flash_port = flash_port_var.get().strip().upper() + station.test_port = test_port_var.get().strip().upper() + station.firmware_kind = kind_var.get() + self._update_row(station) + dialog.destroy() + + ttk.Button(dialog, text="确定", command=apply).grid(row=3, column=0, columnspan=2, pady=8) + + def _update_row(self, station: Station) -> None: + if self.table.exists(station.number): + protocol = station.protocol + self.table.item(station.number, values=( + station.number, station.flash_port, station.test_port, station.status, + protocol.session, protocol.imei, f"{protocol.passed_count}/6", + "产测固件" if station.firmware_kind == "production" else "正式固件", station.message, + )) + + def _refresh_node_table(self, _event=None) -> None: + selected = self.table.selection() + for iid in self.node_table.get_children(): + self.node_table.delete(iid) + if not selected or selected[0] not in self.stations: + return + protocol = self.stations[selected[0]].protocol + for node_id, name in enumerate(NODE_NAMES): + detail = protocol.node_details.get(name, {}) + data = json.dumps(detail.get("data", {}), ensure_ascii=False, separators=(",", ":")) + self.node_table.insert("", END, values=( + node_id, name, protocol.nodes[name], detail.get("code", ""), + detail.get("elapsed_ms", ""), data if data != "{}" else "", + )) + + def start_selected_flash(self) -> None: + selected = self.table.selection() + if len(selected) != 1: + messagebox.showwarning("请选择工位", "请先只选中一个工位，再点击“烧录选中”。") + return + self._start_flash_jobs(selected) + + def start_flash(self) -> None: + selected = self.table.selection() or tuple(self.stations.keys()) + self._start_flash_jobs(selected) + + def _start_flash_jobs(self, selected) -> None: + downloader = find_downloader(self.config_data["flash"].get("downloader_path", ""), self.config_data["flash"].get("aboot_path", "")) + if not downloader: + messagebox.showerror("无法烧录", "未找到 adownload.exe。请把 burn_tool 完整目录放到根目录，或在配置中指定 flash.downloader_path。aboot.exe 是图形界面，不能作为命令行烧录进程。") + return + jobs = [ + self.stations[iid] + for iid in selected + if iid in self.stations + and self.stations[iid].status not in ("排队中", "等待上电", "烧录中") + ] + if not jobs: + self._log("系统", "没有可提交的工位（正在烧录的工位不会重复排队）") + return + configured_args = self.config_data["flash"].get("args", ["{firmware}"]) + if isinstance(configured_args, str): + configured_args = parse_command_template(configured_args) + auto_usb = any(str(arg) in ("-a", "--auto-enable") for arg in configured_args) + uses_named_port = any("{port}" in str(arg) for arg in configured_args) + self.flash_cancel.clear() + workers = max(1, int(self.config_data["flash"].get("workers", 2))) + if auto_usb and not uses_named_port and len(jobs) > 1: + workers = 1 + self._log("系统", "USB 自动等待模式将按工位顺序烧录；请只给当前等待工位上电") + self.executor = ThreadPoolExecutor(max_workers=min(workers, len(jobs))) + for station in jobs: + station.status = "排队中" + station.message = "" + self._update_row(station) + self.executor.submit(self._flash_one, station, downloader) + self._log("系统", f"已提交 {len(jobs)} 个烧录任务，并发数 {workers}") + + def stop_flash(self) -> None: + self.flash_cancel.set() + self._log("系统", "已请求停止后续烧录任务；正在执行的 adownload 将在超时或退出后结束") + + def _flash_one(self, station: Station, downloader: Path) -> None: + if self.flash_cancel.is_set(): + self.events.put(("station", station.number, "已取消", "")) + return + firmware = resolve_project_path(self.config_data["firmwares"].get(station.firmware_kind, "")) + if not firmware.is_file(): + self.events.put(("station", station.number, "失败", f"固件不存在: {firmware}")) + return + configured_args = self.config_data["flash"].get("args", ["{firmware}"]) + if isinstance(configured_args, str): + configured_args = parse_command_template(configured_args) + baudrate = int(self.config_data["serial"].get("baudrate", 115200)) + if not station.flash_port and any("{port}" in str(item) for item in configured_args): + self.events.put(("station", station.number, "失败", "未设置烧录串口")) + return + args = expand_args(configured_args, aboot=downloader, firmware=firmware, port=station.flash_port, station=station.number, baudrate=baudrate) + command = [str(downloader)] + args + auto_usb = any(str(arg) in ("-a", "--auto-enable") for arg in configured_args) + status = "等待上电" if auto_usb else "烧录中" + hint = "下载器已启动，请现在给设备上电" if auto_usb else "" + self.events.put(("station", station.number, status, hint or " ".join(shlex.quote(x) for x in command))) + self.events.put(("log", station.number, "命令: " + " ".join(shlex.quote(x) for x in command))) + try: + process = subprocess.Popen(command, cwd=str(downloader.parent), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1) + output_tail = [] + deadline = time.monotonic() + int(self.config_data["flash"].get("timeout_seconds", 900)) + assert process.stdout is not None + lines: queue.Queue[str | None] = queue.Queue() + + def read_output() -> None: + try: + for raw_line in process.stdout: + lines.put(raw_line.rstrip()) + finally: + lines.put(None) + + threading.Thread(target=read_output, name=f"flash-output-{station.number}", daemon=True).start() + reader_done = False + timed_out = False + while not reader_done or process.poll() is None: + try: + line = lines.get(timeout=0.1) + if line is None: + reader_done = True + elif line: + output_tail.append(line) + output_tail = output_tail[-3:] + self.events.put(("log", station.number, line)) + except queue.Empty: + pass + if self.flash_cancel.is_set() or time.monotonic() > deadline: + timed_out = not self.flash_cancel.is_set() + process.terminate() + break + try: + return_code = process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + return_code = process.wait(timeout=5) + if self.flash_cancel.is_set(): + result, message = "已取消", "用户停止" + elif timed_out: + result, message = "失败", "等待上电或烧录超时" + elif return_code == 0: + result, message = "烧录成功", " | ".join(output_tail[-2:]) + else: + result, message = "失败", f"退出码 {return_code}: {' | '.join(output_tail[-2:])}" + except (OSError, subprocess.SubprocessError) as exc: + result, message = "失败", str(exc) + self.events.put(("station", station.number, result, message)) + + def start_tests(self) -> None: + selected = self.table.selection() or tuple(self.stations.keys()) + for iid in selected: + if iid not in self.stations: + continue + station = self.stations[iid] + if not station.test_port: + station.status, station.message = "失败", "未设置产测 CDC 串口" + self._update_row(station) + continue + station.protocol.reset() + station.report_path = None + station.report_saved = False + station.status = "等待设备" + station.message = "等待 FT1 session_start" + if station.reader: + station.reader.stop() + station.reader = SerialReader( + station.test_port, + int(self.config_data["serial"].get("baudrate", 115200)), + self.config_data["serial"].get("encoding", "utf-8"), + lambda line, s=station: self._serial_line(s, line), + ) + station.reader.start() + self._update_row(station) + self._refresh_node_table() + self._log("系统", "FT1 串口产测已启动；只保存以 FT1 开头的协议行") + + def _serial_line(self, station: Station, line: str) -> None: + self.events.put(("serial", station.number, line)) + + def _process_serial_line(self, station: Station, line: str) -> None: + if line.startswith("[SERIAL_ERROR]"): + station.status, station.message = "失败", line + self._log(station.number, line) + self._update_row(station) + return + try: + message = parse_ft1_line(line) + except ProtocolError as exc: + station.status = "协议错误" + station.message = str(exc) + self._log(station.number, f"[FT1错误] {exc}: {line}") + self._update_row(station) + return + if message is None: + self._log(station.number, f"[SDK] {line}") + return + + previous_session = station.protocol.session + try: + description = station.protocol.apply(message) + except ProtocolError as exc: + station.status = "协议错误" + station.message = str(exc) + self._log(station.number, f"[FT1错误] {exc}") + self._update_row(station) + return + if message["type"] == "session_start" and station.protocol.session != previous_session: + station.report_path = None + station.report_saved = False + + msg_type = message["type"] + if msg_type == "session_start": + station.status = "产测中" + elif msg_type == "node_start": + station.status = f"测试 {message['name']}" + elif msg_type == "node_result" and message["result"] == "FAIL": + station.status = "产测失败" + elif msg_type == "session_result": + station.status = "产测通过" if message["result"] == "PASS" else "产测失败" + elif msg_type == "power_off" and station.protocol.result == "PASS": + station.status = "通过/已关机" + station.message = description + self._log(station.number, f"[FT1] {description}") + self._update_row(station) + self._refresh_node_table() + + if self.config_data.get("protocol", {}).get("auto_save", True) and msg_type in {"session_result", "power_off"}: + self._save_station_session(station, msg_type) + + def _save_station_session(self, station: Station, trigger: str) -> Path | None: + protocol = station.protocol + if not protocol.messages: + return None + safe_session = re.sub(r"[^A-Za-z0-9_.-]+", "_", protocol.session or "UNKNOWN") + if station.report_path is None: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + station.report_path = SESSION_DIR / f"{timestamp}_station-{station.number}_{safe_session}.json" + record = { + "saved_at": datetime.now().isoformat(timespec="seconds"), + "save_trigger": trigger, + "station": station.number, + "flash_port": station.flash_port, + "test_port": station.test_port, + "firmware_kind": station.firmware_kind, + "summary": protocol.summary(), + "messages": protocol.messages, + } + station.report_path.write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8-sig") + if not station.report_saved: + csv_path = LOG_DIR / "factory_results.csv" + new_file = not csv_path.exists() + with csv_path.open("a", newline="", encoding="utf-8-sig") as handle: + writer = csv.writer(handle) + if new_file: + writer.writerow([ + "保存时间", "工位", "产测串口", "会话", "结果", "失败码", "失败原因", + "固件", "IMEI", "ICCID", "NFC_UID", "开始时间", "耗时_ms", *NODE_NAMES, + ]) + summary = protocol.summary() + writer.writerow([ + record["saved_at"], station.number, station.test_port, protocol.session, + summary["result"], protocol.fail_code, summary["fail_reason"], protocol.firmware, + protocol.imei, protocol.iccid, protocol.nfc_uid, protocol.start_time or "", + protocol.duration_ms, *(protocol.nodes[name] for name in NODE_NAMES), + ]) + station.report_saved = True + self._log(station.number, f"记录已保存: {station.report_path}") + return station.report_path + + def stop_serial(self) -> None: + for station in self.stations.values(): + if station.reader: + station.reader.stop() + station.reader = None + if station.protocol.messages and not station.report_saved: + self._save_station_session(station, "manual_stop") + if station.status in ("等待设备", "产测中") or station.status.startswith("测试 "): + station.status = "已停止" + self._update_row(station) + self._log("系统", "串口监听已停止") + + def _drain_events(self) -> None: + try: + while True: + event = self.events.get_nowait() + kind = event[0] + if kind == "station": + _, number, status, message = event + if number in self.stations: + station = self.stations[number] + station.status, station.message = status, message + self._update_row(station) + self._log(number, f"{status}: {message}") + elif kind == "log": + _, number, line = event + self._log(number, line) + elif kind == "serial": + _, number, line = event + if number in self.stations: + self._process_serial_line(self.stations[number], line) + except queue.Empty: + pass + self.after(100, self._drain_events) + + def _log(self, station: str, text: str) -> None: + timestamp = datetime.now().strftime("%H:%M:%S") + self.log_text.insert("", END, values=(timestamp, station, text)) + children = self.log_text.get_children() + if len(children) > 1000: + self.log_text.delete(children[0]) + + def export_report(self) -> None: + path = filedialog.asksaveasfilename(title="导出产测报表", defaultextension=".csv", initialfile=f"factory_{self.session_id}.csv", filetypes=(("CSV", "*.csv"),)) + if not path: + return + with open(path, "w", newline="", encoding="utf-8-sig") as handle: + writer = csv.writer(handle) + writer.writerow([ + "时间", "工位", "烧录口", "产测口", "状态", "会话", "结果", "失败码", "失败原因", + "固件类型", "产测固件版本", "IMEI", "ICCID", "NFC_UID", "开始时间", "耗时_ms", *NODE_NAMES, + ]) + for station in self.stations.values(): + protocol = station.protocol + summary = protocol.summary() + writer.writerow([ + datetime.now().isoformat(timespec="seconds"), station.number, station.flash_port, + station.test_port, station.status, protocol.session, summary["result"], + protocol.fail_code, summary["fail_reason"], station.firmware_kind, protocol.firmware, + protocol.imei, protocol.iccid, protocol.nfc_uid, protocol.start_time or "", + protocol.duration_ms, *(protocol.nodes[name] for name in NODE_NAMES), + ]) + self._log("系统", f"报表已导出: {path}") + + def _on_close(self) -> None: + self.stop_serial() + self.flash_cancel.set() + self.destroy() + + +def main() -> None: + try: + app = FactoryHost() + app.mainloop() + except RuntimeError as exc: + messagebox.showerror("启动失败", str(exc)) + raise + + +if __name__ == "__main__": + main() diff --git a/factory_host/factory_host_config.json b/factory_host/factory_host_config.json new file mode 100644 index 0000000..e9dc1c1 --- /dev/null +++ b/factory_host/factory_host_config.json @@ -0,0 +1,30 @@ +{ + "flash": { + "aboot_path": "burn_tool/aboot.exe", + "downloader_path": "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": "factory_host/dist/factory_firmware/EG800AKCN_11LG-FT1-factory-test.zip", + "official": "version/EG800AKCN_11LG-1.8.4-20260921171244/EG800AKCN_11LG-1.8.4-20260921171244-fac.zip" + }, + "serial": { + "baudrate": 115200, + "encoding": "utf-8", + "line_ending": "\\r\\n" + }, + "protocol": { + "prefix": "FT1 ", + "version": 1, + "auto_save": true + } +} diff --git a/factory_host/ft1_protocol.py b/factory_host/ft1_protocol.py new file mode 100644 index 0000000..f351492 --- /dev/null +++ b/factory_host/ft1_protocol.py @@ -0,0 +1,246 @@ +"""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 = 1 +MESSAGE_TYPES = { + "session_start", + "node_start", + "node_progress", + "node_result", + "session_result", + "power_off", +} +NODE_NAMES = ( + "BOOT_AUDIO", + "POWER_ADC", + "AUDIO_LOOPBACK", + "ESIM", + "DATA_CALL", + "NFC", +) +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 地址", + 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) + 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 + 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 == "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), + } diff --git a/factory_host/requirements.txt b/factory_host/requirements.txt new file mode 100644 index 0000000..0fb1e24 --- /dev/null +++ b/factory_host/requirements.txt @@ -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 diff --git a/factory_host/test_factory_host.py b/factory_host/test_factory_host.py new file mode 100644 index 0000000..aa18415 --- /dev/null +++ b/factory_host/test_factory_host.py @@ -0,0 +1,98 @@ +import json +import tempfile +import unittest +from pathlib import Path + +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_project_root_and_relative_path(self): + self.assertTrue((factory_host.ROOT_DIR / "burn_tool").is_dir()) + 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() + ) + + def test_ft1_ignores_sdk_log_and_parses_node(self): + self.assertIsNone(parse_ft1_line("DP1312EA is OK!")) + message = parse_ft1_line( + 'FT1 {"v":1,"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":1,"type":"session_start","seq":1,"session":"BOOT-1",' + '"uptime_ms":2050,"imei":"860123456789012","iccid":"","nfc_uid":"",' + '"fw":"FACTORY_TEST_1.0.0","start_time":null}' + ) + session.apply(start) + result = parse_ft1_line( + 'FT1 {"v":1,"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"}}' + ) + 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":1,"type":"node_start","seq":2,"session":"BOOT-1",' + '"uptime_ms":2100,"node":1,"name":"NFC"}' + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/ql-application/threadx/interface/factory_test/factory_audio.c b/ql-application/threadx/interface/factory_test/factory_audio.c index 747e972..33b88d9 100644 --- a/ql-application/threadx/interface/factory_test/factory_audio.c +++ b/ql-application/threadx/interface/factory_test/factory_audio.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); diff --git a/ql-application/threadx/interface/factory_test/factory_protocol.c b/ql-application/threadx/interface/factory_test/factory_protocol.c index 889fee6..d16d0e5 100644 --- a/ql-application/threadx/interface/factory_test/factory_protocol.c +++ b/ql-application/threadx/interface/factory_test/factory_protocol.c @@ -32,8 +32,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" }; if (node < 0 || node >= FT_NODE_COUNT) { @@ -131,11 +131,11 @@ 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] ? "\"" : ""); } @@ -191,9 +191,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\"}}", 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, @@ -202,8 +201,7 @@ void ft_protocol_session_result(int passed, ft_error_t code) (unsigned long)(ft_now_ms() - g_protocol.start_ms), 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[4]), ft_state_name(g_protocol.nodes[5])); } void ft_protocol_power_off(void) diff --git a/ql-application/threadx/interface/factory_test/factory_test.c b/ql-application/threadx/interface/factory_test/factory_test.c index 7a88378..66383e3 100644 --- a/ql-application/threadx/interface/factory_test/factory_test.c +++ b/ql-application/threadx/interface/factory_test/factory_test.c @@ -8,7 +8,6 @@ #include "ql_gpio.h" #include "ql_iic.h" #include "ql_nw.h" -#include "ql_ping.h" #include "ql_power.h" #include "ql_rtos.h" #include "ql_sim.h" @@ -21,10 +20,6 @@ #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 +30,14 @@ 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 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) @@ -111,43 +98,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,54 +157,6 @@ 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}; @@ -292,10 +194,8 @@ 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); default: return FT_ERR_NFC_LOCKED; } diff --git a/ql-application/threadx/interface/factory_test/factory_test.h b/ql-application/threadx/interface/factory_test/factory_test.h index aa13073..5929fa4 100644 --- a/ql-application/threadx/interface/factory_test/factory_test.h +++ b/ql-application/threadx/interface/factory_test/factory_test.h @@ -6,7 +6,6 @@ #define FT_PROTOCOL_VERSION 1 #define FT_FIRMWARE_VERSION "FACTORY_TEST_1.0.0" -#define FT_PING_HOST "banban.api.talkingq.com" #define FT_ADC_MIN_MV 3500 #define FT_ADC_MAX_MV 4200 @@ -18,11 +17,10 @@ #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_TRIPLE_KEY_PIN GPIO_PIN_NO_84 +#define FT_LOOPBACK_KEY_PIN GPIO_PIN_NO_84 #define FT_PA_PIN GPIO_PIN_NO_68 #define FT_AUDIO_BOOT_PATH "U:/factory_boot_zh.mp3" @@ -33,10 +31,8 @@ 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_COUNT } ft_node_t; @@ -53,16 +49,12 @@ 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_NFC_LOCKED = 1701, FT_ERR_NFC_CHIP = 1702, FT_ERR_NFC_TIMEOUT = 1703,