adom-desktop-altium-bridge

#!/usr/bin/env python3
"""Adom Altium Bridge — localhost HTTP server (runs on the user's Windows desktop).

Spawned on demand by Adom Desktop (v1.8.0+ dynamic bridge). Implements the standard
dynamic-bridge contract:
    GET  /status            -> {"ok": true, "version": "...", ...}
    POST /command  body {"command"|"verb": "altium_*", "args": {...}}
        -> {"success": true, "output": "<json-string>"}  |  {"success": false, "error", "_hint"}

Phase 0 verbs: altium_ping, altium_detect. Later phases add the IntLib/DbLib/365
component-add verbs and BOM/CPL export (those route to the in-Altium add-in on :8775).

Usage:  python server.py [--port 8801]
"""

import json
import sys
import traceback
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from altium_detect import detect_altium
from addin_client import addin_call

DEFAULT_PORT = 8801

VERSION = (Path(__file__).parent / "BRIDGE_VERSION").read_text().strip() \
    if (Path(__file__).parent / "BRIDGE_VERSION").exists() else "0.0.0"

# Address of the in-Altium add-in (Phase 2+); not used by Phase 0 verbs.
ADDIN_URL = "http://127.0.0.1:8775"


def handle_ping(args: dict) -> dict:
    """Liveness + a quick Altium-presence summary."""
    info = detect_altium()
    return {
        "ok": True,
        "bridge": "altium",
        "version": VERSION,
        "altiumInstalled": info["installed"],
        "altiumVersion": info.get("version"),
        "altiumExe": info.get("exe"),
        "addinUrl": ADDIN_URL,
    }


def handle_detect(args: dict) -> dict:
    """Full Altium install + library-location snapshot for this machine."""
    info = detect_altium()
    info["ok"] = True
    if not info["installed"]:
        info["_hint"] = (
            "Altium Designer not found. Expected C:\\Program Files\\Altium\\AD<NN>\\X2.exe "
            "or an 'Altium Designer' Uninstall registry entry."
        )
    return info


def handle_addin_ping(args: dict) -> dict:
    """Round-trip the in-Altium dispatcher (proves request/response works live)."""
    return addin_call("ping", {}, timeout=args.get("timeout", 60))


def handle_add_component(args: dict) -> dict:
    """Add/refresh a component's identity params (incl. 'LCSC Part #') in a SchLib."""
    if not args.get("schlib"):
        return {"ok": False, "error": "missing 'schlib' (path to the .SchLib on the desktop)"}
    keys = ("schlib", "component", "lcsc", "mpn", "manufacturer",
            "value", "description", "package", "datasheet")
    return addin_call("add_lcsc_param", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 90))


def handle_read_params(args: dict) -> dict:
    """Read back a component's parameters (verification)."""
    return addin_call("read_params",
                      {"schlib": args.get("schlib"), "component": args.get("component")},
                      timeout=args.get("timeout", 60))


def handle_create_and_attach(args: dict) -> dict:
    """Create a fresh SchLib + component, attach LCSC/identity params, read back, save.
    No external symbol source needed — proves the keystone end to end."""
    keys = ("component", "schlib", "lcsc", "mpn", "manufacturer", "value", "description")
    return addin_call("create_and_attach", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 90))


def handle_compile_intlib(args: dict) -> dict:
    """Generate a .LibPkg wrapping the .SchLib, then compile it to .IntLib."""
    import os
    schlib = args.get("schlib") or r"C:\Users\Public\AdomAltium\Adom.SchLib"
    d = os.path.dirname(schlib)
    libname = os.path.splitext(os.path.basename(schlib))[0]
    libpkg = os.path.join(d, libname + ".LibPkg")
    # CreateIntegratedLibrary wants the full .IntLib FILE path, not a folder.
    output = args.get("output_file") or os.path.join(d, libname + ".IntLib")
    # .LibPkg is an INI project file; reference the SchLib by relative name (same dir).
    content = (
        "[Design]\r\n"
        "Version=1.0\r\n"
        "HierarchyMode=1\r\n"
        "\r\n"
        "[Document1]\r\n"
        "DocumentPath=" + os.path.basename(schlib) + "\r\n"
    )
    try:
        with open(libpkg, "w", encoding="utf-8") as fh:
            fh.write(content)
    except OSError as e:
        return {"ok": False, "error": "could not write LibPkg: {}".format(e)}
    return addin_call("compile_intlib", {"libpkg": libpkg, "output": output},
                      timeout=args.get("timeout", 180))


def handle_extract_intlib(args: dict) -> dict:
    """Decompile a compiled .IntLib into editable source libs (.SchLib + .PcbLib);
    returns the extracted document paths so the caller can edit them."""
    if not args.get("intlib"):
        return {"ok": False, "error": "missing 'intlib' (path to the .IntLib on the desktop)"}
    return addin_call("extract_intlib", {"intlib": args.get("intlib")},
                      timeout=args.get("timeout", 150))


def handle_create_resistor(args: dict) -> dict:
    """Create a 2-pin resistor symbol with params + footprint link, save to a Resistors lib."""
    import os
    schlib = args.get("schlib") or r"C:\Users\Public\AdomLibs\Resistors.SchLib"
    try:
        os.makedirs(os.path.dirname(schlib), exist_ok=True)
    except OSError:
        pass
    payload = {
        "schlib": schlib,
        "component": args.get("component") or "R",
        "lcsc": args.get("lcsc"),
        "mpn": args.get("mpn"),
        "manufacturer": args.get("manufacturer"),
        "value": args.get("value"),
        "description": args.get("description"),
        "footprint": args.get("footprint") or "R0603",
        "footprint_lib": args.get("footprint_lib") or "Resistors.PcbLib",
    }
    return addin_call("create_resistor", payload, timeout=args.get("timeout", 120))


def handle_clone_symbol(args: dict) -> dict:
    keys = ("schlib", "template", "component", "footprint", "designator")
    return addin_call("clone_symbol", {k: args.get(k) for k in keys}, timeout=args.get("timeout", 120))


def handle_rename_component(args: dict) -> dict:
    """Rename a SchLib component in place; optionally update Description, footprint
    link, and designator. Preserves the exact symbol graphics/pins/params."""
    keys = ("schlib", "component", "newname", "footprint", "designator", "description")
    return addin_call("rename_component", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 120))


def handle_rename_footprint(args: dict) -> dict:
    """Rename a footprint (IPCB_LibComponent) inside a PcbLib in place."""
    keys = ("pcblib", "footprint", "newname")
    return addin_call("rename_footprint", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 120))


def handle_set_visibility(args: dict) -> dict:
    """Set which parameters are visible on a component (others hidden); controls
    what shows when the symbol is placed on a schematic."""
    keys = ("schlib", "component", "visible")
    return addin_call("set_visibility", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 90))


def handle_add_body(args: dict) -> dict:
    """Attach a simple extruded 3D body (box, sized in mils) to a footprint."""
    keys = ("pcblib", "footprint", "bw_mil", "bh_mil", "h_mil", "so_mil")
    return addin_call("add_body", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 120))


def handle_import_part(args: dict) -> dict:
    """Batched single-round-trip import: from already-extracted src SchLib+PcbLib,
    rename footprint + save-as into the target PcbLib; attach LCSC/params + rename
    symbol + link + save-as into the target SchLib; install both; read back params."""
    keys = ("src_sch", "src_pcb", "tgt_sch", "tgt_pcb", "name", "footprint",
            "designator", "lcsc", "mpn", "manufacturer", "value", "description", "package")
    return addin_call("import_part", {k: args.get(k) for k in keys},
                      timeout=args.get("timeout", 180))


def handle_list_symbols(args: dict) -> dict:
    return addin_call("list_symbols", {"schlib": args.get("schlib")}, timeout=args.get("timeout", 60))


def handle_read_symbol(args: dict) -> dict:
    return addin_call("read_symbol",
                      {"schlib": args.get("schlib"), "component": args.get("component")},
                      timeout=args.get("timeout", 60))


def handle_install_library(args: dict) -> dict:
    """Register a file-based library (already on the desktop) into Available Libraries."""
    return addin_call("install_library", {"path": args.get("path")}, timeout=args.get("timeout", 60))


def handle_build_symbol(args: dict) -> dict:
    """Build/append a symbol (N pins + footprint-implementation link) into a SchLib."""
    return addin_call("build_symbol", args, timeout=args.get("timeout", 120))


def handle_build_footprint(args: dict) -> dict:
    """Build an Altium footprint from a pad spec + optional STEP body. The STEP must
    already be on the desktop (the orchestrator send_files it first). Pads come as
    pad0..padN keys (CommaText: smd,x,y,w,h,drill,name in mils)."""
    return addin_call("build_footprint", args, timeout=args.get("timeout", 150))


def handle_list_docs(args: dict) -> dict:
    return addin_call("list_docs", {}, timeout=args.get("timeout", 60))


def handle_remove_component(args: dict) -> dict:
    return addin_call("remove_component",
                      {"schlib": args.get("schlib"), "component": args.get("component")},
                      timeout=args.get("timeout", 90))


def handle_close_doc(args: dict) -> dict:
    return addin_call("close_doc", {"path": args.get("path")}, timeout=args.get("timeout", 60))


def handle_cleanup_libs(args: dict) -> dict:
    """Uninstall available libraries whose path matches any comma-separated needle."""
    match = args.get("match") or "adomaltium,schlib1,adom.intlib,adom.schlib"
    return addin_call("cleanup_libs", {"match": match}, timeout=args.get("timeout", 60))


COMMAND_HANDLERS = {
    "altium_ping": handle_ping,
    "altium_detect": handle_detect,
    "altium_compile_intlib": handle_compile_intlib,
    "altium_extract_intlib": handle_extract_intlib,
    "altium_cleanup_libs": handle_cleanup_libs,
    "altium_addin_ping": handle_addin_ping,
    "altium_add_component": handle_add_component,
    "altium_read_params": handle_read_params,
    "altium_create_and_attach": handle_create_and_attach,
    "altium_create_resistor": handle_create_resistor,
    "altium_list_docs": handle_list_docs,
    "altium_remove_component": handle_remove_component,
    "altium_close_doc": handle_close_doc,
    "altium_build_footprint": handle_build_footprint,
    "altium_build_symbol": handle_build_symbol,
    "altium_install_library": handle_install_library,
    "altium_list_symbols": handle_list_symbols,
    "altium_read_symbol": handle_read_symbol,
    "altium_clone_symbol": handle_clone_symbol,
    "altium_rename_component": handle_rename_component,
    "altium_rename_footprint": handle_rename_footprint,
    "altium_import_part": handle_import_part,
    "altium_add_body": handle_add_body,
    "altium_set_visibility": handle_set_visibility,
}


def dispatch(verb: str, args: dict) -> dict:
    handler = COMMAND_HANDLERS.get(verb)
    if handler is None:
        return {
            "success": False,
            "error": f"Unknown verb: {verb}",
            "_hint": f"Valid altium verbs: {', '.join(sorted(COMMAND_HANDLERS))}.",
        }
    payload = handler(args or {})
    # Dynamic-bridge contract: success envelope with output as a JSON string.
    return {"success": True, "output": json.dumps(payload)}


class AltiumBridgeHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/status":
            self._respond(200, {"ok": True, "bridge": "altium", "version": VERSION})
        else:
            self._respond(404, {"ok": False, "error": "Not found"})

    def do_POST(self):
        if self.path != "/command":
            self._respond(404, {"success": False, "error": "Not found"})
            return
        length = int(self.headers.get("Content-Length", 0))
        try:
            request = json.loads(self.rfile.read(length) or b"{}")
        except json.JSONDecodeError as e:
            self._respond(400, {"success": False, "error": f"Invalid JSON: {e}"})
            return
        verb = request.get("command") or request.get("verb") or ""
        args = request.get("args", {})
        print(f"[Altium Bridge] {verb} | {json.dumps(args)}")
        try:
            self._respond(200, dispatch(verb, args))
        except Exception as e:
            traceback.print_exc()
            self._respond(500, {"success": False, "error": f"Internal error: {e}"})

    def _respond(self, status: int, data: dict):
        body = json.dumps(data).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        print(f"[Altium Bridge] {args[0]} {args[1]} {args[2]}")


def main():
    port = DEFAULT_PORT
    if "--port" in sys.argv:
        idx = sys.argv.index("--port")
        if idx + 1 < len(sys.argv):
            port = int(sys.argv[idx + 1])
    info = detect_altium()
    print(f"[Altium Bridge] v{VERSION} | Altium installed: {info['installed']} "
          f"({info.get('version')}) exe={info.get('exe')}")
    server = ThreadingHTTPServer(("127.0.0.1", port), AltiumBridgeHandler)
    print(f"[Altium Bridge] Listening on http://127.0.0.1:{port}  (status: /status)")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.server_close()


if __name__ == "__main__":
    main()