134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""高层命令：帧序列编排。协议规范 docs/protocol.md。"""
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
|
||
from . import protocol as P
|
||
|
||
|
||
async def set_time(t, dt: datetime.datetime | None = None):
|
||
dt = dt or datetime.datetime.now()
|
||
await t.write_rxtx(P.build_set_time(dt))
|
||
await t.delay(0.3)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def set_mode(t, mode: str):
|
||
"""模式切换：setri 语义——命令后 300ms 补 e2（ble.js:3956-3962）。"""
|
||
await t.write_rxtx(P.mode_frame(mode))
|
||
await t.delay(0.3)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def toggle(t, name: str):
|
||
"""单字节切换类命令（invert/font/rotate180/hour_format/clock_color），补 e2。"""
|
||
await t.write_rxtx(P.simple_command(name))
|
||
await t.delay(0.3)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def clear(t, param: int = 0x00):
|
||
for frame in P.clear_frames(param):
|
||
await t.write_epd(frame)
|
||
|
||
|
||
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。"""
|
||
if slot is not None:
|
||
await t.write_rxtx(P.slot_select_frame(slot))
|
||
await t.delay(0.3)
|
||
for frame in P.image_blocks(bw, small=small, red=False):
|
||
await t.write_epd(frame)
|
||
if block_delay:
|
||
await t.delay(block_delay)
|
||
if red is not None:
|
||
for frame in P.image_blocks(red, small=small, red=True):
|
||
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))
|
||
|
||
|
||
async def countdown_set(t, d: datetime.date, mode: str, prefix: str | None = None,
|
||
show_calendar: bool = False):
|
||
"""倒计时：前缀文字位图（可选）→ 日期帧 → E2。
|
||
show_calendar=True 走 upload_rili 序列：990* → FA → e10103 → E2（timedjs2.js:308-312）。"""
|
||
from . import textbitmap as TB
|
||
if prefix:
|
||
for frame in TB.prefix_frames(prefix, "calendar" if show_calendar else mode):
|
||
await t.write_rxtx(frame)
|
||
await t.delay(1.0)
|
||
if show_calendar:
|
||
await t.write_rxtx(bytes([0xFA]))
|
||
await t.delay(1.0)
|
||
await t.write_rxtx(P.mode_frame("calendar1"))
|
||
await t.delay(1.0)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
return
|
||
await t.write_rxtx(P.countdown_date_frame(d, mode))
|
||
await t.delay(1.0)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def countdown_off(t):
|
||
for frame in P.countdown_off_frames():
|
||
await t.write_rxtx(frame)
|
||
await t.delay(1.0)
|
||
|
||
|
||
async def parking(t, number: str):
|
||
await t.write_rxtx(P.parking_frame(number))
|
||
await t.delay(0.5)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def sleep(t, on: bool, start_h: int, end_h: int):
|
||
await t.write_rxtx(P.sleep_frame(on, start_h, end_h))
|
||
|
||
|
||
async def rotation(t, count: int | None = None, interval: int | None = None,
|
||
unit: str = "min"):
|
||
if count is not None:
|
||
await t.write_rxtx(P.rotation_count_frame(count))
|
||
if interval is not None:
|
||
await t.write_rxtx(P.rotation_interval_frame(interval, unit))
|
||
|
||
|
||
async def activate(t, code_hex: str):
|
||
await t.write_rxtx(P.activation_frame(code_hex))
|
||
await t.delay(0.4)
|
||
await t.write_rxtx(P.simple_command("commit"))
|
||
|
||
|
||
async def activate_auto(t, mac: bytes | None = None) -> dict:
|
||
"""离线自动激活（firmware-analysis §4.3）：MAC→本地 keygen→EF 下发。
|
||
mac=None 时经 EFEF 设备 ID 反解（字节序无歧义）；返回 {device_id, mac, code}。"""
|
||
device_id = None
|
||
if mac is None:
|
||
device_id = await t.request_device_id()
|
||
mac = P.recover_mac_from_device_id(device_id)
|
||
code = P.keygen_code(mac)
|
||
await activate(t, code.hex())
|
||
return {"device_id": device_id, "mac": mac.hex(), "code": code.hex()}
|
||
|
||
|
||
async def get_device_id(t) -> str:
|
||
return await t.request_device_id()
|
||
|
||
|
||
async def wifi(t, ssid: str, password: str, city: str = ""):
|
||
await t.write_rxtx(P.wifi_frame(ssid, password, city))
|
||
|
||
|
||
async def lut(t, value: int):
|
||
await t.write_rxtx(P.lut_frame(value))
|
||
|
||
|
||
async def raw(t, data: bytes, channel: str = "rxtx"):
|
||
if channel == "epd":
|
||
await t.write_epd(data)
|
||
else:
|
||
await t.write_rxtx(data)
|