Adom Desktop - KiCad Bridge

"""KiCad Feature Tour — main orchestrator.

Usage:
    python tour.py [--speed fast|normal|slow] [--start N] [--no-ipc]
"""

import argparse
import json
import sys
import threading
import time

from narration import (
    NarrationPanel, GREEN, RED, YELLOW, BLUE, ACCENT,
    TEXT_MUTED, TEXT_SECONDARY, FG_DIM, FG_SECONDARY,
)
from steps import (
    TOUR_STEPS, DISPLAY, BRIDGE, BRIDGE_VISUAL, IPC_TEST, IPC_LAYER,
    WIN_CLOSE,
)
import bridge_client
import ipc_client
import win_manager


# ── Speed multipliers ────────────────────────────────────────────
SPEED_MULT = {"fast": 0.5, "normal": 1.0, "slow": 1.5}


# ── Data formatters ──────────────────────────────────────────────

def _extract_result(data: dict):
    """Extract the meaningful payload from a bridge response."""
    if "data" in data:
        return data["data"]
    if "result" in data:
        return data["result"]
    return data


def format_json_response(data: dict, indent: int = 2) -> str:
    """Format a bridge response dict into readable text."""
    if not data:
        return "(no data)"
    if "error" in data and not data.get("success", True):
        return f"Error: {data['error']}"
    result = _extract_result(data)
    if isinstance(result, str):
        return result
    if isinstance(result, dict):
        return json.dumps(result, indent=indent, default=str)
    if isinstance(result, list):
        lines = []
        for i, item in enumerate(result[:30]):
            if isinstance(item, dict):
                line = json.dumps(item, default=str)
                if len(line) > 80:
                    line = line[:77] + "..."
                lines.append(f"  {line}")
            else:
                lines.append(f"  {item}")
        if len(result) > 30:
            lines.append(f"  ... and {len(result) - 30} more")
        return f"[{len(result)} items]\n" + "\n".join(lines)
    return str(result)


def format_pcb_stats(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    if not isinstance(r, dict):
        return format_json_response(data)
    lines = []
    if "board_dimensions" in r:
        d = r["board_dimensions"]
        lines.append(f"Board Size:     {d.get('width_mm', '?')} x {d.get('height_mm', '?')} mm")
    for key, label in [("layer_count", "Copper Layers"), ("track_count", "Tracks"),
                        ("via_count", "Vias"), ("pad_count", "Pads"),
                        ("footprint_count", "Footprints"), ("net_count", "Nets"),
                        ("zone_count", "Zones")]:
        if key in r:
            val = r[key]
            lines.append(f"{label + ':':16s} {val:,}" if isinstance(val, int) else f"{label + ':':16s} {val}")
    return "\n".join(lines) if lines else format_json_response(data)


def format_schematic_stats(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    if not isinstance(r, dict):
        return format_json_response(data)
    lines = []
    for key, label in [
        ("component_count", "Components"), ("symbol_count", "Symbols"),
        ("wire_count", "Wires"), ("label_count", "Labels"),
        ("global_label_count", "Global Labels"),
        ("hierarchical_label_count", "Hier. Labels"),
        ("sheet_count", "Sub-sheets"), ("junction_count", "Junctions"),
        ("no_connect_count", "No-connects"), ("bus_count", "Buses"),
        ("power_port_count", "Power Ports"),
    ]:
        if key in r:
            lines.append(f"{label + ':':20s} {r[key]}")
    return "\n".join(lines) if lines else format_json_response(data)


def format_hierarchy(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)

    def _tree(node, prefix="", is_last=True):
        lines = []
        connector = "+-- " if is_last else "|-- "
        name = node.get("name", node.get("sheet_name", "?"))
        filename = node.get("filename", node.get("file", ""))
        label = f"{name}"
        if filename:
            label += f"  ({filename})"
        lines.append(f"{prefix}{connector}{label}")
        children = node.get("children", node.get("sub_sheets", []))
        child_prefix = prefix + ("    " if is_last else "|   ")
        for i, child in enumerate(children):
            lines.extend(_tree(child, child_prefix, i == len(children) - 1))
        return lines

    if isinstance(r, dict):
        name = r.get("name", r.get("sheet_name", "Root"))
        children = r.get("children", r.get("sub_sheets", []))
        lines = [f"[Root] {name}"]
        for i, child in enumerate(children):
            lines.extend(_tree(child, "  ", i == len(children) - 1))
        return "\n".join(lines)
    elif isinstance(r, list):
        return "\n".join(
            format_hierarchy({"result": item}) if isinstance(item, dict) else str(item)
            for item in r
        )
    return format_json_response(data)


def format_components(data: dict) -> str:
    """Format schematic component list."""
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    items = r if isinstance(r, list) else r.get("components", []) if isinstance(r, dict) else []
    lines = [f"Total components: {len(items)}", ""]
    lines.append(f"{'Ref':8s} {'Value':15s} {'Library Symbol'}")
    lines.append("-" * 50)
    for item in items[:25]:
        ref = str(item.get("reference", item.get("ref", "?")))[:7]
        val = str(item.get("value", "?"))[:14]
        lib = str(item.get("library_symbol", item.get("symbol", item.get("lib_id", "?"))))
        if ":" in lib:
            lib = lib.split(":")[-1]
        lines.append(f"{ref:8s} {val:15s} {lib[:25]}")
    if len(items) > 25:
        lines.append(f"... and {len(items) - 25} more")
    return "\n".join(lines)


def format_bom(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    if isinstance(r, dict):
        items = r.get("items", r.get("components", []))
        summary = r.get("summary", {})
    elif isinstance(r, list):
        items = r
        summary = {}
    else:
        return format_json_response(data)
    lines = []
    if summary:
        lines.append(f"Unique parts:  {summary.get('unique_count', len(items))}")
        lines.append(f"Total placed:  {summary.get('total_count', '?')}")
        lines.append("")
    lines.append(f"{'Ref':10s} {'Value':15s} {'Footprint':25s}")
    lines.append("-" * 52)
    for item in items[:25]:
        ref = str(item.get("reference", item.get("ref", "?")))[:9]
        val = str(item.get("value", "?"))[:14]
        fp = str(item.get("footprint", "?")).split(":")[-1][:24]
        lines.append(f"{ref:10s} {val:15s} {fp:25s}")
    if len(items) > 25:
        lines.append(f"... and {len(items) - 25} more components")
    return "\n".join(lines)


def format_nets(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    nets = r if isinstance(r, list) else r.get("nets", []) if isinstance(r, dict) else []
    lines = [f"Total nets: {len(nets)}", ""]
    lines.append(f"{'ID':6s} {'Net Name'}")
    lines.append("-" * 40)
    for net in nets[:30]:
        if isinstance(net, dict):
            nid = str(net.get("id", net.get("number", "?")))
            name = net.get("name", "unnamed")
        else:
            nid = "-"
            name = str(net)
        lines.append(f"{nid:6s} {name}")
    if len(nets) > 30:
        lines.append(f"... and {len(nets) - 30} more nets")
    return "\n".join(lines)


def format_search_results(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    results = r if isinstance(r, list) else r.get("results", r.get("symbols", [])) if isinstance(r, dict) else []
    lines = [f"Found {len(results)} matching symbols:", ""]
    for item in results[:20]:
        if isinstance(item, dict):
            name = item.get("name", item.get("symbol_name", "?"))
            lib = item.get("library", item.get("lib_name", ""))
            desc = item.get("description", "")
            lines.append(f"  {lib}:{name}" if lib else f"  {name}")
            if desc:
                lines.append(f"    {desc[:60]}")
        else:
            lines.append(f"  {item}")
    return "\n".join(lines)


def format_symbol_info(data: dict) -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    if not isinstance(r, dict):
        return format_json_response(data)
    lines = []
    name = r.get("name", r.get("symbol_name", "?"))
    lines.append(f"Symbol: {name}")
    if "description" in r:
        lines.append(f"  {r['description']}")
    lines.append("")
    pins = r.get("pins", [])
    if pins:
        lines.append(f"{'Pin':6s} {'Name':15s} {'Type':12s} {'Dir'}")
        lines.append("-" * 45)
        for pin in pins[:20]:
            num = str(pin.get("number", pin.get("num", "?")))[:5]
            pname = str(pin.get("name", "?"))[:14]
            ptype = str(pin.get("electrical_type", pin.get("type", "?")))[:11]
            direction = str(pin.get("direction", pin.get("orientation", "")))
            lines.append(f"{num:6s} {pname:15s} {ptype:12s} {direction}")
        if len(pins) > 20:
            lines.append(f"... and {len(pins) - 20} more pins")
    return "\n".join(lines)


def format_drc_erc(data: dict, check_type: str = "DRC") -> str:
    r = _extract_result(data)
    if isinstance(r, str):
        try:
            r = json.loads(r)
        except (json.JSONDecodeError, TypeError):
            return str(r)
    if not isinstance(r, dict):
        return format_json_response(data)
    lines = []
    violations = r.get("violations", r.get("errors", []))
    warnings = r.get("warnings", [])
    total = len(violations) + len(warnings)
    if total == 0:
        lines.append(f"{check_type} passed with no violations!")
    else:
        lines.append(f"{check_type}: {len(violations)} violations, {len(warnings)} warnings")
        lines.append("")
        for v in violations[:15]:
            if isinstance(v, dict):
                sev = v.get("severity", "error")
                msg = v.get("message", v.get("description", str(v)))
                lines.append(f"  [{sev}] {msg[:70]}")
            else:
                lines.append(f"  {str(v)[:70]}")
        for w in warnings[:10]:
            if isinstance(w, dict):
                msg = w.get("message", w.get("description", str(w)))
                lines.append(f"  [warn] {msg[:70]}")
            else:
                lines.append(f"  [warn] {str(w)[:70]}")
        remaining = total - min(len(violations), 15) - min(len(warnings), 10)
        if remaining > 0:
            lines.append(f"  ... and {remaining} more")
    return "\n".join(lines)


# Command-specific formatters
FORMATTERS = {
    "get_pcb_statistics": format_pcb_stats,
    "get_schematic_stats": format_schematic_stats,
    "get_schematic_hierarchy": format_hierarchy,
    "list_schematic_components": format_components,
    "generate_bom": format_bom,
    "list_pcb_nets": format_nets,
    "search_symbols": format_search_results,
    "get_symbol_info": format_symbol_info,
    "run_drc": lambda d: format_drc_erc(d, "DRC"),
    "run_erc": lambda d: format_drc_erc(d, "ERC"),
}


class TourRunner:
    """Runs the tour steps on a background thread, updates the narration panel."""

    def __init__(self, panel: NarrationPanel, speed: float = 1.0,
                 start_step: int = 1, skip_ipc: bool = False):
        self.panel = panel
        self.speed = speed
        self.start_step = start_step
        self.skip_ipc = skip_ipc

        self._pause_event = threading.Event()
        self._pause_event.set()
        self._skip_flag = False
        self._quit_flag = False
        self._ipc_cache: dict = {}

    def start(self):
        t = threading.Thread(target=self._run, daemon=True)
        t.start()

    def pause(self):
        self._pause_event.clear()

    def resume(self):
        self._pause_event.set()

    def skip(self):
        self._skip_flag = True
        self._pause_event.set()

    def quit(self):
        self._quit_flag = True
        self._pause_event.set()

    # ── Internal ─────────────────────────────────────────────────

    def _sleep(self, seconds: float):
        end = time.monotonic() + seconds * self.speed
        while time.monotonic() < end:
            if self._quit_flag or self._skip_flag:
                return
            self._pause_event.wait(timeout=0.2)

    def _update_panel(self, func, *args, **kwargs):
        self.panel.root.after(0, lambda: func(*args, **kwargs))

    def _run(self):
        total = len(TOUR_STEPS)
        self._preflight()

        for i, step in enumerate(TOUR_STEPS):
            step_num = i + 1
            if step_num < self.start_step:
                continue
            if self._quit_flag:
                break

            self._skip_flag = False
            self._update_panel(
                self.panel.set_step, step_num, total,
                step["title"], step["description"], step.get("act", ""),
            )

            self._sleep(step.get("delay_before", 1.0))
            if self._quit_flag:
                break
            if self._skip_flag:
                continue

            # Bring KiCad to foreground before every step so user sees the action
            action = step["action"]
            if action not in (DISPLAY,):
                try:
                    win_manager.position_kicad_left(self.panel.width)
                except Exception:
                    pass
                time.sleep(0.3)

            try:
                self._execute_step(step)
            except Exception as e:
                self._update_panel(self.panel.set_status, f"Error: {e}", RED)
                self._update_panel(self.panel.set_data, str(e))
                self._update_panel(self.panel.set_data_color, RED)

            if self._quit_flag:
                break
            if self._skip_flag:
                continue

            self._sleep(step.get("delay_after", 3.0))

        if not self._quit_flag:
            self._update_panel(self.panel.set_status, "Tour complete!", GREEN)

    def _preflight(self):
        self._update_panel(self.panel.set_status, "Preflight checks...", YELLOW)

        health = bridge_client.health_check()
        if health:
            self._update_panel(
                self.panel.set_data,
                f"Bridge server: ONLINE\n  URL: {bridge_client.BRIDGE_URL}"
            )
            self._update_panel(self.panel.set_data_color, GREEN)
        else:
            self._update_panel(
                self.panel.set_data,
                f"Bridge server: OFFLINE\n  URL: {bridge_client.BRIDGE_URL}\n"
                "  Start with: python plugins/kicad/server.py"
            )
            self._update_panel(self.panel.set_data_color, RED)

        time.sleep(1.5)

        if not self.skip_ipc:
            has_test = ipc_client.has_test_binary()
            has_layer = ipc_client.has_layer_binary()
            self._update_panel(
                self.panel.set_data,
                f"IPC test binary:  {'FOUND' if has_test else 'MISSING'}\n"
                f"IPC layer binary: {'FOUND' if has_layer else 'MISSING'}"
            )
            self._update_panel(
                self.panel.set_data_color, GREEN if (has_test and has_layer) else YELLOW
            )
            time.sleep(1.5)

        self._update_panel(self.panel.set_status, "Starting tour...", GREEN)
        time.sleep(1.0)

    def _execute_step(self, step: dict):
        action = step["action"]

        if action == DISPLAY:
            self._execute_display(step)
        elif action == BRIDGE:
            self._execute_bridge(step)
        elif action == BRIDGE_VISUAL:
            self._execute_bridge_visual(step)
        elif action == IPC_TEST:
            self._execute_ipc_test(step)
        elif action == IPC_LAYER:
            self._execute_ipc_layer(step)
        elif action == WIN_CLOSE:
            self._execute_win_close(step)

    def _execute_display(self, step: dict):
        self._update_panel(self.panel.set_status, "", TEXT_MUTED)
        if "Tour Complete" in step["title"]:
            self._update_panel(
                self.panel.set_data,
                "GALLIA FEATURE SUMMARY\n"
                "======================\n\n"
                "Track A  Python Bridge (37 commands):\n"
                "  Schematic: open, stats, hierarchy, components\n"
                "  PCB: open, stats, nets, BOM, DRC, ERC\n"
                "  Library: search, symbol info, footprints\n"
                "  Editors: symbol editor, footprint editor, 3D\n"
                "  Export: Gerber, drill, STEP, BOM\n\n"
                "Track B  Rust IPC (15 commands):\n"
                "  Connection: ping, version, get_api_version\n"
                "  Board: layers, stackup, nets, items\n"
                "  Editor: active layer, selection, refresh\n\n"
                "All commands available to AI agents\n"
                "via the Gallia bridge."
            )
            self._update_panel(self.panel.set_data_color, GREEN)

    def _execute_bridge(self, step: dict):
        """Execute a data-only bridge command and show results."""
        command = step["command"]
        args = step.get("args", {})

        self._update_panel(self.panel.set_status, f"bridge: {command}", ACCENT)

        result = bridge_client.send_command(command, args)
        formatter = FORMATTERS.get(command, format_json_response)
        display_text = formatter(result)

        is_ok = result.get("success", True) and "error" not in result
        self._update_panel(self.panel.set_data_color, GREEN if is_ok else RED)
        self._update_panel(
            self.panel.set_status,
            f"{command}  OK" if is_ok else f"{command}  Error",
            GREEN if is_ok else RED,
        )
        self._update_panel(self.panel.set_data, display_text)

    def _execute_bridge_visual(self, step: dict):
        """Execute a bridge command that opens/changes KiCad visually.

        Supports:
            close_first:   window title to close before opening
            command/args:  bridge command to send
            wait_window:   window title to wait for and position
            post_command:  optional second bridge command to run after (e.g. run_drc)
        """
        command = step["command"]
        args = step.get("args", {})

        # Close previous editor if requested
        close_first = step.get("close_first")
        if close_first:
            self._update_panel(self.panel.set_status, f"Closing {close_first}...", TEXT_SECONDARY)
            win_manager.close_window_by_title(close_first)
            time.sleep(1.5)

        # Send the visual command
        self._update_panel(self.panel.set_status, f"bridge: {command}", ACCENT)
        result = bridge_client.send_command(command, args)

        if result.get("error"):
            self._update_panel(self.panel.set_data, f"Error: {result['error']}")
            self._update_panel(self.panel.set_data_color, RED)
            self._update_panel(self.panel.set_status, f"{command} failed", RED)
            return

        # Wait for the window to appear
        wait_window = step.get("wait_window")
        if wait_window:
            self._update_panel(self.panel.set_status, f"Waiting for {wait_window}...", YELLOW)
            time.sleep(3.0)
            hwnd = win_manager.position_kicad_left(self.panel.width)
            if hwnd:
                self._update_panel(
                    self.panel.set_status, f"{wait_window} ready", GREEN,
                )
            else:
                self._update_panel(
                    self.panel.set_status, f"{wait_window} (window not positioned)", YELLOW,
                )

        # Run post-command if specified (e.g. run_drc after opening the board)
        post_cmd = step.get("post_command")
        if post_cmd:
            post_args = step.get("post_args", {})
            self._update_panel(self.panel.set_status, f"bridge: {post_cmd}", ACCENT)
            post_result = bridge_client.send_command(post_cmd, post_args)
            formatter = FORMATTERS.get(post_cmd, format_json_response)
            display_text = formatter(post_result)
            is_ok = post_result.get("success", True) and "error" not in post_result
            self._update_panel(self.panel.set_data_color, GREEN if is_ok else RED)
            self._update_panel(
                self.panel.set_status,
                f"{post_cmd}  OK" if is_ok else f"{post_cmd}  Error",
                GREEN if is_ok else RED,
            )
            self._update_panel(self.panel.set_data, display_text)
        else:
            # Show the open result
            formatter = FORMATTERS.get(command, format_json_response)
            self._update_panel(self.panel.set_data, formatter(result))
            self._update_panel(self.panel.set_data_color, GREEN)

    def _execute_ipc_test(self, step: dict):
        if self.skip_ipc:
            self._update_panel(self.panel.set_status, "IPC skipped (--no-ipc)", YELLOW)
            self._update_panel(self.panel.set_data, "IPC steps skipped.")
            return

        if not self._ipc_cache:
            self._update_panel(self.panel.set_status, "Running IPC test suite...", ACCENT)
            self._ipc_cache = ipc_client.run_ipc_test()
            if not self._ipc_cache:
                self._update_panel(self.panel.set_status, "IPC binary not found", RED)
                self._update_panel(self.panel.set_data, "test_kicad_ipc.exe not found.")
                self._update_panel(self.panel.set_data_color, RED)
                self.skip_ipc = True
                return

        section = step.get("section", 1)
        text = self._ipc_cache.get(section, f"(Section [{section}] not found)")
        self._update_panel(self.panel.set_data, text)
        self._update_panel(self.panel.set_data_color, GREEN)
        self._update_panel(self.panel.set_status, f"IPC section [{section}]", GREEN)

    def _execute_ipc_layer(self, step: dict):
        if self.skip_ipc:
            self._update_panel(self.panel.set_status, "IPC skipped", YELLOW)
            return

        layer_id = step["layer_id"]
        layer_name = step.get("layer_name", f"layer {layer_id}")

        self._update_panel(
            self.panel.set_status,
            f"IPC: set_active_layer({layer_id})  {layer_name}", ACCENT,
        )

        # Bring KiCad to front so the user sees the change
        win_manager.position_kicad_left(self.panel.width)
        time.sleep(0.5)

        result = ipc_client.set_active_layer(layer_id)
        if result.get("success"):
            self._update_panel(
                self.panel.set_data,
                f"Active layer: {layer_name}\n\n{result.get('output', '')}",
            )
            self._update_panel(self.panel.set_data_color, GREEN)
            self._update_panel(self.panel.set_status, f"Layer: {layer_name}", GREEN)
        else:
            self._update_panel(
                self.panel.set_data,
                f"Layer switch failed:\n{result.get('output', 'Unknown error')}",
            )
            self._update_panel(self.panel.set_data_color, RED)
            self._update_panel(self.panel.set_status, "Layer switch failed", RED)

    def _execute_win_close(self, step: dict):
        title = step["window_title"]
        self._update_panel(self.panel.set_status, f"Closing {title}...", TEXT_SECONDARY)
        success = win_manager.close_window_by_title(title)
        if success:
            self._update_panel(self.panel.set_data, f"{title} closed.")
            self._update_panel(self.panel.set_data_color, GREEN)
            self._update_panel(self.panel.set_status, "Closed", GREEN)
        else:
            self._update_panel(self.panel.set_data, f"'{title}' not found (may already be closed).")
            self._update_panel(self.panel.set_data_color, YELLOW)
            self._update_panel(self.panel.set_status, "Window not found", YELLOW)


# ── Entry point ──────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="KiCad Feature Tour")
    parser.add_argument(
        "--speed", choices=["fast", "normal", "slow"], default="normal",
        help="Tour speed (fast=0.5x delays, slow=1.5x delays)",
    )
    parser.add_argument(
        "--start", type=int, default=1,
        help="Start at step N (1-based)",
    )
    parser.add_argument(
        "--no-ipc", action="store_true",
        help="Skip IPC steps (if binaries not available)",
    )
    args = parser.parse_args()
    speed = SPEED_MULT[args.speed]

    runner = None

    def on_pause():
        if runner:
            runner.pause()

    def on_resume():
        if runner:
            runner.resume()

    def on_skip():
        if runner:
            runner.skip()

    def on_quit():
        if runner:
            runner.quit()

    panel = NarrationPanel(
        width=480,
        on_pause=on_pause,
        on_resume=on_resume,
        on_skip=on_skip,
        on_quit=on_quit,
    )

    runner = TourRunner(
        panel=panel, speed=speed,
        start_step=args.start, skip_ipc=args.no_ipc,
    )
    runner.start()

    panel.root.mainloop()


if __name__ == "__main__":
    main()