adom-desktop-altium-bridge

#!/usr/bin/env python3
"""addin_client — drive the in-Altium AdomAltium dispatcher from the bridge.

Runs ON the Windows desktop (the bridge process). For each call it:
  1. copies the AdomAltium script project into the fixed exchange dir (no spaces),
  2. writes request.txt (key=value lines) + clears any old response.json,
  3. launches X2.EXE with -RScriptingSystem:RunScript(...) — single-instance
     forwarding runs the script inside the already-open Altium,
  4. polls for response.json and returns the parsed result.

The exchange dir is C:\\Users\\Public\\AdomAltium (deliberately space-free so the
RunScript ProjectName needs no quoting, sidestepping cmd/argv escaping issues).

NOTE on speed: Altium has no headless mode and DelphiScript can't host a resident
callback server (a script proc can't bind to a native VCL event), so each command
is one X2.EXE forward. The way we go fast is FEWER commands — see the batched
`import_part` verb, which does a whole part import in a single forward.

A module lock serializes the file exchange because the bridge HTTP server is
threaded (request.txt / response.json are shared, one command at a time).
"""

import json
import os
import shutil
import subprocess
import threading
import time
from pathlib import Path

from altium_detect import detect_altium

EXCHANGE_DIR = r"C:\Users\Public\AdomAltium"
ADDIN_SRC = Path(__file__).parent / "addin" / "AdomAltium"
SCRIPT_FILES = ("AdomAltium.pas", "AdomAltium.PrjScr")

_LOCK = threading.Lock()


def _ensure_scripts():
    os.makedirs(EXCHANGE_DIR, exist_ok=True)
    for fname in SCRIPT_FILES:
        src = ADDIN_SRC / fname
        if src.exists():
            shutil.copyfile(str(src), os.path.join(EXCHANGE_DIR, fname))


def _write_request(command, params):
    lines = ["command=" + command]
    for k, v in (params or {}).items():
        if v is None or v == "":
            continue
        # key=value lines: keep values single-line so TStringList.Values parses cleanly
        lines.append("{}={}".format(k, str(v).replace("\r", " ").replace("\n", " ")))
    with open(os.path.join(EXCHANGE_DIR, "request.txt"), "w", encoding="utf-8") as fh:
        fh.write("\n".join(lines) + "\n")


def addin_call(command, params=None, timeout=90):
    """Invoke a command in the in-Altium dispatcher and return its parsed JSON result."""
    info = detect_altium()
    if not info.get("installed"):
        return {"ok": False, "error": "Altium Designer not installed",
                "_hint": "Run altium_detect to confirm the install path."}
    exe = info["exe"]

    with _LOCK:
        _ensure_scripts()
        resp_path = os.path.join(EXCHANGE_DIR, "response.json")
        try:
            if os.path.exists(resp_path):
                os.remove(resp_path)
        except OSError:
            pass

        _write_request(command, params)

        prjscr = os.path.join(EXCHANGE_DIR, "AdomAltium.PrjScr")  # no spaces -> no quoting
        arg = "-RScriptingSystem:RunScript(ProjectName={}|ProcName=AdomAltium>Run)".format(prjscr)
        try:
            # list form -> no shell, no cmd escaping; arg has no spaces so it passes verbatim
            subprocess.Popen([exe, arg], close_fds=True)
        except Exception as e:
            return {"ok": False, "error": "failed to launch Altium: {}".format(e)}

        deadline = time.time() + timeout
        while time.time() < deadline:
            if os.path.exists(resp_path):
                time.sleep(0.15)  # let the script finish writing
                try:
                    with open(resp_path, "r", encoding="utf-8") as fh:
                        return json.load(fh)
                except (OSError, ValueError):
                    time.sleep(0.25)
                    continue
            time.sleep(0.4)

    return {"ok": False,
            "error": "timeout after {}s waiting for Altium".format(timeout),
            "_hint": "Is Altium running and not blocked by a modal dialog? "
                     "The script writes C:\\Users\\Public\\AdomAltium\\response.json."}