Serial

{}

{
  "usb_serial_chips": [
    {"vid": "1A86", "pid": "7523", "name": "CH340"},
    {"vid": "1A86", "pid": "55D4", "name": "CH9102"},
    {"vid": "10C4", "pid": "EA60", "name": "CP2102"},
    {"vid": "10C4", "pid": "EA70", "name": "CP2105"},
    {"vid": "0403", "pid": "6001", "name": "FT232R"},
    {"vid": "0403", "pid": "6010", "name": "FT2232"},
    {"vid": "0403", "pid": "6014", "name": "FT232H"},
    {"vid": "067B", "pid": "2303", "name": "PL2303"},
    {"vid": "067B", "pid": "23A3", "name": "PL2303GS"},
    {"vid": "2341", "pid": "0043", "name": "Arduino Uno"},
    {"vid": "2341", "pid": "0001", "name": "Arduino Mega"},
    {"vid": "1366", "pid": "0105", "name": "SEGGER J-Link (CDC)"},
    {"vid": "0D28", "pid": "0204", "name": "DAPLink (CDC)"},
    {"vid": "303A", "pid": "1001", "name": "ESP32-S2 (CDC)"},
    {"vid": "303A", "pid": "4001", "name": "ESP32-S3 (CDC)"}
  ]
}

"""串口 Hex Dump 查看"""

import argparse
import json
import signal
import sys
import time
from datetime import datetime
from pathlib import Path

from serial_runtime import (
    get_serial_config,
    open_serial_port,
    save_project_config,
    update_state_entry,
)

PARITY_MAP = {"none": "N", "even": "E", "odd": "O", "mark": "M", "space": "S"}
IDLE_FLUSH_SEC = 0.2


def output_json(obj):
    sys.stdout.buffer.write(json.dumps(obj, ensure_ascii=False).encode("utf-8"))
    sys.stdout.buffer.write(b"\n")
    sys.stdout.buffer.flush()


def error_exit(code, message, use_json):
    result = {"status": "error", "action": "hex", "error": {"code": code, "message": message}}
    if use_json:
        output_json(result)
    else:
        print(f"错误: {message}", file=sys.stderr)
    sys.exit(1)


def hex_dump_line(data, offset, width, show_ascii):
    """格式化一行 hex dump"""
    hex_part = " ".join(f"{b:02X}" for b in data)
    hex_part = hex_part.ljust(width * 3 - 1)
    line = f"{offset:08X}  {hex_part}"
    if show_ascii:
        ascii_part = "".join(chr(b) if 0x20 <= b < 0x7F else "." for b in data)
        line += f"  |{ascii_part}|"
    return line


def emit_chunk(data, offset, width, show_ascii, use_json):
    now = datetime.now().isoformat(timespec="milliseconds")
    if use_json:
        ascii_str = "".join(chr(b) if 0x20 <= b < 0x7F else "." for b in data)
        output_json({
            "timestamp": now,
            "offset": offset,
            "length": len(data),
            "hex": data.hex(" "),
            "ascii": ascii_str,
        })
    else:
        print(hex_dump_line(data, offset, width, show_ascii))


def main():
    parser = argparse.ArgumentParser(description="串口 Hex Dump 查看")
    parser.add_argument("--port", help="串口号 (如 COM3)")
    parser.add_argument("--baudrate", type=int, help="波特率")
    parser.add_argument("--bytesize", type=int, help="数据位")
    parser.add_argument("--parity", help="校验位 (none/even/odd)")
    parser.add_argument("--stopbits", type=int, help="停止位")
    parser.add_argument("--encoding", help="编码")
    parser.add_argument("--width", type=int, default=16, help="每行字节数")
    parser.add_argument("--timeout", type=float, default=0, help="超时秒数，0=无限")
    parser.add_argument("--no-ascii", action="store_true", help="不显示 ASCII 列")
    parser.add_argument("--direct", action="store_true", help="直连真实串口，跳过 mux")
    parser.add_argument("--json", action="store_true", help="JSON Lines 输出")
    args = parser.parse_args()

    start_time = time.time()

    # 获取配置
    cfg, sources = get_serial_config(
        cli_port=args.port,
        cli_baudrate=args.baudrate,
        cli_bytesize=args.bytesize,
        cli_parity=args.parity,
        cli_stopbits=args.stopbits,
        cli_encoding=args.encoding,
    )

    if cfg is None:
        if sources.get("need_selection"):
            error_exit("multiple_candidates", f"{sources['error']}，请用 --port 指定", args.json)
        else:
            error_exit("config_error", sources.get("error", "配置错误"), args.json)

    # 保存确认的配置
    save_project_config(values={
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "bytesize": cfg["bytesize"],
        "parity": cfg["parity"],
        "stopbits": cfg["stopbits"],
        "encoding": cfg["encoding"],
    })

    try:
        use_mux = not args.direct
        ser = open_serial_port(cfg, use_mux=use_mux)
        if getattr(ser, "_serial_skill_using_mux", False):
            print("[mux] 已通过多路复用连接，请避免在 minicom 中同时写入以免串口数据冲突", file=sys.stderr)
        ser.timeout = 0.1
    except Exception as e:
        error_exit("connect_failed", str(e), args.json)

    total_bytes = 0
    offset = 0
    running = True
    show_ascii = not args.no_ascii
    buffer = bytearray()
    last_data_at = 0.0

    def on_signal(sig, frame):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)

    try:
        while running:
            if args.timeout > 0 and (time.time() - start_time) >= args.timeout:
                break

            read_size = max(1, getattr(ser, "in_waiting", 0) or 1)
            data = ser.read(read_size)
            if not data:
                if buffer and last_data_at and (time.time() - last_data_at) >= IDLE_FLUSH_SEC:
                    chunk = bytes(buffer)
                    emit_chunk(chunk, offset, args.width, show_ascii, args.json)
                    total_bytes += len(chunk)
                    offset += len(chunk)
                    buffer.clear()
                continue

            buffer.extend(data)
            last_data_at = time.time()

            while len(buffer) >= args.width:
                chunk = bytes(buffer[:args.width])
                del buffer[:args.width]
                emit_chunk(chunk, offset, args.width, show_ascii, args.json)
                total_bytes += len(chunk)
                offset += len(chunk)

    except Exception as e:
        error_exit("read_error", str(e), args.json)
    finally:
        if buffer:
            chunk = bytes(buffer)
            emit_chunk(chunk, offset, args.width, show_ascii, args.json)
            total_bytes += len(chunk)
        ser.close()

    duration = round(time.time() - start_time, 1)
    summary = f"Hex 查看结束，共 {total_bytes} 字节，耗时 {duration}s\n"
    sys.stderr.buffer.write(summary.encode("utf-8"))
    sys.stderr.buffer.flush()

    # 更新状态
    update_state_entry("last_observe", {
        "type": "serial_hex",
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "bytes_received": total_bytes,
        "duration_sec": duration,
    })


if __name__ == "__main__":
    main()

"""串口日志记录"""

import argparse
import json
import os
import signal
import sys
import time
from datetime import datetime
from pathlib import Path

from serial_runtime import (
    get_serial_config,
    open_serial_port,
    save_project_config,
    update_state_entry,
    normalize_path,
)

PARITY_MAP = {"none": "N", "even": "E", "odd": "O", "mark": "M", "space": "S"}


def output_json(obj):
    sys.stdout.buffer.write(json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8"))
    sys.stdout.buffer.write(b"\n")
    sys.stdout.buffer.flush()


def error_exit(code, message, use_json):
    result = {"status": "error", "action": "log", "error": {"code": code, "message": message}}
    if use_json:
        output_json(result)
    else:
        print(f"错误: {message}", file=sys.stderr)
    sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="串口日志记录")
    parser.add_argument("--port", help="串口号 (如 COM3)")
    parser.add_argument("--baudrate", type=int, help="波特率")
    parser.add_argument("--bytesize", type=int, help="数据位")
    parser.add_argument("--parity", help="校验位 (none/even/odd)")
    parser.add_argument("--stopbits", type=int, help="停止位")
    parser.add_argument("--encoding", help="编码")
    parser.add_argument("--output", "-o", help="输出文件路径")
    parser.add_argument("--timestamp", action="store_true", help="每行加时间戳")
    parser.add_argument("--max-size", type=float, default=0, help="最大文件大小(MB)，0=无限")
    parser.add_argument("--duration", type=float, default=0, help="记录时长(秒)，0=无限")
    parser.add_argument("--format", choices=["text", "csv", "json"], default="text", help="输出格式")
    parser.add_argument("--console", action="store_true", help="同时输出到控制台(stderr)")
    parser.add_argument("--direct", action="store_true", help="直连真实串口，跳过 mux")
    parser.add_argument("--json", action="store_true", help="最终输出 summary JSON")
    args = parser.parse_args()

    start_time = time.time()

    # 获取配置
    cfg, sources = get_serial_config(
        cli_port=args.port,
        cli_baudrate=args.baudrate,
        cli_bytesize=args.bytesize,
        cli_parity=args.parity,
        cli_stopbits=args.stopbits,
        cli_encoding=args.encoding,
    )

    if cfg is None:
        if sources.get("need_selection"):
            error_exit("multiple_candidates", f"{sources['error']}，请用 --port 指定", args.json)
        else:
            error_exit("config_error", sources.get("error", "配置错误"), args.json)

    # 保存确认的配置
    save_project_config(values={
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "bytesize": cfg["bytesize"],
        "parity": cfg["parity"],
        "stopbits": cfg["stopbits"],
        "encoding": cfg["encoding"],
    })

    log_dir = cfg["log_dir"]
    if not args.output:
        os.makedirs(log_dir, exist_ok=True)
        ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
        ext = {"text": "log", "csv": "csv", "json": "jsonl"}[args.format]
        args.output = os.path.join(log_dir, f"serial_{ts}.{ext}")

    try:
        use_mux = not args.direct
        ser = open_serial_port(cfg, use_mux=use_mux)
        if getattr(ser, "_serial_skill_using_mux", False):
            print("[mux] 已通过多路复用连接，请避免在 minicom 中同时写入以免串口数据冲突", file=sys.stderr)
    except Exception as e:
        error_exit("connect_failed", str(e), args.json)

    line_count = 0
    byte_count = 0
    running = True
    encoding = cfg["encoding"]
    max_bytes = int(args.max_size * 1024 * 1024) if args.max_size > 0 else 0

    def on_signal(sig, frame):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)

    os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)

    try:
        with open(args.output, "w", encoding="utf-8", newline="") as f:
            if args.format == "csv":
                f.write("timestamp,text\n")

            while running:
                if args.duration > 0 and (time.time() - start_time) >= args.duration:
                    break
                if max_bytes > 0 and byte_count >= max_bytes:
                    break

                raw = ser.readline()
                if not raw:
                    continue

                try:
                    text = raw.decode(encoding, errors="replace").rstrip("\r\n")
                except Exception:
                    text = raw.hex()

                now = datetime.now().isoformat(timespec="milliseconds")
                line_count += 1

                if args.format == "text":
                    line = f"[{now}] {text}\n" if args.timestamp else f"{text}\n"
                elif args.format == "csv":
                    escaped = text.replace('"', '""')
                    line = f'{now},"{escaped}"\n'
                else:
                    line = json.dumps({"timestamp": now, "text": text}, ensure_ascii=False) + "\n"

                f.write(line)
                byte_count += len(line.encode("utf-8"))

                if args.console:
                    prefix = f"[{now}] " if args.timestamp else ""
                    print(f"{prefix}{text}", file=sys.stderr)

    except Exception as e:
        error_exit("write_error", str(e), args.json)
    finally:
        ser.close()

    duration = round(time.time() - start_time, 1)
    result = {
        "status": "ok",
        "action": "log",
        "summary": {
            "file": os.path.abspath(args.output),
            "lines": line_count,
            "bytes": byte_count,
            "duration_sec": duration,
            "format": args.format,
        },
    }

    if args.json:
        output_json(result)
    else:
        print(f"\n日志已保存: {os.path.abspath(args.output)}")
        print(f"  共 {line_count} 行, {byte_count} 字节, 耗时 {duration}s")

    # 更新状态
    update_state_entry("last_observe", {
        "type": "serial_log",
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "file": os.path.abspath(args.output),
        "lines": line_count,
        "duration_sec": duration,
    })


if __name__ == "__main__":
    main()

"""串口实时文本监控"""

import argparse
import json
import re
import signal
import sys
import time
from datetime import datetime
from pathlib import Path

from serial_runtime import (
    get_serial_config,
    open_serial_port,
    save_project_config,
    update_state_entry,
    make_timing,
)

PARITY_MAP = {"none": "N", "even": "E", "odd": "O", "mark": "M", "space": "S"}
IDLE_FLUSH_SEC = 0.2


def output_json(obj):
    sys.stdout.buffer.write(json.dumps(obj, ensure_ascii=False).encode("utf-8"))
    sys.stdout.buffer.write(b"\n")
    sys.stdout.buffer.flush()


def error_exit(action, code, message, use_json):
    result = {"status": "error", "action": action, "error": {"code": code, "message": message}}
    if use_json:
        output_json(result)
    else:
        print(f"错误: {message}", file=sys.stderr)
    sys.exit(1)


def emit_line(text, cfg, args, include_re, exclude_re):
    if include_re:
        try:
            if not include_re.search(text):
                return False
        except Exception:
            pass

    if exclude_re:
        try:
            if exclude_re.search(text):
                return False
        except Exception:
            pass

    now = datetime.now().isoformat(timespec="milliseconds")
    if args.json:
        output_json({"timestamp": now, "port": cfg["port"], "baudrate": cfg["baudrate"], "text": text})
    else:
        prefix = f"[{now}] " if args.timestamp else ""
        print(f"{prefix}{text}")
    return True


def main():
    parser = argparse.ArgumentParser(description="串口实时文本监控")
    parser.add_argument("--port", help="串口号 (如 COM3)")
    parser.add_argument("--baudrate", type=int, help="波特率")
    parser.add_argument("--bytesize", type=int, help="数据位")
    parser.add_argument("--parity", help="校验位 (none/even/odd)")
    parser.add_argument("--stopbits", type=int, help="停止位")
    parser.add_argument("--encoding", help="编码")
    parser.add_argument("--timestamp", action="store_true", help="显示时间戳")
    parser.add_argument("--filter", help="正则过滤（仅显示匹配行）")
    parser.add_argument("--exclude", help="正则排除（隐藏匹配行）")
    parser.add_argument("--timeout", type=float, default=0, help="超时秒数，0=无限")
    parser.add_argument("--direct", action="store_true", help="直连真实串口，跳过 mux")
    parser.add_argument("--json", action="store_true", help="JSON Lines 输出")
    args = parser.parse_args()

    start_time = time.time()

    # 获取配置
    cfg, sources = get_serial_config(
        cli_port=args.port,
        cli_baudrate=args.baudrate,
        cli_bytesize=args.bytesize,
        cli_parity=args.parity,
        cli_stopbits=args.stopbits,
        cli_encoding=args.encoding,
    )

    if cfg is None:
        if sources.get("need_selection"):
            # 多候选情况
            error_exit("monitor", "multiple_candidates", f"{sources['error']}，请用 --port 指定", args.json)
        else:
            error_exit("monitor", "config_error", sources.get("error", "配置错误"), args.json)

    # 保存确认的配置
    save_project_config(values={
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "bytesize": cfg["bytesize"],
        "parity": cfg["parity"],
        "stopbits": cfg["stopbits"],
        "encoding": cfg["encoding"],
    })

    include_re = None
    exclude_re = None
    if args.filter:
        try:
            include_re = re.compile(args.filter)
        except re.error:
            error_exit("monitor", "bad_regex", f"无效正则: {args.filter}", args.json)
    if args.exclude:
        try:
            exclude_re = re.compile(args.exclude)
        except re.error:
            error_exit("monitor", "bad_regex", f"无效正则: {args.exclude}", args.json)

    try:
        use_mux = not args.direct
        ser = open_serial_port(cfg, use_mux=use_mux)
        if getattr(ser, "_serial_skill_using_mux", False):
            print("[mux] 已通过多路复用连接，请避免在 minicom 中同时写入以免串口数据冲突", file=sys.stderr)
        ser.timeout = 0.1
    except Exception as e:
        error_exit("monitor", "connect_failed", str(e), args.json)

    line_count = 0
    running = True

    def on_signal(sig, frame):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)

    encoding = cfg.get("encoding", "utf-8")
    text_buffer = ""
    last_data_at = 0.0
    skip_leading_lf = False

    try:
        while running:
            if args.timeout > 0 and (time.time() - start_time) >= args.timeout:
                break

            read_size = max(1, getattr(ser, "in_waiting", 0) or 1)
            raw = ser.read(read_size)
            if not raw:
                if text_buffer and last_data_at and (time.time() - last_data_at) >= IDLE_FLUSH_SEC:
                    if emit_line(text_buffer, cfg, args, include_re, exclude_re):
                        line_count += 1
                    text_buffer = ""
                continue

            try:
                chunk = raw.decode(encoding, errors="replace")
            except Exception:
                chunk = raw.hex()
            if skip_leading_lf and chunk.startswith("\n"):
                chunk = chunk[1:]
            skip_leading_lf = chunk.endswith("\r")
            text_buffer += chunk.replace("\r\n", "\n").replace("\r", "\n")
            last_data_at = time.time()

            parts = text_buffer.split("\n")
            if text_buffer.endswith("\n"):
                complete_lines = parts[:-1]
                text_buffer = ""
            else:
                complete_lines = parts[:-1]
                text_buffer = parts[-1]

            for text in complete_lines:
                if emit_line(text, cfg, args, include_re, exclude_re):
                    line_count += 1

    except Exception as e:
        error_exit("monitor", "read_error", str(e), args.json)
    finally:
        if text_buffer:
            if emit_line(text_buffer, cfg, args, include_re, exclude_re):
                line_count += 1
        ser.close()

    duration = round(time.time() - start_time, 1)
    summary = f"监控结束，共 {line_count} 行，耗时 {duration}s\n"
    sys.stderr.buffer.write(summary.encode("utf-8"))
    sys.stderr.buffer.flush()

    # 更新状态
    update_state_entry("last_observe", {
        "type": "serial_monitor",
        "port": cfg["port"],
        "baudrate": cfg["baudrate"],
        "lines": line_count,
        "duration_sec": duration,
    })


if __name__ == "__main__":
    main()

"""Serial 多路复用管理 — 单串口读者 + TCP 广播 + 虚拟 PTY"""

from __future__ import annotations

import argparse
import os
import shutil
import signal
import socket
import subprocess
import sys
import threading
import time
from pathlib import Path

from serial_runtime import (
    get_serial_config,
    load_workspace_state,
    save_workspace_state,
    save_project_config,
    is_missing,
    make_result,
    output_json,
)

DEFAULT_MUX_PORT = 20001
DEFAULT_VSERIAL_LINK = "/tmp/serial_mux_vserial"
STATE_KEY = "serial_mux"


def find_free_port(start: int = DEFAULT_MUX_PORT) -> int:
    """从 start 开始找空闲 TCP 端口"""
    for offset in range(100):
        port = start + offset
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            try:
                s.bind(("127.0.0.1", port))
                return port
            except OSError:
                continue
    return start


def wait_for_tcp_server(port: int, process: subprocess.Popen, timeout: float = 2.0) -> bool:
    """等待后台 mux TCP 服务可连接。"""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if process.poll() is not None:
            return False
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.2):
                return True
        except OSError:
            time.sleep(0.05)
    return process.poll() is None


class SerialMuxServer:
    """单进程打开真实串口，并把 RX 广播给所有 TCP 客户端。"""

    def __init__(self, config: dict, tcp_port: int):
        self.config = config
        self.tcp_port = tcp_port
        self.clients: set[socket.socket] = set()
        self.clients_lock = threading.Lock()
        self.serial_lock = threading.Lock()
        self.stop_event = threading.Event()
        self.server_sock: socket.socket | None = None
        self.serial_port = None

    def run(self) -> int:
        try:
            import serial

            parity_map = {"none": "N", "even": "E", "odd": "O", "mark": "M", "space": "S"}
            self.serial_port = serial.Serial(
                port=self.config["port"],
                baudrate=self.config["baudrate"],
                bytesize=self.config["bytesize"],
                parity=parity_map.get(self.config.get("parity", "none"), "N"),
                stopbits=self.config["stopbits"],
                timeout=0.1,
            )
            self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_sock.bind(("127.0.0.1", self.tcp_port))
            self.server_sock.listen()
            self.server_sock.settimeout(0.2)
        except Exception as exc:
            print(f"mux serve failed: {exc}", file=sys.stderr)
            self.close()
            return 1

        signal.signal(signal.SIGINT, self._on_signal)
        signal.signal(signal.SIGTERM, self._on_signal)

        threads = [
            threading.Thread(target=self._accept_loop, daemon=True),
            threading.Thread(target=self._serial_read_loop, daemon=True),
        ]
        for thread in threads:
            thread.start()

        while not self.stop_event.is_set():
            time.sleep(0.2)

        self.close()
        return 0

    def _on_signal(self, sig, frame):
        self.stop_event.set()

    def _accept_loop(self):
        while not self.stop_event.is_set():
            try:
                assert self.server_sock is not None
                client, _addr = self.server_sock.accept()
                client.settimeout(0.2)
                with self.clients_lock:
                    self.clients.add(client)
                threading.Thread(target=self._client_read_loop, args=(client,), daemon=True).start()
            except socket.timeout:
                continue
            except OSError:
                break

    def _serial_read_loop(self):
        while not self.stop_event.is_set():
            try:
                assert self.serial_port is not None
                size = max(1, getattr(self.serial_port, "in_waiting", 0) or 1)
                data = self.serial_port.read(size)
                if data:
                    self._broadcast(data)
            except Exception:
                self.stop_event.set()
                break

    def _client_read_loop(self, client: socket.socket):
        while not self.stop_event.is_set():
            try:
                data = client.recv(4096)
                if not data:
                    break
                with self.serial_lock:
                    assert self.serial_port is not None
                    self.serial_port.write(data)
                    self.serial_port.flush()
            except socket.timeout:
                continue
            except OSError:
                break
            except Exception:
                self.stop_event.set()
                break
        self._remove_client(client)

    def _broadcast(self, data: bytes):
        dead = []
        with self.clients_lock:
            clients = list(self.clients)
        for client in clients:
            try:
                client.sendall(data)
            except OSError:
                dead.append(client)
        for client in dead:
            self._remove_client(client)

    def _remove_client(self, client: socket.socket):
        with self.clients_lock:
            self.clients.discard(client)
        try:
            client.close()
        except OSError:
            pass

    def close(self):
        self.stop_event.set()
        with self.clients_lock:
            clients = list(self.clients)
            self.clients.clear()
        for client in clients:
            try:
                client.close()
            except OSError:
                pass
        if self.server_sock is not None:
            try:
                self.server_sock.close()
            except OSError:
                pass
        if self.serial_port is not None:
            try:
                self.serial_port.close()
            except Exception:
                pass


def run_mux_server(config: dict, tcp_port: int) -> int:
    return SerialMuxServer(config, tcp_port).run()


def start_mux(port: str, baudrate: int | None, workspace: str | None, vserial_link: str):
    """启动串口多路复用"""
    if not shutil.which("socat"):
        return make_result(
            success=False,
            action="mux_start",
            summary="socat 未安装",
            error={"code": "socat_missing", "message": "请安装 socat: apt install socat / pacman -S socat"},
        )

    # 检查已运行的 mux
    state = load_workspace_state(workspace)
    existing = state.get(STATE_KEY)
    if existing:
        if is_mux_alive(existing):
            return make_result(
                success=False,
                action="mux_start",
                summary="Mux 已在运行",
                error={"code": "already_running", "message": f"Mux 已在运行 (TCP:{existing['tcp_port']}, PTY:{existing['vserial']})"},
                details=existing,
            )
        else:
            # 清理僵尸状态
            state.pop(STATE_KEY, None)
            save_workspace_state(state, workspace)

    # 获取串口配置
    cfg, sources = get_serial_config(
        cli_port=port,
        cli_baudrate=baudrate,
        workspace=workspace,
    )

    if cfg is None:
        return make_result(
            success=False,
            action="mux_start",
            summary="无法获取串口配置",
            error={"code": "config_error", "message": sources.get("error", "配置错误")},
        )

    if is_missing(cfg["port"]):
        return make_result(
            success=False,
            action="mux_start",
            summary="未指定串口",
            error={"code": "no_port", "message": "请用 --port 指定串口"},
        )

    tcp_port = find_free_port()

    real_port = cfg["port"]

    # Layer 1: Python 后台进程独占真实串口，并向所有 TCP 客户端广播 RX。
    cmd1 = [
        sys.executable,
        str(Path(__file__).resolve()),
        "serve",
        "--port",
        real_port,
        "--baudrate",
        str(cfg["baudrate"]),
        "--bytesize",
        str(cfg["bytesize"]),
        "--parity",
        str(cfg["parity"]),
        "--stopbits",
        str(cfg["stopbits"]),
        "--tcp-port",
        str(tcp_port),
    ]
    # Layer 2: TCP client → 虚拟 PTY (供 minicom)
    cmd2 = ["socat", "-d", "-d", f"PTY,link={vserial_link},raw,echo=0", f"TCP:127.0.0.1:{tcp_port}"]

    try:
        p1 = subprocess.Popen(cmd1, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        if not wait_for_tcp_server(tcp_port, p1):
            return make_result(
                success=False,
                action="mux_start",
                summary=f"无法打开串口 {real_port}",
                error={"code": "port_open_failed", "message": f"串口 {real_port} 打开失败，请检查是否被占用"},
            )

        p2 = subprocess.Popen(cmd2, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        time.sleep(0.3)
        if p2.poll() is not None:
            p1.terminate()
            p1.wait()
            return make_result(
                success=False,
                action="mux_start",
                summary="无法创建虚拟串口",
                error={"code": "pty_failed", "message": "虚拟 PTY 创建失败"},
            )

        if not os.path.exists(vserial_link):
            p1.terminate()
            p2.terminate()
            p1.wait()
            p2.wait()
            return make_result(
                success=False,
                action="mux_start",
                summary="虚拟串口未创建",
                error={"code": "pty_not_created", "message": f"PTY 链接 {vserial_link} 未创建"},
            )

    except Exception as e:
        return make_result(
            success=False,
            action="mux_start",
            summary="启动失败",
            error={"code": "start_failed", "message": str(e)},
        )

    mux_info = {
        "tcp_port": tcp_port,
        "tcp_pid": p1.pid,
        "pty_pid": p2.pid,
        "vserial": vserial_link,
        "real_port": real_port,
        "baudrate": cfg["baudrate"],
        "bytesize": cfg.get("bytesize", 8),
        "parity": cfg.get("parity", "none"),
        "stopbits": cfg.get("stopbits", 1),
        "started_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
    }

    # 保存状态
    state[STATE_KEY] = mux_info
    save_workspace_state(state, workspace)
    save_project_config(workspace, {
        "port": real_port,
        "baudrate": cfg["baudrate"],
        "bytesize": cfg.get("bytesize", 8),
        "parity": cfg.get("parity", "none"),
        "stopbits": cfg.get("stopbits", 1),
        "encoding": cfg.get("encoding", "utf-8"),
    })

    return make_result(
        success=True,
        action="mux_start",
        summary=f"Mux 已启动: {real_port} -> TCP:{tcp_port} -> PTY:{vserial_link}",
        details=mux_info,
    )


def stop_mux(workspace: str | None = None):
    """停止串口多路复用"""
    state = load_workspace_state(workspace)
    mux_info = state.get(STATE_KEY)

    if not mux_info:
        return make_result(
            success=False,
            action="mux_stop",
            summary="未找到运行中的 Mux",
            error={"code": "not_running", "message": "未找到运行中的串口多路复用"},
        )

    killed = []
    failed = []
    for pid_key in ("tcp_pid", "pty_pid"):
        pid = mux_info.get(pid_key)
        if pid:
            try:
                os.kill(pid, signal.SIGTERM)
                killed.append(pid)
            except ProcessLookupError:
                pass
            except Exception:
                failed.append(str(pid))

    # 清理残留的虚拟串口符号链接
    vserial = mux_info.get("vserial")
    if vserial and os.path.islink(vserial):
        try:
            os.unlink(vserial)
        except OSError:
            pass

    # 清理状态
    state.pop(STATE_KEY, None)
    save_workspace_state(state, workspace)

    if failed:
        return make_result(
            success=True,
            action="mux_stop",
            summary=f"已终止 {len(killed)} 个进程，{len(failed)} 个失败",
            details={"killed": killed, "failed": failed, "vserial": mux_info.get("vserial")},
        )
    else:
        return make_result(
            success=True,
            action="mux_stop",
            summary=f"Mux 已停止 ({len(killed)} 个进程已终止)",
            details={"killed": killed, "vserial": mux_info.get("vserial")},
        )


def is_mux_alive(mux_info: dict) -> bool:
    """检查 mux 进程是否存活"""
    for pid_key in ("tcp_pid", "pty_pid"):
        pid = mux_info.get(pid_key)
        if not pid:
            return False
        try:
            os.kill(pid, 0)
        except (ProcessLookupError, PermissionError):
            return False
    return True


def status_mux(workspace: str | None = None):
    """查询 mux 状态"""
    state = load_workspace_state(workspace)
    mux_info = state.get(STATE_KEY)

    if not mux_info:
        return make_result(
            success=True,
            action="mux_status",
            summary="Mux 未运行",
            details={"running": False},
        )

    alive = is_mux_alive(mux_info)
    if not alive:
        state.pop(STATE_KEY, None)
        save_workspace_state(state, workspace)
        return make_result(
            success=True,
            action="mux_status",
            summary="Mux 已停止（清理残留状态）",
            details={"running": False, "cleaned": True},
        )

    return make_result(
        success=True,
        action="mux_status",
        summary=f"Mux 运行中: {mux_info.get('real_port')} -> TCP:{mux_info.get('tcp_port')} -> PTY:{mux_info.get('vserial')}",
        details={
            "running": True,
            "real_port": mux_info.get("real_port"),
            "tcp_port": mux_info.get("tcp_port"),
            "vserial": mux_info.get("vserial"),
            "baudrate": mux_info.get("baudrate"),
            "started_at": mux_info.get("started_at"),
        },
    )


def main():
    if len(sys.argv) > 1 and sys.argv[1] == "serve":
        serve_parser = argparse.ArgumentParser(description="Serial mux 后台服务")
        serve_parser.add_argument("--port", required=True)
        serve_parser.add_argument("--baudrate", type=int, required=True)
        serve_parser.add_argument("--bytesize", type=int, required=True)
        serve_parser.add_argument("--parity", required=True)
        serve_parser.add_argument("--stopbits", type=int, required=True)
        serve_parser.add_argument("--tcp-port", type=int, required=True)
        args = serve_parser.parse_args(sys.argv[2:])
        config = {
            "port": args.port,
            "baudrate": args.baudrate,
            "bytesize": args.bytesize,
            "parity": args.parity,
            "stopbits": args.stopbits,
        }
        sys.exit(run_mux_server(config, args.tcp_port))

    parser = argparse.ArgumentParser(description="Serial 多路复用管理")
    sub = parser.add_subparsers(dest="command", help="子命令")

    p_start = sub.add_parser("start", help="启动多路复用")
    p_start.add_argument("--port", help="真实串口号 (如 /dev/ttyUSB0)")
    p_start.add_argument("--baudrate", type=int, help="波特率")
    p_start.add_argument("--vserial", default=DEFAULT_VSERIAL_LINK, help=f"虚拟串口路径 (默认: {DEFAULT_VSERIAL_LINK})")
    p_start.add_argument("--workspace", help="工作区路径")

    sub.add_parser("stop", help="停止多路复用").add_argument("--workspace", help="工作区路径")

    sub.add_parser("status", help="查询多路复用状态").add_argument("--workspace", help="工作区路径")

    args = parser.parse_args()

    if args.command == "start":
        result = start_mux(args.port, args.baudrate, getattr(args, "workspace", None), args.vserial)
    elif args.command == "stop":
        result = stop_mux(getattr(args, "workspace", None))
    elif args.command == "status":
        result = status_mux(getattr(args, "workspace", None))
    else:
        result = status_mux()
        if result["details"].get("running"):
            print(f"Mux 运行中: {result['details']['real_port']} -> TCP:{result['details']['tcp_port']} -> PTY:{result['details']['vserial']}")
            print(f"  虚拟串口: {result['details']['vserial']}")
            print(f"  TCP 端口: {result['details']['tcp_port']}")
            print(f"  启动时间: {result['details']['started_at']}")
        else:
            print("Mux 未运行. 使用 'start --port <串口>' 启动")

    output_json(result)


if __name__ == "__main__":
    main()

"""serial skill 私有运行时工具。"""

from __future__ import annotations

import json
import os
import signal
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


SKILL_DIR = Path(__file__).resolve().parent.parent
SKILL_NAME = "serial"
STATE_DIR_NAME = ".embeddedskills"
STATE_FILE_NAME = "state.json"
PROJECT_CONFIG_FILE = "config.json"


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def is_missing(value: Any) -> bool:
    return value is None or value == ""


def load_json_file(path: str | Path) -> dict:
    """加载 JSON 文件，不存在返回空字典"""
    file_path = Path(path)
    if not file_path.exists():
        return {}
    try:
        return json.loads(file_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def save_json_file(path: str | Path, data: dict) -> None:
    """保存 JSON 文件，自动创建目录"""
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def load_local_config() -> dict:
    """加载 skill/config.json（环境级配置）"""
    return load_json_file(SKILL_DIR / "config.json")


def save_local_config(data: dict) -> None:
    """保存环境级配置到 skill/config.json"""
    save_json_file(SKILL_DIR / "config.json", data)


def workspace_root(workspace: str | None = None) -> Path:
    if not is_missing(workspace):
        return Path(str(workspace)).expanduser().resolve()
    return Path.cwd().resolve()


def load_project_config(workspace: str | None = None) -> dict:
    """从 workspace/.embeddedskills/config.json 读取本 skill 的工程级配置"""
    proj_config = load_json_file(workspace_root(workspace) / STATE_DIR_NAME / PROJECT_CONFIG_FILE)
    return proj_config.get(SKILL_NAME, {})


def save_project_config(workspace: str | None = None, values: dict | None = None) -> None:
    """写回工程级配置，只更新本 skill 的部分"""
    if values is None:
        return
    proj_path = workspace_root(workspace) / STATE_DIR_NAME / PROJECT_CONFIG_FILE
    proj_config = load_json_file(proj_path)
    proj_config[SKILL_NAME] = {**proj_config.get(SKILL_NAME, {}), **values}
    save_json_file(proj_path, proj_config)


def load_workspace_state(workspace: str | None = None) -> dict:
    """从 workspace/.embeddedskills/state.json 读取状态"""
    return load_json_file(workspace_root(workspace) / STATE_DIR_NAME / STATE_FILE_NAME)


def save_workspace_state(state: dict, workspace: str | None = None) -> Path:
    """保存状态"""
    ws = workspace_root(workspace)
    file_path = ws / STATE_DIR_NAME / STATE_FILE_NAME
    save_json_file(file_path, _serialize_state_value(state, ws))
    return file_path


def update_state_entry(category: str, record: dict, workspace: str | None = None) -> dict:
    """更新状态条目"""
    ws = workspace_root(workspace)
    state = load_workspace_state(workspace)
    state[category] = _serialize_state_value({**record, "timestamp": record.get("timestamp") or now_iso()}, ws)
    file_path = save_workspace_state(state, workspace)
    return {
        "workspace": str(ws),
        "file": str(file_path),
        "updated_keys": [category],
        category: state[category],
    }


def normalize_path(value: str | None, base: str | Path | None = None) -> str:
    """路径规范化"""
    if is_missing(value):
        return ""
    path = Path(str(value)).expanduser()
    if base and not path.is_absolute():
        path = Path(base) / path
    return str(path.resolve()) if path.is_absolute() else str(path)


def _serialize_state_value(value: Any, workspace: Path) -> Any:
    if isinstance(value, dict):
        return {key: _serialize_state_value(item, workspace) for key, item in value.items()}
    if isinstance(value, list):
        return [_serialize_state_value(item, workspace) for item in value]
    if not isinstance(value, str) or "://" in value:
        return value

    path = Path(value).expanduser()
    if not path.is_absolute():
        return value
    try:
        return Path(os.path.relpath(path.resolve(), workspace)).as_posix()
    except ValueError:
        return value


def _first_resolved(mapping: dict, keys: list[str]) -> tuple[Any, str | None]:
    for key in keys:
        value = mapping.get(key)
        if not is_missing(value):
            return value, key
    return None, None


def resolve_param(
    name: str,
    cli_value: Any = None,
    local_config: dict | None = None,
    local_keys: list[str] | None = None,
    project_config: dict | None = None,
    project_keys: list[str] | None = None,
    state: dict | None = None,
    state_keys: list[str] | None = None,
    default: Any = None,
) -> tuple[Any, str]:
    """统一参数解析，优先级: CLI > 环境级 > 工程级 > state > default"""
    if not is_missing(cli_value):
        return cli_value, "cli"

    if local_config and local_keys:
        value, key = _first_resolved(local_config, local_keys)
        if not is_missing(value):
            return value, f"local:{key}"

    if project_config and project_keys:
        value, key = _first_resolved(project_config, project_keys)
        if not is_missing(value):
            return value, f"project:{key}"

    if state and state_keys:
        value, key = _first_resolved(state, state_keys)
        if not is_missing(value):
            return value, f"state:{key}"

    if not is_missing(default):
        return default, "default"

    return None, ""


def parameter_context(name: str, value: Any, source: str) -> dict:
    """记录参数来源"""
    return {"name": name, "value": value, "source": source}


def make_result(
    success: bool = True,
    action: str = "",
    summary: str = "",
    details: dict | None = None,
    error: dict | None = None,
) -> dict:
    """统一结果格式"""
    result = {
        "status": "ok" if success else "error",
        "action": action,
        "summary": summary,
    }
    if details:
        result["details"] = details
    if error:
        result["error"] = error
    return result


def make_timing(start_time: float) -> dict:
    """执行时间记录"""
    elapsed = datetime.now().timestamp() - start_time
    return {
        "started_at": datetime.fromtimestamp(start_time).astimezone().isoformat(timespec="seconds"),
        "finished_at": now_iso(),
        "elapsed_ms": int(elapsed * 1000),
    }


def scan_serial_ports(filter_keyword: str | None = None) -> tuple[list[dict], str | None]:
    """扫描系统串口，返回 (ports, error)"""
    try:
        from serial.tools.list_ports import comports
    except ImportError:
        return [], "pyserial 未安装，请执行 pip install pyserial"

    # 加载 VID/PID -> 芯片名称映射
    chip_map = {}
    try:
        common_devices_path = SKILL_DIR / "references" / "common_devices.json"
        data = json.loads(common_devices_path.read_text(encoding="utf-8"))
        for entry in data.get("usb_serial_chips", []):
            key = (entry["vid"].upper(), entry["pid"].upper())
            chip_map[key] = entry["name"]
    except Exception:
        pass

    ports = []
    for p in sorted(comports(), key=lambda x: x.device):
        vid = f"{p.vid:04X}" if p.vid else ""
        pid = f"{p.pid:04X}" if p.pid else ""
        chip_name = chip_map.get((vid, pid), "")

        info = {
            "port": p.device,
            "description": p.description or "",
            "vid": vid,
            "pid": pid,
            "chip": chip_name,
            "serial_number": p.serial_number or "",
            "location": p.location or "",
        }

        if filter_keyword:
            text = " ".join(str(v) for v in info.values()).lower()
            if filter_keyword.lower() not in text:
                continue

        ports.append(info)

    return ports, None


def get_serial_config(
    cli_port: str | None = None,
    cli_baudrate: int | None = None,
    cli_bytesize: int | None = None,
    cli_parity: str | None = None,
    cli_stopbits: int | None = None,
    cli_encoding: str | None = None,
    cli_timeout: float | None = None,
    workspace: str | None = None,
) -> tuple[dict, dict]:
    """
    获取串口配置，按优先级解析参数。
    返回 (config_dict, sources_dict)
    """
    local_cfg = load_local_config()
    proj_cfg = load_project_config(workspace)
    state = load_workspace_state(workspace)

    sources = {}

    # 解析各个参数
    port, src = resolve_param(
        "port", cli_port,
        project_config=proj_cfg, project_keys=["port"],
        state=state, state_keys=["last_serial_port"],
    )
    sources["port"] = src or "unknown"

    baudrate, src = resolve_param(
        "baudrate", cli_baudrate,
        project_config=proj_cfg, project_keys=["baudrate"],
        state=state, state_keys=["last_baudrate"],
        default=115200,
    )
    sources["baudrate"] = src or "default"

    bytesize, src = resolve_param(
        "bytesize", cli_bytesize,
        project_config=proj_cfg, project_keys=["bytesize"],
        default=8,
    )
    sources["bytesize"] = src or "default"

    parity, src = resolve_param(
        "parity", cli_parity,
        project_config=proj_cfg, project_keys=["parity"],
        default="none",
    )
    sources["parity"] = src or "default"

    stopbits, src = resolve_param(
        "stopbits", cli_stopbits,
        project_config=proj_cfg, project_keys=["stopbits"],
        default=1,
    )
    sources["stopbits"] = src or "default"

    encoding, src = resolve_param(
        "encoding", cli_encoding,
        project_config=proj_cfg, project_keys=["encoding"],
        default="utf-8",
    )
    sources["encoding"] = src or "default"

    timeout, src = resolve_param(
        "timeout_sec", cli_timeout,
        project_config=proj_cfg, project_keys=["timeout_sec"],
        default=1.0,
    )
    sources["timeout_sec"] = src or "default"

    # 如果没有指定 port，尝试扫描
    if is_missing(port):
        ports, err = scan_serial_ports()
        if err:
            return None, {"error": err}
        if len(ports) == 1:
            # 唯一候选，自动写入配置
            port = ports[0]["port"]
            sources["port"] = "auto_scan"
            save_project_config(workspace, {"port": port})
        elif len(ports) > 1:
            return None, {
                "error": "找到多个串口，请指定一个",
                "candidates": ports,
                "need_selection": True,
            }
        else:
            return None, {"error": "未找到可用串口"}

    log_dir, src = resolve_param(
        "log_dir", None,
        project_config=proj_cfg, project_keys=["log_dir"],
        default=".embeddedskills/logs/serial",
    )
    sources["log_dir"] = src or "default"

    config = {
        "port": port,
        "baudrate": baudrate,
        "bytesize": bytesize,
        "parity": parity,
        "stopbits": stopbits,
        "encoding": encoding,
        "timeout_sec": timeout,
        "log_dir": log_dir,
    }

    return config, sources


def is_mux_alive(mux_info: dict) -> bool:
    """检查 mux 进程是否存活"""
    for pid_key in ("tcp_pid", "pty_pid"):
        pid = mux_info.get(pid_key, 0)
        if not pid:
            return False
        try:
            os.kill(pid, 0)
        except (ProcessLookupError, PermissionError):
            return False
    return True


def get_mux_info(workspace: str | None = None) -> dict | None:
    """获取运行中的 mux 连接信息，未运行返回 None"""
    state = load_workspace_state(workspace)
    mux_info = state.get("serial_mux")
    if not mux_info:
        return None
    if not is_mux_alive(mux_info):
        state.pop("serial_mux", None)
        save_workspace_state(state, workspace)
        return None
    return mux_info


def _normalize_serial_port(value: Any) -> str:
    text = str(value or "")
    if not text:
        return ""
    if os.name == "nt":
        return os.path.normcase(text)
    if text.startswith("/"):
        return os.path.realpath(text)
    return text


def config_matches_mux(config: dict, mux_info: dict) -> bool:
    """确认当前串口配置与运行中的 mux 指向同一串口。"""
    if _normalize_serial_port(config.get("port")) != _normalize_serial_port(mux_info.get("real_port")):
        return False

    checks = (
        ("baudrate", 115200),
        ("bytesize", 8),
        ("parity", "none"),
        ("stopbits", 1),
    )
    for key, default in checks:
        if str(config.get(key, default)).lower() != str(mux_info.get(key, default)).lower():
            return False
    return True


def get_matching_mux_info(config: dict, workspace: str | None = None) -> dict | None:
    """仅在 mux 与本次解析出的串口配置一致时返回 mux 信息。"""
    mux_info = get_mux_info(workspace)
    if mux_info and config_matches_mux(config, mux_info):
        return mux_info
    return None


def open_serial_port(config: dict, use_mux: bool = True):
    """根据配置打开串口连接。

    当 mux 运行且串口配置匹配时自动通过 socket:// 连接 TCP 端口，
    从而实现与 minicom 同时访问串口。
    use_mux=False 时跳过 mux 检测，直接打开真实串口。
    """
    import serial

    if use_mux:
        mux = get_matching_mux_info(config)
        if mux:
            url = f"socket://127.0.0.1:{mux['tcp_port']}"
            ser = serial.serial_for_url(url)
            ser.timeout = config.get("timeout_sec", 1.0)
            setattr(ser, "_serial_skill_using_mux", True)
            return ser

    PARITY_MAP = {"none": "N", "even": "E", "odd": "O", "mark": "M", "space": "S"}
    parity = PARITY_MAP.get(config.get("parity", "none"), "N")

    return serial.Serial(
        port=config["port"],
        baudrate=config["baudrate"],
        bytesize=config["bytesize"],
        parity=parity,
        stopbits=config["stopbits"],
        timeout=config["timeout_sec"],
    )


def output_json(data: dict, *, indent: int = 2) -> None:
    """输出 JSON 到 stdout"""
    sys.stdout.reconfigure(encoding="utf-8")
    print(json.dumps(data, ensure_ascii=False, indent=indent), flush=True)

"""串口扫描：枚举系统串口并展示设备信息"""

import argparse
import json
import sys
from pathlib import Path

from serial_runtime import get_mux_info

COMMON_DEVICES_PATH = Path(__file__).parent.parent / "references" / "common_devices.json"


def load_chip_map():
    """加载 VID/PID -> 芯片名称映射"""
    chip_map = {}
    try:
        data = json.loads(COMMON_DEVICES_PATH.read_text(encoding="utf-8"))
        for entry in data.get("usb_serial_chips", []):
            key = (entry["vid"].upper(), entry["pid"].upper())
            chip_map[key] = entry["name"]
    except Exception:
        pass
    return chip_map


def scan_ports(filter_keyword=None):
    """扫描系统串口"""
    try:
        from serial.tools.list_ports import comports
    except ImportError:
        return None, "pyserial 未安装，请执行 pip install pyserial"

    chip_map = load_chip_map()
    ports = []

    for p in sorted(comports(), key=lambda x: x.device):
        vid = f"{p.vid:04X}" if p.vid else ""
        pid = f"{p.pid:04X}" if p.pid else ""
        chip_name = chip_map.get((vid, pid), "")

        info = {
            "port": p.device,
            "description": p.description or "",
            "vid": vid,
            "pid": pid,
            "chip": chip_name,
            "serial_number": p.serial_number or "",
            "location": p.location or "",
        }

        if filter_keyword:
            text = " ".join(str(v) for v in info.values()).lower()
            if filter_keyword.lower() not in text:
                continue

        ports.append(info)

    return ports, None


def output_json(result):
    sys.stdout.buffer.write(json.dumps(result, ensure_ascii=False, indent=2).encode("utf-8"))
    sys.stdout.buffer.write(b"\n")
    sys.stdout.buffer.flush()


def main():
    parser = argparse.ArgumentParser(description="扫描系统串口")
    parser.add_argument("--filter", help="按关键词过滤")
    parser.add_argument("--json", action="store_true", help="JSON 输出")
    args = parser.parse_args()

    ports, err = scan_ports(args.filter)

    if err:
        result = {"status": "error", "action": "scan", "error": {"code": "import_error", "message": err}}
        if args.json:
            output_json(result)
        else:
            print(f"错误: {err}", file=sys.stderr)
        sys.exit(1)

    mux_info = get_mux_info()
    result = {
        "status": "ok",
        "action": "scan",
        "summary": f"发现 {len(ports)} 个串口",
        "details": {"ports": ports},
    }
    if mux_info:
        result["details"]["mux"] = {
            "running": True,
            "vserial": mux_info["vserial"],
            "tcp_port": mux_info["tcp_port"],
            "real_port": mux_info["real_port"],
        }
        result["summary"] += f" (Mux 运行中: {mux_info['vserial']})"

    if args.json:
        output_json(result)
    else:
        if not ports:
            print("未发现可用串口")
        else:
            print(f"发现 {len(ports)} 个串口:\n")
            for p in ports:
                chip = f" [{p['chip']}]" if p["chip"] else ""
                vid_pid = f" (VID:{p['vid']} PID:{p['pid']})" if p["vid"] else ""
                print(f"  {p['port']}: {p['description']}{chip}{vid_pid}")
        if mux_info:
            print(f"\nMux 运行中: {mux_info['real_port']} -> TCP:{mux_info['tcp_port']} -> PTY:{mux_info['vserial']}")


if __name__ == "__main__":
    main()