feat(ppclock): protocol.py 协议编解码 23 测试全绿(BCD语义经JS证据校正)
This commit is contained in:
1 parent
190182fb2e
commit
6f5b0c60be
11 files changed
+32843
No files matched your search
@@ -0,0 +1,164 @@
|
||||
"""BLE 协议编解码(纯函数,无 BLE 依赖)。规范:docs/protocol.md;证据:analysis/web/protocol-raw.md。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
from typing import Iterator
|
||||
|
||||
# GATT(docs/protocol.md §1)
|
||||
EPD_SERVICE_UUID = "13187b10-eba9-a3ba-044e-83d3217d9a38"
|
||||
EPD_CHAR_UUID = "4b646063-6264-f3a7-8941-e65356ea82fe"
|
||||
RXTX_SERVICE_UUID = "00001f10-0000-1000-8000-00805f9b34fb"
|
||||
RXTX_CHAR_UUID = "00001f1f-0000-1000-8000-00805f9b34fb"
|
||||
DEVICE_NAME_PREFIX = "NRF-"
|
||||
|
||||
BLOCK_PAYLOAD = 240 # my_step=480 hex 字符 = 240 数据字节
|
||||
COMMIT = 0xE2
|
||||
|
||||
MODES = {
|
||||
**{f"image{i}": bytes([0xE1, 0x00, 0x00 + i]) for i in range(4)},
|
||||
"calendar1": bytes.fromhex("e10103"),
|
||||
"calendar2": bytes.fromhex("e10104"),
|
||||
"calendar3": bytes.fromhex("e10105"),
|
||||
"clock1": bytes.fromhex("e10201"),
|
||||
"clock2": bytes.fromhex("e10202"),
|
||||
"clock3": bytes.fromhex("e10203"),
|
||||
"tricolor": bytes.fromhex("2300"),
|
||||
"mono": bytes.fromhex("2301"),
|
||||
}
|
||||
|
||||
SIMPLE_COMMANDS = {
|
||||
"hour_format": bytes([0x14]),
|
||||
"clock_color": bytes([0x15]),
|
||||
"invert": bytes([0xE3]),
|
||||
"font": bytes([0xE4]),
|
||||
"rotate180": bytes([0xE5]),
|
||||
"commit": bytes([COMMIT]),
|
||||
}
|
||||
|
||||
|
||||
def image_blocks(plane: bytes, *, small: bool, red: bool) -> Iterator[bytes]:
|
||||
"""平面字节流 → 图像块帧迭代器。cod: 03整图/04小图;code: ff=bw 00=红。"""
|
||||
cod = 0x04 if small else 0x03
|
||||
code = 0x00 if red else 0xFF
|
||||
for offset in range(0, len(plane), BLOCK_PAYLOAD):
|
||||
chunk = plane[offset : offset + BLOCK_PAYLOAD]
|
||||
yield bytes([cod, code]) + offset.to_bytes(2, "big") + chunk
|
||||
|
||||
|
||||
def end_frame(*, small: bool) -> bytes:
|
||||
return bytes([0xAA if small else 0x01])
|
||||
|
||||
|
||||
def clear_frames(param: int = 0x00) -> list[bytes]:
|
||||
return [bytes([0x00, param]), bytes([0x01])]
|
||||
|
||||
|
||||
def build_set_time(dt: datetime.datetime) -> bytes:
|
||||
"""dd + unix(4B,本地墙钟按UTC计) + year(2B BE) + month + day + week(周一=1)。"""
|
||||
unix_local = int(dt.replace(tzinfo=datetime.timezone.utc).timestamp())
|
||||
return (
|
||||
bytes([0xDD])
|
||||
+ unix_local.to_bytes(4, "big")
|
||||
+ dt.year.to_bytes(2, "big")
|
||||
+ bytes([dt.month, dt.day, dt.isoweekday()])
|
||||
)
|
||||
|
||||
|
||||
def parking_frame(number: str) -> bytes:
|
||||
"""停车牌:每位数字双写,'12' → EF 11 22。"""
|
||||
if not number.isdigit() or not number:
|
||||
raise ValueError("停车牌号码必须为纯数字")
|
||||
return bytes([0xEF]) + bytes(int(f"{d}{d}", 16) for d in number)
|
||||
|
||||
|
||||
def rotation_interval_frame(n: int, unit: str) -> bytes:
|
||||
"""e901 + n + 01(分钟)/00(小时)。"""
|
||||
if not 0 <= n <= 0xFF:
|
||||
raise ValueError("轮播间隔超出范围")
|
||||
if unit not in ("min", "hour"):
|
||||
raise ValueError("unit 必须是 min 或 hour")
|
||||
return bytes([0xE9, 0x01, n, 0x01 if unit == "min" else 0x00])
|
||||
|
||||
|
||||
def slot_select_frame(slot: int) -> bytes:
|
||||
if not 0 <= slot <= 3:
|
||||
raise ValueError("槽位 0–3")
|
||||
return bytes([0xEA, slot])
|
||||
|
||||
|
||||
def rotation_count_frame(count: int) -> bytes:
|
||||
if not 1 <= count <= 4:
|
||||
raise ValueError("轮播数量 1–4")
|
||||
return bytes([0xEA, 0x00, 0x00 + (count - 1)])
|
||||
|
||||
|
||||
def _bcd(v: int) -> int:
|
||||
"""十进制值 → BCD 字节(JS 中 '0'+n 十进制直拼后经 hexToBytes 的语义,
|
||||
证据 timedjs2.js:242-243、ble.js:4013-4015)。如 12 → 0x12。"""
|
||||
if not 0 <= v <= 99:
|
||||
raise ValueError("BCD 值范围 0–99")
|
||||
return int(f"{v:02d}", 16)
|
||||
|
||||
|
||||
def countdown_date_frame(d: datetime.date, mode: str) -> bytes:
|
||||
"""ff00(时钟)/ffff(日历) + 年份4位十进制BCD + 月(BCD) + 日(BCD)。"""
|
||||
if mode == "clock":
|
||||
head = bytes([0xFF, 0x00])
|
||||
elif mode == "calendar":
|
||||
head = bytes([0xFF, 0xFF])
|
||||
else:
|
||||
raise ValueError("mode 必须是 clock 或 calendar")
|
||||
y = f"{d.year:04d}"
|
||||
return head + bytes([int(y[:2], 16), int(y[2:], 16), _bcd(d.month), _bcd(d.day)])
|
||||
|
||||
|
||||
def countdown_off_frames() -> list[bytes]:
|
||||
return [bytes([0xFA]), bytes([COMMIT])]
|
||||
|
||||
|
||||
def activation_frame(code_hex: str) -> bytes:
|
||||
return bytes([0xEF]) + bytes.fromhex(code_hex.upper())
|
||||
|
||||
|
||||
def request_id_frame() -> bytes:
|
||||
return bytes([0xEF, 0xEF])
|
||||
|
||||
|
||||
def lut_frame(value: int) -> bytes:
|
||||
"""E6: 01–0f 设红,10–f0 设黑;0 无效。"""
|
||||
if not 0x01 <= value <= 0xF0:
|
||||
raise ValueError("LUT 值范围 0x01–0xF0")
|
||||
return bytes([0xE6, value])
|
||||
|
||||
|
||||
def sleep_frame(on: bool, start_h: int, end_h: int) -> bytes:
|
||||
"""FB01/FB00 + 开始小时(BCD) + 结束小时(BCD)(证据 ble.js:4013-4015 十进制直拼)。"""
|
||||
for h in (start_h, end_h):
|
||||
if not 0 <= h <= 23:
|
||||
raise ValueError("小时范围 0–23")
|
||||
return bytes([0xFB, 0x01 if on else 0x00, _bcd(start_h), _bcd(end_h)])
|
||||
|
||||
|
||||
def mode_frame(mode: str) -> bytes:
|
||||
try:
|
||||
return MODES[mode]
|
||||
except KeyError:
|
||||
raise ValueError(f"未知模式 {mode!r},可选:{sorted(MODES)}") from None
|
||||
|
||||
|
||||
def simple_command(name: str) -> bytes:
|
||||
try:
|
||||
return SIMPLE_COMMANDS[name]
|
||||
except KeyError:
|
||||
raise ValueError(f"未知命令 {name!r},可选:{sorted(SIMPLE_COMMANDS)}") from None
|
||||
|
||||
|
||||
def wifi_frame(ssid: str, password: str, city: str) -> bytes:
|
||||
return f"WIFI_SSID:{ssid};WIFI_PASS:{password};CITY:{city}".encode("utf-8")
|
||||
|
||||
|
||||
def parse_device_id(data: bytes) -> str:
|
||||
"""14 字节设备 ID notify 应答 → 字符串(可打印即 ASCII,否则大写 hex)。"""
|
||||
if all(0x20 <= b < 0x7F for b in data):
|
||||
return data.decode("ascii")
|
||||
return data.hex().upper()
|
||||
Reference in new issue
Block a user