From 2919ebe31201ff93f3304d5133cb6f464873c1d7 Mon Sep 17 00:00:00 2001 From: yi-bot Date: Mon, 3 Aug 2026 03:12:48 +0000 Subject: [PATCH] fix: use HM42 vendor image upload session --- src/ppclock/commands.py | 17 ++++++++++++++--- src/ppclock/protocol.py | 22 ++++++++++++++++++++++ tests/test_client.py | 11 +++++++++-- tests/test_commands.py | 20 +++++++++++++++++++- tests/test_protocol.py | 11 +++++++++++ 5 files changed, 75 insertions(+), 6 deletions(-) diff --git a/src/ppclock/commands.py b/src/ppclock/commands.py index 32fb189..ce6764f 100644 --- a/src/ppclock/commands.py +++ b/src/ppclock/commands.py @@ -34,10 +34,16 @@ async def clear(t, param: int = 0x00): async def upload_image(t, bw: bytes, red: bytes | None, *, small: bool, slot: int | None = None, block_delay: float = 0.0): - """传图：可选槽位选择 → bw 平面块 → 红平面块 → 300ms → 01/AA。""" + """传图；HM42_AIO 使用厂商 App 的 EPD 会话，其他固件保留旧协议。""" + hm42_session = P.uses_hm42_image_session( + getattr(t, "firmware_revision", None) + ) if slot is not None: await t.write_rxtx(P.slot_select_frame(slot)) await t.delay(0.3) + if hm42_session: + for frame in P.hm42_image_begin_frames(): + await t.write_epd(frame) for frame in P.image_blocks(bw, small=small, red=False): await t.write_epd(frame) if block_delay: @@ -47,8 +53,13 @@ async def upload_image(t, bw: bytes, red: bytes | None, *, small: bool, await t.write_epd(frame) if block_delay: await t.delay(block_delay) - await t.delay(0.3) - await t.write_epd(P.end_frame(small=small)) + if hm42_session: + await t.write_epd(P.hm42_image_finish_frame()) + await t.delay(0.3) + await t.write_rxtx(P.simple_command("commit")) + else: + await t.delay(0.3) + await t.write_epd(P.end_frame(small=small)) async def countdown_set(t, d: datetime.date, mode: str, prefix: str | None = None, diff --git a/src/ppclock/protocol.py b/src/ppclock/protocol.py index 2781857..e6b8338 100644 --- a/src/ppclock/protocol.py +++ b/src/ppclock/protocol.py @@ -40,6 +40,28 @@ def requires_complete_image_planes(firmware_revision: str | None) -> bool: revision = (firmware_revision or "").strip().upper() return revision.startswith("HM42_AIO") + +def uses_hm42_image_session(firmware_revision: str | None) -> bool: + """Whether image transfer needs the newer HM42 EPD session framing. + + The vendor Android provider ``轻舞/时光`` (``d8.e``) wraps both image + planes with ``8001, 0000, 020000`` and closes with ``0101`` followed by + RXTX ``e2`` for a normal (non-calendar) image. The older Web-BLE path has + no such preamble and ends with the legacy one-byte EPD frame. + """ + revision = (firmware_revision or "").strip().upper() + return revision.startswith("HM42_AIO") + + +def hm42_image_begin_frames() -> tuple[bytes, ...]: + """Vendor-app EPD image-session preamble for HM42_AIO firmware.""" + return (bytes.fromhex("8001"), bytes.fromhex("0000"), bytes.fromhex("020000")) + + +def hm42_image_finish_frame() -> bytes: + """Vendor-app EPD image-session terminator for HM42_AIO firmware.""" + return bytes.fromhex("0101") + MODES = { **{f"image{i}": bytes([0xE1, 0x00, 0x00 + i]) for i in range(4)}, "calendar1": bytes.fromhex("e10103"), diff --git a/tests/test_client.py b/tests/test_client.py index 4d95c0d..ce8a5b6 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -101,10 +101,17 @@ class TestCommands: ) assert result["bytes_bw"] == 15000 assert result["bytes_red"] == 15000 - assert len(client._t.epd_writes) == 63 + 63 + 1 - first_red = client._t.epd_writes[63] + assert client._t.epd_writes[:3] == [ + bytes.fromhex("8001"), + bytes.fromhex("0000"), + bytes.fromhex("020000"), + ] + assert len(client._t.epd_writes) == 3 + 63 + 63 + 1 + first_red = client._t.epd_writes[3 + 63] assert first_red[:4] == bytes.fromhex("03000000") assert set(first_red[4:]) == {0} + assert client._t.epd_writes[-1] == bytes.fromhex("0101") + assert client._t.rxtx_writes[-1] == bytes.fromhex("e2") @pytest.mark.asyncio async def test_render_template(self, client): diff --git a/tests/test_commands.py b/tests/test_commands.py index 2eb1538..c7a44a3 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -7,10 +7,11 @@ from ppclock import commands as C class FakeTransport: - def __init__(self, id_response="DEVICE12345678"): + def __init__(self, id_response="DEVICE12345678", firmware_revision=None): self.epd_writes = [] self.rxtx_writes = [] self.id_response = id_response + self.firmware_revision = firmware_revision self.delays = [] async def write_epd(self, data, response=True): @@ -80,6 +81,23 @@ class TestUploadImage: assert t.epd_writes[-1] == bytes([0xAA]) assert t.epd_writes[0][0] == 0x04 + @pytest.mark.asyncio + async def test_hm42_uses_vendor_epd_session(self): + t = FakeTransport(firmware_revision="HM42_AIO_V1.0.9") + await C.upload_image( + t, bytes(15000), bytes(15000), small=False, slot=1 + ) + assert t.rxtx_writes == [bytes.fromhex("EA01"), bytes.fromhex("E2")] + assert t.epd_writes[:3] == [ + bytes.fromhex("8001"), + bytes.fromhex("0000"), + bytes.fromhex("020000"), + ] + assert t.epd_writes[3][:4] == bytes.fromhex("03FF0000") + assert t.epd_writes[66][:4] == bytes.fromhex("03000000") + assert t.epd_writes[-1] == bytes.fromhex("0101") + assert len(t.epd_writes) == 3 + 63 + 63 + 1 + class TestOtherCommands: @pytest.mark.asyncio diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 4fb52eb..f3487b9 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -73,6 +73,17 @@ class TestCommandFrames: assert P.requires_complete_image_planes("PP_da14585_4.2") is False assert P.requires_complete_image_planes(None) is False + def test_hm42_image_session_by_firmware_family(self): + assert P.uses_hm42_image_session("HM42_AIO_V1.0.9") is True + assert P.uses_hm42_image_session("PP_da14585_4.2") is False + assert P.uses_hm42_image_session(None) is False + assert P.hm42_image_begin_frames() == ( + bytes.fromhex("8001"), + bytes.fromhex("0000"), + bytes.fromhex("020000"), + ) + assert P.hm42_image_finish_frame() == bytes.fromhex("0101") + def test_parking(self): """停车牌 "12" → EF1122（向量#4）""" assert P.parking_frame("12") == bytes([0xEF, 0x11, 0x22])