Adom Desktop - KiCad Bridge

"""bridge_client.py - talk to the adom_bridge.py plugin running inside
each KiCad GUI process.

The Phase 2 reverse bridge (see PLAN.md and `adom_bridge.py` deployed
at KiCad's USER_SITE / `usercustomize.py`) exposes a small HTTP/JSON
RPC endpoint inside every running KiCad GUI exe. This module gives the
kicad-bridge HTTP server (this server.py) a thin client for those
endpoints, plus a `bridge_status` handler that enumerates which
processes have an active plugin connection.

Discovery
=========
Each running KiCad process writes a JSON file at
%TEMP%/adom-kicad-bridge-<exeName>.json
with `{port, pid, version, exeName, endpoint, ...}`. We enumerate
those files, then probe each one's /status to verify it's still live
(the file might be stale from a killed KiCad).
"""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Optional

_TEMP = Path(os.environ.get("TEMP") or os.environ.get("TMP") or "/tmp")
# v1.7.6+: discovery files are per-PID. Match BOTH the new per-PID
# pattern AND the v1.7.5 exe-only pattern so the bridge keeps working
# during version skew (user on v1.7.5 desktop binary + v1.7.6 plugin
# payload, or vice versa).
_DISCOVERY_PATTERN = "adom-kicad-bridge-*.json"


def _discovery_files() -> list[Path]:
    return sorted(_TEMP.glob(_DISCOVERY_PATTERN))


def _parse_discovery_name(path: Path) -> tuple[str, int | None]:
    """Parse exeName and optional pid from a discovery filename.

    Per-PID format (v1.7.6+): `adom-kicad-bridge-<exe>-<pid>.json`
    Legacy format (v1.7.5):  `adom-kicad-bridge-<exe>.json`

    Returns (exeName, pid_or_None). exeName is "" if parse fails.
    """
    stem = path.stem  # e.g. "adom-kicad-bridge-eeschema-12345"
    prefix = "adom-kicad-bridge-"
    if not stem.startswith(prefix):
        return ("", None)
    remainder = stem[len(prefix):]
    if not remainder:
        return ("", None)
    parts = remainder.rsplit("-", 1)
    if len(parts) == 2 and parts[1].isdigit():
        return (parts[0], int(parts[1]))
    return (remainder, None)


def _files_for_exe(exe_name: str) -> list[Path]:
    """All discovery files matching this exe (per-PID variants + legacy)."""
    out = []
    for p in _discovery_files():
        ex, _pid = _parse_discovery_name(p)
        if ex == exe_name:
            out.append(p)
    return out


def _read_discovery(path: Path) -> Optional[dict]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None


def _probe(port: int, timeout: float = 1.0) -> Optional[dict]:
    """GET http://127.0.0.1:<port>/status. Returns None if unreachable."""
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/status", timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except (urllib.error.URLError, OSError, json.JSONDecodeError):
        return None


# Per-call probe cache + deadline so handle_bridge_status can never hang no
# matter how many discovery files have accumulated. Many dead python processes
# all register the SAME reverse-plugin port, so probing each FILE is wasteful
# (and a single slow/half-open port × N files = a multi-minute hang — observed
# 2026-06-25 with 288 stale files). We probe each UNIQUE port at most once, and
# stop probing entirely once a wall-clock budget is exhausted.
_PROBE_BUDGET_S = 3.0


def _make_port_prober():
    cache: dict[int, Optional[dict]] = {}
    deadline = time.monotonic() + _PROBE_BUDGET_S

    def probe_port(port: int) -> Optional[dict]:
        if port in cache:
            return cache[port]
        if time.monotonic() > deadline:
            cache[port] = None  # over budget — treat as not-alive without probing
            return None
        result = _probe(port)
        cache[port] = result
        return result

    return probe_port


def call(port: int, method: str, params: Optional[dict] = None, timeout: float = 10.0) -> dict:
    """Make a POST /rpc call. Returns the parsed JSON-RPC response."""
    body = json.dumps({
        "id": f"{method}-{time.time_ns()}",
        "method": method,
        "params": params or {},
    }).encode("utf-8")
    req = urllib.request.Request(
        f"http://127.0.0.1:{port}/rpc",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        # Server returned an HTTP error code; body still has JSON-RPC error
        try:
            return json.loads(e.read().decode("utf-8"))
        except (OSError, json.JSONDecodeError, AttributeError):
            return {"id": None, "error": {"code": -32000, "message": f"http {e.code}: {e.reason}"}}
    except urllib.error.URLError as e:
        return {"id": None, "error": {"code": -32002, "message": f"unreachable: {e.reason}"}}
    except (OSError, json.JSONDecodeError) as e:
        return {"id": None, "error": {"code": -32000, "message": f"transport error: {e}"}}


def handle_bridge_status(kicad_info: dict, args: dict) -> dict:
    """Enumerate all adom-kicad-bridge discovery files; probe each.

    v1.7.6+: handles per-PID discovery files; multiple instances of the
    same exe each get their own row. Optionally filter to a specific
    exeName or pid via args.

    Args (all optional):
      exeName:    filter to one exe (kicad / pcbnew / eeschema / kicad-cli)
      pid:        filter to a specific PID
      pruneStale: if true, delete discovery files for plugins that don't
                  respond to /status (defaults false; left for explicit
                  cleanup runs to avoid racing on a freshly-started bridge)
    """
    filter_exe = args.get("exeName")
    filter_pid = args.get("pid")
    prune_stale = bool(args.get("pruneStale"))

    probe_port = _make_port_prober()  # unique-port cache + wall-clock budget
    plugins = []
    for path in _discovery_files():
        meta = _read_discovery(path)
        if not meta:
            plugins.append({
                "discoveryFile": str(path),
                "alive": False,
                "_error": "could not parse discovery JSON",
            })
            continue
        exe = meta.get("exeName")
        pid = meta.get("pid")
        if filter_exe and exe != filter_exe:
            continue
        if filter_pid is not None and pid != filter_pid:
            continue
        entry = {
            "exeName": exe,
            "pid": pid,
            "port": meta.get("port"),
            "version": meta.get("version"),
            "endpoint": meta.get("endpoint"),
            "startedAt": meta.get("startedAt"),
            "discoveryFile": str(path),
            "alive": False,
        }
        status = probe_port(meta.get("port", 0)) if meta.get("port") else None
        if status:
            entry["alive"] = True
            entry["uptimeMs"] = status.get("uptimeMs")
            entry["requestCount"] = status.get("requestCount")
        elif prune_stale:
            try:
                path.unlink()
                entry["pruned"] = True
            except OSError:
                pass
        plugins.append(entry)

    alive = sum(1 for p in plugins if p.get("alive"))
    return {
        "success": True,
        "plugins": plugins,
        "summary": {"total": len(plugins), "alive": alive},
        "_hint": (
            "If alive=0 but you expect a running KiCad, the adom_bridge.py plugin "
            "isn't deployed at its USER_SITE location. Call kicad_install_plugin "
            "to force a fresh deploy (auto-install runs once per bridge boot; "
            "the explicit verb bypasses the cache). Otherwise the file is stale "
            "— pass pruneStale:true to clean those up. v1.7.6+ supports multiple "
            "instances per exe (one row per PID)."
        ),
    }


def _resolve_plugin(exe_name: str, pid: int | None = None) -> tuple[dict | None, dict | None]:
    """Resolve which discovery file to use for an (exeName, optional pid)
    pair.

    Returns (meta_dict, error_dict). On success: (meta, None). On failure:
    (None, error_dict) where error_dict is ready to return to the caller.

    Picks ALIVE plugins first; if multiple alive instances of the same exe,
    picks the newest one (most recent startedAt).
    """
    paths = _files_for_exe(exe_name)
    if not paths:
        return None, {
            "success": False,
            "error": f"no live plugin for exe '{exe_name}'",
            "errorCode": "plugin_not_running",
            "_hint": (
                f"Either no instance of {exe_name}.exe is running, or it doesn't have "
                "the adom_bridge plugin deployed (Phase 2 not installed for this KiCad). "
                "Run kicad_bridge_status to see what plugins are active. If you expected "
                "the plugin to be there, call kicad_install_plugin to force-deploy + "
                "then ask the user to restart KiCad."
            ),
        }
    candidates: list[dict] = []
    for path in paths:
        meta = _read_discovery(path)
        if not meta:
            continue
        if pid is not None and meta.get("pid") != pid:
            continue
        candidates.append(meta)
    if not candidates:
        return None, {
            "success": False,
            "error": f"no plugin for exe '{exe_name}' pid={pid}",
            "errorCode": "plugin_not_running",
            "_hint": (
                f"Found {len(paths)} discovery file(s) for {exe_name}, but none matched "
                f"pid={pid}. Drop the pid arg to talk to any instance, or call "
                "kicad_bridge_status to see live PIDs."
            ),
        }
    # Probe to find ALIVE ones; prefer the most recently started alive instance.
    alive = []
    for meta in candidates:
        port = meta.get("port")
        if not port:
            continue
        status = _probe(port)
        if status:
            meta = dict(meta)
            meta["_uptimeMs"] = status.get("uptimeMs")
            alive.append(meta)
    if not alive:
        return None, {
            "success": False,
            "error": f"plugin(s) for '{exe_name}' have stale discovery files (not responding to /status)",
            "errorCode": "plugin_not_running",
            "_hint": (
                f"Found {len(candidates)} discovery file(s) for {exe_name} but none responded. "
                f"Likely cause: the KiCad process exited without cleaning up its discovery file. "
                "Call kicad_bridge_status with pruneStale:true to clean them up, then ask the "
                "user to relaunch KiCad."
            ),
        }
    # Pick the newest alive instance.
    alive.sort(key=lambda m: m.get("startedAt", ""), reverse=True)
    return alive[0], None


def handle_bridge_call(kicad_info: dict, args: dict) -> dict:
    """Generic passthrough: call any RPC method on a specific plugin.

    Args:
      exeName:   which KiCad exe to talk to (kicad, pcbnew, eeschema, etc.)
      method:    RPC method name (ping, list_frames, get_menu_ids, wm_command,
                                  get_board_info, get_schematic_info)
      params:    optional dict of method params
      timeout:   optional float (default 10s)
      pid:       optional — pin to a specific KiCad process PID. Without this,
                 picks the newest alive instance.

    Returns the JSON-RPC response wrapped in {"success": ...}.
    """
    exe_name = args.get("exeName")
    method = args.get("method")
    pid = args.get("pid")
    if not exe_name:
        return {"success": False, "error": "exeName is required",
                "_hint": "exeName must be one of: kicad, pcbnew, eeschema, kicad-cli"}
    if not method:
        return {"success": False, "error": "method is required",
                "_hint": "method must be one of: ping, list_frames, get_menu_ids, wm_command, get_board_info, get_schematic_info, get_open_editors"}

    meta, err = _resolve_plugin(exe_name, pid if isinstance(pid, int) else None)
    if err:
        return err
    port = meta["port"]

    timeout = float(args.get("timeout") or 10.0)
    rpc_response = call(port, method, args.get("params") or {}, timeout=timeout)
    if "error" in rpc_response:
        return {
            "success": False,
            "error": rpc_response["error"].get("message", "rpc error"),
            "errorCode": "rpc_error",
            "rpcError": rpc_response["error"],
            "exeName": exe_name,
            "pid": meta.get("pid"),
            "port": port,
        }
    return {
        "success": True,
        "result": rpc_response.get("result"),
        "exeName": exe_name,
        "pid": meta.get("pid"),
        "port": port,
    }


def handle_open_editors(kicad_info: dict, args: dict) -> dict:
    """Cross-process editor inventory: hit every alive plugin's
    get_open_editors method and concatenate. Replaces the old
    handle_window_info screen-scraping for the common case.

    Returns:
      {success, editors:[{kind, exeName, pid, frameTitle, hwnd, isShown}],
       byKind:{schematic_editor:[...], pcb_editor:[...], ...}, summary:{total, byKind:{...}}}

    Args (all optional):
      includeStale: include exited-process discovery files (default false)
    """
    out_editors: list[dict] = []
    by_kind: dict[str, list[dict]] = {}
    probed_plugins = 0
    failed_plugins = 0

    # Use the BUDGETED port-prober (3s wall-clock cap + per-port dedup), not raw
    # per-file 1s probes: many dead python processes register the SAME plugin port,
    # and probing each of N stale discovery files at 1s each hung open_editors for
    # the full 30s client timeout when no KiCad was live. (AdomLapper 2026-07.)
    probe_port = _make_port_prober()
    seen_ports: set = set()
    for path in _discovery_files():
        meta = _read_discovery(path)
        if not meta:
            continue
        port = meta.get("port")
        if not port or port in seen_ports:
            continue
        seen_ports.add(port)
        # Probe alive first (budgeted — returns None instantly once over budget)
        status = probe_port(port)
        if not status:
            continue
        probed_plugins += 1
        rpc = call(port, "get_open_editors", {}, timeout=2.0)
        if "error" in rpc:
            failed_plugins += 1
            continue
        result = rpc.get("result") or {}
        if not result.get("available"):
            continue
        for ed in result.get("editors", []):
            row = {
                "kind": ed.get("kind"),
                "exeName": result.get("exeName"),
                "pid": result.get("pid"),
                "frameClass": ed.get("frameClass"),
                "frameTitle": ed.get("frameTitle"),
                "hwnd": ed.get("hwnd"),
                "isShown": ed.get("isShown"),
            }
            out_editors.append(row)
            by_kind.setdefault(row["kind"], []).append(row)

    return {
        "success": True,
        "editors": out_editors,
        "byKind": by_kind,
        "summary": {
            "totalEditors": len(out_editors),
            "probedPlugins": probed_plugins,
            "failedPlugins": failed_plugins,
            "byKindCounts": {k: len(v) for k, v in by_kind.items()},
        },
        "_hint": (
            "Cross-process inventory: each KiCad GUI exe's plugin contributes its own "
            "wx top-level frames. If totalEditors=0 but the user has KiCad windows open, "
            "the plugins likely haven't deployed yet OR KiCad was running before "
            "v1.7.5 plugin install (those instances need a restart to load the bridge). "
            "Call kicad_bridge_status to confirm plugins are alive."
        ),
    }


# ── Phase 3 — three-tier fallback for menu-driven actions ──────────────

def try_wm_command_via_bridge(exe_name: str, menu_id: int, frame_index: int = 0,
                               pid: int | None = None, timeout: float = 10.0) -> dict:
    """Tier 1 of Phase 3's 3-tier fallback chain.

    Send a `wm_command` RPC to the bridge. Returns:
      {success: True,  pathway: "plugin", pid, port, ...}  on success
      {success: False, pathway: "plugin", reason: "..."}    on any failure

    Callers should fall through to tier-2 (Win32 PostMessage from outside)
    when success is False. Tier-3 is SendKeys via menu nav (out of MVP).
    """
    meta, err = _resolve_plugin(exe_name, pid)
    if err:
        return {"success": False, "pathway": "plugin",
                "reason": err.get("error", "plugin not available"),
                "errorCode": err.get("errorCode")}
    rpc = call(meta["port"], "wm_command",
               {"frame_index": frame_index, "menu_id": menu_id},
               timeout=timeout)
    if "error" in rpc:
        return {"success": False, "pathway": "plugin",
                "reason": rpc["error"].get("message", "rpc error"),
                "errorCode": "rpc_error"}
    result = rpc.get("result") or {}
    return {
        "success": bool(result.get("posted")),
        "pathway": "plugin",
        "exeName": exe_name,
        "pid": meta.get("pid"),
        "port": meta.get("port"),
        "frame": {
            "frameClass": result.get("frameClass"),
            "frameTitle": result.get("frameTitle"),
            "hwnd": result.get("hwnd"),
        },
        "menuId": menu_id,
    }


def resolve_frame_index_via_bridge(exe_name: str, title_contains: str,
                                   pid: int | None = None, timeout: float = 10.0) -> int | None:
    """Return the index of the plugin's frame whose title contains `title_contains`
    (case-insensitive), or None. In KiCad 10 the Footprint Editor is a frame INSIDE
    the pcbnew process, so `list_frames` exposes it at some index (usually 1 when a
    board is also open); `wm_command` honors that frame_index even though
    `get_menu_ids` currently only reports frame 0."""
    meta, err = _resolve_plugin(exe_name, pid)
    if err:
        return None
    rpc = call(meta["port"], "list_frames", {}, timeout=timeout)
    if "error" in rpc:
        return None
    needle = title_contains.lower()
    for i, fr in enumerate((rpc.get("result") or {}).get("frames", [])):
        if needle in str(fr.get("frameTitle", "")).lower():
            return i
    return None


def resolve_menu_id_via_bridge(exe_name: str, contains: str, frame_index: int = 0,
                               pid: int | None = None, timeout: float = 10.0) -> int | None:
    """Look up a live menu-command id by matching text — menu ids DRIFT across KiCad
    builds/versions (e.g. the PCB '3D Viewer' id was 20563 on one build, 20568 on
    another), so hardcoding one silently breaks on the next release. Ask the plugin
    for the real menu tree (`get_menu_ids`) and return the id of the first item whose
    label / helpString / accelerator contains `contains` (case-insensitive).
    Returns None if the plugin isn't reachable or nothing matches."""
    meta, err = _resolve_plugin(exe_name, pid)
    if err:
        return None
    rpc = call(meta["port"], "get_menu_ids", {"frame_index": frame_index}, timeout=timeout)
    if "error" in rpc:
        return None
    result = rpc.get("result") or {}
    needle = contains.lower()
    for frame in result.get("frames", []):
        for it in frame.get("menuItems", []):
            hay = " ".join(str(it.get(k, "")) for k in ("label", "helpString", "accelerator")).lower()
            if needle in hay:
                try:
                    return int(it.get("id"))
                except (TypeError, ValueError):
                    continue
    return None