329 lines
13 KiB
Python
329 lines
13 KiB
Python
"""ppclock CLI — 离线、agent 友好（--json 机器可读输出）。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import datetime
|
||
import json
|
||
import sys
|
||
|
||
from . import commands as C
|
||
from . import image_pipeline as IP
|
||
from . import protocol as P
|
||
|
||
EXIT_OK = 0
|
||
EXIT_USAGE = 1
|
||
EXIT_TRANSPORT = 2
|
||
EXIT_ERROR = 3
|
||
|
||
|
||
def _emit(args, cmd, data=None, error=None):
|
||
if args.json:
|
||
out = {"ok": error is None, "cmd": cmd}
|
||
if data is not None:
|
||
out["data"] = data
|
||
if error is not None:
|
||
out["error"] = error
|
||
print(json.dumps(out, ensure_ascii=False))
|
||
else:
|
||
if error is not None:
|
||
print(f"错误: {error}", file=sys.stderr)
|
||
elif data is not None:
|
||
if isinstance(data, (dict, list)):
|
||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(data)
|
||
else:
|
||
print("OK")
|
||
|
||
|
||
def _make_transport(args):
|
||
"""工厂函数——测试用 monkeypatch 替换为 FakeTransport。"""
|
||
from .transport import BLETransport
|
||
return BLETransport(address=args.mac, timeout=args.timeout)
|
||
|
||
|
||
async def _run_with_transport(args, fn):
|
||
async with _make_transport(args) as t:
|
||
return await fn(t)
|
||
|
||
|
||
def _need_image(args):
|
||
from PIL import Image
|
||
img = Image.open(args.path)
|
||
size = tuple(int(v) for v in args.size.split("x")) if args.size else (400, 300)
|
||
bw, red = IP.process_image(
|
||
img, size=size, algorithm=args.algo, tricolor=not args.mono,
|
||
threshold=args.threshold, diffusion=args.diffusion / 100,
|
||
brightness=args.brightness, contrast=args.contrast,
|
||
saturation=args.saturation, rotate=args.rotate,
|
||
)
|
||
small = len(bw) in (3500, 9000)
|
||
return bw, red, small
|
||
|
||
|
||
async def _cmd_image(args, t):
|
||
bw, red, small = _need_image(args)
|
||
await C.upload_image(t, bw, red, small=small, slot=args.slot,
|
||
block_delay=args.block_delay)
|
||
return {"bytes_bw": len(bw), "bytes_red": len(red) if red else 0,
|
||
"small": small, "slot": args.slot}
|
||
|
||
|
||
async def _cmd_template(args, t):
|
||
from .templates import render_template
|
||
payload = json.loads(args.data) if args.data else {}
|
||
img = render_template(args.name, payload)
|
||
bw, red = IP.process_image(img, size=(400, 300), algorithm=args.algo,
|
||
tricolor=not args.mono)
|
||
await C.upload_image(t, bw, red, small=False, slot=args.slot,
|
||
block_delay=args.block_delay)
|
||
return {"template": args.name, "slot": args.slot}
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(prog="ppclock", description="4.2寸墨水屏离线 CLI 上位机")
|
||
p.add_argument("--mac", help="设备 MAC（省略则自动扫描 NRF- 前缀设备）")
|
||
p.add_argument("--timeout", type=float, default=10.0)
|
||
p.add_argument("--json", action="store_true", help="机器可读 JSON 输出")
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
sub.add_parser("scan", help="扫描设备")
|
||
|
||
sp = sub.add_parser("time", help="对时（默认当前本地时间）")
|
||
sp.add_argument("datetime", nargs="?", help="ISO8601，如 2026-07-24T15:30:00")
|
||
|
||
sp = sub.add_parser("mode", help="显示模式")
|
||
sp.add_argument("name", choices=sorted(P.MODES))
|
||
|
||
sp = sub.add_parser("toggle", help="切换类命令")
|
||
sp.add_argument("name", choices=["invert", "font", "rotate180", "hour_format", "clock_color"])
|
||
|
||
sub.add_parser("clear", help="刷屏（EPD 通道）")
|
||
|
||
sp = sub.add_parser("image", help="传图")
|
||
sp.add_argument("path")
|
||
sp.add_argument("--slot", type=int, choices=[0, 1, 2, 3])
|
||
sp.add_argument("--algo", default="atkinson", choices=IP.ALGORITHMS)
|
||
sp.add_argument("--mono", action="store_true", help="黑白模式（不发红平面）")
|
||
sp.add_argument("--threshold", type=int, default=125)
|
||
sp.add_argument("--diffusion", type=int, default=100, help="0–100")
|
||
sp.add_argument("--brightness", type=int, default=0)
|
||
sp.add_argument("--contrast", type=int, default=0)
|
||
sp.add_argument("--saturation", type=int, default=100)
|
||
sp.add_argument("--rotate", type=int, default=0, choices=[0, 90, 180, 270])
|
||
sp.add_argument("--size", help="WxH，默认 400x300")
|
||
sp.add_argument("--block-delay", type=float, default=0.0)
|
||
|
||
sp = sub.add_parser("template", help="本地模板渲染并上传")
|
||
sp.add_argument("name", choices=["schedule", "businesscard", "memo", "course", "qrcode", "custom"])
|
||
sp.add_argument("--data", help="模板参数 JSON")
|
||
sp.add_argument("--slot", type=int, choices=[0, 1, 2, 3])
|
||
sp.add_argument("--algo", default="atkinson", choices=IP.ALGORITHMS)
|
||
sp.add_argument("--mono", action="store_true")
|
||
sp.add_argument("--block-delay", type=float, default=0.0)
|
||
|
||
sp = sub.add_parser("countdown", help="倒计时")
|
||
sp.add_argument("date", nargs="?", help="YYYY-MM-DD")
|
||
sp.add_argument("--mode", default="clock", choices=["clock", "calendar"])
|
||
sp.add_argument("--prefix", help="前缀文字（时钟≤8单位/日历≤4字）")
|
||
sp.add_argument("--show-calendar", action="store_true",
|
||
help="设置日历中文字并切日历模式一（upload_rili 序列）")
|
||
sp.add_argument("--off", action="store_true")
|
||
|
||
sp = sub.add_parser("sleep", help="休眠时段")
|
||
sp.add_argument("--on", action="store_true")
|
||
sp.add_argument("--off", action="store_true")
|
||
sp.add_argument("--start", type=int, default=0)
|
||
sp.add_argument("--end", type=int, default=0)
|
||
|
||
sp = sub.add_parser("parking", help="停车牌号码")
|
||
sp.add_argument("number")
|
||
|
||
sp = sub.add_parser("rotation", help="轮播设置")
|
||
sp.add_argument("--count", type=int, choices=[1, 2, 3, 4])
|
||
sp.add_argument("--interval", type=int)
|
||
sp.add_argument("--unit", default="min", choices=["min", "hour"])
|
||
|
||
sp = sub.add_parser("lut", help="LUT 校准（0x01-0f 红，0x10-f0 黑）")
|
||
sp.add_argument("value", type=lambda v: int(v, 0))
|
||
|
||
sp = sub.add_parser("activate", help="激活")
|
||
sp.add_argument("code", nargs="?", help="激活码 hex")
|
||
sp.add_argument("--show-id", action="store_true", help="仅读取设备 ID")
|
||
sp.add_argument("--auto", action="store_true",
|
||
help="离线自动激活：EFEF 反解 MAC + 本地 keygen（无需服务器）")
|
||
|
||
sp = sub.add_parser("wifi", help="WiFi 配置")
|
||
sp.add_argument("ssid")
|
||
sp.add_argument("password")
|
||
sp.add_argument("--city", default="")
|
||
|
||
sp = sub.add_parser("raw", help="高级命令：直发 hex")
|
||
sp.add_argument("hex")
|
||
sp.add_argument("--channel", default="rxtx", choices=["rxtx", "epd"])
|
||
sp.add_argument("--no-response", action="store_true")
|
||
|
||
sp = sub.add_parser("ota", help="SUOTA 固件升级（高危，需确认）")
|
||
sp.add_argument("file", help="固件 .img（pQ 头格式）")
|
||
sp.add_argument("--yes", action="store_true", help="确认执行（否则仅预检）")
|
||
|
||
sp = sub.add_parser("batch", help="JSONL 批量执行（单次连接，agent 编排）")
|
||
sp.add_argument("file", help="JSONL 文件，每行 {\"argv\": [...]}；'-' 读 stdin")
|
||
sp.add_argument("--stop-on-error", action="store_true")
|
||
return p
|
||
|
||
|
||
async def _run_command(args, t):
|
||
"""在已有连接上执行单条命令（body 供 batch 复用）。"""
|
||
c = args.cmd
|
||
if c == "time":
|
||
dt = (datetime.datetime.fromisoformat(args.datetime)
|
||
if args.datetime else None)
|
||
await C.set_time(t, dt)
|
||
return {"set": (dt or datetime.datetime.now()).isoformat()}
|
||
if c == "mode":
|
||
await C.set_mode(t, args.name)
|
||
return {"mode": args.name}
|
||
if c == "toggle":
|
||
await C.toggle(t, args.name)
|
||
return {"toggle": args.name}
|
||
if c == "clear":
|
||
await C.clear(t)
|
||
return None
|
||
if c == "image":
|
||
return await _cmd_image(args, t)
|
||
if c == "template":
|
||
return await _cmd_template(args, t)
|
||
if c == "countdown":
|
||
if args.off:
|
||
await C.countdown_off(t)
|
||
return {"countdown": "off"}
|
||
if args.show_calendar:
|
||
if not args.prefix:
|
||
raise ValueError("--show-calendar 需要 --prefix 提供日历文字")
|
||
await C.countdown_set(t, None, args.mode, prefix=args.prefix,
|
||
show_calendar=True)
|
||
return {"calendar_text": args.prefix}
|
||
if not args.date:
|
||
raise ValueError("缺少日期（YYYY-MM-DD）或 --off")
|
||
d = datetime.date.fromisoformat(args.date)
|
||
await C.countdown_set(t, d, args.mode, prefix=args.prefix)
|
||
return {"date": d.isoformat(), "mode": args.mode,
|
||
"prefix": args.prefix}
|
||
if c == "sleep":
|
||
on = args.on or not args.off
|
||
await C.sleep(t, on, args.start, args.end)
|
||
return {"on": on, "start": args.start, "end": args.end}
|
||
if c == "parking":
|
||
await C.parking(t, args.number)
|
||
return {"number": args.number}
|
||
if c == "rotation":
|
||
await C.rotation(t, count=args.count, interval=args.interval, unit=args.unit)
|
||
return {"count": args.count, "interval": args.interval, "unit": args.unit}
|
||
if c == "lut":
|
||
await C.lut(t, args.value)
|
||
return {"lut": hex(args.value)}
|
||
if c == "activate":
|
||
if args.show_id:
|
||
return {"device_id": await C.get_device_id(t)}
|
||
if args.auto:
|
||
mac = None
|
||
if args.mac:
|
||
# 显示序 MAC → NVDS 字节序（LSB 在前）；不确定时优先省略 --mac 走 EFEF 反解
|
||
mac = bytes.fromhex(args.mac.replace(":", ""))[::-1]
|
||
return await C.activate_auto(t, mac)
|
||
if not args.code:
|
||
raise ValueError("缺少激活码、--auto 或 --show-id")
|
||
await C.activate(t, args.code)
|
||
return None
|
||
if c == "wifi":
|
||
await C.wifi(t, args.ssid, args.password, args.city)
|
||
return None
|
||
if c == "raw":
|
||
data = bytes.fromhex(args.hex)
|
||
if args.channel == "epd":
|
||
await t.write_epd(data, response=not args.no_response)
|
||
else:
|
||
await t.write_rxtx(data, response=not args.no_response)
|
||
return {"sent": len(data)}
|
||
if c == "ota":
|
||
from pathlib import Path
|
||
image = Path(args.file).read_bytes()
|
||
_ota_precheck(args) # 不合格直接抛 ValueError
|
||
progress_log = []
|
||
result = await t.run_ota(image, on_progress=progress_log.append)
|
||
return {"ota": "done", **result, "blocks_ok": len(progress_log)}
|
||
raise ValueError(f"未知命令 {c}")
|
||
|
||
|
||
def _ota_precheck(args) -> dict:
|
||
"""无设备场景的固件预检（--yes 才需要连接）。"""
|
||
import zlib
|
||
from pathlib import Path
|
||
image = Path(args.file).read_bytes()
|
||
if len(image) < 64 or image[0:2] != b"pQ":
|
||
raise ValueError("固件格式错误（缺 pQ 头）")
|
||
body_size = int.from_bytes(image[4:8], "little")
|
||
crc_ok = (zlib.crc32(image[64:64 + body_size]) & 0xFFFFFFFF
|
||
== int.from_bytes(image[8:12], "little"))
|
||
if not crc_ok:
|
||
raise ValueError("固件 CRC 校验失败，拒绝烧录")
|
||
return {"precheck": "ok", "body_size": body_size,
|
||
"hint": "加 --yes 执行烧录"}
|
||
|
||
|
||
async def _run_batch(args) -> list:
|
||
lines = (sys.stdin.read() if args.file == "-"
|
||
else open(args.file, encoding="utf-8").read())
|
||
results = []
|
||
parser = build_parser()
|
||
async with _make_transport(args) as t:
|
||
for lineno, line in enumerate(lines.splitlines(), 1):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
item = json.loads(line)
|
||
argv = item["argv"] if isinstance(item, dict) else item
|
||
sub_args = parser.parse_args(argv)
|
||
data = await _run_command(sub_args, t)
|
||
results.append({"line": lineno, "ok": True, "cmd": sub_args.cmd,
|
||
"data": data})
|
||
except Exception as e:
|
||
results.append({"line": lineno, "ok": False, "error": str(e)})
|
||
if args.stop_on_error:
|
||
break
|
||
return results
|
||
|
||
|
||
async def _dispatch(args) -> dict | list | None:
|
||
if args.cmd == "scan":
|
||
from .transport import scan
|
||
return {"devices": await scan(timeout=args.timeout)}
|
||
if args.cmd == "batch":
|
||
return {"results": await _run_batch(args)}
|
||
if args.cmd == "ota" and not args.yes:
|
||
return _ota_precheck(args) # 预检无需设备
|
||
return await _run_with_transport(args, lambda t: _run_command(args, t))
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
args = build_parser().parse_args(argv)
|
||
try:
|
||
data = asyncio.run(_dispatch(args))
|
||
except ValueError as e:
|
||
_emit(args, args.cmd, error=str(e))
|
||
return EXIT_USAGE
|
||
except Exception as e:
|
||
from .transport import TransportError
|
||
code = EXIT_TRANSPORT if isinstance(e, TransportError) else EXIT_ERROR
|
||
_emit(args, args.cmd, error=f"{type(e).__name__}: {e}")
|
||
return code
|
||
_emit(args, args.cmd, data=data)
|
||
return EXIT_OK
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|