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