#!/usr/bin/env python3 # -*- coding: utf-8 -*- # SPEC: PJ2026-010104 AI网关; PJ2026-010405 云端控制台。 # Implementation reference: draft-2026-07-13-p0-cloud-console. """面向 Windows HWPOD 的单文件 HWLAB 图形节点。 用户入口不接受命令行参数，也不依赖 HWLAB_* 环境变量。 持久配置统一存放在 ~/.hwlab/，并由设置界面或受控部署管理。 """ from __future__ import annotations import base64 import calendar import ctypes import ctypes.wintypes as wt import hashlib import json import logging from logging.handlers import RotatingFileHandler import os from pathlib import Path import queue import random import re import shutil import socket import ssl import struct import subprocess import sys import threading import time import traceback import urllib.parse import urllib.request WIN_WORD = getattr(wt, "WORD", ctypes.c_ushort) WIN_HBITMAP = getattr(wt, "HBITMAP", wt.HANDLE) WIN_HBRUSH = getattr(wt, "HBRUSH", wt.HANDLE) WIN_HICON = getattr(wt, "HICON", wt.HANDLE) WIN_HINSTANCE = getattr(wt, "HINSTANCE", wt.HANDLE) WIN_WPARAM = ctypes.c_size_t WIN_LPARAM = ctypes.c_ssize_t WIN_LRESULT = ctypes.c_ssize_t try: import tkinter as tk from tkinter import messagebox, ttk except Exception: # pragma: no cover - import failure is reported in main. tk = None ttk = None messagebox = None APP_VERSION = "0.1.3" NODE_VERSION = f"{APP_VERSION}-python-gui" CONTRACT_VERSION = "hwpod-node-ops-v1" APP_NAME = "HWLAB Node" CONFIG_DIR = Path.home() / ".hwlab" LOG_DIR = CONFIG_DIR / "logs" UPDATE_DIR = CONFIG_DIR / "update" CONFIG_PATH = CONFIG_DIR / "config.json" STATE_PATH = CONFIG_DIR / "state.json" OPERATION_LEDGER_PATH = CONFIG_DIR / "hwpod-operations.json" LOG_PATH = LOG_DIR / "hwlab-node.log" DEFAULT_CONFIG = { "schemaVersion": 1, "serverUrl": "https://hwlab.pikapython.com", "nodeId": "node-d601-f103-v2", "autoConnect": False, "requireCredential": False, "credentialFile": str(CONFIG_DIR / "hwpod-node.token"), "allowedWorkspaceRoots": [], "defaultWorkspaceRoot": "", "update": { "enabled": True, "autoApplyWhenIdle": False, "checkIntervalSeconds": 300, "channel": "stable", "metadataUrl": "https://hwlab.pikapython.com/v1/hwlab-node/update", "downloadHostAllowlist": ["hwlab.pikapython.com"], }, "logging": { "maxUiLines": 1000, "maxFileBytes": 5 * 1024 * 1024, "redactSecrets": True, }, } SECRET_PATTERNS = [ re.compile(r"(authorization\s*:\s*bearer\s+)[^\s]+", re.I), re.compile(r"(api[_-]?key\s*[=:]\s*)[^\s,;]+", re.I), re.compile(r"(token\s*[=:]\s*)[^\s,;]+", re.I), re.compile(r"(password\s*[=:]\s*)[^\s,;]+", re.I), re.compile(r"hwl_live_[A-Za-z0-9_\-]+"), ] def ensure_dirs() -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR.mkdir(parents=True, exist_ok=True) UPDATE_DIR.mkdir(parents=True, exist_ok=True) def deep_merge(default: dict, current: dict) -> dict: result = dict(default) for key, value in current.items(): if isinstance(value, dict) and isinstance(result.get(key), dict): result[key] = deep_merge(result[key], value) else: result[key] = value return result def atomic_write_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(tmp, path) def redact(text: object) -> str: value = str(text) for pattern in SECRET_PATTERNS: value = pattern.sub(lambda match: (match.group(1) if match.groups() else "") + "", value) return value def utc_now() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def safe_int(value: object, default: int) -> int: try: number = int(value) return number if number > 0 else default except Exception: return default class ConfigStore: def __init__(self, path: Path = CONFIG_PATH): self.path = path self.config = self.load() def load(self) -> dict: ensure_dirs() if not self.path.exists(): atomic_write_json(self.path, DEFAULT_CONFIG) return deep_merge(DEFAULT_CONFIG, {}) try: loaded = json.loads(self.path.read_text(encoding="utf-8")) if not isinstance(loaded, dict): loaded = {} except Exception: backup = self.path.with_suffix(".invalid.json") try: shutil.copy2(self.path, backup) except Exception: pass loaded = {} merged = deep_merge(DEFAULT_CONFIG, loaded) if merged != loaded: atomic_write_json(self.path, merged) return merged def save(self, config: dict) -> None: self.config = deep_merge(DEFAULT_CONFIG, config) atomic_write_json(self.path, self.config) def setup_logger() -> logging.Logger: ensure_dirs() logger = logging.getLogger("hwlab-node") logger.setLevel(logging.INFO) logger.handlers.clear() handler = RotatingFileHandler(LOG_PATH, maxBytes=DEFAULT_CONFIG["logging"]["maxFileBytes"], backupCount=3, encoding="utf-8") handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) logger.addHandler(handler) return logger class DiagnosticBuffer: def __init__(self, limit: int = 200): self.limit = limit self.lock = threading.Lock() self.events: list[dict] = [] self.sequence = 0 def add(self, source: str, message: str, level: str = "ERROR", details: dict | None = None) -> dict: with self.lock: self.sequence += 1 event = { "seq": self.sequence, "level": str(level or "ERROR"), "source": str(source or "unknown")[:80], "message": redact(message)[:2000], "details": details or {}, "observedAt": utc_now(), } self.events.append(event) if len(self.events) > self.limit: self.events = self.events[-self.limit:] return dict(event) def recent(self, limit: int = 50) -> list[dict]: with self.lock: return [dict(item) for item in self.events[-max(1, min(limit, self.limit)):]] def since(self, sequence: int, limit: int = 20) -> list[dict]: with self.lock: selected = [dict(item) for item in self.events if int(item.get("seq") or 0) > sequence] return selected[:max(1, limit)] def summary(self) -> dict: with self.lock: last = dict(self.events[-1]) if self.events else None return {"count": len(self.events), "lastSeq": self.sequence, "last": last} DIAGNOSTICS = DiagnosticBuffer() _EXCEPTION_HOOKS_INSTALLED = False _EXCEPTION_HOOKS_LOCK = threading.Lock() _DIAGNOSTIC_LOGGER: logging.Logger | None = None _DIAGNOSTIC_BUS: "UiEventBus | None" = None _ORIGINAL_STDERR = sys.stderr def report_diagnostic(source: str, message: object, *, level: str = "ERROR", details: dict | None = None, logger: logging.Logger | None = None, bus: "UiEventBus | None" = None) -> dict: event = DIAGNOSTICS.add(source, str(message), level=level, details=details) target_logger = logger or _DIAGNOSTIC_LOGGER if target_logger is not None: log_level = logging.ERROR if str(level).upper() in ("ERROR", "CRITICAL") else logging.WARNING target_logger.log(log_level, "diagnostic[%s] %s", event["source"], event["message"]) target_bus = bus or _DIAGNOSTIC_BUS if target_bus is not None: target_bus.emit("log", f"诊断[{event['source']}]: {event['message']}") return event class DiagnosticStderr: def __init__(self, original, logger: logging.Logger): self.original = original self.logger = logger self._buffer = "" def write(self, text: str) -> int: if self.original is not None: self.original.write(text) self._buffer += str(text) while "\n" in self._buffer: line, self._buffer = self._buffer.split("\n", 1) if line.strip(): report_diagnostic("stderr", line.rstrip(), logger=self.logger) return len(text) def flush(self) -> None: if self.original is not None: self.original.flush() def isatty(self) -> bool: return bool(getattr(self.original, "isatty", lambda: False)()) def install_exception_hooks(logger: logging.Logger, bus: "UiEventBus | None" = None) -> None: global _EXCEPTION_HOOKS_INSTALLED, _DIAGNOSTIC_LOGGER, _DIAGNOSTIC_BUS with _EXCEPTION_HOOKS_LOCK: _DIAGNOSTIC_LOGGER = logger _DIAGNOSTIC_BUS = bus if _EXCEPTION_HOOKS_INSTALLED: return def excepthook(exc_type, exc, tb): report_diagnostic("sys.excepthook", exc, logger=logger, bus=bus, details={"trace": "".join(traceback.format_exception(exc_type, exc, tb))}) def threading_hook(args): report_diagnostic("threading.excepthook", args.exc_value, logger=logger, bus=bus, details={"thread": getattr(args.thread, "name", None), "trace": "".join(traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback))}) def unraisable_hook(args): obj = getattr(args, "object", None) report_diagnostic("sys.unraisablehook", args.exc_value, logger=logger, bus=bus, details={"object": repr(obj)[:300], "errMsg": getattr(args, "err_msg", None), "trace": "".join(traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback))}) sys.excepthook = excepthook threading.excepthook = threading_hook sys.unraisablehook = unraisable_hook sys.stderr = DiagnosticStderr(_ORIGINAL_STDERR, logger) _EXCEPTION_HOOKS_INSTALLED = True class UiEventBus: def __init__(self): self.queue: queue.Queue[tuple[str, object]] = queue.Queue() def emit(self, kind: str, payload: object = None) -> None: self.queue.put((kind, payload)) class NodeOpsExecutor: def __init__(self, config: dict, logger: logging.Logger): self.config = config self.logger = logger self.uart = None self.uart_lock = threading.Lock() @property def node_id(self) -> str: return str(self.config.get("nodeId") or socket.gethostname()) def execute(self, plan: dict) -> dict: if not isinstance(plan, dict): raise ValueError("plan must be an object") if plan.get("contractVersion") != CONTRACT_VERSION: raise ValueError(f"contractVersion must be {CONTRACT_VERSION}") ops = plan.get("ops") if not isinstance(ops, list) or not ops: raise ValueError("ops must be a non-empty array") requested_node = str(plan.get("nodeId") or self.node_id) if requested_node and requested_node != self.node_id: return self._node_mismatch(plan, requested_node) results = [self._execute_op(op if isinstance(op, dict) else {}) for op in ops] failed = any(item.get("ok") is False for item in results) return { "ok": not failed, "status": "failed" if failed else "completed", "contractVersion": CONTRACT_VERSION, "serviceId": "hwlab-node", "nodeVersion": NODE_VERSION, "nodeRole": "single-host-hwpod-node", "acceptedInput": "hwpod-node-ops", "specAuthority": "none", "planId": plan.get("planId"), "hwpodId": plan.get("hwpodId"), "nodeId": requested_node, "localNodeId": self.node_id, "results": results, "observedAt": utc_now(), } def capabilities(self) -> list[str]: return [ "hwpod-node-ops", "node.health", "node.version", "node.inventory", "node.diagnostics", "workspace.ls", "workspace.cat", "workspace.rg", "workspace.write", "workspace.replace", "workspace.insert-after", "cmd.run", "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]: values = self.config.get("allowedWorkspaceRoots") if not isinstance(values, list): return [] roots = [] for value in values: if str(value).strip(): roots.append(Path(str(value)).expanduser().resolve(strict=False)) return roots def _require_allowed_path(self, path: Path) -> Path: resolved = path.expanduser().resolve(strict=False) roots = self.allowed_workspace_roots() if not roots: return resolved resolved_key = os.path.normcase(str(resolved)) for root in roots: root_key = os.path.normcase(str(root)) try: if os.path.commonpath([resolved_key, root_key]) == root_key: return resolved except ValueError: continue raise PermissionError("workspace path is outside allowedWorkspaceRoots") def _node_mismatch(self, plan: dict, requested_node: str) -> dict: blocker = { "code": "hwpod_node_id_mismatch", "layer": "hwpod-node", "retryable": True, "summary": f"hwlab-node {self.node_id} cannot execute plan for {requested_node}", "details": {"localNodeId": self.node_id, "requestedNodeId": requested_node, "platform": sys.platform, "hostname": socket.gethostname()}, } results = [] for index, op in enumerate(plan.get("ops") if isinstance(plan.get("ops"), list) else []): results.append({"opId": str(op.get("opId") or f"op_{index + 1}"), "op": str(op.get("op") or "unknown"), "ok": False, "status": "blocked", "blocker": blocker}) return {"ok": False, "status": "blocked", "contractVersion": CONTRACT_VERSION, "serviceId": "hwlab-node", "nodeVersion": NODE_VERSION, "nodeId": requested_node, "localNodeId": self.node_id, "results": results, "blocker": blocker, "observedAt": utc_now()} def _execute_op(self, op: dict) -> dict: op_id = str(op.get("opId") or "op_unknown") name = str(op.get("op") or "") args = op.get("args") if isinstance(op.get("args"), dict) else {} try: if name == "node.health": return self._ok(op_id, name, {"platform": sys.platform, "hostname": socket.gethostname(), "cwd": os.getcwd(), "diagnostics": DIAGNOSTICS.summary()}) if name == "node.version": return self._ok(op_id, name, {"version": NODE_VERSION}) if name == "node.inventory": return self._ok(op_id, name, self._node_inventory(args)) if name == "node.diagnostics": return self._ok(op_id, name, self._node_diagnostics(args)) if name == "workspace.ls": return self._ok(op_id, name, self._workspace_ls(args)) if name == "workspace.cat": return self._ok(op_id, name, self._workspace_cat(args)) if name == "workspace.rg": return self._ok(op_id, name, self._workspace_rg(args)) if name == "workspace.write": return self._ok(op_id, name, self._workspace_write(args)) if name == "workspace.replace": return self._ok(op_id, name, self._workspace_replace(args)) if name == "workspace.insert-after": return self._ok(op_id, name, self._workspace_insert_after(args)) if name == "cmd.run": 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)}) def _ok(self, op_id: str, name: str, output: dict) -> dict: return {"opId": op_id, "op": name, "ok": True, "status": "completed", "output": output} def _failed(self, op_id: str, name: str, output: dict, code: str, summary: str) -> dict: return {"opId": op_id, "op": name, "ok": False, "status": "failed", "output": output, "blocker": {"code": code, "layer": "hwpod-node", "retryable": True, "summary": summary}} def _blocked(self, op_id: str, name: str, code: str, summary: str, details: dict | None = None) -> dict: blocker = {"code": str(code), "layer": "hwpod-node", "retryable": True, "summary": summary} if details is not None: blocker["details"] = details return {"opId": op_id, "op": name, "ok": False, "status": "blocked", "blocker": blocker} def _command_result(self, op_id: str, name: str, output: dict) -> dict: if output.get("blockerCode"): return self._blocked(op_id, name, output["blockerCode"], output.get("stderr") or "command failed before process start", {"output": output}) if output.get("ok"): return self._ok(op_id, name, output) return self._failed(op_id, name, output, "hwpod_node_command_failed", f"{name} node-side command exited with {output.get('exitCode')}") def _workspace_root(self, args: dict) -> Path: value = ( args.get("workspacePath") or args.get("workspaceRoot") or args.get("cwd") or self.config.get("defaultWorkspaceRoot") or os.getcwd() ) return self._require_allowed_path(Path(str(value))) def _resolve_workspace_path(self, args: dict, relative: object) -> Path: text = str(relative or ".") candidate = Path(text) if candidate.is_absolute() or re.match(r"^[A-Za-z]:[\\/]", text): return self._require_allowed_path(candidate) return self._require_allowed_path(self._workspace_root(args) / candidate) def _node_inventory(self, args: dict) -> dict: root = self._workspace_root(args) return { "workspaceRoot": str(root), "workspaceExists": root.exists(), "workspaceIsDirectory": root.is_dir(), "allowedWorkspaceRootCount": len(self.allowed_workspace_roots()), "capabilities": self.capabilities(), "platform": sys.platform, "hostname": socket.gethostname(), } def _node_diagnostics(self, args: dict) -> dict: max_events = safe_int(args.get("maxEvents"), 50) max_log_lines = safe_int(args.get("maxLogLines"), 80) output = {"events": DIAGNOSTICS.recent(max_events), "summary": DIAGNOSTICS.summary(), "logPath": str(LOG_PATH)} if args.get("includeLogTail") is not False: try: output["logTail"] = LOG_PATH.read_text(encoding="utf-8", errors="replace").splitlines()[-max_log_lines:] except Exception as exc: output["logTailError"] = str(exc) return output 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(): 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 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 "") if not pattern: raise ValueError("pattern is required") flags = re.I if args.get("ignoreCase") is True else 0 matcher = re.compile(pattern, flags) root = self._workspace_root(args) target = self._resolve_workspace_path(args, args.get("path") or ".") max_files = safe_int(args.get("maxFiles"), 5000) max_matches = safe_int(args.get("maxMatches"), 200) max_bytes = safe_int(args.get("maxBytesPerFile"), 1024 * 1024) matches: list[dict] = [] state = {"scannedFiles": 0, "skippedFiles": 0, "skippedDirectories": 0, "truncated": False} files = [target] if target.is_file() else (path for path in target.rglob("*") if path.is_file()) skipped_dirs = {".git", "node_modules", ".worktree", "dist", "build", "__pycache__"} for file_path in files: parts = set(file_path.parts) if parts & skipped_dirs: state["skippedDirectories"] += 1 continue if state["scannedFiles"] >= max_files or len(matches) >= max_matches: state["truncated"] = True break try: if file_path.stat().st_size > max_bytes: state["skippedFiles"] += 1 continue data = file_path.read_bytes() if b"\0" in data: state["skippedFiles"] += 1 continue lines = data.decode("utf-8", errors="replace").splitlines() except Exception: state["skippedFiles"] += 1 continue state["scannedFiles"] += 1 display = str(file_path.relative_to(root)) if file_path.is_relative_to(root) else str(file_path) for number, line in enumerate(lines, start=1): if matcher.search(line): matches.append({"path": display, "line": number, "text": line[:600]}) if len(matches) >= max_matches: state["truncated"] = True break return {"root": str(root), "path": str(target), "pattern": pattern, "matches": matches, **state} def _expected_sha_ok(self, target: Path, expected_sha: object) -> None: if not expected_sha: return current = sha256_bytes(target.read_bytes()) if target.exists() else None if current != str(expected_sha): raise ValueError(f"expectedSha mismatch for {target}: current={current} expected={expected_sha}") def _workspace_write(self, args: dict) -> dict: target = self._resolve_workspace_path(args, args.get("path")) content = str(args.get("content") or "") if args.get("lineEnding") == "crlf": content = content.replace("\r\n", "\n").replace("\r", "\n").replace("\n", "\r\n") if args.get("finalNewline") is True and not content.endswith(("\n", "\r\n")): content += "\r\n" if args.get("lineEnding") == "crlf" else "\n" before_sha = sha256_bytes(target.read_bytes()) if target.exists() else None if args.get("dryRun") is True: return {"path": str(target), "dryRun": True, "beforeSha": before_sha, "afterSha": sha256_bytes(content.encode("utf-8"))} self._expected_sha_ok(target, args.get("expectedSha")) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return {"path": str(target), "beforeSha": before_sha, "afterSha": sha256_bytes(target.read_bytes()), "bytes": target.stat().st_size} def _workspace_replace(self, args: dict) -> dict: target = self._resolve_workspace_path(args, args.get("path")) find = str(args.get("find") or "") replace_with = str(args.get("replace") or "") if not find: raise ValueError("find is required") self._expected_sha_ok(target, args.get("expectedSha")) text = target.read_text(encoding="utf-8") count = text.count(find) if count == 0: raise ValueError("find text was not found") limit = safe_int(args.get("count"), count) new_text = text.replace(find, replace_with, limit) if args.get("dryRun") is not True: target.write_text(new_text, encoding="utf-8") return {"path": str(target), "replacements": min(count, limit), "dryRun": args.get("dryRun") is True, "afterSha": sha256_bytes(new_text.encode("utf-8"))} def _workspace_insert_after(self, args: dict) -> dict: target = self._resolve_workspace_path(args, args.get("path")) anchor = str(args.get("anchor") or "") line = str(args.get("line") or "") if not anchor: raise ValueError("anchor is required") self._expected_sha_ok(target, args.get("expectedSha")) text = target.read_text(encoding="utf-8") index = text.find(anchor) if index < 0: raise ValueError("anchor was not found") insert_at = text.find("\n", index) insert_at = len(text) if insert_at < 0 else insert_at + 1 suffix = "" if line.endswith("\n") else "\n" new_text = text[:insert_at] + line + suffix + text[insert_at:] if args.get("dryRun") is not True: target.write_text(new_text, encoding="utf-8") return {"path": str(target), "anchor": anchor, "dryRun": args.get("dryRun") is True, "afterSha": sha256_bytes(new_text.encode("utf-8"))} def _cmd_run(self, args: dict, default_timeout_ms: int) -> dict: command = args.get("command") argv = args.get("argv") if isinstance(args.get("argv"), list) else [] if not command: return {"ok": False, "blockerCode": "invalid_command", "stderr": "command is required", "exitCode": None} command_binding = args.get("commandBinding") if isinstance(args.get("commandBinding"), dict) else {} if command_binding.get("nativeHelper") == "io-probe-observation": return self._io_probe_observation(args, command_binding, default_timeout_ms) output = self._spawn_output([str(command), *[str(item) for item in argv]], self._workspace_root(args), safe_int(args.get("timeoutMs"), default_timeout_ms)) observation = self._command_observation(command_binding, output) return {**output, **({"observation": observation} if observation else {})} def _command_observation(self, command_binding: dict, output: dict) -> dict: if command_binding.get("kind") not in ("io-probe", "board-comm"): return {} return self._parse_json(output.get("stdout")) def _io_probe_observation(self, args: dict, command_binding: dict, default_timeout_ms: int) -> dict: argv = args.get("argv") if isinstance(args.get("argv"), list) else [] input_data = self._parse_json(argv[1] if len(argv) > 1 else None) command_base = input_data.get("commandBase") if isinstance(input_data.get("commandBase"), list) else [] if not command_base: return {"ok": False, "blockerCode": "invalid_io_probe_binding", "stderr": "io-probe commandBase is required", "exitCode": None} count = max(1, min(safe_int(input_data.get("count"), 1), 50)) interval_ms = self._nonnegative_int(input_data.get("intervalMs")) settle_ms = self._nonnegative_int(input_data.get("settleMs")) timeout_ms = safe_int(args.get("timeoutMs"), default_timeout_ms) samples = [] response = None exit_code = 0 commands = [] if settle_ms: time.sleep(settle_ms / 1000) for index in range(count): command = self._io_probe_command(input_data, command_base) output = self._spawn_output(command, self._workspace_root(args), timeout_ms) commands.append(output) response = self._parse_json(output.get("stdout")) value = self._io_probe_value(response, str(input_data.get("valuePath") or "")) if value is not None: samples.append(value) if not output.get("ok"): exit_code = output.get("exitCode") or 1 if index < count - 1 and interval_ms: time.sleep(interval_ms / 1000) stats = {} if samples: stats = {"count": len(samples), "mean": sum(samples) / len(samples), "min": min(samples), "max": max(samples)} observation = { "observationId": f"obs_{input_data.get('probeId')}_{int(time.time() * 1000)}", "probeId": input_data.get("probeId") or command_binding.get("probeId"), "quantity": input_data.get("quantity") or command_binding.get("quantity"), "unit": input_data.get("unit") or command_binding.get("unit"), "samples": samples, "stats": stats, "response": response, "rawArtifactRef": response.get("log_path") or response.get("logPath") or response.get("artifactRef") if isinstance(response, dict) else None, } final_exit_code = exit_code if exit_code or samples else 1 return { "ok": final_exit_code == 0, "command": [str(item) for item in command_base], "cwd": str(self._workspace_root(args)), "exitCode": final_exit_code, "stdout": json.dumps(observation, ensure_ascii=False) + "\n", "stderr": "".join(str(item.get("stderr") or "") for item in commands), "observation": observation, } def _io_probe_command(self, input_data: dict, command_base: list) -> list[str]: return [ *[str(item) for item in command_base], "jrpctcp", "--host", str(input_data.get("host") or ""), "--port", str(input_data.get("port") or 8000), *(["--transport", str(input_data["transport"])] if input_data.get("transport") else []), *(["--target-node-id", str(input_data["targetNodeId"])] if input_data.get("targetNodeId") is not None else []), str(input_data.get("method") or "get"), str(input_data.get("path") or ""), *[str(item) for item in input_data.get("params", [])], ] def _io_probe_value(self, response: dict, value_path: str) -> float | None: candidates = [] nested_response = response.get("response") if isinstance(response, dict) else None if isinstance(nested_response, dict) and isinstance(nested_response.get("response"), dict): candidates.append(nested_response["response"]) if isinstance(nested_response, dict): candidates.append(nested_response) if isinstance(response, dict): candidates.append(response) keys = [item for item in value_path.removeprefix("$").removeprefix(".").split(".") if item] for candidate in candidates: value = candidate for key in keys: value = value.get(key) if isinstance(value, dict) else None try: return float(value) except (TypeError, ValueError): continue return None def _nonnegative_int(self, value: object) -> int: try: return max(0, int(value or 0)) except (TypeError, ValueError): return 0 def _debug_command(self, op: str, args: dict) -> dict: runs = args.get("commandRuns") if isinstance(args.get("commandRuns"), list) else None if runs: outputs = [] ok = True exit_code = 0 for index, run in enumerate(runs): if not isinstance(run, dict) or not run.get("command"): return {"ok": False, "blockerCode": "invalid_command_run", "stderr": f"commandRuns[{index}].command is required", "exitCode": None} argv = run.get("argv") if isinstance(run.get("argv"), list) else [] output = self._spawn_output([str(run["command"]), *[str(item) for item in argv]], self._workspace_root(args), safe_int(run.get("timeoutMs") or args.get("timeoutMs"), 120000)) outputs.append({"index": index, "command": [str(run["command"]), *[str(item) for item in argv]], **output}) ok = ok and bool(output.get("ok")) exit_code = output.get("exitCode") if not output.get("ok"): break return {"ok": ok, "cwd": str(self._workspace_root(args)), "op": op, "bindingSource": "hwpod-node-ops.commandRuns", "commands": outputs, "commandCount": len(outputs), "exitCode": exit_code, "stdout": "".join(item.get("stdout", "") for item in outputs), "stderr": "".join(item.get("stderr", "") for item in outputs)} if args.get("command"): output = self._cmd_run(args, default_timeout_ms=120000) return {"cwd": str(self._workspace_root(args)), "op": op, "bindingSource": "hwpod-node-ops.command", **output} return {"ok": False, "blockerCode": "hwpod_node_op_not_configured", "stderr": f"{op} requires command or commandRuns", "exitCode": None} def _spawn_output(self, command: list[str], cwd: Path, timeout_ms: int) -> dict: started = time.time() try: completed = subprocess.run(command, cwd=str(cwd), capture_output=True, text=True, timeout=max(1, timeout_ms / 1000), shell=False) return {"ok": completed.returncode == 0, "command": command, "cwd": str(cwd), "exitCode": completed.returncode, "stdout": completed.stdout, "stderr": completed.stderr, "elapsedMs": int((time.time() - started) * 1000)} except FileNotFoundError as exc: return {"ok": False, "blockerCode": "hwpod_node_command_spawn_failed", "command": command, "cwd": str(cwd), "exitCode": None, "stdout": "", "stderr": str(exc), "elapsedMs": int((time.time() - started) * 1000)} except subprocess.TimeoutExpired as exc: return {"ok": False, "blockerCode": "hwpod_node_command_timeout", "command": command, "cwd": str(cwd), "exitCode": None, "stdout": exc.stdout or "", "stderr": exc.stderr or "timeout", "elapsedMs": int((time.time() - started) * 1000)} def _serial_command_base(self, args: dict) -> list[str]: value = args.get("serialMonitorCommand") if isinstance(value, list) and value: return [str(item) for item in value] return ["bun", "scripts/serial-monitor-cli.ts"] def _uart_control(self, args: dict, action: str) -> dict: if str(args.get("uartBackend") or "").lower() == "pyserial": return self._pyserial_control(args, action) 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")) server_start = None if action == "start" and self._serial_server_unreachable(body): server_start = self._spawn_output([*command_base, "server", "start"], serial_dir, timeout_ms) server_body = self._parse_json(server_start.get("stdout")) server_start = {**server_start, "response": server_body or None} if server_start.get("ok") and server_body.get("success") is True: 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, "serverStart": server_start} def _serial_server_unreachable(self, body: dict) -> bool: data = body.get("data") if isinstance(body.get("data"), dict) else {} request = data.get("request") if isinstance(data.get("request"), dict) else {} diagnosis = data.get("diagnosis") if isinstance(data.get("diagnosis"), dict) else {} return body.get("success") is False and (request.get("code") == "ECONNREFUSED" or diagnosis.get("status") == "not_running") 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} if str(args.get("uartBackend") or "").lower() == "pyserial": return self._pyserial_write(args, data) 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: if str(args.get("uartBackend") or "").lower() == "pyserial": return self._pyserial_read(args) 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 args.get("resolvedPort") or args.get("targetPort") or "") baud = args.get("baudRate") or args.get("baudrate") status = self._spawn_output([*command_base, "monitor", "status"], serial_dir, timeout_ms) status_body = self._parse_json(status.get("stdout")) data = status_body.get("data") if isinstance(status_body, dict) and isinstance(status_body.get("data"), dict) else {} active_port = str(data.get("port") or "") active_baud = data.get("baudRate") monitor_start = None matches = status.get("ok") and status_body.get("success") is True and data.get("isMonitoring") is True and (not port or active_port.lower() == port.lower()) and (not baud or int(active_baud or 0) == int(baud)) if not matches: start_args = ["monitor", "start", *(["-p", port] if port else []), *(["-b", str(baud)] if baud else [])] start = self._spawn_output([*command_base, *start_args], serial_dir, timeout_ms) start_body = self._parse_json(start.get("stdout")) monitor_start = start_body if isinstance(start_body, dict) and start_body else start if not start.get("ok") or not isinstance(start_body, dict) or start_body.get("success") is not True: ports = self._spawn_output([*command_base, "ports"], serial_dir, timeout_ms) return {"ok": False, "blockerCode": "hwpod_uart_monitor_start_failed", "summary": f"serial-monitor could not start on {port or 'the requested UART'}{('/' + str(baud)) if baud else ''}", "details": {"requestedPort": port or None, "baudRate": baud, "serialMonitor": {"dir": str(serial_dir), "command": command_base, "monitorStatus": status_body or status, "start": monitor_start, "ports": self._parse_json(ports.get("stdout")) or ports}}} confirmed_status = self._spawn_output([*command_base, "monitor", "status"], serial_dir, timeout_ms) confirmed_body = self._parse_json(confirmed_status.get("stdout")) confirmed_data = confirmed_body.get("data") if isinstance(confirmed_body, dict) and isinstance(confirmed_body.get("data"), dict) else {} confirmed_port = str(confirmed_data.get("port") or "") confirmed_baud = confirmed_data.get("baudRate") confirmed_matches = confirmed_status.get("ok") and isinstance(confirmed_body, dict) and confirmed_body.get("success") is True and confirmed_data.get("isMonitoring") is True and (not port or confirmed_port.lower() == port.lower()) and (not baud or int(confirmed_baud or 0) == int(baud)) if not confirmed_matches: return {"ok": False, "blockerCode": "hwpod_uart_monitor_binding_mismatch", "summary": "serial-monitor did not confirm the requested UART binding after start", "details": {"requestedPort": port or None, "baudRate": baud, "serialMonitor": {"dir": str(serial_dir), "command": command_base, "monitorStatus": status_body or status, "start": monitor_start, "confirmedStatus": confirmed_body or confirmed_status}}} status = confirmed_status status_body = confirmed_body data = confirmed_data active_port = confirmed_port active_baud = confirmed_baud limit = max(1, min(safe_int(args.get("limit"), max(1, safe_int(args.get("maxBytes"), 4096) // 80)), 200)) fetch_args = ["fetch", "-l", str(limit), "--session-only"] if args.get("since"): fetch_args.extend(["-s", str(args["since"])]) fetch = self._spawn_output([*command_base, *fetch_args], serial_dir, timeout_ms) fetch_body = self._parse_json(fetch.get("stdout")) if not fetch.get("ok") or fetch_body.get("success") is not True: return {"ok": False, "blockerCode": "hwpod_uart_read_failed", "summary": "serial-monitor fetch failed for io.uart.read", "details": {"serialMonitor": {"dir": str(serial_dir), "command": command_base, "fetch": fetch_body or fetch}}} 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) 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 _pyserial_control(self, args: dict, action: str) -> dict: with self.uart_lock: if action == "stop": if self.uart is not None: self.uart.close() self.uart = None return {"ok": True, "bindingSource": "pyserial", "status": "closed"} import serial port = str(args.get("port") or "") baud = safe_int(args.get("baudRate") or args.get("baudrate"), 115200) if self.uart is not None and (self.uart.port != port or self.uart.baudrate != baud): self.uart.close() self.uart = None if self.uart is None or not self.uart.is_open: self.uart = serial.Serial(port=port, baudrate=baud, timeout=0) return {"ok": True, "bindingSource": "pyserial", "status": "open", "port": port, "baudRate": baud} def _pyserial_write(self, args: dict, data: str) -> dict: with self.uart_lock: if self.uart is None or not self.uart.is_open: return {"ok": False, "blockerCode": "hwpod_uart_not_open", "stderr": "UART must be opened before write", "exitCode": None} payload = data.encode(str(args.get("encoding") or "utf-8")) written = self.uart.write(payload) return {"ok": written == len(payload), "bindingSource": "pyserial", "bytes": written, "port": self.uart.port, "baudRate": self.uart.baudrate} def _pyserial_read(self, args: dict) -> dict: with self.uart_lock: if self.uart is None or not self.uart.is_open: return {"ok": False, "blockerCode": "hwpod_uart_not_open", "summary": "UART must be opened before read"} max_bytes = max(1, min(safe_int(args.get("maxBytes"), 16384), 1048576)) available = min(safe_int(getattr(self.uart, "in_waiting", 0), 0), max_bytes) payload = self.uart.read(available) if available else b"" text = payload.decode(str(args.get("encoding") or "utf-8"), errors="replace") return {"ok": True, "bindingSource": "pyserial", "port": self.uart.port, "baudRate": self.uart.baudrate, "bytes": len(payload), "text": text, "dataBase64": base64.b64encode(payload).decode("ascii"), "truncated": safe_int(getattr(self.uart, "in_waiting", 0), 0) > 0} def _parse_json(self, text: object) -> dict: try: value = json.loads(str(text or "")) return value if isinstance(value, dict) else {} except Exception: return {} class WebSocketNodeClient: def __init__(self, config: dict, executor: NodeOpsExecutor, logger: logging.Logger, bus: UiEventBus): self.config = config self.executor = executor self.logger = logger self.bus = bus self.stop_event = threading.Event() self.thread: threading.Thread | None = None self.sock: socket.socket | None = None self.manual_stop = False self.connected = False self.diagnostic_cursor = 0 self.operation_retention_seconds = 0 def start(self) -> None: if self.thread and self.thread.is_alive(): return self.manual_stop = False self.stop_event.clear() self.thread = threading.Thread(target=self._run, name="hwlab-node-ws", daemon=True) self.thread.start() def stop(self) -> None: self.manual_stop = True self.stop_event.set() self._close_socket() def _run(self) -> None: while not self.stop_event.is_set(): try: self.bus.emit("connection", "connecting") self._connect_once() except Exception as exc: self.logger.warning("websocket connection failed: %s", redact(exc)) self.bus.emit("log", f"连接失败: {redact(exc)}") finally: self.connected = False self.bus.emit("connection", "manual-disconnected" if self.manual_stop else "reconnecting") self._close_socket() if self.manual_stop or self.stop_event.is_set(): break time.sleep(3) self.bus.emit("connection", "manual-disconnected") def _connect_once(self) -> None: ws_url = self._ws_url() parsed = urllib.parse.urlparse(ws_url) port = parsed.port or (443 if parsed.scheme == "wss" else 80) host = parsed.hostname or "" raw = socket.create_connection((host, port), timeout=10) self.sock = ssl.create_default_context().wrap_socket(raw, server_hostname=host) if parsed.scheme == "wss" else raw self._handshake(parsed, host, port) self.sock.settimeout(1.0) self.bus.emit("log", f"WebSocket 已打开: {ws_url}") self._send_json({ "type": "register", "nodeId": self.executor.node_id, "name": self.executor.node_id, "capabilities": self.executor.capabilities(), "maxInFlight": 1, "runtime": { "kind": "python-tkinter-gui", "installed": True, "desktopVisible": True, "currentVersion": APP_VERSION, "updateChannel": str(self.config.get("update", {}).get("channel") or "stable"), }, "labels": { "platform": sys.platform, "hostname": socket.gethostname(), "version": NODE_VERSION, "startedAt": utc_now(), "allowedWorkspaceRootCount": len(self.executor.allowed_workspace_roots()), "diagnostics": DIAGNOSTICS.summary(), }, }) last_heartbeat = 0.0 last_diagnostic_flush = 0.0 while not self.stop_event.is_set(): if time.time() - last_heartbeat > 15: self._send_json({"type": "heartbeat", "nodeId": self.executor.node_id, "at": utc_now()}) last_heartbeat = time.time() if time.time() - last_diagnostic_flush > 5: self._flush_diagnostics() last_diagnostic_flush = time.time() try: opcode, payload = self._recv_frame() except socket.timeout: continue if opcode == 0x8: break if opcode == 0x9: self._send_frame(payload, opcode=0xA) continue if opcode != 0x1: continue self._handle_message(payload.decode("utf-8", errors="replace")) def _ws_url(self) -> str: base = str(self.config.get("serverUrl") or DEFAULT_CONFIG["serverUrl"]).rstrip("/") parsed = urllib.parse.urlparse(base) scheme = "wss" if parsed.scheme == "https" else "ws" return urllib.parse.urlunparse((scheme, parsed.netloc, "/v1/hwpod-node/ws", "", "", "")) def _handshake(self, parsed: urllib.parse.ParseResult, host: str, port: int) -> None: key = base64.b64encode(os.urandom(16)).decode("ascii") path = parsed.path or "/" if parsed.query: path += "?" + parsed.query credential = self._credential() credential_header = f"X-HWPOD-Node-Token: {credential}\r\n" if credential else "" request = ( f"GET {path} HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n" f"{credential_header}\r\n" ) assert self.sock is not None self.sock.sendall(request.encode("ascii")) response = b"" while b"\r\n\r\n" not in response: chunk = self.sock.recv(4096) if not chunk: break response += chunk if len(response) > 16384: break if b" 101 " not in response.split(b"\r\n", 1)[0]: raise RuntimeError(f"websocket upgrade failed: {response[:200]!r}") def _credential(self) -> str: path_value = str(self.config.get("credentialFile") or "").strip() required = self.config.get("requireCredential") is True if not path_value: if required: raise RuntimeError("credentialFile is required") return "" path = Path(path_value).expanduser() try: credential = path.read_text(encoding="utf-8").strip() except FileNotFoundError: credential = "" if required and not credential: raise RuntimeError("HWPOD node credential is missing") return credential def _handle_message(self, text: str) -> None: message = json.loads(text) if message.get("type") == "ack" and message.get("requestId") == "register": self.connected = message.get("ok") is True self.operation_retention_seconds = safe_int(message.get("operationRetentionSeconds"), 0) self.bus.emit("connection", "connected" if self.connected else "error") self.bus.emit("log", f"注册结果: ok={message.get('ok')} message={message.get('message')}") return if message.get("type") != "hwpod-node-ops": return request_id = str(message.get("requestId") or "req_unknown") plan = message.get("plan") or {} plan_id = str(plan.get("planId") or request_id) ledger = self._operation_ledger() existing = ledger.get(plan_id) self._send_json({"type": "hwpod-node-ops-accepted", "nodeId": self.executor.node_id, "requestId": request_id, "planId": plan_id, "observedAt": utc_now()}) if isinstance(existing, dict) and existing.get("status") == "completed" and isinstance(existing.get("result"), dict): self._send_json({"type": "hwpod-node-ops-result", "nodeId": self.executor.node_id, "requestId": request_id, "result": existing["result"], "observedAt": utc_now()}) return ledger[plan_id] = {"status": "accepted", "acceptedAt": utc_now(), "result": None} self._save_operation_ledger(ledger) try: result = self.executor.execute(plan) except Exception as exc: result = {"ok": False, "status": "failed", "results": [], "blocker": {"code": "hwpod_node_ws_run_failed", "layer": "hwpod-node", "retryable": True, "summary": str(exc)}} ledger[plan_id] = {"status": "completed", "acceptedAt": ledger[plan_id]["acceptedAt"], "completedAt": utc_now(), "result": result} self._save_operation_ledger(ledger) self._send_json({"type": "hwpod-node-ops-result", "nodeId": self.executor.node_id, "requestId": request_id, "result": result, "observedAt": utc_now()}) def _operation_ledger(self) -> dict: try: value = json.loads(OPERATION_LEDGER_PATH.read_text(encoding="utf-8")) ledger = value if isinstance(value, dict) else {} except Exception: ledger = {} if self.operation_retention_seconds > 0: cutoff = time.time() - self.operation_retention_seconds ledger = {key: value for key, value in ledger.items() if not isinstance(value, dict) or self._operation_timestamp(value) >= cutoff} return ledger def _save_operation_ledger(self, ledger: dict) -> None: atomic_write_json(OPERATION_LEDGER_PATH, ledger) def _operation_timestamp(self, value: dict) -> float: timestamp = str(value.get("completedAt") or value.get("acceptedAt") or "") try: return calendar.timegm(time.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ")) except Exception: return time.time() def _flush_diagnostics(self) -> None: for event in DIAGNOSTICS.since(self.diagnostic_cursor, 20): self._send_json({"type": "hwpod-node-diagnostic", "nodeId": self.executor.node_id, "diagnostic": event, "observedAt": utc_now()}) self.diagnostic_cursor = max(self.diagnostic_cursor, safe_int(event.get("seq"), self.diagnostic_cursor)) def _send_json(self, value: dict) -> None: self._send_frame(json.dumps(value, ensure_ascii=False).encode("utf-8"), opcode=0x1) def _send_frame(self, payload: bytes, opcode: int = 0x1) -> None: assert self.sock is not None first = 0x80 | opcode length = len(payload) mask = os.urandom(4) if length < 126: header = struct.pack("!BB", first, 0x80 | length) elif length < 65536: header = struct.pack("!BBH", first, 0x80 | 126, length) else: header = struct.pack("!BBQ", first, 0x80 | 127, length) masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)) self.sock.sendall(header + mask + masked) def _recv_exact(self, length: int) -> bytes: assert self.sock is not None chunks = [] remaining = length while remaining > 0: chunk = self.sock.recv(remaining) if not chunk: raise RuntimeError("websocket closed") chunks.append(chunk) remaining -= len(chunk) return b"".join(chunks) def _recv_frame(self) -> tuple[int, bytes]: head = self._recv_exact(2) first, second = head[0], head[1] opcode = first & 0x0F masked = bool(second & 0x80) length = second & 0x7F if length == 126: length = struct.unpack("!H", self._recv_exact(2))[0] elif length == 127: length = struct.unpack("!Q", self._recv_exact(8))[0] mask = self._recv_exact(4) if masked else b"" payload = self._recv_exact(length) if length else b"" if masked: payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)) return opcode, payload def _close_socket(self) -> None: sock, self.sock = self.sock, None if sock: try: sock.close() except Exception: pass class UpdateManager: def __init__(self, config: dict, logger: logging.Logger): self.config = config self.logger = logger def check(self) -> dict: update_config = self.config.get("update") if isinstance(self.config.get("update"), dict) else {} url = str(update_config.get("metadataUrl") or DEFAULT_CONFIG["update"]["metadataUrl"]) channel = str(update_config.get("channel") or "stable") parsed = urllib.parse.urlparse(url) query = dict(urllib.parse.parse_qsl(parsed.query)) query.update({"platform": "windows" if sys.platform == "win32" else sys.platform, "channel": channel, "current": APP_VERSION}) request_url = urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) request = urllib.request.Request(request_url, headers={"user-agent": f"hwlab-node/{APP_VERSION}"}) try: with urllib.request.urlopen(request, timeout=10) as response: if response.status == 204: return {"ok": True, "updateAvailable": False, "latestVersion": APP_VERSION, "checkedAt": utc_now()} body = json.loads(response.read().decode("utf-8")) if not isinstance(body, dict): raise ValueError("metadata response is not an object") body["checkedAt"] = utc_now() return body except Exception as exc: return {"ok": False, "updateAvailable": False, "error": str(exc), "checkedAt": utc_now()} def apply_update(self, metadata: dict) -> dict: if not metadata.get("updateAvailable"): return {"ok": False, "error": "no update available"} download_url = str(metadata.get("downloadUrl") or "") parsed = urllib.parse.urlparse(download_url) allowlist = self.config.get("update", {}).get("downloadHostAllowlist", ["hwlab.pikapython.com"]) if parsed.hostname not in allowlist or not parsed.path.endswith(".py"): return {"ok": False, "error": f"download host/path is not allowed: {parsed.hostname}{parsed.path}"} tmp = UPDATE_DIR / "hwlab-node.py.tmp" backup = UPDATE_DIR / "hwlab-node.py.bak" current = Path(__file__).resolve() try: with urllib.request.urlopen(download_url, timeout=30) as response: data = response.read() expected = metadata.get("sha256") actual = sha256_bytes(data) if expected and actual.lower() != str(expected).lower(): return {"ok": False, "error": f"sha256 mismatch: {actual}"} if b"HWLAB" not in data[:4096] or b"APP_VERSION" not in data[:4096]: return {"ok": False, "error": "downloaded file does not look like hwlab-node.py"} tmp.write_bytes(data) shutil.copy2(current, backup) os.replace(tmp, current) atomic_write_json(STATE_PATH, {"updatedAt": utc_now(), "fromVersion": APP_VERSION, "toVersion": metadata.get("latestVersion"), "backup": str(backup)}) subprocess.Popen([sys.executable, str(current)], cwd=str(current.parent), close_fds=True) return {"ok": True, "restarting": True} except Exception as exc: return {"ok": False, "error": str(exc)} class TrayIcon: def __init__(self, bus: UiEventBus, logger: logging.Logger): self.bus = bus self.logger = logger self.available = False self.connection_state = "manual-disconnected" self.update_available = False self._hwnd = None self._notify_data = None self._thread = None self._ready = threading.Event() self._wndproc_ref = None self._icon = None self._icon_cache: dict[tuple[str, bool], int] = {} def start(self) -> None: if sys.platform != "win32": return self._thread = threading.Thread(target=self._run, name="hwlab-node-tray", daemon=True) self._thread.start() self._ready.wait(2) def update(self, connection_state: str | None = None, update_available: bool | None = None) -> None: if connection_state is not None: self.connection_state = connection_state if update_available is not None: self.update_available = update_available if self.available: try: self._modify_icon() except Exception as exc: report_diagnostic("tray.update", exc, logger=self.logger, bus=self.bus, details={"trace": traceback.format_exc()}) def remove(self) -> None: if not self.available or not self._notify_data: return try: shell32 = ctypes.windll.shell32 shell32.Shell_NotifyIconW.argtypes = [wt.DWORD, ctypes.c_void_p] shell32.Shell_NotifyIconW.restype = wt.BOOL shell32.Shell_NotifyIconW(0x00000002, ctypes.byref(self._notify_data)) except Exception as exc: report_diagnostic("tray.remove", exc, logger=self.logger, bus=self.bus, details={"trace": traceback.format_exc()}) finally: self.available = False self._destroy_icon_cache() def _run(self) -> None: try: self._create_window() self._add_icon() self.available = True self._ready.set() msg = wt.MSG() user32 = ctypes.windll.user32 while user32.GetMessageW(ctypes.byref(msg), None, 0, 0) != 0: user32.TranslateMessage(ctypes.byref(msg)) user32.DispatchMessageW(ctypes.byref(msg)) except Exception as exc: report_diagnostic("tray.init", exc, logger=self.logger, bus=self.bus, details={"trace": traceback.format_exc()}) self._ready.set() def _create_window(self) -> None: user32 = ctypes.windll.user32 self._configure_win32_api() hinst = ctypes.windll.kernel32.GetModuleHandleW(None) wndproc_type = ctypes.WINFUNCTYPE(WIN_LRESULT, wt.HWND, wt.UINT, WIN_WPARAM, WIN_LPARAM) def wndproc(hwnd, msg, wparam, lparam): try: if msg == 0x0400 + 20: if int(lparam) == 0x0203: self.bus.emit("tray", "show") elif int(lparam) == 0x0205: self._show_menu() return 0 if msg == 0x0010: self.remove() user32.PostQuitMessage(0) return 0 return self._def_window_proc(user32, hwnd, msg, wparam, lparam) except Exception as exc: report_diagnostic("tray.wndproc", exc, logger=self.logger, bus=self.bus, details={"msg": int(msg), "wparam": int(wparam), "lparam": int(lparam), "trace": traceback.format_exc()}) try: return self._def_window_proc(user32, hwnd, msg, wparam, lparam) except Exception: return 0 self._wndproc_ref = wndproc_type(wndproc) class WNDCLASSW(ctypes.Structure): _fields_ = [("style", wt.UINT), ("lpfnWndProc", wndproc_type), ("cbClsExtra", ctypes.c_int), ("cbWndExtra", ctypes.c_int), ("hInstance", WIN_HINSTANCE), ("hIcon", WIN_HICON), ("hCursor", wt.HANDLE), ("hbrBackground", WIN_HBRUSH), ("lpszMenuName", wt.LPCWSTR), ("lpszClassName", wt.LPCWSTR)] class_name = "HWLABNodeTrayWindow" wc = WNDCLASSW(0, self._wndproc_ref, 0, 0, hinst, None, None, None, None, class_name) user32.RegisterClassW(ctypes.byref(wc)) self._hwnd = user32.CreateWindowExW(0, class_name, APP_NAME, 0, 0, 0, 0, 0, None, None, hinst, None) if not self._hwnd: raise ctypes.WinError() def _configure_win32_api(self) -> None: user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 shell32 = ctypes.windll.shell32 kernel32.GetModuleHandleW.argtypes = [wt.LPCWSTR] kernel32.GetModuleHandleW.restype = WIN_HINSTANCE user32.DefWindowProcW.argtypes = [wt.HWND, wt.UINT, WIN_WPARAM, WIN_LPARAM] user32.DefWindowProcW.restype = WIN_LRESULT user32.PostQuitMessage.argtypes = [ctypes.c_int] user32.GetMessageW.argtypes = [ctypes.POINTER(wt.MSG), wt.HWND, wt.UINT, wt.UINT] user32.GetMessageW.restype = ctypes.c_int user32.TranslateMessage.argtypes = [ctypes.POINTER(wt.MSG)] user32.TranslateMessage.restype = wt.BOOL user32.DispatchMessageW.argtypes = [ctypes.POINTER(wt.MSG)] user32.DispatchMessageW.restype = WIN_LRESULT user32.RegisterClassW.argtypes = [ctypes.c_void_p] user32.RegisterClassW.restype = WIN_WORD user32.CreateWindowExW.argtypes = [wt.DWORD, wt.LPCWSTR, wt.LPCWSTR, wt.DWORD, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, wt.HWND, wt.HANDLE, WIN_HINSTANCE, ctypes.c_void_p] user32.CreateWindowExW.restype = wt.HWND shell32.Shell_NotifyIconW.argtypes = [wt.DWORD, ctypes.c_void_p] shell32.Shell_NotifyIconW.restype = wt.BOOL def _def_window_proc(self, user32, hwnd, msg, wparam, lparam): return user32.DefWindowProcW(hwnd, msg, WIN_WPARAM(int(wparam)).value, WIN_LPARAM(int(lparam)).value) def _add_icon(self) -> None: self._notify_data = self._make_notify_data() if not ctypes.windll.shell32.Shell_NotifyIconW(0x00000000, ctypes.byref(self._notify_data)): raise ctypes.WinError() def _modify_icon(self) -> None: if not self._notify_data: return self._notify_data.hIcon = self._make_icon() self._icon = self._notify_data.hIcon self._notify_data.szTip = self._tooltip() if not ctypes.windll.shell32.Shell_NotifyIconW(0x00000001, ctypes.byref(self._notify_data)): raise ctypes.WinError() def _make_notify_data(self): class GUID(ctypes.Structure): _fields_ = [("Data1", wt.DWORD), ("Data2", WIN_WORD), ("Data3", WIN_WORD), ("Data4", ctypes.c_ubyte * 8)] class NOTIFYICONDATAW(ctypes.Structure): _fields_ = [("cbSize", wt.DWORD), ("hWnd", wt.HWND), ("uID", wt.UINT), ("uFlags", wt.UINT), ("uCallbackMessage", wt.UINT), ("hIcon", WIN_HICON), ("szTip", wt.WCHAR * 128), ("dwState", wt.DWORD), ("dwStateMask", wt.DWORD), ("szInfo", wt.WCHAR * 256), ("uTimeoutOrVersion", wt.UINT), ("szInfoTitle", wt.WCHAR * 64), ("dwInfoFlags", wt.DWORD), ("guidItem", GUID), ("hBalloonIcon", WIN_HICON)] data = NOTIFYICONDATAW() data.cbSize = ctypes.sizeof(NOTIFYICONDATAW) data.hWnd = self._hwnd data.uID = 1 data.uFlags = 0x00000001 | 0x00000002 | 0x00000004 data.uCallbackMessage = 0x0400 + 20 data.hIcon = self._make_icon() self._icon = data.hIcon data.szTip = self._tooltip() return data def _tooltip(self) -> str: state_map = {"connected": "已连接", "connecting": "连接中", "reconnecting": "重连中", "error": "错误"} state = state_map.get(self.connection_state, "未连接") update = "有更新" if self.update_available else "已是最新" return f"HWLAB Node - {state} - {update}"[:127] def _make_icon(self): key = (self.connection_state, bool(self.update_available)) cached = self._icon_cache.get(key) if cached: return cached try: icon = self._make_color_icon() if icon: self._icon_cache[key] = icon return icon raise RuntimeError("CreateIconIndirect returned empty icon handle") except Exception as exc: report_diagnostic("tray.icon", exc, level="WARN", logger=self.logger, bus=self.bus, details={"trace": traceback.format_exc()}) if self._icon: return self._icon user32 = ctypes.windll.user32 user32.LoadIconW.argtypes = [WIN_HINSTANCE, ctypes.c_void_p] user32.LoadIconW.restype = WIN_HICON return user32.LoadIconW(None, ctypes.c_void_p(32512)) def _destroy_icon_cache(self) -> None: if not self._icon_cache: return user32 = ctypes.windll.user32 user32.DestroyIcon.argtypes = [WIN_HICON] user32.DestroyIcon.restype = wt.BOOL for icon in set(self._icon_cache.values()): try: user32.DestroyIcon(icon) except Exception: pass self._icon_cache.clear() self._icon = None def _make_color_icon(self): size = 16 colors = { "connected": (40, 180, 80), "connecting": (60, 130, 230), "reconnecting": (230, 170, 35), "error": (220, 55, 55), } rgb = colors.get(self.connection_state, (150, 150, 150)) pixels = bytearray(size * size * 4) cx = cy = 7.5 for y in range(size): for x in range(size): i = (y * size + x) * 4 dist = ((x - cx) ** 2 + (y - cy) ** 2) ** 0.5 if dist <= 7: pixels[i:i + 4] = bytes((rgb[2], rgb[1], rgb[0], 255)) else: pixels[i:i + 4] = b"\x00\x00\x00\x00" if self.update_available: for y in range(0, 7): for x in range(9, 16): if (x - 12) ** 2 + (y - 3) ** 2 <= 12: i = (y * size + x) * 4 pixels[i:i + 4] = bytes((0, 220, 255, 255)) class BITMAPINFOHEADER(ctypes.Structure): _fields_ = [("biSize", wt.DWORD), ("biWidth", ctypes.c_long), ("biHeight", ctypes.c_long), ("biPlanes", WIN_WORD), ("biBitCount", WIN_WORD), ("biCompression", wt.DWORD), ("biSizeImage", wt.DWORD), ("biXPelsPerMeter", ctypes.c_long), ("biYPelsPerMeter", ctypes.c_long), ("biClrUsed", wt.DWORD), ("biClrImportant", wt.DWORD)] class BITMAPINFO(ctypes.Structure): _fields_ = [("bmiHeader", BITMAPINFOHEADER), ("bmiColors", wt.DWORD * 3)] class ICONINFO(ctypes.Structure): _fields_ = [("fIcon", wt.BOOL), ("xHotspot", wt.DWORD), ("yHotspot", wt.DWORD), ("hbmMask", WIN_HBITMAP), ("hbmColor", WIN_HBITMAP)] gdi32 = ctypes.windll.gdi32 user32 = ctypes.windll.user32 user32.GetDC.argtypes = [wt.HWND] user32.GetDC.restype = wt.HDC user32.ReleaseDC.argtypes = [wt.HWND, wt.HDC] user32.ReleaseDC.restype = ctypes.c_int user32.CreateIconIndirect.argtypes = [ctypes.POINTER(ICONINFO)] user32.CreateIconIndirect.restype = WIN_HICON gdi32.CreateDIBSection.argtypes = [wt.HDC, ctypes.c_void_p, wt.UINT, ctypes.POINTER(ctypes.c_void_p), wt.HANDLE, wt.DWORD] gdi32.CreateDIBSection.restype = WIN_HBITMAP gdi32.CreateBitmap.argtypes = [ctypes.c_int, ctypes.c_int, wt.UINT, wt.UINT, ctypes.c_void_p] gdi32.CreateBitmap.restype = WIN_HBITMAP gdi32.DeleteObject.argtypes = [wt.HANDLE] gdi32.DeleteObject.restype = wt.BOOL hdc = user32.GetDC(None) if not hdc: raise ctypes.WinError() bits = ctypes.c_void_p() bmi = BITMAPINFO() bmi.bmiHeader.biSize = ctypes.sizeof(BITMAPINFOHEADER) bmi.bmiHeader.biWidth = size bmi.bmiHeader.biHeight = -size bmi.bmiHeader.biPlanes = 1 bmi.bmiHeader.biBitCount = 32 bmi.bmiHeader.biCompression = 0 color = None mask = None try: color = gdi32.CreateDIBSection(hdc, ctypes.byref(bmi), 0, ctypes.byref(bits), None, 0) if not color or not bits.value: raise ctypes.WinError() ctypes.memmove(bits.value, bytes(pixels), len(pixels)) mask_bits = (ctypes.c_ubyte * ((size * size) // 8))() mask = gdi32.CreateBitmap(size, size, 1, 1, mask_bits) if not mask: raise ctypes.WinError() icon_info = ICONINFO(True, 0, 0, mask, color) icon = user32.CreateIconIndirect(ctypes.byref(icon_info)) if not icon: raise ctypes.WinError() return icon finally: if color: gdi32.DeleteObject(color) if mask: gdi32.DeleteObject(mask) user32.ReleaseDC(None, hdc) def _show_menu(self) -> None: user32 = ctypes.windll.user32 menu = user32.CreatePopupMenu() items = [(1001, "打开"), (1002, "断开" if self.connection_state == "connected" else "连接"), (1003, "检查更新"), (1004, "退出")] for item_id, text in items: user32.AppendMenuW(menu, 0, item_id, text) point = wt.POINT() user32.GetCursorPos(ctypes.byref(point)) user32.SetForegroundWindow(self._hwnd) command = user32.TrackPopupMenu(menu, 0x0100, point.x, point.y, 0, self._hwnd, None) ctypes.windll.user32.DestroyMenu(menu) actions = {1001: "show", 1002: "toggle", 1003: "check_update", 1004: "exit"} if command in actions: self.bus.emit("tray", actions[command]) class SettingsDialog: def __init__(self, parent, config: dict, on_save): self.parent = parent self.config = json.loads(json.dumps(config)) self.on_save = on_save self.window = tk.Toplevel(parent) self.window.title("HWLAB Node 设置") self.window.resizable(False, False) self.server_var = tk.StringVar(value=self.config.get("serverUrl")) self.node_var = tk.StringVar(value=self.config.get("nodeId")) self.auto_connect_var = tk.BooleanVar(value=bool(self.config.get("autoConnect"))) update = self.config.get("update", {}) self.update_enabled_var = tk.BooleanVar(value=bool(update.get("enabled", True))) self.auto_update_var = tk.BooleanVar(value=bool(update.get("autoApplyWhenIdle", False))) self.interval_var = tk.StringVar(value=str(update.get("checkIntervalSeconds", 300))) self.channel_var = tk.StringVar(value=str(update.get("channel", "stable"))) self._build() self.window.transient(parent) self.window.grab_set() def _build(self) -> None: frame = ttk.Frame(self.window, padding=12) frame.grid(row=0, column=0, sticky="nsew") labels = [("服务器 URL", self.server_var), ("节点 ID", self.node_var), ("更新频道", self.channel_var), ("检查间隔秒", self.interval_var)] for row, (label, var) in enumerate(labels): ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", pady=4) ttk.Entry(frame, textvariable=var, width=46).grid(row=row, column=1, sticky="ew", pady=4) ttk.Checkbutton(frame, text="启动后自动连接", variable=self.auto_connect_var).grid(row=4, column=0, columnspan=2, sticky="w", pady=4) ttk.Checkbutton(frame, text="启用更新检查", variable=self.update_enabled_var).grid(row=5, column=0, columnspan=2, sticky="w", pady=4) ttk.Checkbutton(frame, text="自动更新（空闲时）", variable=self.auto_update_var).grid(row=6, column=0, columnspan=2, sticky="w", pady=4) buttons = ttk.Frame(frame) buttons.grid(row=7, column=0, columnspan=2, sticky="e", pady=(12, 0)) ttk.Button(buttons, text="保存", command=self._save).grid(row=0, column=0, padx=4) ttk.Button(buttons, text="取消", command=self.window.destroy).grid(row=0, column=1, padx=4) def _save(self) -> None: update = self.config.setdefault("update", {}) self.config["serverUrl"] = self.server_var.get().strip() or DEFAULT_CONFIG["serverUrl"] self.config["nodeId"] = self.node_var.get().strip() or DEFAULT_CONFIG["nodeId"] self.config["autoConnect"] = bool(self.auto_connect_var.get()) update["enabled"] = bool(self.update_enabled_var.get()) update["autoApplyWhenIdle"] = bool(self.auto_update_var.get()) update["checkIntervalSeconds"] = safe_int(self.interval_var.get(), 300) update["channel"] = self.channel_var.get().strip() or "stable" self.on_save(self.config) self.window.destroy() class HwlabNodeApp: def __init__(self): if tk is None: raise RuntimeError("tkinter is not available") ensure_dirs() self.store = ConfigStore() self.config = self.store.config self.logger = setup_logger() self.bus = UiEventBus() install_exception_hooks(self.logger, self.bus) self.executor = NodeOpsExecutor(self.config, self.logger) self.client: WebSocketNodeClient | None = None self.update_manager = UpdateManager(self.config, self.logger) self.update_metadata: dict | None = None self.connection_state = "manual-disconnected" self.update_available = False self.root = tk.Tk() self.root.title(APP_NAME) self.root.geometry("720x460") self.status_var = tk.StringVar(value="未连接") self.server_var = tk.StringVar(value=self._display_server()) self.version_var = tk.StringVar(value=f"v{APP_VERSION}") self._build_ui() self.tray = TrayIcon(self.bus, self.logger) self.tray.start() self.root.protocol("WM_DELETE_WINDOW", self._hide_to_tray) self.root.after(150, self._poll_events) self.root.after(800, self._startup_actions) def _build_ui(self) -> None: outer = ttk.Frame(self.root, padding=12) outer.pack(fill="both", expand=True) top = ttk.Frame(outer) top.pack(fill="x") ttk.Label(top, text=APP_NAME, font=("Segoe UI", 16, "bold")).grid(row=0, column=0, sticky="w") ttk.Label(top, textvariable=self.version_var).grid(row=0, column=1, sticky="w", padx=10) ttk.Label(top, text="状态:").grid(row=1, column=0, sticky="w", pady=(8, 0)) ttk.Label(top, textvariable=self.status_var).grid(row=1, column=1, sticky="w", pady=(8, 0)) ttk.Label(top, text="服务器:").grid(row=2, column=0, sticky="w") ttk.Label(top, textvariable=self.server_var).grid(row=2, column=1, sticky="w") buttons = ttk.Frame(outer) buttons.pack(fill="x", pady=10) self.connect_button = ttk.Button(buttons, text="连接", command=self.connect) self.connect_button.pack(side="left", padx=(0, 6)) self.disconnect_button = ttk.Button(buttons, text="断开", command=self.disconnect) self.disconnect_button.pack(side="left", padx=6) ttk.Button(buttons, text="设置", command=self.show_settings).pack(side="left", padx=6) ttk.Button(buttons, text="检查更新", command=self.check_update).pack(side="left", padx=6) self.update_button = ttk.Button(buttons, text="更新", command=self.apply_update) self.update_button.pack(side="left", padx=6) self.update_button.state(["disabled"]) self.log_text = tk.Text(outer, height=18, wrap="word", state="disabled") self.log_text.pack(fill="both", expand=True) self._append_log(f"{APP_NAME} v{APP_VERSION} 已启动，配置目录: {CONFIG_DIR}") def _startup_actions(self) -> None: if self.config.get("autoConnect") is True: self.connect() if self.config.get("update", {}).get("enabled", True): self.root.after(1000, self.check_update) self._schedule_update_timer() def _display_server(self) -> str: parsed = urllib.parse.urlparse(str(self.config.get("serverUrl") or "")) return parsed.netloc or str(self.config.get("serverUrl") or "") def connect(self) -> None: if self.client and self.client.thread and self.client.thread.is_alive(): return self.executor = NodeOpsExecutor(self.config, self.logger) self.client = WebSocketNodeClient(self.config, self.executor, self.logger, self.bus) self.client.start() def disconnect(self) -> None: if self.client: self.client.stop() self._set_connection("manual-disconnected") def show_settings(self) -> None: SettingsDialog(self.root, self.config, self._save_settings) def _save_settings(self, config: dict) -> None: self.store.save(config) self.config = self.store.config self.server_var.set(self._display_server()) self.update_manager = UpdateManager(self.config, self.logger) self._append_log("设置已保存") def check_update(self) -> None: self._append_log("正在检查更新...") threading.Thread(target=self._check_update_worker, daemon=True).start() def _check_update_worker(self) -> None: result = self.update_manager.check() self.bus.emit("update_checked", result) def apply_update(self) -> None: if not self.update_metadata: return self._append_log("正在应用更新...") threading.Thread(target=self._apply_update_worker, daemon=True).start() def _apply_update_worker(self) -> None: result = self.update_manager.apply_update(self.update_metadata or {}) self.bus.emit("update_applied", result) def _schedule_update_timer(self) -> None: interval = safe_int(self.config.get("update", {}).get("checkIntervalSeconds"), 300) self.root.after(interval * 1000, self._scheduled_update_check) def _scheduled_update_check(self) -> None: if self.config.get("update", {}).get("enabled", True): self.check_update() self._schedule_update_timer() def _poll_events(self) -> None: while True: try: kind, payload = self.bus.queue.get_nowait() except queue.Empty: break if kind == "log": self._append_log(str(payload)) elif kind == "connection": self._set_connection(str(payload)) elif kind == "update_checked": self._handle_update_checked(payload if isinstance(payload, dict) else {}) elif kind == "update_applied": self._handle_update_applied(payload if isinstance(payload, dict) else {}) elif kind == "tray": self._handle_tray(str(payload)) self.root.after(150, self._poll_events) def _set_connection(self, state: str) -> None: self.connection_state = state labels = {"manual-disconnected": "未连接", "connecting": "连接中", "connected": "已连接", "reconnecting": "重连中", "error": "错误"} self.status_var.set(labels.get(state, state)) self.tray.update(connection_state=state, update_available=self.update_available) def _handle_update_checked(self, result: dict) -> None: if not result.get("ok", True) and result.get("error"): self._append_log(f"更新检查失败: {redact(result.get('error'))}") return self.update_available = result.get("updateAvailable") is True self.update_metadata = result if self.update_available else None self.tray.update(update_available=self.update_available) if self.update_available: self.update_button.state(["!disabled"]) self._append_log(f"发现新版本: {result.get('latestVersion')}") if self.config.get("update", {}).get("autoApplyWhenIdle") is True and self.connection_state in ("manual-disconnected", "error"): self.apply_update() else: self.update_button.state(["disabled"]) self._append_log("当前已是最新版本") def _handle_update_applied(self, result: dict) -> None: if result.get("ok"): self._append_log("更新已应用，正在重启") self.root.after(300, self.quit) else: self._append_log(f"更新失败: {redact(result.get('error'))}") def _handle_tray(self, action: str) -> None: if action == "show": self.root.deiconify() self.root.lift() elif action == "toggle": if self.connection_state == "connected": self.disconnect() else: self.connect() elif action == "check_update": self.check_update() elif action == "exit": self.quit() def _append_log(self, text: str) -> None: line = f"[{time.strftime('%H:%M:%S')}] {redact(text)}" self.logger.info(redact(text)) self.log_text.configure(state="normal") self.log_text.insert("end", line + "\n") max_lines = safe_int(self.config.get("logging", {}).get("maxUiLines"), 1000) current_lines = int(self.log_text.index("end-1c").split(".")[0]) if current_lines > max_lines: self.log_text.delete("1.0", f"{current_lines - max_lines}.0") self.log_text.see("end") self.log_text.configure(state="disabled") def _hide_to_tray(self) -> None: if self.tray.available: self.root.withdraw() self._append_log("已最小化到托盘") else: self.root.iconify() self._append_log("托盘不可用，已最小化到任务栏") def quit(self) -> None: if self.client: self.client.stop() self.tray.remove() atomic_write_json(STATE_PATH, {"lastExitAt": utc_now(), "connectionState": self.connection_state, "updateAvailable": self.update_available}) self.root.destroy() def run(self) -> None: self.root.mainloop() def main() -> int: ensure_dirs() logger = setup_logger() install_exception_hooks(logger) try: app = HwlabNodeApp() app.run() return 0 except Exception as exc: report_diagnostic("startup.fatal", exc, logger=logger, details={"trace": traceback.format_exc()}) logger.error("fatal startup error: %s\n%s", redact(exc), traceback.format_exc()) if messagebox is not None: try: messagebox.showerror(APP_NAME, f"启动失败: {exc}\n日志: {LOG_PATH}") except Exception: pass return 1 if __name__ == "__main__": sys.exit(main())