Altium - the Altium Bridge

"""taskbar.py — "Adom is driving Altium" on the Windows taskbar, the same way the
KiCad and Fusion bridges do it.

Two signals, both painted by ab-core's generic `desktop_taskbar` verb (never by an
in-process ITaskbarList3 call — that returns S_OK cross-process and the shell
silently ignores it; pup, fusion and kicad each paid for that lesson):

  * OVERLAY BADGE — the Adom mark (resources/adom-overlay-32.png, anchored top-right
    of a transparent 32 px canvas) on Altium's taskbar button while the bridge is
    driving it; cleared when the bridge goes idle.
        set:   desktop_taskbar {hwnd, overlay:{icon:<b64 png>, tooltip}}
        clear: desktop_taskbar {hwnd, overlay:{badge:"none"}}
  * PROGRESS BAR — for slow verbs, paced by this machine's MEASURED estimate
    (handlers/progress.py); marquee when there is no history yet; red flash on
    failure; cleared when the verb returns. Also cancels Windows' own orange
    attention flash on the button (John: "can u not orange flash the taskbar").

NO AUMID / window-identity stamping. KiCad learned the hard way that a per-window
AppUserModelID poisons the shell's icon cache; Altium's own taskbar identity is
left exactly as Altium set it. Badge + progress ride ITaskbarList3 by hwnd only.
"""
from __future__ import annotations

import base64
import os
import threading
import time

_LOCK = threading.Lock()
_B64: str | None = None
_BADGED: set[int] = set()
_LAST_ERROR = [None]
_PROGRESS_STOP: dict[str, threading.Event] = {}


def last_error():
    return _LAST_ERROR[0]


def _badge_b64() -> str:
    global _B64
    if _B64 is None:
        p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "resources", "adom-overlay-32.png")
        try:
            _B64 = base64.b64encode(open(p, "rb").read()).decode()
        except Exception:
            _B64 = ""
    return _B64


def _taskbar(args: dict) -> bool:
    try:
        import caller_identity
        from handlers import ad_client
        if not ad_client.available():
            _LAST_ERROR[0] = "ab direct API unavailable"
            return False
        r = caller_identity.ad_callback("desktop_taskbar", args, timeout=15)
        raw = str(r)
        if r and ("Taskbar updated" in raw or (isinstance(r, dict) and r.get("success"))):
            _LAST_ERROR[0] = None
            return True
        _LAST_ERROR[0] = f"desktop_taskbar reply: {raw[:160]}"
        return False
    except Exception as e:
        _LAST_ERROR[0] = f"exception: {str(e)[:160]}"
        return False


def _main_hwnd() -> int | None:
    try:
        from handlers import altium_detect
        mw = altium_detect.main_window()
        return mw["hwnd"] if mw else None
    except Exception:
        return None


# ---- badge -----------------------------------------------------------------------

def set_driving_badge(hwnd: int | None = None) -> bool:
    hwnd = hwnd or _main_hwnd()
    if not hwnd:
        return False
    if hwnd in _BADGED:
        return True
    icon = _badge_b64()
    if not icon:
        _LAST_ERROR[0] = "badge asset missing (resources/adom-overlay-32.png)"
        return False
    with _LOCK:
        ok = _taskbar({"hwnd": int(hwnd), "overlay": {"icon": icon, "tooltip": "Adom is driving Altium Designer"}})
        if ok:
            _BADGED.add(int(hwnd))
        return ok


def clear_driving_badge(hwnd: int | None = None) -> bool:
    targets = [int(hwnd)] if hwnd else list(_BADGED)
    ok = True
    with _LOCK:
        for h in targets:
            if _taskbar({"hwnd": h, "overlay": {"badge": "none"}}):
                _BADGED.discard(h)
            else:
                ok = False
    return ok


def badge_active() -> bool:
    return bool(_BADGED)


# ---- progress ---------------------------------------------------------------------

def _taskbar_progress(state: str, value=None) -> bool:
    hwnd = _main_hwnd()
    if not hwnd:
        return False
    prog = {"state": state}
    if value is not None:
        prog["value"] = max(0, min(100, int(value)))
    return _taskbar({"hwnd": int(hwnd), "progress": prog})


def start_activity_progress(verb: str, estimate_s: float | None, confidence: str = "typical"):
    """Begin a progress bar for `verb`, paced by its measured estimate. Returns the stopper."""
    stop = threading.Event()
    _PROGRESS_STOP[verb] = stop

    def _tick():
        started = time.time()
        if not estimate_s or confidence == "unknown":
            _taskbar_progress("indeterminate")
            while not stop.wait(1.0):
                pass
            return
        fails = 0
        while not stop.wait(0.7):
            pct = min(95, int(95 * (time.time() - started) / estimate_s)) if estimate_s > 0 else 0
            if _taskbar_progress("normal", pct):
                fails = 0
            else:
                fails += 1          # window may not exist yet during a cold launch — keep trying
                if fails > 120:
                    return

    threading.Thread(target=_tick, daemon=True).start()
    return stop


def end_activity_progress(verb: str, failed: bool = False) -> None:
    stop = _PROGRESS_STOP.pop(verb, None)
    if stop:
        stop.set()
    try:
        if failed:
            _taskbar_progress("error", 100)
            time.sleep(1.2)
        _taskbar_progress("none")
        hwnd = _main_hwnd()
        if hwnd:
            _taskbar({"hwnd": int(hwnd), "flash": "stop"})
    except Exception:
        pass


# ---- idle auto-clear (John, 2026-09-01: "do you auto-remove that overlay after 5 minutes of
# verb inactivity?") ---------------------------------------------------------------------------
#
# He asked because the answer was NO: the badge was set on the first driving verb and never
# cleared until the bridge died — the header above always PROMISED "cleared when the bridge goes
# idle", but nothing implemented it. A permanent overlay is worse than none: it stops meaning
# "an AI is driving right now" and starts meaning nothing.

IDLE_CLEAR_SEC = 300.0
_LAST_DRIVING = [0.0]
_IDLE_WATCHER = [False]


def note_driving() -> None:
    """Called by dispatch on every driving verb; feeds the idle clock."""
    _LAST_DRIVING[0] = time.time()


def start_idle_watcher() -> bool:
    """Clear the overlay after IDLE_CLEAR_SEC without a driving verb. Idempotent.
    The next driving verb re-badges instantly (set_driving_badge re-adds after a clear)."""
    if _IDLE_WATCHER[0]:
        return False
    _IDLE_WATCHER[0] = True

    def _loop():
        while True:
            time.sleep(15.0)
            try:
                if badge_active() and _LAST_DRIVING[0] and time.time() - _LAST_DRIVING[0] > IDLE_CLEAR_SEC:
                    clear_driving_badge()
            except Exception:
                pass

    threading.Thread(target=_loop, daemon=True, name="altium-badge-idle-clear").start()
    return True