Altium - the Altium Bridge

"""altium_detect.py — where is Altium Designer, is it running, which windows are its own.

Detection facts (measured on AD 26.9.1, 2026-08-29):
  * install root  C:\\Program Files\\Altium\\AD<nn>\\X2.EXE   (one dir per major version)
  * process name  X2.exe; the main window class is TDocumentForm and its title ends
    with "- Altium Designer (26.9.1) - <licence state>" ("No License" / "Trial" / …)
  * a second X2.EXE launched with a -R<Server>:<Command>(…) argument FORWARDS the
    command to the running instance and exits (single-instance); it only spawns a
    new instance when none is running.
  * modal script errors are owned dialogs titled exactly "X2" (VCL TMessageForm) —
    see dialogs.py.

Never read a user-profile path from ONE env var (ab strips the bridge's env; see
kicad-bridge's LOCALAPPDATA lesson) — derive from several and take the first hit.
"""
from __future__ import annotations

import glob
import os
import re

import caller_identity
import subprocess
import sys
from pathlib import Path

IS_WINDOWS = sys.platform == "win32"
EXE_NAME = "X2.EXE"
PROC_NAME = "X2"
HOST_APP = "Altium Designer"

_TITLE_RE = re.compile(
    r"Altium Designer"
    r"(?: (?P<edition>Develop|Viewer|Standard|Professional|Pro|Enterprise))?"
    r"(?: \((?P<ver>[\d.]+)\))?"
    r"(?: - (?P<lic>.+))?$"
)  # issue #2/PR #64: ADDevelop titles as "…- Altium Designer Develop" (no version parens)


def _program_files_roots() -> list[Path]:
    roots = []
    for var in ("ProgramFiles", "ProgramW6432", "ProgramFiles(x86)"):
        v = os.environ.get(var)
        if v:
            roots.append(Path(v))
    for fixed in (r"C:\Program Files", r"C:\Program Files (x86)"):
        roots.append(Path(fixed))
    seen, out = set(), []
    for r in roots:
        k = str(r).lower()
        if k not in seen:
            seen.add(k)
            out.append(r)
    return out


def _user_profile_candidates() -> list[Path]:
    c = []
    for var in ("USERPROFILE",):
        v = os.environ.get(var)
        if v:
            c.append(Path(v))
    try:
        c.append(Path.home())
    except Exception:
        pass
    return c


def appdata_dir() -> Path | None:
    """%APPDATA%\\Altium — where prefs/logs live. Tolerates a stripped env."""
    for var in ("APPDATA",):
        v = os.environ.get(var)
        if v and (Path(v) / "Altium").exists():
            return Path(v) / "Altium"
    for up in _user_profile_candidates():
        p = up / "AppData" / "Roaming" / "Altium"
        if p.exists():
            return p
    return None


def _version_from_dir(d: Path) -> str:
    m = re.search(r"AD(\d+)$", d.name)
    return m.group(1) if m else d.name


def _exe_file_version(exe: Path) -> str | None:
    if not IS_WINDOWS:
        return None
    try:
        import ctypes
        from ctypes import wintypes
        ver = ctypes.windll.version
        size = ver.GetFileVersionInfoSizeW(str(exe), None)
        if not size:
            return None
        buf = ctypes.create_string_buffer(size)
        ver.GetFileVersionInfoW(str(exe), 0, size, buf)
        lp = ctypes.c_void_p()
        ln = wintypes.UINT()
        ver.VerQueryValueW(buf, "\\", ctypes.byref(lp), ctypes.byref(ln))
        class VS_FIXEDFILEINFO(ctypes.Structure):
            _fields_ = [("dwSignature", wintypes.DWORD), ("dwStrucVersion", wintypes.DWORD),
                        ("dwFileVersionMS", wintypes.DWORD), ("dwFileVersionLS", wintypes.DWORD),
                        ("dwProductVersionMS", wintypes.DWORD), ("dwProductVersionLS", wintypes.DWORD)]
        ffi = ctypes.cast(lp, ctypes.POINTER(VS_FIXEDFILEINFO)).contents
        ms, ls = ffi.dwFileVersionMS, ffi.dwFileVersionLS
        return f"{ms >> 16}.{ms & 0xFFFF}.{ls >> 16}.{ls & 0xFFFF}"
    except Exception:
        return None


def detect_installs() -> list[dict]:
    """Every Altium install on the box, newest major first."""
    found = {}
    for root in _program_files_roots():
        for exe in glob.glob(str(root / "Altium" / "AD*" / EXE_NAME)):
            p = Path(exe)
            d = p.parent
            found[str(d).lower()] = {
                "installed": True,
                "base_dir": str(d),
                "exe": str(p),
                "system_dir": str(d / "System"),
                "major": _version_from_dir(d),
                "fileVersion": _exe_file_version(p),
            }
    installs = list(found.values())
    installs.sort(key=lambda i: int(i["major"]) if str(i["major"]).isdigit() else 0, reverse=True)
    return installs


def running_pids() -> list[int]:
    """PIDs of every X2.exe — via a Toolhelp32 snapshot (ctypes, no subprocess).
    The earlier `tasklist` probe took 7+ s under load and timed out, which made
    every verb report 'Altium not running' while its window was plainly there."""
    if not IS_WINDOWS:
        return []
    import ctypes
    from ctypes import wintypes
    TH32CS_SNAPPROCESS = 0x00000002

    class PROCESSENTRY32W(ctypes.Structure):
        _fields_ = [("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD),
                    ("th32ProcessID", wintypes.DWORD), ("th32DefaultHeapID", ctypes.POINTER(ctypes.c_ulong)),
                    ("th32ModuleID", wintypes.DWORD), ("cntThreads", wintypes.DWORD),
                    ("th32ParentProcessID", wintypes.DWORD), ("pcPriClassBase", ctypes.c_long),
                    ("dwFlags", wintypes.DWORD), ("szExeFile", ctypes.c_wchar * 260)]

    k32 = ctypes.windll.kernel32
    snap = k32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
    if snap == wintypes.HANDLE(-1).value or snap == -1:
        return []
    pids = []
    try:
        pe = PROCESSENTRY32W()
        pe.dwSize = ctypes.sizeof(PROCESSENTRY32W)
        ok = k32.Process32FirstW(snap, ctypes.byref(pe))
        while ok:
            if pe.szExeFile.lower() == EXE_NAME.lower():
                pids.append(int(pe.th32ProcessID))
            ok = k32.Process32NextW(snap, ctypes.byref(pe))
    finally:
        k32.CloseHandle(snap)
    return pids


# ---- windows owned by Altium ------------------------------------------------

def _enum_windows_for_pids(pids: set[int]) -> list[dict]:
    """Top-level windows whose owning process is one of `pids` (ctypes, no deps)."""
    if not IS_WINDOWS or not pids:
        return []
    import ctypes
    from ctypes import wintypes
    user32 = ctypes.windll.user32
    res = []

    @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
    def cb(hwnd, _):
        pid = wintypes.DWORD()
        user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
        if pid.value in pids:
            length = user32.GetWindowTextLengthW(hwnd)
            buf = ctypes.create_unicode_buffer(length + 1)
            user32.GetWindowTextW(hwnd, buf, length + 1)
            cls = ctypes.create_unicode_buffer(256)
            user32.GetClassNameW(hwnd, cls, 256)
            rect = wintypes.RECT()
            user32.GetWindowRect(hwnd, ctypes.byref(rect))
            res.append({
                "hwnd": int(hwnd), "pid": int(pid.value),
                "title": buf.value, "className": cls.value,
                "visible": bool(user32.IsWindowVisible(hwnd)),
                "enabled": bool(user32.IsWindowEnabled(hwnd)),
                "rect": {"left": rect.left, "top": rect.top, "right": rect.right, "bottom": rect.bottom,
                         "width": rect.right - rect.left, "height": rect.bottom - rect.top},
            })
        return True

    user32.EnumWindows(cb, 0)
    return res


def altium_windows() -> list[dict]:
    """Every visible top-level window owned by an X2.exe, classified."""
    wins = _enum_windows_for_pids(set(running_pids()))
    out = []
    for w in wins:
        if not w["visible"] or (w["rect"]["width"] < 20 and w["rect"]["height"] < 20):
            continue
        t = w["title"] or ""
        m = _TITLE_RE.search(t)
        kind = "other"
        # A match must carry a version, an edition word, or a licence tail — or BE the whole
        # title — so a bare mid-string "Altium Designer" inside a dialog cannot pose as main.
        if m and (m.group("ver") or m.group("edition") or m.group("lic") or m.group(0) == t):
            kind = "main"
            w["adVersion"] = m.group("ver")
            w["edition"] = m.group("edition")
            w["licenceState"] = (m.group("lic") or "").strip() or None
            w["document"] = t.split(" - Altium Designer")[0].strip() if " - Altium Designer" in t else None
        elif t == "X2":
            kind = "script_error_modal"
        elif w["className"] in ("TMessageForm", "#32770"):
            kind = "dialog"
        w["kind"] = kind
        out.append(w)
    # main window first, then dialogs, then the rest
    order = {"main": 0, "script_error_modal": 1, "dialog": 2, "other": 3}
    out.sort(key=lambda w: order.get(w["kind"], 9))
    return out


def main_window() -> dict | None:
    """The main window of the SELECTED instance, or the only one.

    Instance-awareness has to reach this far down: `licence_state()` is built on it, and
    `_need_licensed()` is built on that. With two Altiums running, returning "the first main window"
    meant every licence gate judged the WRONG instance — a verb targeting Develop was refused
    because Designer held no seat (measured 2026-09-01, right after the instance layer landed)."""
    wins = [w for w in altium_windows() if w["kind"] == "main"]
    if not wins:
        return None
    try:
        from handlers import agent as _agent
        sel = _agent.current_instance()
    except Exception:
        sel = None
    if sel and sel.get("pid"):
        for w in wins:
            if w.get("pid") == sel["pid"]:
                return w
    return wins[0]


def licence_state() -> dict:
    """Parsed from the main window title (the cheapest truthful source)."""
    mw = main_window()
    if not mw:
        return {"running": False, "licensed": None, "state": None}
    st = mw.get("licenceState")
    licensed = None
    if st:
        licensed = st.lower() not in ("no license", "no licence", "viewer")
    elif mw.get("edition") == "Develop":
        # Altium Designer DEVELOP does not put a licence tail in its title when it HOLDS a seat —
        # it only writes one when it does not ("- Viewer"). Reading "no tail" as "no licence" made
        # `instances` report ADD unlicensed while its agent happily imported five parts
        # (measured 2026-09-01). Absent tail on Develop means licensed, kind unknown.
        licensed, st = True, "Licensed (Develop reports no licence tail)"
    return {"running": True, "licensed": licensed, "state": st, "adVersion": mw.get("adVersion"),
            "hwnd": mw["hwnd"], "pid": mw["pid"]}


# ---- instances: more than one Altium can run at once (John, 2026-09-01) ---------------------
#
# John runs Altium Designer Develop alongside Altium Designer so two AI threads can drive both.
# Everything below therefore treats "the Altium" as a CHOICE, not a given. The bridge previously
# assumed one instance in three places and each was wrong with two: the launcher hardcoded the
# AD26 exe path, `licence_state()` reported whichever main window sorted first, and the dialog
# sweep covered every X2 pid (so one thread could clear a modal belonging to the other).

def _exe_of_pid(pid: int) -> str | None:
    """The process's own image path — never a constant. Develop and Designer live in different
    install folders, so the exe to spawn a -R helper against depends on WHICH instance you mean."""
    try:
        r = caller_identity.ad_callback("process_list", {"pid": pid}, on_behalf=False, timeout=10) or {}
        for p in r.get("processes") or []:
            if p.get("pid") == pid:
                cl = (p.get("commandLine") or "").strip()
                if cl.startswith('"'):
                    return cl[1:].split('"', 1)[0]
                return cl.split(" ")[0] or None
    except Exception:
        pass
    return None


def _key_for(exe: str | None, pid: int) -> str:
    """A STABLE per-instance key, used for the agent workspace folder.

    The install folder name ("AD26", "ADDevelop") survives a restart, which a pid does not — so a
    resident agent is still recognised after Altium is relaunched. Two instances from the SAME
    install fall back to the pid, which is the honest answer: they are genuinely distinguishable
    only by process."""
    if exe:
        parts = [q for q in exe.replace("/", "\\").split("\\") if q]
        if len(parts) >= 2:
            return parts[-2]
    return f"pid{pid}"


def instances() -> list[dict]:
    """Every running Altium, each with the identity a caller needs to choose between them."""
    out = []
    seen = set()
    for w in altium_windows():
        if w["kind"] != "main" or w["pid"] in seen:
            continue
        seen.add(w["pid"])
        exe = _exe_of_pid(w["pid"])
        st = w.get("licenceState")
        licensed = (st.lower() not in ("no license", "no licence", "viewer")) if st else None
        if st is None and (w.get("edition") == "Develop"):
            # Develop writes NO licence tail when it HOLDS the seat (only "- Viewer" when it does
            # not) — reading the absence as "unknown" reported ADD unlicensed for a whole test while
            # its agent imported five parts (2026-09-01).
            licensed, st = True, "Licensed (Develop reports no licence tail)"
        out.append({
            "pid": w["pid"],
            "key": _key_for(exe, w["pid"]),
            "edition": w.get("edition") or "Designer",
            "adVersion": w.get("adVersion"),
            "licenceState": st,
            "licensed": licensed,
            "exe": exe,
            "hwnd": w["hwnd"],
            "document": w.get("document"),
            "title": w.get("title"),
        })
    out.sort(key=lambda i: i["pid"])
    return out


def resolve_instance(selector=None) -> dict:
    """Pick the instance a verb should act on.

    No selector + exactly one Altium  -> that one (the normal case; nothing changes for users).
    No selector + several            -> REFUSE and list them. Guessing here means one thread's verb
                                        lands in another thread's Altium, which is worse than an error.
    Selector                         -> match on pid, key, or a case-insensitive substring of the
                                        edition/title (so "develop" works).
    """
    found = instances()
    if not found:
        return {"error": "no Altium is running", "errorCode": "altium_not_running", "instances": []}
    if selector in (None, "", "auto"):
        if len(found) == 1:
            return {"instance": found[0]}
        return {"error": f"{len(found)} Altium instances are running — say which one",
                "errorCode": "ambiguous_instance", "instances": found,
                "_hint": "pass instance: a pid, a key (e.g. 'ADDevelop'), or a word from the edition/title (e.g. 'develop')"}
    s = str(selector).strip().lower()
    for i in found:
        if s == str(i["pid"]) or s == (i["key"] or "").lower():
            return {"instance": i}
    for i in found:
        hay = f"{i.get('edition') or ''} {i.get('key') or ''} {i.get('title') or ''}".lower()
        if s in hay:
            return {"instance": i}
    return {"error": f"no running Altium matches instance={selector!r}",
            "errorCode": "instance_not_found", "instances": found}