fix(screen): streamline runtime display states
This commit is contained in:
admin
@@ -45,12 +45,14 @@ class ProductTestClient:
|
||||
baud: int,
|
||||
timeout: float,
|
||||
echo_logs: bool = False,
|
||||
verbose: bool = False,
|
||||
jsonl_path: Optional[Path] = None,
|
||||
) -> None:
|
||||
self.port = port
|
||||
self.baud = baud
|
||||
self.timeout = timeout
|
||||
self.echo_logs = echo_logs
|
||||
self.verbose = verbose
|
||||
self.jsonl_path = jsonl_path
|
||||
self.seq = 1
|
||||
self.serial = serial.Serial(
|
||||
@@ -204,8 +206,9 @@ class ProductTestClient:
|
||||
raise ProtocolError(f"{command} 执行期间设备重启，等待结果失败 seq={seq}")
|
||||
continue
|
||||
if frame.token == TOKEN_PROGRESS:
|
||||
stage = frame.payload.get("stage", "-")
|
||||
print(f" - {command}: {stage}")
|
||||
if self.verbose:
|
||||
stage = frame.payload.get("stage", "-")
|
||||
print(f" - {command}: {stage}")
|
||||
continue
|
||||
if frame.token == TOKEN_RESULT:
|
||||
return frame.payload
|
||||
@@ -230,7 +233,8 @@ def default_steps(args: argparse.Namespace) -> List[Dict[str, Any]]:
|
||||
{"cmd": "PING", "timeout": 10},
|
||||
{"cmd": "STATUS", "timeout": 10},
|
||||
{"cmd": "WAIT_WIFI", "payload": {"timeout_ms": args.wifi_timeout_ms}, "timeout": args.wifi_timeout_ms / 1000 + 5},
|
||||
{"cmd": "SCREEN", "payload": {"text": "TQ Printer\nProduct Test\nScreen OK"}, "timeout": 15},
|
||||
{"cmd": "SCREEN", "payload": {"text": "小Q打印机\n屏幕测试\n显示正常"}, "timeout": 15},
|
||||
{"cmd": "SCREEN_FLOW", "payload": {"step_ms": 400}, "timeout": 20},
|
||||
{"cmd": "PRINTER_STATUS", "timeout": 10},
|
||||
{"cmd": "AUDIO_PROBE", "payload": {"samples": 320, "timeout_ms": 3000}, "timeout": 15},
|
||||
{"cmd": "VOICE_SESSION", "payload": {"timeout_ms": args.voice_timeout_ms}, "timeout": args.voice_timeout_ms / 1000 + 10},
|
||||
@@ -321,30 +325,25 @@ def _printer_status_summary(result: Dict[str, Any]) -> str:
|
||||
temp_max = printer.get("temperature_max_c")
|
||||
battery = printer.get("battery_percent")
|
||||
has_paper = printer.get("has_paper")
|
||||
ntc_raw = printer.get("ntc_adc_raw")
|
||||
|
||||
parts: List[str] = []
|
||||
if has_paper is not None:
|
||||
parts.append(f"paper={'ok' if has_paper else 'missing'}")
|
||||
parts.append("paper=ok" if has_paper else "paper=missing")
|
||||
if battery is not None:
|
||||
parts.append(f"batt={battery}%")
|
||||
if temp is not None:
|
||||
if temp_min is not None and temp_max is not None:
|
||||
parts.append(f"temp={float(temp):.1f}C[{temp_min}-{temp_max}]")
|
||||
else:
|
||||
parts.append(f"temp={float(temp):.1f}C")
|
||||
if ntc_raw is not None:
|
||||
parts.append(f"ntc_raw={ntc_raw}")
|
||||
parts.append(f"temp={float(temp):.1f}C")
|
||||
return " printer={" + " ".join(parts) + "}" if parts else ""
|
||||
|
||||
|
||||
def summarize_result(result: Dict[str, Any]) -> str:
|
||||
status = result.get("status", "-")
|
||||
rc = result.get("rc", "-")
|
||||
duration = result.get("duration_ms", "-")
|
||||
message = result.get("message", "")
|
||||
suffix = f" msg={message}" if message else ""
|
||||
return f"status={status} rc={rc} duration_ms={duration}{suffix}{_printer_status_summary(result)}"
|
||||
if status == "ok" and rc == 0 and not suffix:
|
||||
return f"ok{_printer_status_summary(result)}"
|
||||
return f"status={status} rc={rc}{suffix}{_printer_status_summary(result)}"
|
||||
|
||||
|
||||
def run_steps(
|
||||
@@ -357,19 +356,18 @@ def run_steps(
|
||||
cmd = str(step["cmd"])
|
||||
payload = step.get("payload") or {}
|
||||
timeout = float(step.get("timeout", client.timeout))
|
||||
print(f"[{index}] {cmd}")
|
||||
try:
|
||||
result = client.command(cmd, payload=payload, timeout_s=timeout)
|
||||
except ProtocolError as exc:
|
||||
failures += 1
|
||||
print(f" FAIL {exc}")
|
||||
print(f"[{index}] {cmd}: FAIL {exc}")
|
||||
if not continue_on_fail:
|
||||
break
|
||||
continue
|
||||
|
||||
status = result.get("status")
|
||||
ok = status == "ok"
|
||||
print(f" {'OK' if ok else 'FAIL'} {summarize_result(result)}")
|
||||
print(f"[{index}] {cmd}: {'OK' if ok else 'FAIL'} {summarize_result(result)}")
|
||||
if not ok:
|
||||
failures += 1
|
||||
if not continue_on_fail:
|
||||
@@ -388,6 +386,7 @@ def parse_args(argv: List[str]) -> argparse.Namespace:
|
||||
parser.add_argument("--reset-delay", type=float, default=2.0, help="硬复位后等待设备启动的秒数")
|
||||
parser.add_argument("--jsonl", type=Path, help="保存收发帧 JSONL")
|
||||
parser.add_argument("--echo-logs", action="store_true", help="打印非协议串口日志")
|
||||
parser.add_argument("--verbose", action="store_true", help="显示每条命令的中间阶段")
|
||||
parser.add_argument("--continue-on-fail", action="store_true", help="失败后继续执行后续步骤")
|
||||
parser.add_argument("--plan", type=Path, help="从 JSON 测试计划读取步骤")
|
||||
|
||||
@@ -424,17 +423,18 @@ def main(argv: List[str]) -> int:
|
||||
baud=args.baud,
|
||||
timeout=args.timeout,
|
||||
echo_logs=args.echo_logs,
|
||||
verbose=args.verbose,
|
||||
jsonl_path=args.jsonl,
|
||||
) as client:
|
||||
if args.reset:
|
||||
print(f"复位设备: {args.port}")
|
||||
if args.verbose:
|
||||
print(f"复位设备: {args.port}")
|
||||
client.hard_reset(args.reset_delay)
|
||||
print(f"等待设备协议就绪: {args.port} @ {args.baud}")
|
||||
ready = client.wait_ready(args.ready_timeout)
|
||||
print(
|
||||
"READY "
|
||||
+ json.dumps(ready, ensure_ascii=False, separators=(",", ":"))
|
||||
)
|
||||
if args.verbose:
|
||||
print("READY " + json.dumps(ready, ensure_ascii=False, separators=(",", ":")))
|
||||
else:
|
||||
print("设备就绪")
|
||||
failures = run_steps(client, steps, args.continue_on_fail)
|
||||
|
||||
if failures:
|
||||
|
||||
Reference in New Issue
Block a user