diff --git a/pyproject.toml b/pyproject.toml index 0274a08..7e675a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dev = ["pytest>=8"] [project.scripts] ppclock = "ppclock.cli:main" +ppclock-bridge = "ppclock.bridge_server:run" [tool.setuptools.packages.find] where = ["src"] diff --git a/tools/bridge_server.py b/src/ppclock/bridge_server.py similarity index 99% rename from tools/bridge_server.py rename to src/ppclock/bridge_server.py index cd8d2dd..aaba5e3 100644 --- a/tools/bridge_server.py +++ b/src/ppclock/bridge_server.py @@ -325,8 +325,13 @@ async def main(): await srv.serve_forever() -if __name__ == "__main__": +def run(): + """console_script 入口:ppclock-bridge。""" try: asyncio.run(main()) except KeyboardInterrupt: sys.exit(0) + + +if __name__ == "__main__": + run() diff --git a/src/ppclock/cli.py b/src/ppclock/cli.py index 4aaa926..05fc4ab 100644 --- a/src/ppclock/cli.py +++ b/src/ppclock/cli.py @@ -42,13 +42,13 @@ def _emit(args, cmd, data=None, error=None): def _make_transport(args): """工厂函数——测试用 monkeypatch 替换为 FakeTransport。""" - from .bridge import BridgeTransport, parse_via + from .transports.bridge import BridgeTransport, parse_via via = parse_via(getattr(args, "via", None)) if via: host, port = via return BridgeTransport(address=args.mac, timeout=args.timeout, host=host, port=port) - from .transport import BLETransport + from .transports.local import BLETransport return BLETransport(address=args.mac, timeout=args.timeout) @@ -316,12 +316,12 @@ async def _run_batch(args) -> list: async def _dispatch(args) -> dict | list | None: if args.cmd == "scan": - from .bridge import parse_via + from .transports.bridge import parse_via via = parse_via(getattr(args, "via", None)) if via: - from .bridge import scan as bridge_scan + from .transports.bridge import scan as bridge_scan return {"devices": await bridge_scan(via[0], via[1], timeout=args.timeout)} - from .transport import scan + from .transports.local import scan return {"devices": await scan(timeout=args.timeout)} if args.cmd == "batch": return {"results": await _run_batch(args)} @@ -338,7 +338,7 @@ def main(argv=None) -> int: _emit(args, args.cmd, error=str(e)) return EXIT_USAGE except Exception as e: - from .transport import TransportError + from .transports.base import TransportError code = EXIT_TRANSPORT if isinstance(e, TransportError) else EXIT_ERROR _emit(args, args.cmd, error=f"{type(e).__name__}: {e}") return code diff --git a/src/ppclock/firmware/__init__.py b/src/ppclock/firmware/__init__.py new file mode 100644 index 0000000..1fb3649 --- /dev/null +++ b/src/ppclock/firmware/__init__.py @@ -0,0 +1,8 @@ +"""固件域:DA14585 镜像解析/重打包(image)与 SUOTA 升级(ota)。""" +from .image import parse_image, unpack_body, pack_image +from .ota import run_ota, OtaError, package_image, xor_checksum + +__all__ = [ + "parse_image", "unpack_body", "pack_image", + "run_ota", "OtaError", "package_image", "xor_checksum", +] diff --git a/src/ppclock/firmware/image.py b/src/ppclock/firmware/image.py new file mode 100644 index 0000000..095c0ee --- /dev/null +++ b/src/ppclock/firmware/image.py @@ -0,0 +1,54 @@ +"""DA14585 SUOTA 镜像解析/重打包(tools 交付物)。 +格式(证据:app/ 样本 + CRC 反推验证): + [0:2] magic 'pQ' = 0x7051(LE) + [2:4] flags = 0x01AA + [4:8] body 长度(= 文件长 - 64) + [8:12] CRC32(body)(zlib 多项式,LE) + [12:64] 0xFF 填充 + [64:] 镜像体,起始为 Cortex-M 向量表(SP, Reset, ...),加载基址 0x07FC0000 +""" +from __future__ import annotations + +import struct +import zlib + +MAGIC_BYTES = b"pQ" # 文件头两字节 70 51 +MAGIC = 0x7051 +FLAGS_DEFAULT = 0x01AA +HEADER_SIZE = 64 +LOAD_BASE = 0x07FC0000 + + +def parse_image(data: bytes) -> dict: + if len(data) < HEADER_SIZE: + raise ValueError("文件小于 64 字节头") + if data[0:2] != MAGIC_BYTES: + raise ValueError(f"magic 错误: {data[0:2].hex()}(期望 {MAGIC_BYTES.hex()}='pQ')") + flags, body_size, crc = struct.unpack_from(" bytes: + info = parse_image(data) + return data[HEADER_SIZE:HEADER_SIZE + info["body_size"]] + + +def pack_image(body: bytes, flags: int = FLAGS_DEFAULT) -> bytes: + header = MAGIC_BYTES + struct.pack(" bytes: ... + async def request_device_id(self, timeout: float = 8.0) -> str: ... + async def delay(self, seconds: float): ... + async def run_ota(self, image: bytes, on_progress=None): ... diff --git a/src/ppclock/bridge.py b/src/ppclock/transports/bridge.py similarity index 100% rename from src/ppclock/bridge.py rename to src/ppclock/transports/bridge.py diff --git a/src/ppclock/transport.py b/src/ppclock/transports/local.py similarity index 98% rename from src/ppclock/transport.py rename to src/ppclock/transports/local.py index a4812f9..1692279 100644 --- a/src/ppclock/transport.py +++ b/src/ppclock/transports/local.py @@ -6,7 +6,7 @@ import asyncio from bleak import BleakClient, BleakScanner from bleak.exc import BleakError -from . import protocol as P +from .. import protocol as P class TransportError(Exception): @@ -92,5 +92,5 @@ class BLETransport: async def run_ota(self, image: bytes, on_progress=None): """SUOTA 固件升级(证据 analysis/apk/apk-analysis.md §e)。""" - from . import ota + from ..firmware import ota return await ota.run_ota(self._client, image, on_progress) diff --git a/tests/test_bridge.py b/tests/test_bridge.py index 8b39b5f..e957569 100644 --- a/tests/test_bridge.py +++ b/tests/test_bridge.py @@ -5,7 +5,7 @@ import json import pytest import pytest_asyncio -from ppclock import bridge +from ppclock.transports import bridge class FakeBridgeServer: diff --git a/tests/test_cli.py b/tests/test_cli.py index c4b9b1b..36c47bc 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -154,7 +154,7 @@ class TestCommands: assert FakeTransport.instances[0].epd_writes == [bytes([0x01])] def test_scan(self, capsys, monkeypatch): - import ppclock.transport as tr + import ppclock.transports.local as tr async def fake_scan(timeout=5.0, name_prefix="NRF-"): return [{"name": "NRF-TEST", "address": "AA:BB", "rssi": -50}] monkeypatch.setattr(tr, "scan", fake_scan) @@ -242,7 +242,7 @@ _ORIG_MAKE = cli._make_transport class TestViaBridge: def test_via_bridge_creates_bridge_transport(self): import argparse - from ppclock.bridge import BridgeTransport + from ppclock.transports.bridge import BridgeTransport args = argparse.Namespace(mac="AA:BB", timeout=5.0, via="bridge://192.168.61.35:8971") t = _ORIG_MAKE(args) assert isinstance(t, BridgeTransport) @@ -257,12 +257,12 @@ class TestViaBridge: def test_via_local_creates_ble(self): import argparse - from ppclock.transport import BLETransport + from ppclock.transports.local import BLETransport args = argparse.Namespace(mac=None, timeout=5.0, via=None) assert isinstance(_ORIG_MAKE(args), BLETransport) def test_scan_uses_bridge(self, capsys, monkeypatch): - import ppclock.bridge as br + import ppclock.transports.bridge as br called = {} async def fake_scan(host, port, timeout=5.0): @@ -275,7 +275,7 @@ class TestViaBridge: assert out["data"]["devices"][0]["name"] == "NRF-BR" def test_env_var_fallback(self, capsys, monkeypatch): - import ppclock.bridge as br + import ppclock.transports.bridge as br called = {} async def fake_scan(host, port, timeout=5.0): diff --git a/tests/test_ota.py b/tests/test_ota.py index 2770f2a..62eec65 100644 --- a/tests/test_ota.py +++ b/tests/test_ota.py @@ -1,7 +1,7 @@ """OTA/SUOTA 打包逻辑测试(证据:analysis/apk/apk-analysis.md §e,z7.h0/z7.b0)。""" import pytest -from ppclock import ota +from ppclock.firmware import ota class TestPackaging: diff --git a/tools/fw.py b/tools/fw.py index 095c0ee..1c857ee 100644 --- a/tools/fw.py +++ b/tools/fw.py @@ -1,54 +1,5 @@ -"""DA14585 SUOTA 镜像解析/重打包(tools 交付物)。 -格式(证据:app/ 样本 + CRC 反推验证): - [0:2] magic 'pQ' = 0x7051(LE) - [2:4] flags = 0x01AA - [4:8] body 长度(= 文件长 - 64) - [8:12] CRC32(body)(zlib 多项式,LE) - [12:64] 0xFF 填充 - [64:] 镜像体,起始为 Cortex-M 向量表(SP, Reset, ...),加载基址 0x07FC0000 +"""兼容垫片:fw 工具已迁入 ppclock.firmware.image(SDK 域)。 +保留本模块以兼容 tools/field_test.py 的 `from tools.fw import ...`。 """ -from __future__ import annotations - -import struct -import zlib - -MAGIC_BYTES = b"pQ" # 文件头两字节 70 51 -MAGIC = 0x7051 -FLAGS_DEFAULT = 0x01AA -HEADER_SIZE = 64 -LOAD_BASE = 0x07FC0000 - - -def parse_image(data: bytes) -> dict: - if len(data) < HEADER_SIZE: - raise ValueError("文件小于 64 字节头") - if data[0:2] != MAGIC_BYTES: - raise ValueError(f"magic 错误: {data[0:2].hex()}(期望 {MAGIC_BYTES.hex()}='pQ')") - flags, body_size, crc = struct.unpack_from(" bytes: - info = parse_image(data) - return data[HEADER_SIZE:HEADER_SIZE + info["body_size"]] - - -def pack_image(body: bytes, flags: int = FLAGS_DEFAULT) -> bytes: - header = MAGIC_BYTES + struct.pack("