mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 补齐 CaseRun UART 权威观测

Browse Source
This commit is contained in:
root
2026-07-16 19:49:17 +02:00
parent af4962e138
commit a9570ceedf
3 changed files with 81 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/hwlab-node.py
+8 -1
View File
@@ -803,7 +803,14 @@ class NodeOpsExecutor:
rows = fetch_body.get("data") if isinstance(fetch_body.get("data"), list) else []
text = "\n".join(str(row.get("data")) for row in rows if isinstance(row, dict) and row.get("data"))
max_bytes = safe_int(args.get("maxBytes"), 16384)
return {"ok": True, "bindingSource": "serial-monitor-cli", "serialMonitorDir": str(serial_dir), "requestedPort": port or None, "resolvedPort": active_port or port or None, "baudRate": baud or active_baud, "command": [*command_base, *fetch_args], "data": rows, "count": fetch_body.get("count", len(rows)), "totalCount": fetch_body.get("totalCount"), "hasMore": fetch_body.get("hasMore", False), "text": text[:max_bytes], "truncated": len(text) > max_bytes or fetch_body.get("truncated") is True, "monitorStatus": status_body, "monitorStarted": not matches, "monitorStart": monitor_start, "sourceFile": fetch_body.get("sourceFile")}
truncated_text = text[:max_bytes]
resolved_port = active_port or port or None
resolved_baud = baud or active_baud
source_file = str(fetch_body.get("sourceFile") or "")
observation = {"observationId": f"uart:{resolved_port or 'unknown'}:{source_file or 'session'}", "port": resolved_port, "baudRate": resolved_baud, "data": rows, "text": truncated_text, "truncated": len(text) > max_bytes or fetch_body.get("truncated") is True}
if source_file:
observation["rawArtifactRef"] = source_file
return {"ok": True, "workspacePath": str(self._workspace_root(args)), "bindingSource": "serial-monitor-cli", "serialMonitorDir": str(serial_dir), "requestedPort": port or None, "resolvedPort": resolved_port, "baudRate": resolved_baud, "command": [*command_base, *fetch_args], "data": rows, "count": fetch_body.get("count", len(rows)), "totalCount": fetch_body.get("totalCount"), "hasMore": fetch_body.get("hasMore", False), "text": truncated_text, "truncated": len(text) > max_bytes or fetch_body.get("truncated") is True, "monitorStatus": status_body, "monitorStarted": not matches, "monitorStart": monitor_start, "sourceFile": source_file or None, "observation": observation}
def _parse_json(self, text: object) -> dict:
try: