From 122023cdf4ff917558b2e61d8c1eb0ae4dd463cd Mon Sep 17 00:00:00 2001
From: agent <agent@local>
Date: Thu, 30 Jul 2026 14:53:59 +0000
Subject: [PATCH] =?UTF-8?q?feat(sdk):=20PPClient=20=E9=97=A8=E9=9D=A2=20AP?=
 =?UTF-8?q?I=EF=BC=88via=5Flocal/via=5Fbridge/via=5Furi/scan=5Fdevices=20+?=
 =?UTF-8?q?=20=E5=85=A8=E8=AE=BE=E5=A4=87=E6=93=8D=E4=BD=9C=EF=BC=89+=20?=
 =?UTF-8?q?=5F=5Fversion=5F=5F=200.1.0=EF=BC=8C165=20=E6=B5=8B=E8=AF=95?=
 =?UTF-8?q?=E5=85=A8=E7=BB=BF?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/ppclock/__init__.py |  30 +++++++
 src/ppclock/client.py   | 171 ++++++++++++++++++++++++++++++++++++++++
 tests/test_client.py    | 148 ++++++++++++++++++++++++++++++++++
 3 files changed, 349 insertions(+)
 create mode 100644 src/ppclock/client.py
 create mode 100644 tests/test_client.py

diff --git a/src/ppclock/__init__.py b/src/ppclock/__init__.py
index e69de29..8809447 100644
--- a/src/ppclock/__init__.py
+++ b/src/ppclock/__init__.py
@@ -0,0 +1,30 @@
+"""ppclock SDK —— 4.2寸墨水屏设备（DA14585 BLE）离线开发包。
+
+快速上手：
+    from ppclock import PPClient
+    async with PPClient.via_bridge("192.168.61.35:8971") as dev:
+        await dev.set_time(tz=8)
+        await dev.upload_image("photo.jpg", slot=0)
+
+域：
+- client.PPClient：统一门面（推荐入口）
+- protocol：BLE 协议编解码（纯函数）
+- image_pipeline：图像调整/抖动/双平面打包
+- templates：本地 6 模板渲染
+- transports：local(bleak)/bridge(RPC) 传输实现
+- firmware：镜像解析重打包(image)与 SUOTA(ota)
+- bridge_server：BLE 桥服务端（ppclock-bridge 命令）
+"""
+__version__ = "0.1.0"
+
+from .client import PPClient
+from .transports.base import Transport, TransportError
+from .transports.bridge import BridgeError
+
+__all__ = [
+    "PPClient",
+    "Transport",
+    "TransportError",
+    "BridgeError",
+    "__version__",
+]
diff --git a/src/ppclock/client.py b/src/ppclock/client.py
new file mode 100644
index 0000000..4a2fdbe
--- /dev/null
+++ b/src/ppclock/client.py
@@ -0,0 +1,171 @@
+"""PPClient —— ppclock SDK 门面（二次开发统一入口）。
+
+用法：
+    from ppclock import PPClient
+
+    async with PPClient.via_bridge("192.168.61.35:8971", mac="18:BC:5A:5D:BF:28") as dev:
+        await dev.set_time(tz=8)
+        await dev.set_mode("clock1")
+        await dev.upload_image("photo.jpg", slot=0, algo="atkinson")
+        await dev.activate_auto()
+
+    # 本机蓝牙：
+    async with PPClient.via_local() as dev: ...
+"""
+from __future__ import annotations
+
+import datetime
+from pathlib import Path
+
+from . import commands as C
+from . import image_pipeline as IP
+from .transports import bridge as _bridge_mod
+from .transports import local as _local_mod
+from .transports.base import TransportError  # noqa: F401  (SDK 公开异常)
+from .transports.bridge import BridgeTransport, parse_via
+
+
+class PPClient:
+    """设备门面：持有一个传输层，编排 commands/image/templates/firmware 操作。"""
+
+    def __init__(self, transport):
+        self._t = transport
+
+    # ---------- 生命周期 ----------
+
+    async def __aenter__(self):
+        await self._t.__aenter__()
+        return self
+
+    async def __aexit__(self, *exc):
+        return await self._t.__aexit__(*exc)
+
+    # ---------- 工厂 ----------
+
+    @classmethod
+    def via_local(cls, mac: str | None = None, timeout: float = 10.0) -> "PPClient":
+        """本机 BLE（bleak）。mac 省略则自动扫描 NRF- 前缀设备。"""
+        return cls(_local_mod.BLETransport(address=mac, timeout=timeout))
+
+    @classmethod
+    def via_bridge(cls, host: str, mac: str | None = None,
+                   timeout: float = 10.0, port: int | None = None) -> "PPClient":
+        """经 bridge 服务端远程用他机蓝牙。host 可带 ':port'（默认 8971）。"""
+        if ":" in host and port is None:
+            host, port_s = host.rsplit(":", 1)
+            port = int(port_s)
+        return cls(BridgeTransport(address=mac, timeout=timeout,
+                                   host=host, port=port or 8971))
+
+    @classmethod
+    def via_uri(cls, uri: str, mac: str | None = None,
+                timeout: float = 10.0) -> "PPClient":
+        """'local' 或 'bridge://host[:port]'。"""
+        via = parse_via(uri)
+        if via is None:
+            return cls.via_local(mac=mac, timeout=timeout)
+        return cls.via_bridge(via[0], mac=mac, timeout=timeout, port=via[1])
+
+    @staticmethod
+    async def scan_devices(uri: str = "local", timeout: float = 5.0):
+        """按 uri 扫描：'local' → 本机；'bridge://host[:port]' → 经桥。"""
+        via = parse_via(uri)
+        if via is None:
+            return await _local_mod.scan(timeout=timeout)
+        return await _bridge_mod.scan(via[0], via[1], timeout=timeout)
+
+    # ---------- 时钟/显示 ----------
+
+    async def set_time(self, dt: datetime.datetime | None = None,
+                       tz: float | None = None):
+        """对时。tz 为小时偏移（如 8=北京），仅 dt 缺省时生效。"""
+        if dt is None and tz is not None:
+            z = datetime.timezone(datetime.timedelta(hours=tz))
+            dt = datetime.datetime.now(z).replace(tzinfo=None)
+        await C.set_time(self._t, dt)
+
+    async def set_mode(self, mode: str):
+        """clock1-3/calendar1-3/image0-3/tricolor/mono。"""
+        await C.set_mode(self._t, mode)
+
+    async def toggle(self, name: str):
+        """invert/font/rotate180/hour_format/clock_color。"""
+        await C.toggle(self._t, name)
+
+    async def clear(self):
+        await C.clear(self._t)
+
+    # ---------- 图像 ----------
+
+    async def upload_image(self, source, slot: int | None = None,
+                           algo: str = "atkinson", mono: bool = False,
+                           size: tuple[int, int] = (400, 300), **adjust) -> dict:
+        """传图。source: 路径或 PIL.Image；adjust 支持 threshold/diffusion/
+        brightness/contrast/saturation/rotate。返回字节统计。"""
+        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)
+        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,
+                "small": small, "slot": slot}
+
+    async def render_template(self, name: str, payload: dict | None = None,
+                              slot: int | None = None, algo: str = "atkinson") -> dict:
+        """本地渲染 6 模板并上传：schedule/businesscard/memo/course/qrcode/custom。"""
+        from .templates import render_template
+        img = render_template(name, payload or {})
+        bw, red = IP.process_image(img, size=(400, 300), algorithm=algo, tricolor=True)
+        await C.upload_image(self._t, bw, red, small=False, slot=slot)
+        return {"template": name, "slot": slot}
+
+    # ---------- 倒计时/文字 ----------
+
+    async def countdown(self, d: datetime.date, mode: str = "clock",
+                        prefix: str | None = None):
+        await C.countdown_set(self._t, d, mode, prefix=prefix)
+
+    async def countdown_off(self):
+        await C.countdown_off(self._t)
+
+    async def calendar_text(self, text: str):
+        """设置日历中文字（upload_rili 序列）。"""
+        await C.countdown_set(self._t, None, "calendar", prefix=text,
+                              show_calendar=True)
+
+    # ---------- 其他命令 ----------
+
+    async def parking(self, number: str):
+        await C.parking(self._t, number)
+
+    async def sleep(self, on: bool, start_h: int, end_h: int):
+        await C.sleep(self._t, on, start_h, end_h)
+
+    async def rotation(self, count: int | None = None,
+                       interval: int | None = None, unit: str = "min"):
+        await C.rotation(self._t, count=count, interval=interval, unit=unit)
+
+    async def lut(self, value: int):
+        await C.lut(self._t, value)
+
+    async def wifi(self, ssid: str, password: str, city: str = ""):
+        await C.wifi(self._t, ssid, password, city)
+
+    async def raw(self, data: bytes, channel: str = "rxtx"):
+        await C.raw(self._t, data, channel)
+
+    # ---------- 激活/固件 ----------
+
+    async def activate(self, code_hex: str):
+        await C.activate(self._t, code_hex)
+
+    async def activate_auto(self) -> dict:
+        """离线激活：EFEF 反解 MAC + 本地 keygen（无需服务器）。"""
+        return await C.activate_auto(self._t)
+
+    async def get_device_id(self) -> str:
+        return await C.get_device_id(self._t)
+
+    async def ota(self, image: bytes, on_progress=None) -> dict:
+        """SUOTA 固件升级（高危：调用方需先完成授权流程）。"""
+        return await self._t.run_ota(image, on_progress=on_progress)
diff --git a/tests/test_client.py b/tests/test_client.py
new file mode 100644
index 0000000..22e311d
--- /dev/null
+++ b/tests/test_client.py
@@ -0,0 +1,148 @@
+"""PPClient 门面 API 测试。"""
+import datetime
+
+import pytest
+
+from ppclock.client import PPClient
+
+
+class FakeTransport:
+    def __init__(self):
+        self.epd_writes = []
+        self.rxtx_writes = []
+        self.address = "AA:BB:CC:DD:EE:FF"
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def write_epd(self, data, response=True):
+        self.epd_writes.append(data)
+
+    async def write_rxtx(self, data, response=True):
+        self.rxtx_writes.append(data)
+
+    async def write_alt(self, data, response=True):
+        self.rxtx_writes.append(data)
+
+    async def read_rxtx(self):
+        return b"\x00"
+
+    async def request_device_id(self, timeout=8.0):
+        self.rxtx_writes.append(bytes([0xEF, 0xEF]))
+        return "81233F3C267112"
+
+    async def run_ota(self, image, on_progress=None):
+        return {"blocks": 1, "bytes": len(image)}
+
+    async def delay(self, seconds):
+        pass
+
+
+@pytest.fixture
+def client():
+    return PPClient(FakeTransport())
+
+
+class TestLifecycle:
+    @pytest.mark.asyncio
+    async def test_context_manager(self, client):
+        async with client as dev:
+            assert dev is client
+
+    @pytest.mark.asyncio
+    async def test_device_id(self, client):
+        assert await client.get_device_id() == "81233F3C267112"
+
+
+class TestCommands:
+    @pytest.mark.asyncio
+    async def test_set_time_with_tz(self, client):
+        await client.set_time(tz=8)
+        assert client._t.rxtx_writes[0][0] == 0xDD
+        assert client._t.rxtx_writes[-1] == bytes([0xE2])
+
+    @pytest.mark.asyncio
+    async def test_set_mode(self, client):
+        await client.set_mode("clock1")
+        assert client._t.rxtx_writes == [bytes.fromhex("e10201"), bytes([0xE2])]
+
+    @pytest.mark.asyncio
+    async def test_upload_image_pil(self, client):
+        from PIL import Image
+        img = Image.new("RGB", (400, 300), (255, 255, 255))
+        result = await client.upload_image(img, slot=0, algo="none")
+        assert result["bytes_bw"] == 15000
+        assert client._t.rxtx_writes[0] == bytes([0xEA, 0x00])
+        assert len(client._t.epd_writes) == 63 + 63 + 1
+
+    @pytest.mark.asyncio
+    async def test_render_template(self, client):
+        result = await client.render_template("custom", {"text": "你好"})
+        assert result["template"] == "custom"
+
+    @pytest.mark.asyncio
+    async def test_countdown_and_off(self, client):
+        await client.countdown(datetime.date(2026, 12, 31), prefix="目标")
+        await client.countdown_off()
+        assert client._t.rxtx_writes[-2] == bytes([0xFA])
+
+    @pytest.mark.asyncio
+    async def test_calendar_text(self, client):
+        await client.calendar_text("倒计时")
+        kinds = [w[0] for w in client._t.rxtx_writes if len(w) >= 1]
+        assert 0x99 in kinds and 0xFA in kinds
+
+    @pytest.mark.asyncio
+    async def test_activate_auto(self, client):
+        result = await client.activate_auto()
+        assert result["device_id"] == "81233F3C267112"
+        assert result["mac"] == "28bf5d5abc18"
+        # EF + keygen 码已下发
+        assert any(w[0] == 0xEF and len(w) == 7 for w in client._t.rxtx_writes)
+
+    @pytest.mark.asyncio
+    async def test_parking_sleep_lut_wifi_raw(self, client):
+        await client.parking("13800138000")
+        await client.sleep(True, 23, 7)
+        await client.lut(0x05)
+        await client.wifi("AP", "pw", "101010100")
+        await client.raw(bytes([0xE2]))
+        w = client._t.rxtx_writes
+        assert w[0][0] == 0xEF                      # 停车牌帧
+        assert w[1] == bytes([0xE2])                # 停车牌提交
+        assert w[2] == bytes([0xFB, 0x01, 0x23, 0x07])
+        assert w[3] == bytes([0xE6, 0x05])
+        assert w[4].startswith(b"WIFI_SSID:")
+        assert w[5] == bytes([0xE2])
+
+    @pytest.mark.asyncio
+    async def test_ota_path(self, client, tmp_path):
+        f = tmp_path / "fw.img"
+        f.write_bytes(b"\x70\x51" + b"\x00" * 100)
+        result = await client.ota(f.read_bytes())
+        assert result["bytes"] == 102
+
+
+class TestFactories:
+    def test_via_local_returns_client(self):
+        c = PPClient.via_local(mac="AA:BB")
+        assert isinstance(c, PPClient)
+
+    def test_via_bridge_parses_host_port(self):
+        c = PPClient.via_bridge("192.168.61.35", mac="AA:BB")
+        assert c._t.host == "192.168.61.35"
+        assert c._t.port == 8971
+        c2 = PPClient.via_bridge("192.168.61.35:9999")
+        assert c2._t.port == 9999
+
+    @pytest.mark.asyncio
+    async def test_scan_bridge(self, monkeypatch):
+        from ppclock.transports import bridge as br
+        async def fake_scan(host, port, timeout=5.0):
+            return [{"name": "NRF-X", "address": "AA", "rssi": -1}]
+        monkeypatch.setattr(br, "scan", fake_scan)
+        devices = await PPClient.scan_devices("bridge://192.168.61.35")
+        assert devices[0]["name"] == "NRF-X"