""" Togen Windows Display Agent (DVI-1230 / DVI-1227 P2) A thin kiosk agent: enrolls a Windows PC as a Togen Display, renders its assigned dashboard in a WebView2 window, and applies Windows-only window layer control (topmost / bottom / windowed / wallpaper). Functionality parity target = togen/signage_agent/ (the Linux Display Agent) -- ALL dashboard/interface/splash/layout/Present logic is server- and browser-side; this agent stays thin (register, heartbeat, render, apply layer). Enroll (admin-minted enrollment token -> permanent agent_token; used by install.ps1, --enroll-only exits without starting the kiosk window): display_agent.exe --server https://togen.example.com --enroll --enroll-only Run (after enrollment; agent_token is stored via Windows Credential Manager with an HKLM/HKCU registry fallback, same convention as the Vision agent): display_agent.exe --server https://togen.example.com """ from __future__ import annotations import argparse import logging import logging.handlers import os import signal import socket import sys import threading import time from pathlib import Path import requests import server_client import winlayer _AGENT_VERSION = server_client.AGENT_VERSION _SHARE_DIR = Path(os.environ.get("PROGRAMDATA", r"C:\ProgramData")) / "Togen" / "Display" _LOG_DIR = _SHARE_DIR / "Logs" _LOG_FILE = _LOG_DIR / "agent.log" # Poll/retry cadence (server contract: checkin ~30s, config poll ~15s). HEARTBEAT_INTERVAL_S = 30 CONFIG_POLL_INTERVAL_S = 15 KEEP_AWAKE_INTERVAL_S = 60 UNAUTHORIZED_RETRY_S = 30 # Mirrors the Linux kiosk-launch.sh respawn `sleep 3` between relaunches. RESPAWN_BACKOFF_S = 3 log = logging.getLogger("display_agent") _SHUTDOWN = threading.Event() def _setup_logging(level: str): handlers = [] try: _LOG_DIR.mkdir(parents=True, exist_ok=True) handlers.append(logging.handlers.RotatingFileHandler( _LOG_FILE, maxBytes=2 * 1024 * 1024, backupCount=2, encoding="utf-8")) except OSError: pass handlers.append(logging.StreamHandler()) logging.basicConfig( level=getattr(logging, level.upper(), logging.INFO), format="%(asctime)s %(levelname)s %(name)s %(message)s", handlers=handlers, ) def _install_signal_handlers(): def _handle(_signum, _frame): log.info("Shutdown signal received") _SHUTDOWN.set() for sig_name in ("SIGTERM", "SIGINT", "SIGBREAK"): sig = getattr(signal, sig_name, None) if sig is None: continue try: signal.signal(sig, _handle) except (ValueError, OSError): pass def _safe_error_body(response) -> str: if response is None: return "(no response)" try: data = response.json() return str(data.get("error", response.text[:200])) except Exception: return response.text[:200] if response.text else f"HTTP {response.status_code}" # --------------------------------------------------------------------------- # Enrollment # --------------------------------------------------------------------------- def enroll(base_url: str, enrollment_token: str, hostname: str) -> int: """POST /signage/agents/register and persist the returned agent_token. Returns a process exit code: 0 on success, 1 on any failure -- so install.ps1 can branch on $LASTEXITCODE the same way it does for the Vision agent's --enroll-only path. """ client = server_client.DisplayAgentClient(base_url) log.info("Enrolling display with %s", base_url) try: data = client.register(enrollment_token, hostname) except requests.exceptions.ConnectTimeout: log.error( "ENROLLMENT FAILED -- connection to %s timed out. Check that this " "machine can reach the Togen server (Test-NetConnection).", base_url) return 1 except requests.exceptions.ConnectionError as exc: log.error("ENROLLMENT FAILED -- could not connect to %s: %s", base_url, exc) return 1 except requests.exceptions.Timeout as exc: log.error("ENROLLMENT FAILED -- request to %s timed out: %s", base_url, exc) return 1 except requests.exceptions.HTTPError as exc: log.error("ENROLLMENT FAILED -- server returned HTTP %s: %s", exc.response.status_code if exc.response is not None else "?", _safe_error_body(exc.response)) return 1 except ValueError as exc: log.error("ENROLLMENT FAILED -- %s", exc) return 1 token = data["agent_token"] server_client.store_agent_token(token) log.info("Enrolled as display %s -- agent_token stored.", data.get("agent_id")) return 0 # --------------------------------------------------------------------------- # Kiosk window lifecycle # --------------------------------------------------------------------------- WINDOW_TITLE = "Togen Display" def _hwnd_from_native(native) -> int: """pywebview's edgechromium backend exposes `window.native` as a WinForms Form (via pythonnet); its .Handle is the HWND as a .NET IntPtr.""" handle = getattr(native, "Handle", None) if handle is None: raise RuntimeError("pywebview window.native has no Handle attribute") to_int64 = getattr(handle, "ToInt64", None) if callable(to_int64): return int(to_int64()) return int(handle) def _hwnd_by_window_scan(title: str = WINDOW_TITLE, timeout_s: float = 15.0) -> int: """pywebview >= 6 removed Window.native, so scan for the HWND instead: enumerate top-level windows and match this process's visible window by title. WebView2's own windows are children of the WinForms host form and are not top-level, so this uniquely finds the form. Retries briefly because the window may not be visible yet when the start callback runs.""" import ctypes import ctypes.wintypes as wt user32 = ctypes.windll.user32 pid = os.getpid() enum_proc = ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM) deadline = time.monotonic() + timeout_s while True: found: list[int] = [] def _cb(hwnd, _lparam): wnd_pid = wt.DWORD() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(wnd_pid)) if wnd_pid.value != pid or not user32.IsWindowVisible(hwnd): return True buf = ctypes.create_unicode_buffer(256) user32.GetWindowTextW(hwnd, buf, 256) if buf.value == title: found.append(int(hwnd)) return False # stop enumeration return True user32.EnumWindows(enum_proc(_cb), 0) if found: return found[0] if time.monotonic() >= deadline: raise RuntimeError( f"no visible top-level window titled {title!r} in pid {pid}") time.sleep(0.25) def _resolve_hwnd(window) -> int: """Try the pywebview<6 native.Handle path first, then the EnumWindows scan (pywebview>=6). Raises when both fail.""" try: return _hwnd_from_native(window.native) except Exception as exc: log.info("window.native HWND path unavailable (%s) -- " "falling back to EnumWindows scan", exc) return _hwnd_by_window_scan() def _safe_destroy(window): try: window.destroy() except Exception: log.exception("window.destroy() failed") def _heartbeat_loop(client, token, shared, stop_event, window): while not stop_event.is_set(): with shared["lock"]: effective_layer = shared.get("effective_layer") try: client.checkin(token, effective_layer=effective_layer) except requests.exceptions.HTTPError as exc: if exc.response is not None and exc.response.status_code == 401: log.warning("Checkin unauthorized (revoked/invalid token) -- " "closing display window per contract") _safe_destroy(window) stop_event.set() return log.warning("Checkin failed: %s", exc) except Exception as exc: log.warning("Checkin error: %s", exc) stop_event.wait(HEARTBEAT_INTERVAL_S) def _config_poll_loop(client, token, layer_engine, hwnd, shared, stop_event, window): last_applied = None while not stop_event.is_set(): try: cfg = client.get_display_config(token) window_cfg = server_client.sanitize_window_config(cfg.get("window")) if window_cfg != last_applied: log.info("Applying window config %s to hwnd %s", window_cfg, hwnd) effective = layer_engine.apply(hwnd, window_cfg) log.info("Effective layer: %s", effective) # Only remember the config once apply succeeded, so a # transient failure is retried on the next poll. last_applied = window_cfg with shared["lock"]: shared["effective_layer"] = effective except requests.exceptions.HTTPError as exc: if exc.response is not None and exc.response.status_code == 401: log.warning("Config poll unauthorized -- closing display window " "per contract") _safe_destroy(window) stop_event.set() return log.warning("Config poll failed: %s", exc) except Exception as exc: log.warning("Config poll error: %s", exc) stop_event.wait(CONFIG_POLL_INTERVAL_S) def _keep_awake_loop(stop_event): while not stop_event.is_set(): winlayer.prevent_sleep() stop_event.wait(KEEP_AWAKE_INTERVAL_S) def _run_kiosk_window(client, token, layer_engine): """Create the frameless WebView2 window and block until it closes (user action, crash, or an unauthorized-token close from the poll threads). Imports pywebview lazily so this module still imports (for unit testing the rest of the agent) on a machine without pywebview/WebView2 installed. """ import webview url = client.display_url(token) window = webview.create_window( WINDOW_TITLE, url=url, frameless=True, easy_drag=False, on_top=False, confirm_close=False, ) shared = {"effective_layer": None, "lock": threading.Lock()} stop_event = threading.Event() window.events.closed += lambda: stop_event.set() def _started(): winlayer.prevent_sleep() try: hwnd = _resolve_hwnd(window) log.info("Resolved kiosk window HWND: %s", hwnd) except Exception: log.exception("Could not resolve a Win32 HWND for this window -- " "layer control is disabled for this run") hwnd = None threading.Thread( target=_heartbeat_loop, args=(client, token, shared, stop_event, window), name="display-agent-heartbeat", daemon=True).start() if hwnd is not None: threading.Thread( target=_config_poll_loop, args=(client, token, layer_engine, hwnd, shared, stop_event, window), name="display-agent-config-poll", daemon=True).start() threading.Thread( target=_keep_awake_loop, args=(stop_event,), name="display-agent-keep-awake", daemon=True).start() # gui='edgechromium' pins pywebview to the WebView2 backend per the # board-approved plan (D1-A) rather than letting it auto-detect. webview.start(func=_started, gui="edgechromium", debug=False) def _wait_until_authorized(client, token) -> bool: """Block (with retry) until a checkin succeeds, or shutdown is requested. Returns False only when shutdown was requested mid-wait.""" first = True while not _SHUTDOWN.is_set(): try: client.checkin(token) return True except requests.exceptions.HTTPError as exc: status = exc.response.status_code if exc.response is not None else "?" if first: log.warning("Not authorized yet (HTTP %s) -- retrying every %ss", status, UNAUTHORIZED_RETRY_S) first = False except Exception as exc: if first: log.warning("Checkin error (%s) -- retrying every %ss", exc, UNAUTHORIZED_RETRY_S) first = False _SHUTDOWN.wait(UNAUTHORIZED_RETRY_S) return False def run(base_url: str): token = server_client.load_agent_token() if not token: log.error("No agent_token stored -- run with --enroll --enroll-only first.") sys.exit(1) client = server_client.DisplayAgentClient(base_url) layer_engine = winlayer.LayerEngine() _install_signal_handlers() while not _SHUTDOWN.is_set(): if not _wait_until_authorized(client, token): break try: _run_kiosk_window(client, token, layer_engine) except Exception: log.exception("Kiosk window loop crashed") if _SHUTDOWN.is_set(): break log.warning("Display window exited; respawning in %ss", RESPAWN_BACKOFF_S) _SHUTDOWN.wait(RESPAWN_BACKOFF_S) winlayer.allow_sleep() log.info("Display agent shutting down.") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def _parse_args(argv): p = argparse.ArgumentParser(description="Togen Windows Display Agent") p.add_argument("--server", required=True, help="Base URL of the Togen server") p.add_argument("--enroll", default="", help="Enrollment token (from Admin -> Agents -> Display)") p.add_argument("--enroll-only", action="store_true", help="Register and exit without starting the kiosk window (used by install.ps1)") p.add_argument("--log-level", default="INFO") return p.parse_args(argv) def main(argv=None): args = _parse_args(argv) _setup_logging(args.log_level) if args.enroll: rc = enroll(args.server, args.enroll, socket.gethostname()) if args.enroll_only or rc != 0: sys.exit(rc) run(args.server) if __name__ == "__main__": main()