Can

{
  "slcan_serial_port": "",
  "slcan_serial_baudrate": 115200
}

{
  "usb_can_devices": [
    {"vid": "0C72", "pid": "000C", "name": "PEAK PCAN-USB", "interface": "pcan", "channel": "PCAN_USBBUS1"},
    {"vid": "0C72", "pid": "0012", "name": "PEAK PCAN-USB FD", "interface": "pcan", "channel": "PCAN_USBBUS1"},
    {"vid": "0C72", "pid": "0014", "name": "PEAK PCAN-USB Pro FD", "interface": "pcan", "channel": "PCAN_USBBUS1"},
    {"vid": "1D50", "pid": "606F", "name": "CANable / canable (gs_usb)", "interface": "gs_usb", "channel": "0"},
    {"vid": "1D50", "pid": "606D", "name": "CANtact (gs_usb)", "interface": "gs_usb", "channel": "0"},
    {"vid": "AD50", "pid": "60C4", "name": "candleLight (gs_usb)", "interface": "gs_usb", "channel": "0"},
    {"vid": "1CD2", "pid": "0018", "name": "Kvaser Leaf Light v2", "interface": "kvaser", "channel": "0"},
    {"vid": "1CD2", "pid": "0008", "name": "Kvaser USBcan II", "interface": "kvaser", "channel": "0"}
  ],
  "known_interfaces": [
    {"interface": "pcan", "platform": "windows", "note": "需安装 PEAK 驱动"},
    {"interface": "vector", "platform": "windows", "note": "需安装 Vector XL Driver Library"},
    {"interface": "ixxat", "platform": "windows", "note": "需安装 IXXAT VCI 驱动"},
    {"interface": "kvaser", "platform": "windows/linux", "note": "需安装 Kvaser CANlib"},
    {"interface": "slcan", "platform": "windows/linux", "note": "串口转 CAN，需 pyserial"},
    {"interface": "socketcan", "platform": "linux", "note": "Linux 内核原生支持"},
    {"interface": "gs_usb", "platform": "linux", "note": "candleLight/CANable 等 gs_usb 固件设备"},
    {"interface": "virtual", "platform": "all", "note": "虚拟总线，用于测试"}
  ]
}

"""CAN 数据库解码：用 DBC/ARXML/KCD 等文件解码单帧或日志"""

import argparse
import json
import sys
from pathlib import Path


def parse_hex_data(s):
    s = s.replace(" ", "").replace(",", "")
    return bytes.fromhex(s)


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 load_database(db_path, db_format="auto"):
    """加载数据库文件"""
    import cantools

    path = Path(db_path)
    if not path.exists():
        return None, f"数据库文件不存在: {db_path}"

    try:
        if db_format == "auto":
            db = cantools.database.load_file(str(path))
        else:
            db = cantools.database.Database()
            with open(path, "r", encoding="utf-8") as f:
                content = f.read()
            format_map = {
                "dbc": "dbc",
                "arxml": "autosar",
                "kcd": "kcd",
                "sym": "sym",
                "cdd": "cdd",
            }
            fmt = format_map.get(db_format)
            if fmt:
                db.add_dbc_string(content) if fmt == "dbc" else db.add_autosar_string(content) if fmt == "autosar" else None
                # 对于非 dbc/autosar，回退到 load_file
                if fmt not in ("dbc", "autosar"):
                    db = cantools.database.load_file(str(path))
            else:
                return None, f"不支持的数据库格式: {db_format}"
        return db, None
    except Exception as e:
        return None, f"加载数据库失败: {e}"


def list_messages(db, signal_filter=None):
    """列出数据库中的报文定义"""
    messages = []
    for msg in db.messages:
        signals = []
        for sig in msg.signals:
            if signal_filter and signal_filter.lower() not in sig.name.lower():
                continue
            sig_info = {
                "name": sig.name,
                "start_bit": sig.start,
                "length": sig.length,
                "unit": sig.unit or "",
                "min": sig.minimum,
                "max": sig.maximum,
            }
            signals.append(sig_info)

        if signal_filter and not signals:
            continue

        messages.append({
            "name": msg.name,
            "id": f"0x{msg.frame_id:03X}",
            "dlc": msg.length,
            "signals": signals,
        })
    return messages


def decode_single(db, arb_id, data):
    """解码单帧"""
    try:
        msg = db.get_message_by_frame_id(arb_id)
    except KeyError:
        return None, f"数据库中未找到 ID 0x{arb_id:03X} 的定义"

    try:
        decoded = msg.decode(data)
        signals = []
        for k, v in decoded.items():
            sig_info = {"name": k, "value": round(v, 6) if isinstance(v, float) else v}
            # 查找信号的单位
            for sig in msg.signals:
                if sig.name == k:
                    sig_info["unit"] = sig.unit or ""
                    break
            signals.append(sig_info)
        return {"message": msg.name, "id": f"0x{arb_id:03X}", "signals": signals}, None
    except Exception as e:
        return None, f"解码失败: {e}"


def decode_log_file(db, log_path, signal_filter=None):
    """解码日志文件"""
    import can

    path = Path(log_path)
    if not path.exists():
        return None, f"日志文件不存在: {log_path}"

    results = []
    errors = 0

    try:
        reader = can.LogReader(str(path))
        for msg in reader:
            try:
                db_msg = db.get_message_by_frame_id(msg.arbitration_id)
                decoded = db_msg.decode(msg.data)
                if signal_filter:
                    decoded = {k: v for k, v in decoded.items() if signal_filter.lower() in k.lower()}
                    if not decoded:
                        continue
                decoded = {k: round(v, 6) if isinstance(v, float) else v for k, v in decoded.items()}
                results.append({
                    "timestamp": round(msg.timestamp, 6),
                    "message": db_msg.name,
                    "id": f"0x{msg.arbitration_id:03X}",
                    "decoded": decoded,
                })
            except (KeyError, Exception):
                errors += 1
                continue
    except Exception as e:
        return None, f"读取日志失败: {e}"

    return {"frames": results, "decoded_count": len(results), "error_count": errors}, None


def main():
    parser = argparse.ArgumentParser(description="CAN 数据库解码")
    parser.add_argument("db_file", help="数据库文件路径（DBC/ARXML/KCD/SYM/CDD）")
    parser.add_argument("--db-format", default="auto", help="数据库格式（auto|dbc|arxml|kcd|sym|cdd）")
    parser.add_argument("--id", help="单帧 CAN ID（支持 0x 前缀）")
    parser.add_argument("--data", help="单帧数据（Hex）")
    parser.add_argument("--log", help="日志文件路径")
    parser.add_argument("--signal", help="按信号名过滤")
    parser.add_argument("--list", action="store_true", help="列出数据库中所有报文定义")
    parser.add_argument("--json", action="store_true", help="JSON 输出")
    args = parser.parse_args()

    try:
        import cantools  # noqa: F401
    except ImportError:
        err = {"status": "error", "action": "decode", "error": {"code": "import_error", "message": "cantools 未安装，请执行 pip install cantools"}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    db, err = load_database(args.db_file, args.db_format)
    if err:
        result = {"status": "error", "action": "decode", "error": {"code": "db_load_failed", "message": err}}
        if args.json:
            output_json(result)
        else:
            print(f"错误: {err}", file=sys.stderr)
        sys.exit(1)

    # 列出报文定义
    if args.list:
        messages = list_messages(db, args.signal)
        result = {
            "status": "ok",
            "action": "decode",
            "summary": f"数据库包含 {len(messages)} 个报文",
            "details": {"messages": messages},
        }
        if args.json:
            output_json(result)
        else:
            for m in messages:
                print(f"\n{m['name']} ({m['id']}) DLC={m['dlc']}")
                for s in m["signals"]:
                    unit = f" [{s['unit']}]" if s["unit"] else ""
                    print(f"  {s['name']}: bit {s['start_bit']}+{s['length']}{unit} ({s['min']}~{s['max']})")
        return

    # 解码单帧
    if args.id and args.data:
        arb_id = int(args.id, 0)
        data = parse_hex_data(args.data)
        decoded, err = decode_single(db, arb_id, data)
        if err:
            result = {"status": "error", "action": "decode", "error": {"code": "decode_failed", "message": err}}
            if args.json:
                output_json(result)
            else:
                print(f"错误: {err}", file=sys.stderr)
            sys.exit(1)

        result = {
            "status": "ok",
            "action": "decode",
            "summary": f"已解码 {decoded['message']} ({decoded['id']})",
            "details": decoded,
        }
        if args.json:
            output_json(result)
        else:
            print(f"\n{decoded['message']} ({decoded['id']})")
            for s in decoded["signals"]:
                unit = f" {s.get('unit', '')}" if s.get("unit") else ""
                print(f"  {s['name']} = {s['value']}{unit}")
        return

    # 解码日志
    if args.log:
        log_result, err = decode_log_file(db, args.log, args.signal)
        if err:
            result = {"status": "error", "action": "decode", "error": {"code": "log_decode_failed", "message": err}}
            if args.json:
                output_json(result)
            else:
                print(f"错误: {err}", file=sys.stderr)
            sys.exit(1)

        result = {
            "status": "ok",
            "action": "decode",
            "summary": f"已解码 {log_result['decoded_count']} 帧（{log_result['error_count']} 帧无法解码）",
            "details": log_result,
        }
        if args.json:
            output_json(result)
        else:
            for f in log_result["frames"]:
                sigs = ", ".join(f"{k}={v}" for k, v in f["decoded"].items())
                print(f"[{f['timestamp']:.6f}] {f['message']} ({f['id']}): {sigs}")
            print(f"\n解码 {log_result['decoded_count']} 帧, {log_result['error_count']} 帧无法解码")
        return

    # 未指定操作时提示
    print("请指定操作：--list 列出定义、--id + --data 解码单帧、--log 解码日志", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":
    main()

"""CAN 总线日志记录：保存报文到 ASC / BLF / CSV / LOG 文件"""

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

from can_runtime import (
    add_can_connection_args,
    get_can_config,
    open_can_bus,
    save_project_config,
    update_state_entry,
)


def parse_id_list(s):
    if not s:
        return None
    ids = set()
    for part in s.split(","):
        part = part.strip()
        if part:
            ids.add(int(part, 0))
    return ids


def format_data(data):
    return " ".join(f"{b:02X}" for b in data)


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="CAN 总线日志记录")
    add_can_connection_args(parser)
    parser.add_argument("--output", "-o", help="输出文件路径（按扩展名选格式：.asc/.blf/.csv/.log）")
    parser.add_argument("--duration", type=float, help="记录时长（秒）")
    parser.add_argument("--max-count", type=int, help="最大记录帧数")
    parser.add_argument("--filter-id", help="只记录指定 ID（逗号分隔）")
    parser.add_argument("--console", action="store_true", help="同时输出到控制台")
    parser.add_argument("--json", action="store_true", help="JSON 输出")
    args = parser.parse_args()

    try:
        import can
    except ImportError:
        err = {"status": "error", "action": "log", "error": {"code": "import_error", "message": "python-can 未安装，请执行 pip install python-can"}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    # 获取配置
    config, sources = get_can_config(
        cli_interface=args.interface,
        cli_channel=args.channel,
        cli_bitrate=args.bitrate,
    )

    if config is None:
        if sources.get("need_selection"):
            err = {"status": "error", "action": "log", "error": {"code": "multiple_candidates", "message": f"{sources['error']}，请用 --interface 和 --channel 指定"}}
        else:
            err = {"status": "error", "action": "log", "error": {"code": "config_error", "message": sources.get("error", "配置错误")}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    interface = config["interface"]
    channel = config["channel"]
    bitrate = config["bitrate"]
    log_dir = config["log_dir"]

    # 保存确认的配置
    save_project_config(values={
        "interface": interface,
        "channel": channel,
        "bitrate": bitrate,
    })

    # 确定输出文件
    if args.output:
        output_path = Path(args.output)
    else:
        os.makedirs(log_dir, exist_ok=True)
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        output_path = Path(log_dir) / f"can_{timestamp}.asc"

    output_path.parent.mkdir(parents=True, exist_ok=True)

    filter_ids = parse_id_list(args.filter_id)

    try:
        bus = open_can_bus(config)
    except Exception as e:
        err = {"status": "error", "action": "log", "error": {"code": "interface_open_failed", "message": str(e)}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: 无法打开 CAN 接口 — {e}", file=sys.stderr)
        sys.exit(1)

    # 根据扩展名选择 logger
    try:
        logger = can.Logger(str(output_path))
    except Exception as e:
        bus.shutdown()
        err = {"status": "error", "action": "log", "error": {"code": "logger_init_failed", "message": str(e)}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: 无法创建日志文件 — {e}", file=sys.stderr)
        sys.exit(1)

    start = time.time()
    count = 0

    print(f"开始记录到 {output_path} ...", file=sys.stderr)

    try:
        while True:
            if args.duration and (time.time() - start) >= args.duration:
                break
            if args.max_count and count >= args.max_count:
                break

            remaining = None
            if args.duration:
                remaining = args.duration - (time.time() - start)
                if remaining <= 0:
                    break

            msg = bus.recv(timeout=min(remaining, 1.0) if remaining else 1.0)
            if msg is None:
                continue

            if filter_ids and msg.arbitration_id not in filter_ids:
                continue

            logger.on_message_received(msg)
            count += 1

            if args.console:
                print(f"[{msg.timestamp:.6f}] 0x{msg.arbitration_id:03X} [{msg.dlc}] {format_data(msg.data)}", file=sys.stderr)

    except KeyboardInterrupt:
        pass
    finally:
        logger.stop()
        bus.shutdown()

    elapsed = time.time() - start
    result = {
        "status": "ok",
        "action": "log",
        "summary": f"记录 {count} 帧到 {output_path}",
        "details": {
            "file": str(output_path),
            "frames": count,
            "duration_sec": round(elapsed, 1),
        },
    }

    if args.json:
        output_json(result)
    else:
        print(f"\n记录完成: {count} 帧, {elapsed:.1f} 秒 -> {output_path}", file=sys.stderr)

    # 更新状态
    update_state_entry("last_observe", {
        "type": "can_log",
        "interface": interface,
        "channel": channel,
        "file": str(output_path),
        "frames": count,
        "duration_sec": round(elapsed, 1),
    })


if __name__ == "__main__":
    main()

"""CAN 总线监控：持续读取报文，支持过滤、DBC 解码和 CAN-FD"""

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

from can_runtime import (
    add_can_connection_args,
    get_can_config,
    open_can_bus,
    save_project_config,
    update_state_entry,
)


def parse_id_list(s):
    """解析逗号分隔的 CAN ID 列表，支持 0x 前缀"""
    if not s:
        return None
    ids = set()
    for part in s.split(","):
        part = part.strip()
        if part:
            ids.add(int(part, 0))
    return ids


def output_json_line(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 format_data(data):
    return " ".join(f"{b:02X}" for b in data)


def main():
    parser = argparse.ArgumentParser(description="CAN 总线实时监控")
    add_can_connection_args(parser, include_data_bitrate=True)
    parser.add_argument("--fd", action="store_true", help="启用 CAN-FD 模式")
    parser.add_argument("--filter-id", help="只显示指定 ID（逗号分隔）")
    parser.add_argument("--exclude-id", help="排除指定 ID（逗号分隔）")
    parser.add_argument("--dbc", help="DBC 数据库文件路径，用于解码")
    parser.add_argument("--timeout", type=float, help="监控时长（秒）")
    parser.add_argument("--json", action="store_true", help="JSON Lines 输出")
    args = parser.parse_args()

    try:
        import can
    except ImportError:
        err = {"status": "error", "action": "monitor", "error": {"code": "import_error", "message": "python-can 未安装，请执行 pip install python-can"}}
        if args.json:
            output_json_line(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    # 获取配置
    config, sources = get_can_config(
        cli_interface=args.interface,
        cli_channel=args.channel,
        cli_bitrate=args.bitrate,
        cli_data_bitrate=args.data_bitrate,
    )

    if config is None:
        if sources.get("need_selection"):
            err = {"status": "error", "action": "monitor", "error": {"code": "multiple_candidates", "message": f"{sources['error']}，请用 --interface 和 --channel 指定"}}
        else:
            err = {"status": "error", "action": "monitor", "error": {"code": "config_error", "message": sources.get("error", "配置错误")}}
        if args.json:
            output_json_line(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    interface = config["interface"]
    channel = config["channel"]
    bitrate = config["bitrate"]
    data_bitrate = config["data_bitrate"]

    # 保存确认的配置
    save_project_config(values={
        "interface": interface,
        "channel": channel,
        "bitrate": bitrate,
        "data_bitrate": data_bitrate,
    })

    filter_ids = parse_id_list(args.filter_id)
    exclude_ids = parse_id_list(args.exclude_id)

    # 加载 DBC
    db = None
    if args.dbc:
        try:
            import cantools
            db = cantools.database.load_file(args.dbc)
        except ImportError:
            print("警告: cantools 未安装，DBC 解码不可用", file=sys.stderr)
        except Exception as e:
            print(f"警告: 加载 DBC 失败: {e}", file=sys.stderr)

    # 连接总线
    try:
        bus = open_can_bus(config)
        if args.fd and data_bitrate:
            # 重新打开以启用 FD 模式
            bus = can.Bus(
                interface=interface,
                channel=channel,
                bitrate=bitrate,
                fd=True,
                data_bitrate=data_bitrate,
            )
    except Exception as e:
        err = {"status": "error", "action": "monitor", "error": {"code": "interface_open_failed", "message": str(e)}}
        if args.json:
            output_json_line(err)
        else:
            print(f"错误: 无法打开 CAN 接口 — {e}", file=sys.stderr)
        sys.exit(1)

    start = time.time()
    count = 0

    try:
        while True:
            if args.timeout and (time.time() - start) >= args.timeout:
                break

            remaining = None
            if args.timeout:
                remaining = args.timeout - (time.time() - start)
                if remaining <= 0:
                    break

            msg = bus.recv(timeout=min(remaining, 1.0) if remaining else 1.0)
            if msg is None:
                continue

            arb_id = msg.arbitration_id
            if filter_ids and arb_id not in filter_ids:
                continue
            if exclude_ids and arb_id in exclude_ids:
                continue

            count += 1

            # 尝试 DBC 解码
            decoded = None
            if db:
                try:
                    db_msg = db.get_message_by_frame_id(arb_id)
                    decoded = db_msg.decode(msg.data)
                    decoded = {k: round(v, 6) if isinstance(v, float) else v for k, v in decoded.items()}
                except Exception:
                    pass

            if args.json:
                obj = {
                    "timestamp": round(msg.timestamp, 6),
                    "id": f"0x{arb_id:03X}",
                    "dlc": msg.dlc,
                    "data": format_data(msg.data),
                    "is_fd": msg.is_fd,
                }
                if decoded:
                    obj["decoded"] = decoded
                output_json_line(obj)
            else:
                fd_flag = " [FD]" if msg.is_fd else ""
                dec_str = ""
                if decoded:
                    dec_str = " | " + ", ".join(f"{k}={v}" for k, v in decoded.items())
                print(f"[{msg.timestamp:.6f}] 0x{arb_id:03X} [{msg.dlc}] {format_data(msg.data)}{fd_flag}{dec_str}")

    except KeyboardInterrupt:
        pass
    finally:
        bus.shutdown()
        elapsed = time.time() - start
        print(f"\n监控结束: {count} 帧, {elapsed:.1f} 秒", file=sys.stderr)

    # 更新状态
    update_state_entry("last_observe", {
        "type": "can_monitor",
        "interface": interface,
        "channel": channel,
        "frames": count,
        "duration_sec": round(elapsed, 1),
    })


if __name__ == "__main__":
    main()

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

from __future__ import annotations

import json
import platform
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


SKILL_DIR = Path(__file__).resolve().parent.parent
SKILL_NAME = "can"
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 add_can_connection_args(parser, *, include_data_bitrate: bool = False) -> None:
    """为 CLI 补充通用 CAN 连接参数。"""
    parser.add_argument("--interface", help="CAN 后端接口名，如 pcan、vector、socketcan、slcan")
    parser.add_argument("--channel", help="接口通道，如 PCAN_USBBUS1、0、can0")
    parser.add_argument("--bitrate", type=int, help="CAN 仲裁域波特率，默认 500000")
    if include_data_bitrate:
        parser.add_argument("--data-bitrate", dest="data_bitrate", type=int, help="CAN-FD 数据域波特率，默认 2000000")


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:
    """保存状态"""
    file_path = workspace_root(workspace) / STATE_DIR_NAME / STATE_FILE_NAME
    save_json_file(file_path, state)
    return file_path


def update_state_entry(category: str, record: dict, workspace: str | None = None) -> dict:
    """更新状态条目"""
    state = load_workspace_state(workspace)
    state[category] = {**record, "timestamp": record.get("timestamp") or now_iso()}
    file_path = save_workspace_state(state, workspace)
    return {
        "workspace": str(workspace_root(workspace)),
        "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 _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 load_known_devices() -> tuple[list, list]:
    """加载已知的 CAN 设备和接口"""
    try:
        common_path = SKILL_DIR / "references" / "common_interfaces.json"
        data = json.loads(common_path.read_text(encoding="utf-8"))
        return data.get("usb_can_devices", []), data.get("known_interfaces", [])
    except Exception:
        return [], []


def check_interface_available(interface_name: str) -> bool:
    """尝试导入对应后端，判断是否可用"""
    try:
        from can.interfaces import VALID_INTERFACES
        return interface_name in VALID_INTERFACES
    except Exception:
        return False


def scan_usb_can_devices() -> list[dict]:
    """扫描 USB-CAN 设备"""
    known_devices, _ = load_known_devices()
    if not known_devices:
        return []

    found = []
    is_windows = platform.system() == "Windows"

    if is_windows:
        try:
            result = subprocess.run(
                ["powershell", "-Command",
                 "Get-PnpDevice -Class USB -Status OK | Select-Object -Property InstanceId,FriendlyName | ConvertTo-Json"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0 and result.stdout.strip():
                devices = json.loads(result.stdout)
                if isinstance(devices, dict):
                    devices = [devices]
                for dev in devices:
                    instance_id = dev.get("InstanceId", "").upper()
                    friendly = dev.get("FriendlyName", "")
                    for known in known_devices:
                        vid = known["vid"].upper()
                        pid = known["pid"].upper()
                        if f"VID_{vid}" in instance_id and f"PID_{pid}" in instance_id:
                            found.append({
                                "name": known["name"],
                                "vid": vid,
                                "pid": pid,
                                "interface": known["interface"],
                                "channel": known["channel"],
                                "friendly_name": friendly,
                            })
        except Exception:
            pass
    else:
        # Linux: 检查 lsusb
        try:
            result = subprocess.run(["lsusb"], capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                for line in result.stdout.splitlines():
                    line_upper = line.upper()
                    for known in known_devices:
                        vid = known["vid"].upper()
                        pid = known["pid"].upper()
                        if f"{vid}:{pid}" in line_upper:
                            found.append({
                                "name": known["name"],
                                "vid": vid,
                                "pid": pid,
                                "interface": known["interface"],
                                "channel": known["channel"],
                                "friendly_name": line.strip(),
                            })
        except Exception:
            pass

    return found


def scan_socketcan() -> list[dict]:
    """Linux: 扫描 SocketCAN 接口"""
    if platform.system() != "Linux":
        return []
    interfaces = []
    try:
        result = subprocess.run(
            ["ip", "-j", "link", "show", "type", "can"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode == 0 and result.stdout.strip():
            for iface in json.loads(result.stdout):
                interfaces.append({
                    "interface": "socketcan",
                    "channel": iface.get("ifname", ""),
                    "status": iface.get("operstate", "unknown").lower(),
                })
    except Exception:
        pass
    return interfaces


def scan_can_interfaces() -> tuple[list[dict], str | None]:
    """扫描所有可用 CAN 接口"""
    try:
        import can  # noqa: F401
    except ImportError:
        return [], "python-can 未安装，请执行 pip install python-can"

    results = []

    # 1. 扫描 USB-CAN 设备
    for dev in scan_usb_can_devices():
        results.append({
            "interface": dev["interface"],
            "channel": dev["channel"],
            "device": dev["name"],
            "vid": dev["vid"],
            "pid": dev["pid"],
            "status": "detected",
        })

    # 2. 扫描 SocketCAN
    for iface in scan_socketcan():
        results.append({
            "interface": iface["interface"],
            "channel": iface["channel"],
            "device": "",
            "vid": "",
            "pid": "",
            "status": iface["status"],
        })

    # 3. 检查已知后端可用性
    _, known_interfaces = load_known_devices()
    backends_found = {r["interface"] for r in results}
    for ki in known_interfaces:
        iface_name = ki["interface"]
        if iface_name not in backends_found:
            if check_interface_available(iface_name):
                results.append({
                    "interface": iface_name,
                    "channel": "",
                    "device": "",
                    "vid": "",
                    "pid": "",
                    "status": "backend_available",
                })

    return results, None


def get_can_config(
    cli_interface: str | None = None,
    cli_channel: str | None = None,
    cli_bitrate: int | None = None,
    cli_data_bitrate: int | None = None,
    workspace: str | None = None,
) -> tuple[dict, dict]:
    """
    获取 CAN 配置，按优先级解析参数。
    返回 (config_dict, sources_dict)
    """
    local_cfg = load_local_config()
    proj_cfg = load_project_config(workspace)
    state = load_workspace_state(workspace)

    sources = {}

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

    channel, src = resolve_param(
        "channel", cli_channel,
        project_config=proj_cfg, project_keys=["channel"],
        state=state, state_keys=["last_can_channel"],
    )
    sources["channel"] = src or "unknown"

    bitrate, src = resolve_param(
        "bitrate", cli_bitrate,
        project_config=proj_cfg, project_keys=["bitrate"],
        state=state, state_keys=["last_can_bitrate"],
        default=500000,
    )
    sources["bitrate"] = src or "default"

    data_bitrate, src = resolve_param(
        "data_bitrate", cli_data_bitrate,
        project_config=proj_cfg, project_keys=["data_bitrate"],
        state=state, state_keys=["last_can_data_bitrate"],
        default=2000000,
    )
    sources["data_bitrate"] = src or "default"

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

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

    # 获取 slcan 相关环境级配置
    slcan_serial_port = local_cfg.get("slcan_serial_port", "")
    slcan_serial_baudrate = local_cfg.get("slcan_serial_baudrate", 115200)

    config = {
        "interface": interface,
        "channel": channel,
        "bitrate": bitrate,
        "data_bitrate": data_bitrate,
        "log_dir": log_dir,
        "slcan_serial_port": slcan_serial_port,
        "slcan_serial_baudrate": slcan_serial_baudrate,
    }

    return config, sources


def open_can_bus(config: dict):
    """根据配置打开 CAN 总线"""
    import can

    bus_kwargs = {
        "interface": config["interface"],
        "channel": config["channel"],
    }
    if config.get("bitrate"):
        bus_kwargs["bitrate"] = config["bitrate"]

    return can.Bus(**bus_kwargs)


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)

"""CAN 接口扫描：枚举系统可用 CAN 后端与 USB-CAN 设备"""

import argparse
import json
import sys
import platform
from pathlib import Path

CONFIG_PATH = Path(__file__).parent.parent / "config.json"
COMMON_INTERFACES_PATH = Path(__file__).parent.parent / "references" / "common_interfaces.json"


def load_config():
    try:
        return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}


def load_known_devices():
    try:
        data = json.loads(COMMON_INTERFACES_PATH.read_text(encoding="utf-8"))
        return data.get("usb_can_devices", []), data.get("known_interfaces", [])
    except Exception:
        return [], []


def check_interface_available(interface_name):
    """尝试导入对应后端，判断是否可用"""
    try:
        import can
        # 利用 python-can 的接口枚举检测后端是否注册
        from can.interfaces import VALID_INTERFACES
        return interface_name in VALID_INTERFACES
    except Exception:
        return False


def scan_usb_devices():
    """扫描 USB 设备，匹配已知 USB-CAN 适配器"""
    known_devices, _ = load_known_devices()
    if not known_devices:
        return []

    found = []
    is_windows = platform.system() == "Windows"

    if is_windows:
        try:
            import subprocess
            result = subprocess.run(
                ["powershell", "-Command",
                 "Get-PnpDevice -Class USB -Status OK | Select-Object -Property InstanceId,FriendlyName | ConvertTo-Json"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0 and result.stdout.strip():
                devices = json.loads(result.stdout)
                if isinstance(devices, dict):
                    devices = [devices]
                for dev in devices:
                    instance_id = dev.get("InstanceId", "").upper()
                    friendly = dev.get("FriendlyName", "")
                    for known in known_devices:
                        vid = known["vid"].upper()
                        pid = known["pid"].upper()
                        if f"VID_{vid}" in instance_id and f"PID_{pid}" in instance_id:
                            found.append({
                                "name": known["name"],
                                "vid": vid,
                                "pid": pid,
                                "interface": known["interface"],
                                "channel": known["channel"],
                                "friendly_name": friendly,
                            })
        except Exception:
            pass
    else:
        # Linux: 检查 /sys/bus/usb/devices
        try:
            import subprocess
            result = subprocess.run(["lsusb"], capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                for line in result.stdout.splitlines():
                    line_upper = line.upper()
                    for known in known_devices:
                        vid = known["vid"].upper()
                        pid = known["pid"].upper()
                        if f"{vid}:{pid}" in line_upper:
                            found.append({
                                "name": known["name"],
                                "vid": vid,
                                "pid": pid,
                                "interface": known["interface"],
                                "channel": known["channel"],
                                "friendly_name": line.strip(),
                            })
        except Exception:
            pass

    return found


def scan_socketcan():
    """Linux: 扫描 SocketCAN 接口"""
    if platform.system() != "Linux":
        return []
    interfaces = []
    try:
        import subprocess
        result = subprocess.run(
            ["ip", "-j", "link", "show", "type", "can"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode == 0 and result.stdout.strip():
            for iface in json.loads(result.stdout):
                interfaces.append({
                    "interface": "socketcan",
                    "channel": iface.get("ifname", ""),
                    "status": iface.get("operstate", "unknown").lower(),
                })
    except Exception:
        pass
    return interfaces


def scan_interfaces():
    """综合扫描所有可用 CAN 接口"""
    try:
        import can  # noqa: F401
    except ImportError:
        return None, "python-can 未安装，请执行 pip install python-can"

    results = []

    # 1. 扫描 USB-CAN 设备
    usb_devices = scan_usb_devices()
    for dev in usb_devices:
        results.append({
            "interface": dev["interface"],
            "channel": dev["channel"],
            "device": dev["name"],
            "vid": dev["vid"],
            "pid": dev["pid"],
            "status": "detected",
        })

    # 2. 扫描 SocketCAN（Linux）
    for iface in scan_socketcan():
        results.append({
            "interface": iface["interface"],
            "channel": iface["channel"],
            "device": "",
            "vid": "",
            "pid": "",
            "status": iface["status"],
        })

    # 3. 检查已知后端可用性
    _, known_interfaces = load_known_devices()
    backends_found = {r["interface"] for r in results}
    for ki in known_interfaces:
        iface_name = ki["interface"]
        if iface_name not in backends_found:
            if check_interface_available(iface_name):
                results.append({
                    "interface": iface_name,
                    "channel": "",
                    "device": "",
                    "vid": "",
                    "pid": "",
                    "status": "backend_available",
                })

    return results, 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="扫描可用 CAN 接口")
    parser.add_argument("--json", action="store_true", help="JSON 输出")
    args = parser.parse_args()

    interfaces, err = scan_interfaces()

    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)

    result = {
        "status": "ok",
        "action": "scan",
        "summary": f"发现 {len(interfaces)} 个 CAN 接口",
        "details": {"interfaces": interfaces},
    }

    if args.json:
        output_json(result)
    else:
        if not interfaces:
            print("未发现可用 CAN 接口")
        else:
            print(f"发现 {len(interfaces)} 个 CAN 接口:\n")
            for iface in interfaces:
                dev = f" [{iface['device']}]" if iface["device"] else ""
                vid_pid = f" (VID:{iface['vid']} PID:{iface['pid']})" if iface["vid"] else ""
                print(f"  {iface['interface']}:{iface['channel']}{dev}{vid_pid} — {iface['status']}")


if __name__ == "__main__":
    main()

"""CAN 报文发送：支持标准帧、扩展帧、远程帧、CAN-FD 帧"""

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

from can_runtime import (
    add_can_connection_args,
    get_can_config,
    open_can_bus,
    save_project_config,
    update_state_entry,
)


def parse_hex_data(s):
    """解析 hex 数据字符串，支持空格或无空格"""
    s = s.replace(" ", "").replace(",", "")
    return bytes.fromhex(s)


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 output_json_line(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 format_data(data):
    return " ".join(f"{b:02X}" for b in data)


def main():
    parser = argparse.ArgumentParser(description="CAN 报文发送")
    add_can_connection_args(parser, include_data_bitrate=True)
    parser.add_argument("id", help="CAN ID（支持 0x 前缀）")
    parser.add_argument("data", help="数据（Hex 字符串，如 'DE AD BE EF'）")
    parser.add_argument("--extended", action="store_true", help="扩展帧（29位 ID）")
    parser.add_argument("--remote", action="store_true", help="远程帧")
    parser.add_argument("--fd", action="store_true", help="CAN-FD 帧")
    parser.add_argument("--repeat", type=int, default=1, help="重复发送次数")
    parser.add_argument("--interval", type=float, default=0, help="重复发送间隔（秒）")
    parser.add_argument("--periodic", type=float, help="周期发送间隔（毫秒），Ctrl+C 停止")
    parser.add_argument("--listen", action="store_true", help="发送后监听响应")
    parser.add_argument("--json", action="store_true", help="JSON 输出")
    args = parser.parse_args()

    try:
        import can
    except ImportError:
        err = {"status": "error", "action": "send", "error": {"code": "import_error", "message": "python-can 未安装，请执行 pip install python-can"}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    # 获取配置
    config, sources = get_can_config(
        cli_interface=args.interface,
        cli_channel=args.channel,
        cli_bitrate=args.bitrate,
        cli_data_bitrate=args.data_bitrate,
    )

    if config is None:
        if sources.get("need_selection"):
            err = {"status": "error", "action": "send", "error": {"code": "multiple_candidates", "message": f"{sources['error']}，请用 --interface 和 --channel 指定"}}
        else:
            err = {"status": "error", "action": "send", "error": {"code": "config_error", "message": sources.get("error", "配置错误")}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: {err['error']['message']}", file=sys.stderr)
        sys.exit(1)

    interface = config["interface"]
    channel = config["channel"]
    bitrate = config["bitrate"]
    data_bitrate = config["data_bitrate"]

    # 保存确认的配置
    save_project_config(values={
        "interface": interface,
        "channel": channel,
        "bitrate": bitrate,
        "data_bitrate": data_bitrate,
    })

    arb_id = int(args.id, 0)
    data = parse_hex_data(args.data) if not args.remote else b""

    try:
        bus = open_can_bus(config)
        if args.fd and data_bitrate:
            # 重新打开以启用 FD 模式
            bus = can.Bus(
                interface=interface,
                channel=channel,
                bitrate=bitrate,
                fd=True,
                data_bitrate=data_bitrate,
            )
    except Exception as e:
        err = {"status": "error", "action": "send", "error": {"code": "interface_open_failed", "message": str(e)}}
        if args.json:
            output_json(err)
        else:
            print(f"错误: 无法打开 CAN 接口 — {e}", file=sys.stderr)
        sys.exit(1)

    msg = can.Message(
        arbitration_id=arb_id,
        data=data,
        is_extended_id=args.extended,
        is_remote_frame=args.remote,
        is_fd=args.fd,
    )

    tx_count = 0

    try:
        if args.periodic is not None:
            # 周期发送
            period_sec = args.periodic / 1000.0
            print(f"周期发送: 0x{arb_id:03X} 每 {args.periodic:.0f}ms，Ctrl+C 停止", file=sys.stderr)
            while True:
                bus.send(msg)
                tx_count += 1
                time.sleep(period_sec)
        else:
            # 普通发送（可重复）
            for i in range(args.repeat):
                bus.send(msg)
                tx_count += 1
                if args.interval > 0 and i < args.repeat - 1:
                    time.sleep(args.interval)
    except KeyboardInterrupt:
        pass

    # 发送结果
    tx_info = {
        "id": f"0x{arb_id:03X}",
        "data": format_data(data),
        "dlc": len(data),
        "extended": args.extended,
        "remote": args.remote,
        "fd": args.fd,
        "count": tx_count,
    }

    # 监听响应
    rx_list = []
    if args.listen:
        listen_timeout = 2.0
        listen_start = time.time()
        while (time.time() - listen_start) < listen_timeout:
            resp = bus.recv(timeout=0.5)
            if resp and resp.arbitration_id != arb_id:
                rx_entry = {
                    "timestamp": round(resp.timestamp, 6),
                    "id": f"0x{resp.arbitration_id:03X}",
                    "dlc": resp.dlc,
                    "data": format_data(resp.data),
                    "is_fd": resp.is_fd,
                }
                rx_list.append(rx_entry)
                if args.json:
                    output_json_line(rx_entry)
                else:
                    print(f"  <- [{resp.timestamp:.6f}] 0x{resp.arbitration_id:03X} [{resp.dlc}] {format_data(resp.data)}")

    bus.shutdown()

    result = {
        "status": "ok",
        "action": "send",
        "summary": f"已发送 {tx_count} 帧到 0x{arb_id:03X}",
        "details": {"tx": tx_info},
    }
    if args.listen:
        result["details"]["rx"] = rx_list
        result["summary"] += f"，收到 {len(rx_list)} 帧响应"

    if args.json:
        output_json(result)
    else:
        print(f"\n已发送 {tx_count} 帧: 0x{arb_id:03X} [{len(data)}] {format_data(data)}")
        if args.listen:
            print(f"收到 {len(rx_list)} 帧响应")

    # 更新状态
    update_state_entry("last_can_send", {
        "interface": interface,
        "channel": channel,
        "id": f"0x{arb_id:03X}",
        "count": tx_count,
    })


if __name__ == "__main__":
    main()