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

feat: add HWPOD native workspace operations

Browse Source
This commit is contained in:
root
2026-07-21 10:08:55 +02:00
parent 692e4fcc80
commit b33430eadf
18 changed files with 602 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/hwlab-node.py
+67 -4
View File
@@ -53,7 +53,7 @@ except Exception: # pragma: no cover - import failure is reported in main.
messagebox = None
APP_VERSION = "0.1.2"
APP_VERSION = "0.1.3"
NODE_VERSION = f"{APP_VERSION}-python-gui"
CONTRACT_VERSION = "hwpod-node-ops-v1"
APP_NAME = "HWLAB Node"
@@ -357,7 +357,10 @@ class NodeOpsExecutor:
"debug.build",
"debug.download",
"debug.reset",
"io.uart.open",
"io.uart.close",
"io.uart.read",
"io.uart.write",
]
def allowed_workspace_roots(self) -> list[Path]:
@@ -427,11 +430,17 @@ class NodeOpsExecutor:
return self._command_result(op_id, name, self._cmd_run(args, default_timeout_ms=30000))
if name in ("debug.build", "debug.download", "debug.reset"):
return self._command_result(op_id, name, self._debug_command(name, args))
if name == "io.uart.open":
return self._command_result(op_id, name, self._uart_control(args, "start"))
if name == "io.uart.close":
return self._command_result(op_id, name, self._uart_control(args, "stop"))
if name == "io.uart.read":
output = self._uart_read(args)
if output.get("ok"):
return self._ok(op_id, name, output)
return self._blocked(op_id, name, output.get("blockerCode", "hwpod_uart_read_failed"), output.get("summary", "io.uart.read failed"), output.get("details", output))
if name == "io.uart.write":
return self._command_result(op_id, name, self._uart_write(args))
return self._blocked(op_id, name, "unsupported_hwpod_node_op", f"unsupported hwpod-node op: {name}")
except Exception as exc:
return self._blocked(op_id, name, getattr(exc, "errno", None) or exc.__class__.__name__, str(exc), {"trace": traceback.format_exc(limit=4)})
@@ -497,17 +506,48 @@ class NodeOpsExecutor:
def _workspace_ls(self, args: dict) -> dict:
target = self._resolve_workspace_path(args, args.get("path") or ".")
if not target.is_dir():
raise NotADirectoryError(str(args.get("path") or "."))
max_entries = max(1, min(safe_int(args.get("maxEntries"), 500), 2000))
entries = []
for child in target.iterdir():
entries.append({"name": child.name, "type": "dir" if child.is_dir() else "file" if child.is_file() else "other"})
return {"path": str(target), "entries": sorted(entries, key=lambda item: item["name"].lower())}
stat = child.stat()
entries.append({
"name": child.name,
"path": self._relative_workspace_path(args, child),
"type": "dir" if child.is_dir() else "file" if child.is_file() else "other",
"sizeBytes": stat.st_size if child.is_file() else None,
"modifiedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)),
})
entries.sort(key=lambda item: (item["type"] != "dir", item["name"].lower()))
truncated = len(entries) > max_entries
return {"path": self._relative_workspace_path(args, target), "entries": entries[:max_entries], "truncated": truncated, "totalEntries": len(entries)}
def _workspace_cat(self, args: dict) -> dict:
target = self._resolve_workspace_path(args, args.get("path"))
max_bytes = safe_int(args.get("maxBytes"), 65536)
data = target.read_bytes()
truncated = len(data) > max_bytes
return {"path": str(target), "content": data[:max_bytes].decode("utf-8", errors="replace"), "truncated": truncated}
sample = data[:max_bytes]
binary = b"\0" in sample
return {
"path": self._relative_workspace_path(args, target),
"sizeBytes": len(data),
"contentType": "application/octet-stream" if binary else "text/plain; charset=utf-8",
"previewable": not binary,
"content": None if binary else sample.decode("utf-8", errors="replace"),
"truncated": truncated,
"blocker": {"code": "hwpod_workspace_binary_preview_unsupported", "summary": "二进制文件不支持文本预览"} if binary else None,
}
def _relative_workspace_path(self, args: dict, target: Path) -> str:
root = self._workspace_root(args)
try:
relative = target.relative_to(root)
except ValueError:
raise PermissionError("workspace path is outside the selected workspace root")
text = relative.as_posix()
return text if text else "."
def _workspace_rg(self, args: dict) -> dict:
pattern = str(args.get("pattern") or "")
@@ -760,6 +800,29 @@ class NodeOpsExecutor:
return [str(item) for item in value]
return ["bun", "scripts/serial-monitor-cli.ts"]
def _uart_control(self, args: dict, action: str) -> dict:
serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
command_base = self._serial_command_base(args)
timeout_ms = safe_int(args.get("timeoutMs"), 10000)
port = str(args.get("port") or "")
baud = args.get("baudRate") or args.get("baudrate")
command = [*command_base, "monitor", action]
if action == "start":
command.extend([*(["-p", port] if port else []), *(["-b", str(baud)] if baud else [])])
output = self._spawn_output(command, serial_dir, timeout_ms)
body = self._parse_json(output.get("stdout"))
return {**output, "ok": output.get("ok") and body.get("success") is True, "bindingSource": "serial-monitor-cli", "port": port or None, "baudRate": baud, "response": body or None}
def _uart_write(self, args: dict) -> dict:
data = str(args.get("data") or "")
if not data:
return {"ok": False, "blockerCode": "hwpod_uart_data_required", "stderr": "UART data is required", "exitCode": None}
serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
command = [*self._serial_command_base(args), "send", data]
output = self._spawn_output(command, serial_dir, safe_int(args.get("timeoutMs"), 10000))
body = self._parse_json(output.get("stdout"))
return {**output, "ok": output.get("ok") and body.get("success") is True, "bindingSource": "serial-monitor-cli", "bytes": len(data.encode("utf-8")), "response": body or None}
def _uart_read(self, args: dict) -> dict:
serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
command_base = self._serial_command_base(args)