diff --git a/src/ppclock/client.py b/src/ppclock/client.py index 923681d..4260103 100644 --- a/src/ppclock/client.py +++ b/src/ppclock/client.py @@ -19,6 +19,7 @@ from pathlib import Path from . import commands as C from . import image_pipeline as IP +from . import protocol as P from .transports import bridge as _bridge_mod from .transports import local as _local_mod from .transports.base import TransportError # noqa: F401 (SDK 公开异常) @@ -117,6 +118,11 @@ class PPClient: img = IP.Image.open(source) if isinstance(source, (str, Path)) else source bw, red = IP.process_image(img, size=size, algorithm=algo, tricolor=not mono, **adjust) + if red is None and P.requires_complete_image_planes( + getattr(self._t, "firmware_revision", None)): + # HM42 stores the two 400x300 planes as one sequential image. Its + # Web-BLE client sends an empty red plane even for monochrome input. + red = bytes(len(bw)) small = len(bw) in (3500, 9000) await C.upload_image(self._t, bw, red, small=small, slot=slot) return {"bytes_bw": len(bw), "bytes_red": len(red) if red else 0, diff --git a/src/ppclock/protocol.py b/src/ppclock/protocol.py index d15f2d5..2781857 100644 --- a/src/ppclock/protocol.py +++ b/src/ppclock/protocol.py @@ -29,6 +29,17 @@ def select_rxtx_uuid(firmware_revision: str | None) -> str: return RXTX_CHAR_UUID_LEGACY return RXTX_CHAR_UUID + +def requires_complete_image_planes(firmware_revision: str | None) -> bool: + """Whether a nominally mono full-screen upload still needs both planes. + + The HM42 Web-BLE implementation always sends the 400x300 black/white plane + followed by a red plane before opcode 01. Omitting the second plane leaves + HM42 rendering stale/uninitialised red data (observed as a red noisy screen). + """ + revision = (firmware_revision or "").strip().upper() + return revision.startswith("HM42_AIO") + 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 22e311d..4d95c0d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -7,10 +7,11 @@ from ppclock.client import PPClient class FakeTransport: - def __init__(self): + def __init__(self, firmware_revision=None): self.epd_writes = [] self.rxtx_writes = [] self.address = "AA:BB:CC:DD:EE:FF" + self.firmware_revision = firmware_revision async def __aenter__(self): return self @@ -78,6 +79,33 @@ class TestCommands: assert client._t.rxtx_writes[0] == bytes([0xEA, 0x00]) assert len(client._t.epd_writes) == 63 + 63 + 1 + @pytest.mark.asyncio + async def test_upload_mono_non_hm42_sends_one_plane(self): + from PIL import Image + client = PPClient(FakeTransport("PP_da14585_4.2")) + result = await client.upload_image( + Image.new("RGB", (400, 300), "white"), slot=1, + algo="none", mono=True, + ) + assert result["bytes_bw"] == 15000 + assert result["bytes_red"] == 0 + assert len(client._t.epd_writes) == 63 + 1 + + @pytest.mark.asyncio + async def test_upload_mono_hm42_appends_empty_red_plane(self): + from PIL import Image + client = PPClient(FakeTransport("HM42_AIO_V1.0.9")) + result = await client.upload_image( + Image.new("RGB", (400, 300), "white"), slot=1, + algo="none", mono=True, + ) + 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 first_red[:4] == bytes.fromhex("03000000") + assert set(first_red[4:]) == {0} + @pytest.mark.asyncio async def test_render_template(self, client): result = await client.render_template("custom", {"text": "你好"}) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 811dfca..4fb52eb 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -67,6 +67,12 @@ class TestCommandFrames: assert P.select_rxtx_uuid("PP_da14585_4.2") == P.RXTX_CHAR_UUID assert P.select_rxtx_uuid(None) == P.RXTX_CHAR_UUID + def test_complete_image_planes_by_firmware_family(self): + assert P.requires_complete_image_planes("HM42_AIO_V1.0.9") is True + assert P.requires_complete_image_planes("hm42_aio_v2") is True + assert P.requires_complete_image_planes("PP_da14585_4.2") is False + assert P.requires_complete_image_planes(None) is False + def test_parking(self): """停车牌 "12" → EF1122(向量#4)""" assert P.parking_frame("12") == bytes([0xEF, 0x11, 0x22])