fix: use HM42 vendor image upload session

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,

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"),

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):

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

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])