Compare commits

..

4 Commits

Author SHA1 Message Date
alexei.dolgolyov fe82836f4d chore: release v0.2.6
Lint & Test / test (push) Has been skipped
Release / create-release (push) Successful in 3s
Release / build-linux (push) Successful in 30s
Release / build-windows (push) Successful in 55s
Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-05-18 03:19:07 +03:00
alexei.dolgolyov eeab9b2a26 style: sort Xlib import in foreground_service
Resolves the ruff I001 warning introduced by 61cdce9.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-05-18 03:19:02 +03:00
alexei.dolgolyov 61cdce9b60 feat(foreground): track topmost process + browser page title
Lint & Test / test (push) Failing after 8s
Adds cross-platform foreground-window tracking and exposes it over REST
(/api/foreground) and the existing WebSocket feed.

- foreground_service.py: Windows probe via ctypes (HANDLE-correct argtypes
  to avoid 64-bit handle truncation); macOS via AppKit; Linux via Xlib
  (Wayland returns unavailable). TTL cache + per-platform fallback.
- browser_url_service.py: when foreground is a recognised browser, extract
  the page title from the window title (browser-name suffix stripped) and
  surface `is_browser` + `browser_page_title`. Optional UIA-based URL
  extraction behind MEDIA_SERVER_BROWSER_UIA env flag (off by default —
  Chromium browsers keep their accessibility tree dormant otherwise).
- websocket_manager: poll foreground every 1s inside the existing status
  loop, broadcast `foreground` on connect and `foreground_update` on
  change. Diff only on user-visible fields to avoid geometry spam.
- WebUI: new editorial card rendered under the monitor list on the
  Display tab — process name, window title, fullscreen/minimized/monitor
  chips, browser block when applicable, exe path, PID, started-ago,
  geometry, platform. 16px inter-section gap matches Settings cadence.
- i18n: 25 new keys added to both en.json and ru.json.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-05-18 03:11:59 +03:00
alexei.dolgolyov 0cf49deac0 fix(config): secure-by-default loopback bind and startup-error logging
- Default `host: 127.0.0.1` in config.example.yaml; require explicit
  api_tokens or `allow_lan_without_auth: true` before binding LAN.
- Mirror pre-uvicorn fatal errors to startup-errors.log in the config
  dir so silent boot failures via wscript/pythonw are diagnosable.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-05-18 03:11:08 +03:00
20 changed files with 1619 additions and 66 deletions
+14 -45
View File
@@ -1,55 +1,25 @@
## v0.2.5 (2026-05-16)
## v0.2.6 (2026-05-18)
### Features
- **Foreground-window tracker (cross-platform):** New `/api/foreground` endpoint plus live `foreground` / `foreground_update` messages on the existing WebSocket feed. Windows uses a ctypes `GetForegroundWindow` probe with HANDLE-correct `argtypes` so 64-bit window handles aren't truncated; macOS uses AppKit's `NSWorkspace.frontmostApplication`; Linux uses Xlib's `_NET_ACTIVE_WINDOW` (Wayland sessions return a structured "unavailable" so the UI can render gracefully). Each platform has a TTL cache and per-platform fallbacks. ([61cdce9](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/61cdce9))
- **Browser page-title surfacing:** When the foreground process is a recognised browser, the window title is stripped of the trailing browser-name suffix and exposed as `browser_page_title` alongside `is_browser`. Optional UIA-based URL extraction sits behind the `MEDIA_SERVER_BROWSER_UIA` env flag (off by default — Chromium browsers keep their accessibility tree dormant unless something asks, and enabling it has a measurable cost). ([61cdce9](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/61cdce9))
- **Foreground card in the Web UI:** New editorial card under the monitor list on the Display tab renders process name, window title, fullscreen / minimized / monitor chips, the browser block when applicable, exe path, PID, started-ago, geometry, and platform. 16px inter-section gap matches the Settings cadence. 25 new i18n keys added to both `en.json` and `ru.json`. ([61cdce9](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/61cdce9))
- **WebSocket integration:** The existing 1s status loop now polls foreground every tick, broadcasts `foreground` on connect, and emits `foreground_update` only when user-visible fields actually change — geometry deltas alone don't spam the channel. ([61cdce9](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/61cdce9))
### Security
- **Loopback-by-default + auto-generated token:** Server now binds `127.0.0.1` by default; first-run bootstrap generates a random `api_token` and refuses to bind a non-loopback interface without auth unless explicitly opted in. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Browser path-traversal hardening:** `BrowserService.validate_path` now rejects absolute paths, drive letters, UNC paths, and NUL bytes. `/api/browser/{play,metadata,thumbnail}` require a `folder_id` plus a folder-relative path — arbitrary filesystem reads via the browser API are no longer possible. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Strict input validation on links/scripts:** Pydantic validators reject non-http(s) URLs and any icon outside the `mdi:<slug>` namespace. Create/update/delete on scripts, callbacks, and links is gated by the corresponding `*_management` flags. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Hardened response headers + CORS:** Strict `Content-Security-Policy`, `X-Frame-Options: DENY`, `Referrer-Policy: no-referrer`, `X-Content-Type-Options: nosniff`. CORS locked to `localhost:<port>` + `127.0.0.1:<port>` by default; configurable for trusted origins. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Atomic config writes with restrictive permissions:** `config.yaml` writes go through a temp file + `os.replace` and land with `0o600` on POSIX, so a crash mid-write can never leave a half-written token on disk readable to other users. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Subprocess process-group isolation:** Spawned scripts/callbacks now get their own process group (`CREATE_NEW_PROCESS_GROUP` on Windows, `start_new_session=True` on POSIX), so a timeout actually kills the whole tree instead of orphaning child processes. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Frontend XSS hardening:** Monitor name + details are `escapeHtml`'d, the power button moved to a delegated `data-action` handler, and remote MDI SVGs are parsed and sanitized (strip `<script>`, `<foreignObject>`, `on*` handlers, `javascript:` hrefs) before they touch `innerHTML`. All dynamic URL segments now go through `encodeURIComponent`. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **CSP-compliant event wiring:** Strict `script-src 'self'` was blocking every inline `onclick`/`onchange`/`oninput`/`onsubmit` in the UI, leaving buttons and forms silently dead. All 53 inline handler attributes in `index.html` were renamed to `data-on*` and a new `wireInlineHandlers()` in `app.js` parses each expression on `DOMContentLoaded` and attaches a real `addEventListener` — supports no-arg calls, string/number/bool/null literals, and the `event` token. No `unsafe-inline` or `unsafe-hashes` needed. ([eaeebb6](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/eaeebb6))
- **Loopback-by-default in the shipped config:** `config.example.yaml` now defaults `host: 127.0.0.1`. The server still refuses to bind a non-loopback interface without either `api_tokens` configured or an explicit `allow_lan_without_auth: true` opt-in — this hardens fresh installs where a user copies the example config verbatim. ([0cf49de](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/0cf49de))
### Bug Fixes
### Reliability
- **WebSocket reconnect robustness:** Close the previous socket before opening a new one, clear the ping interval per-socket, clear `reconnectTimeout` up-front, retry on `online`/`visibilitychange`, and wrap `JSON.parse` in try/catch — eliminates the stale-socket leaks and "stuck offline after sleep" cases. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Artwork fetch race:** `AbortController` + generation guard so a rapid track change can no longer paint the previous track's artwork over the current one. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Audio analyzer no longer spins infinitely without a loopback device:** A sticky `_unavailable` flag short-circuits start/stop; cleared by `set_device()` so the user can recover once a device appears. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Volume short-circuit cache invalidation:** Cache is now busted when the server reports a remote volume change, so the UI no longer ignores volume updates that happened outside the app. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Browser thumbnail race:** Per-folder generation counter + `isConnected` checks; in-flight fetches are aborted on navigation, so thumbnails from a folder you already left can't paint into the current view. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Track-skip uses cached title** instead of a full WinRT status round-trip — skip feedback is now instant. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Browser list column alignment:** `.browser-list` switched to CSS grid + subgrid so header and rows share column tracks, eliminating the misaligned columns when content widths differed between rows. Matching responsive column overrides applied at the parent. Root-folder SVG sizing (hardcoded 24×24 in `browser.js`) now fills the 56px icon box instead of rendering at ~43%. Compact-grid icon fills its thumb wrapper so the emoji centers instead of being stranded top-left. Premature `isConnected` bail removed from `loadThumbnail` — the img element is intentionally detached when called from `renderBrowserGrid/List`, and the post-await checks already handle navigation-away correctly. ([982dda4](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/982dda4))
### Performance
- **Blocking IO off the event loop:** Linux MPRIS/`pactl` calls, `/api/display` DDC/CI handlers, and `browse_directory` are all wrapped in `asyncio.to_thread` — slow SMB shares or laggy monitors can no longer stall the entire async runtime. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Windows status poll loop reuse:** The 0.5s status poll now caches one asyncio loop per worker thread via `threading.local` instead of `new_event_loop`/`close` on every tick. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **WebSocket broadcast: serialize once:** `broadcast()` serializes JSON a single time and uses `send_text` to fan out to all clients. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Thumbnail cache cleanup actually runs:** The hourly cleanup task was defined but never scheduled — it is now wired into the lifespan handler so the cache no longer grows unbounded. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Progress drag listeners attached only while dragging** — no more global `mousemove` handler firing on every cursor twitch. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
### UI/UX
- **Copper accent consistency:** Green leftover focus rings (`rgba(29,185,84,…)`) replaced with copper (`rgba(var(--copper-rgb),…)`) across the UI. Dialogs now have square corners and a copper top hairline so they read as part of the editorial chrome. `.browser-item` is transparent with a copper hover border (was a filled card). ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Audio device select** uses `var(--sans)` instead of the generic system font so it matches surrounding controls. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Mobile padding tuned for ≤480px screens.** ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Accessible breadcrumb home:** Now a real `<button>` with `aria-label`, and `aria-current` is set on the root. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **i18n gaps filled:** `display.msg.power_*`, `execution.*`, `scripts.params.execute`, `callbacks.empty` now have proper en + ru strings. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Diagnosable silent boot failures:** Pre-uvicorn fatal errors (config parse failures, port conflicts, missing dependencies on Windows) are now mirrored to `startup-errors.log` in the config directory. Previously, launches via `wscript`/`pythonw` (the hidden Startup-folder shortcut path) had no console attached, so any startup crash was effectively invisible — the user just saw "nothing happened". ([0cf49de](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/0cf49de))
---
### Development / Internal
#### Quality
- All `asyncio.get_event_loop()` in coroutines migrated to `get_running_loop()` (the former is deprecated in Python 3.12+). ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- `ThreadPoolExecutor`s now shut down cleanly during lifespan teardown. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- `config_manager` dedup: 12 near-identical CRUD methods collapsed onto generic `_upsert`/`_delete` helpers — about **290 lines removed** with no behavior change. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- Service worker no longer pass-throughs every fetch. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- M3U playlist written via `NamedTemporaryFile` so a fixed-path symlink can no longer clobber it. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- `__version__` prefers live `pyproject.toml` in dev checkouts so `pip install -e .` users see the source-of-truth version, not the stale metadata baked in at install time. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- `_broadcast_after_open` hardening: initialize status, swallow per-poll errors, and track background tasks in a strong-ref set with done-callback cleanup so they aren't garbage-collected mid-flight. ([bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40))
- **Lint cleanup:** Sorted the Xlib import in `foreground_service.py` so `ruff check` is clean. Same module, no behaviour change.
---
@@ -58,8 +28,7 @@
| Hash | Message | Author |
|------|---------|--------|
| [982dda4](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/982dda4) | fix(browser): align list columns via subgrid and fix icon sizing | alexei.dolgolyov |
| [eaeebb6](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/eaeebb6) | fix(csp): replace inline on* handlers with data-on* + JS wiring | alexei.dolgolyov |
| [bcc6d40](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/bcc6d40) | fix: comprehensive security, bug, performance, and UI/UX audit | alexei.dolgolyov |
| [61cdce9](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/61cdce9) | feat(foreground): track topmost process + browser page title | alexei.dolgolyov |
| [0cf49de](https://git.dolgolyov-family.by/alexei.dolgolyov/media-player-server/commit/0cf49de) | fix(config): secure-by-default loopback bind and startup-error logging | alexei.dolgolyov |
</details>
+10 -3
View File
@@ -1,7 +1,13 @@
# Media Server Configuration
# Copy this file to config.yaml and customize as needed.
# By default, authentication is DISABLED (no tokens = open access).
# To enable auth, uncomment and configure the api_tokens section below.
#
# Secure-by-default: the server binds to loopback (127.0.0.1) only and refuses
# to bind a non-loopback address with no tokens configured.
#
# To expose on the LAN you must do ONE of:
# 1. Configure api_tokens below AND change host to "0.0.0.0", OR
# 2. Set `allow_lan_without_auth: true` (LAN-open, no auth — insecure on
# hostile networks, only acceptable on a trusted home LAN).
# API Tokens - Multiple tokens with friendly labels
# This allows you to identify which client is making requests in the logs
@@ -11,8 +17,9 @@
# web_ui: "your-web-ui-token-here"
# Server settings
host: "0.0.0.0"
host: "127.0.0.1"
port: 8765
# allow_lan_without_auth: true # uncomment + change host to 0.0.0.0 for LAN-open mode
# Custom scripts
scripts:
+22 -11
View File
@@ -22,6 +22,7 @@ from .routes import (
browser_router,
callbacks_router,
display_router,
foreground_router,
health_router,
links_router,
media_router,
@@ -241,6 +242,7 @@ def create_app() -> FastAPI:
app.include_router(browser_router)
app.include_router(callbacks_router)
app.include_router(display_router)
app.include_router(foreground_router)
app.include_router(health_router)
app.include_router(links_router)
app.include_router(media_router)
@@ -320,44 +322,53 @@ def main():
print("\nAuthentication is DISABLED (no tokens configured)")
return
# Stderr is invisible when launched via wscript / pythonw (Start Menu shortcut,
# autostart). Mirror pre-uvicorn failures to a file in the config dir so the
# next silent boot failure is diagnosable.
def _fatal(msg: str, exit_code: int = 1) -> None:
print(msg, file=sys.stderr)
try:
log_path = get_config_dir() / "startup-errors.log"
from datetime import datetime
with open(log_path, "a", encoding="utf-8") as f:
f.write(f"[{datetime.now().isoformat(timespec='seconds')}] {msg}\n")
except OSError:
pass
sys.exit(exit_code)
# First-run bootstrap: if no config has ever been written, generate one
# with a random token instead of starting in the insecure "no-auth" mode.
config_path = get_config_dir() / "config.yaml"
if not config_path.exists() and not settings.api_tokens:
try:
generate_default_config(config_path)
print(
_fatal(
f"\nFirst run: generated default config at {config_path}.\n"
"Run --show-token to retrieve the API token, then restart.",
file=sys.stderr,
exit_code=0,
)
sys.exit(0)
except OSError as e:
print(f"WARNING: could not bootstrap config: {e}", file=sys.stderr)
# Refuse to bind a non-loopback address with no tokens, unless explicitly opted in.
non_loopback = args.host not in ("127.0.0.1", "localhost", "::1")
if non_loopback and not settings.api_tokens and not settings.allow_lan_without_auth:
print(
_fatal(
"ERROR: refusing to bind a non-loopback address with no api_tokens configured.\n"
"Either set api_tokens in config.yaml, bind to 127.0.0.1,"
" or set allow_lan_without_auth: true in config.yaml to override.",
file=sys.stderr,
" or set allow_lan_without_auth: true in config.yaml to override."
)
sys.exit(1)
# Check if port is available before starting
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.bind((args.host if args.host != "0.0.0.0" else "127.0.0.1", args.port))
except OSError:
print(
_fatal(
f"ERROR: Port {args.port} is already in use. "
f"Another instance of Media Server may be running.\n"
f"Stop the other process or use --port to pick a different port.",
file=sys.stderr,
f"Stop the other process or use --port to pick a different port."
)
sys.exit(1)
from .tray import PYSTRAY_AVAILABLE, TrayManager
+2
View File
@@ -4,6 +4,7 @@ from .audio import router as audio_router
from .browser import router as browser_router
from .callbacks import router as callbacks_router
from .display import router as display_router
from .foreground import router as foreground_router
from .health import router as health_router
from .links import router as links_router
from .media import router as media_router
@@ -14,6 +15,7 @@ __all__ = [
"browser_router",
"callbacks_router",
"display_router",
"foreground_router",
"health_router",
"links_router",
"media_router",
+26
View File
@@ -0,0 +1,26 @@
"""Foreground (topmost) window/process API."""
import asyncio
import logging
from fastapi import APIRouter, Depends
from ..auth import verify_token
from ..services.foreground_service import get_foreground_info
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/foreground", tags=["foreground"])
@router.get("")
async def get_foreground(
refresh: bool = False, _: str = Depends(verify_token)
) -> dict:
"""Return metadata about the foreground window and owning process.
The probe is cached for ~500ms server-side; pass ``?refresh=1`` to bypass
the cache for one-shot queries.
"""
info = await asyncio.to_thread(get_foreground_info, refresh)
return info.to_dict()
@@ -0,0 +1,296 @@
"""Extract page-level metadata from a focused desktop web browser.
The browser's window title is the reliable signal — every major browser
formats it as ``"<page title> - <Browser Name>"``, so stripping the suffix
gives us the page title for free.
URL extraction was attempted via UI Automation (UIA), but Chromium-based
browsers (Chrome/Edge/Brave/Vivaldi) keep their accessibility tree dormant
unless a screen reader is active or ``--force-renderer-accessibility`` is
set — neither is something we want to require from end users. The UIA
machinery is still here behind a feature flag in case a future caller
opts into the accessibility-flag path; by default we just return the
page title and leave ``url=None``.
Other platforms (macOS via AppleScript, Linux via AT-SPI) are out of scope
for this iteration.
"""
from __future__ import annotations
import logging
import os
import platform
import threading
from dataclasses import dataclass
logger = logging.getLogger(__name__)
# UIA URL extraction is opt-in because Chromium browsers keep their
# accessibility tree dormant unless the user starts the browser with
# ``--force-renderer-accessibility`` (or a screen reader is running).
# Without that, `FindAll` throws and we'd burn 5s per probe retrying.
# Set MEDIA_SERVER_BROWSER_UIA=1 to enable; default off.
_UIA_ENABLED = os.environ.get("MEDIA_SERVER_BROWSER_UIA", "").lower() in (
"1", "true", "yes", "on"
)
# Known browser executables (lowercase, .exe-stripped). Used to decide
# whether to spend the UIA query budget on this foreground process.
BROWSER_PROCESS_HINTS: frozenset[str] = frozenset({
"chrome",
"msedge",
"firefox",
"brave",
"opera",
"vivaldi",
"yandex",
"browser", # Yandex Browser sometimes reports as browser.exe
"arc",
"thorium",
})
@dataclass(frozen=True)
class BrowserPageInfo:
url: str | None = None
page_title: str | None = None
_EMPTY = BrowserPageInfo()
def is_browser_process(process_name: str | None) -> bool:
"""Return True when ``process_name`` looks like a supported browser."""
if not process_name:
return False
base = process_name.lower()
if base.endswith(".exe"):
base = base[:-4]
return base in BROWSER_PROCESS_HINTS
def _strip_browser_suffix(title: str | None, process_name: str | None) -> str | None:
"""Pull the page title out of the browser's window title.
Most browsers format their window title as ``"<page> - <Browser Name>"``.
We strip the trailing suffix so consumers get the page title alone. If
the suffix can't be matched, return the raw title unchanged.
"""
if not title:
return None
suffixes = (
" - Google Chrome",
" — Google Chrome",
" - MicrosoftEdge",
" - Microsoft Edge",
" — Mozilla Firefox",
" - Mozilla Firefox",
" - Brave",
" - Opera",
" - Vivaldi",
" - Yandex",
)
for s in suffixes:
if title.endswith(s):
return title[: -len(s)].strip() or None
return title
# ─── UIA lookup (Windows) ───────────────────────────────────────────
# UIA control type / property constants we need. Avoiding the full
# UIAutomationClient typelib generation — those constants are stable.
_UIA_EditControlTypeId = 50004
_UIA_ControlTypePropertyId = 30003
_UIA_ValueValuePropertyId = 30045
_UIA_NamePropertyId = 30005
_UIA_ValuePatternId = 10002
_TreeScope_Descendants = 4
_PropertyConditionFlags_IgnoreCase = 1
# Lazy import + per-thread COM init.
_uia_lock = threading.Lock()
_uia_singleton = None
_uia_load_error: str | None = None
_uia_thread_local = threading.local()
def _ensure_com() -> None:
"""Initialise COM on the current thread (idempotent per thread)."""
if getattr(_uia_thread_local, "initialised", False):
return
try:
import comtypes # type: ignore
# COINIT_APARTMENTTHREADED is required by UIA; comtypes' default
# CoInitializeEx already passes that flag.
comtypes.CoInitialize()
_uia_thread_local.initialised = True
except Exception as e:
logger.debug("CoInitialize failed: %s", e)
def _get_uia():
"""Return the IUIAutomation singleton, or None if unavailable."""
global _uia_singleton, _uia_load_error
if _uia_singleton is not None:
return _uia_singleton
if _uia_load_error is not None:
return None
with _uia_lock:
if _uia_singleton is not None:
return _uia_singleton
try:
import comtypes.client # type: ignore
# CLSID for CUIAutomation. Using GetActiveObject would fail,
# so we cocreate. comtypes.client.CreateObject keeps the COM
# plumbing tidy.
_uia_singleton = comtypes.client.CreateObject(
"{ff48dba4-60ef-4201-aa87-54103eef594e}",
interface=comtypes.client.GetModule(
"UIAutomationCore.dll"
).IUIAutomation,
)
return _uia_singleton
except Exception as e:
_uia_load_error = str(e)
logger.info("UIA unavailable; browser URL extraction disabled: %s", e)
return None
def _find_address_bar_value(hwnd: int) -> str | None:
"""Walk the UIA tree under ``hwnd`` looking for the URL Edit control.
Strategy: find every descendant Edit control, then pick the first one
whose Name contains an address-bar hint, or — failing that — the first
one whose value parses as a URL-ish string. Browsers expose extra Edit
controls (search bars, find-in-page) so name matching is the reliable
signal; the URL-ish fallback covers locale variants we haven't seen.
"""
_ensure_com()
uia = _get_uia()
if uia is None:
return None
try:
element = uia.ElementFromHandle(hwnd)
if not element:
return None
# Build a condition matching ControlType=Edit, then enumerate.
edit_condition = uia.CreatePropertyCondition(
_UIA_ControlTypePropertyId, _UIA_EditControlTypeId
)
edits = element.FindAll(_TreeScope_Descendants, edit_condition)
count = edits.Length if edits else 0
if count == 0:
return None
# Hints (lowercase) used to identify the address bar by its Name
# property. Covers en-US plus a few common locales / browsers.
name_hints = (
"address", # Chrome/Edge: "Address and search bar"
"адрес", # Chrome ru: "Адресная строка и строка поиска"
"адресная",
"search with", # Firefox: "Search with Google or enter address"
"поиск или ввод", # Firefox ru
"url",
"location",
)
# First pass: name-based match (high confidence).
candidates: list[tuple[int, str]] = []
for i in range(count):
edit = edits.GetElement(i)
try:
name = (edit.CurrentName or "").lower()
except Exception:
name = ""
try:
value = edit.GetCurrentPropertyValue(_UIA_ValueValuePropertyId)
except Exception:
value = None
if value is None:
continue
value_str = str(value)
for h in name_hints:
if h in name:
return value_str
candidates.append((i, value_str))
# Second pass: URL-ish fallback. Pick the first candidate that
# looks like a URL; this catches browser/locale combos we haven't
# listed above.
for _i, v in candidates:
lv = v.lower()
if (
lv.startswith("http://")
or lv.startswith("https://")
or lv.startswith("about:")
or lv.startswith("chrome://")
or lv.startswith("edge://")
or lv.startswith("brave://")
or lv.startswith("file://")
or lv.startswith("ftp://")
):
return v
return None
except Exception as e:
logger.debug("UIA address-bar lookup failed: %s", e)
return None
# ─── Per-(hwnd, title) cache ────────────────────────────────────────
_cache_lock = threading.Lock()
_cache_key: tuple[int | None, str | None] = (None, None)
_cache_value: BrowserPageInfo = _EMPTY
def get_browser_page(
*,
hwnd: int | None,
process_name: str | None,
window_title: str | None,
) -> BrowserPageInfo:
"""Return the URL + page title for the foreground browser tab, if any.
Callers pass the already-resolved foreground HWND/title/process_name so
this service doesn't re-walk Win32 to find them. Returns ``_EMPTY`` for
non-browser processes or when UIA can't resolve the URL.
"""
if not is_browser_process(process_name):
return _EMPTY
if platform.system() != "Windows":
# macOS/Linux paths not implemented in this iteration.
return _EMPTY
if not hwnd:
return _EMPTY
global _cache_key, _cache_value
key = (hwnd, window_title)
with _cache_lock:
if key == _cache_key and _cache_value is not _EMPTY:
return _cache_value
url = _find_address_bar_value(hwnd) if _UIA_ENABLED else None
page_title = _strip_browser_suffix(window_title, process_name)
info = BrowserPageInfo(url=url, page_title=page_title)
with _cache_lock:
_cache_key = key
_cache_value = info
return info
def reset_cache() -> None:
"""Reset the cache. Useful in tests."""
global _cache_key, _cache_value
with _cache_lock:
_cache_key = (None, None)
_cache_value = _EMPTY
+514
View File
@@ -0,0 +1,514 @@
"""Foreground (topmost) window/process tracking.
Reports the process that currently owns the foreground window, plus useful
metadata (window title, executable path, monitor index, whether the window
covers a full monitor, process start time).
All probes happen behind a short TTL cache so the WebSocket status poll and
per-entity HA polls don't pay the OS call cost on every tick.
Windows uses the Win32 API via ``ctypes`` (no extra dependency) and falls back
gracefully when individual probes fail. Linux/macOS implementations are
best-effort and return ``available=False`` when the required tooling is
missing, so the rest of the stack keeps working.
"""
from __future__ import annotations
import logging
import platform
import threading
import time
from dataclasses import asdict, dataclass, field
logger = logging.getLogger(__name__)
_CACHE_TTL = 0.5 # seconds — fast enough for WebSocket broadcast loop
@dataclass(frozen=True)
class ForegroundInfo:
"""Snapshot of the foreground window/process."""
available: bool
pid: int | None = None
process_name: str | None = None
executable_path: str | None = None
window_title: str | None = None
window_handle: int | None = None
is_fullscreen: bool = False
is_minimized: bool = False
monitor_id: int | None = None
monitor_geometry: dict[str, int] | None = None
window_geometry: dict[str, int] | None = None
started_at: float | None = None
platform: str = field(default_factory=lambda: platform.system())
error: str | None = None
# Populated only when the foreground process is a recognised web
# browser. ``browser_page_title`` is derived from the window title
# (suffix stripped); ``browser_url`` requires UIA to succeed.
is_browser: bool = False
browser_url: str | None = None
browser_page_title: str | None = None
def to_dict(self) -> dict:
return asdict(self)
_UNAVAILABLE = ForegroundInfo(available=False)
class _Cache:
"""Single-slot TTL cache shared across callers."""
def __init__(self) -> None:
self._lock = threading.Lock()
self._value: ForegroundInfo | None = None
self._fetched_at: float = 0.0
def get(self, ttl: float, fetch) -> ForegroundInfo:
with self._lock:
now = time.monotonic()
if self._value is not None and (now - self._fetched_at) < ttl:
return self._value
# Fetch outside the lock — OS calls can take tens of ms.
value = fetch()
with self._lock:
self._value = value
self._fetched_at = time.monotonic()
return value
def invalidate(self) -> None:
with self._lock:
self._value = None
self._fetched_at = 0.0
_cache = _Cache()
def _probe_windows() -> ForegroundInfo:
"""Probe foreground window state on Windows via Win32 API."""
import ctypes
import ctypes.wintypes as wt
user32 = ctypes.WinDLL("user32", use_last_error=True)
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
psapi = ctypes.WinDLL("psapi", use_last_error=True)
# CRITICAL: declare argtypes/restype on every Win32 call that returns a
# HANDLE/HWND/HMONITOR. ctypes defaults to `c_int` (32-bit) which
# silently truncates 64-bit pointer values on x64 — that corrupts the
# handle so `CloseHandle()` can either fail or close the wrong kernel
# object, and pointer-equality comparisons (monitor index lookup) miss.
user32.GetForegroundWindow.restype = wt.HWND
user32.GetWindowThreadProcessId.argtypes = [wt.HWND, ctypes.POINTER(wt.DWORD)]
user32.GetWindowThreadProcessId.restype = wt.DWORD
user32.GetWindowTextLengthW.argtypes = [wt.HWND]
user32.GetWindowTextLengthW.restype = ctypes.c_int
user32.GetWindowTextW.argtypes = [wt.HWND, wt.LPWSTR, ctypes.c_int]
user32.GetWindowTextW.restype = ctypes.c_int
user32.IsIconic.argtypes = [wt.HWND]
user32.IsIconic.restype = wt.BOOL
user32.GetWindowRect.argtypes = [wt.HWND, ctypes.POINTER(wt.RECT)]
user32.GetWindowRect.restype = wt.BOOL
user32.MonitorFromWindow.argtypes = [wt.HWND, wt.DWORD]
user32.MonitorFromWindow.restype = wt.HMONITOR
user32.GetMonitorInfoW.argtypes = [wt.HMONITOR, ctypes.c_void_p]
user32.GetMonitorInfoW.restype = wt.BOOL
kernel32.OpenProcess.argtypes = [wt.DWORD, wt.BOOL, wt.DWORD]
kernel32.OpenProcess.restype = wt.HANDLE
kernel32.CloseHandle.argtypes = [wt.HANDLE]
kernel32.CloseHandle.restype = wt.BOOL
kernel32.QueryFullProcessImageNameW.argtypes = [
wt.HANDLE, wt.DWORD, wt.LPWSTR, ctypes.POINTER(wt.DWORD)
]
kernel32.QueryFullProcessImageNameW.restype = wt.BOOL
kernel32.GetProcessTimes.argtypes = [
wt.HANDLE,
ctypes.POINTER(wt.FILETIME),
ctypes.POINTER(wt.FILETIME),
ctypes.POINTER(wt.FILETIME),
ctypes.POINTER(wt.FILETIME),
]
kernel32.GetProcessTimes.restype = wt.BOOL
psapi.GetModuleFileNameExW.argtypes = [wt.HANDLE, wt.HMODULE, wt.LPWSTR, wt.DWORD]
psapi.GetModuleFileNameExW.restype = wt.DWORD
hwnd = user32.GetForegroundWindow()
if not hwnd:
return ForegroundInfo(available=True, error="no foreground window")
# PID + window thread.
pid = wt.DWORD(0)
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
pid_val = int(pid.value) if pid.value else None
# Window title — Unicode.
length = user32.GetWindowTextLengthW(hwnd)
title_buf = ctypes.create_unicode_buffer(length + 1)
user32.GetWindowTextW(hwnd, title_buf, length + 1)
window_title = title_buf.value or None
# Minimized flag.
is_minimized = bool(user32.IsIconic(hwnd))
# Window rect (screen coords).
rect = wt.RECT()
window_geometry: dict[str, int] | None = None
if user32.GetWindowRect(hwnd, ctypes.byref(rect)):
window_geometry = {
"left": int(rect.left),
"top": int(rect.top),
"right": int(rect.right),
"bottom": int(rect.bottom),
"width": int(rect.right - rect.left),
"height": int(rect.bottom - rect.top),
}
# Monitor under the window + its geometry.
monitor_geometry: dict[str, int] | None = None
monitor_id: int | None = None
is_fullscreen = False
try:
MONITOR_DEFAULTTONEAREST = 2
class MONITORINFO(ctypes.Structure):
_fields_ = [
("cbSize", wt.DWORD),
("rcMonitor", wt.RECT),
("rcWork", wt.RECT),
("dwFlags", wt.DWORD),
]
hmon = user32.MonitorFromWindow(hwnd, MONITOR_DEFAULTTONEAREST)
if hmon:
mi = MONITORINFO()
mi.cbSize = ctypes.sizeof(mi)
if user32.GetMonitorInfoW(hmon, ctypes.byref(mi)):
monitor_geometry = {
"left": int(mi.rcMonitor.left),
"top": int(mi.rcMonitor.top),
"right": int(mi.rcMonitor.right),
"bottom": int(mi.rcMonitor.bottom),
"width": int(mi.rcMonitor.right - mi.rcMonitor.left),
"height": int(mi.rcMonitor.bottom - mi.rcMonitor.top),
}
# Fullscreen heuristic: window rect equals monitor rect AND
# not minimized. Many media players (VLC, browser fullscreen)
# set themselves to exactly the monitor bounds.
if window_geometry and not is_minimized:
is_fullscreen = (
window_geometry["left"] == monitor_geometry["left"]
and window_geometry["top"] == monitor_geometry["top"]
and window_geometry["right"] == monitor_geometry["right"]
and window_geometry["bottom"] == monitor_geometry["bottom"]
)
# Resolve monitor index by enumerating displays in order. Coerce
# both the foreground hmon and the per-enum hmon to int so the
# equality compare uses 64-bit values consistently regardless of
# how ctypes represents the handle internally.
try:
indexed: list[int] = []
def _cb(hm, _hdc, _rect, _data):
indexed.append(int(hm) if hm else 0)
return True
MONITORENUMPROC = ctypes.WINFUNCTYPE(
ctypes.c_int,
wt.HMONITOR,
wt.HDC,
ctypes.POINTER(wt.RECT),
wt.LPARAM,
)
user32.EnumDisplayMonitors.argtypes = [
wt.HDC, ctypes.POINTER(wt.RECT), MONITORENUMPROC, wt.LPARAM
]
user32.EnumDisplayMonitors.restype = wt.BOOL
user32.EnumDisplayMonitors(None, None, MONITORENUMPROC(_cb), 0)
target = int(hmon) if hmon else 0
if target and target in indexed:
monitor_id = indexed.index(target)
except Exception as e:
logger.debug("Monitor index resolution failed: %s", e)
except Exception as e:
logger.debug("Monitor info probe failed: %s", e)
# Process executable path + start time.
executable_path: str | None = None
process_name: str | None = None
started_at: float | None = None
if pid_val:
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
h_proc = kernel32.OpenProcess(
PROCESS_QUERY_LIMITED_INFORMATION, False, pid_val
)
if h_proc:
try:
# Image filename — full path. QueryFullProcessImageNameW works
# across 32/64-bit boundaries, unlike GetModuleFileNameExW.
buf = ctypes.create_unicode_buffer(1024)
size = wt.DWORD(len(buf))
if kernel32.QueryFullProcessImageNameW(
h_proc, 0, buf, ctypes.byref(size)
):
executable_path = buf.value or None
else:
# Fallback via psapi. Return value is the length copied
# into the buffer (0 on failure); ignoring it would leave
# `executable_path` as an empty string from the freshly
# allocated buffer instead of None.
written = psapi.GetModuleFileNameExW(h_proc, None, buf, len(buf))
if written:
executable_path = buf.value or None
else:
logger.debug(
"QueryFullProcessImageNameW + psapi fallback both "
"failed for pid=%s (err=%d)",
pid_val,
ctypes.get_last_error(),
)
if executable_path:
import os
process_name = os.path.basename(executable_path)
# Process creation time (FILETIME, 100ns ticks since 1601).
creation = wt.FILETIME()
exit_t = wt.FILETIME()
kernel_t = wt.FILETIME()
user_t = wt.FILETIME()
if kernel32.GetProcessTimes(
h_proc,
ctypes.byref(creation),
ctypes.byref(exit_t),
ctypes.byref(kernel_t),
ctypes.byref(user_t),
):
ticks = (creation.dwHighDateTime << 32) | creation.dwLowDateTime
# Convert to Unix epoch seconds (1601-01-01 → 1970-01-01).
if ticks:
started_at = (ticks - 116444736000000000) / 10_000_000
finally:
kernel32.CloseHandle(h_proc)
return ForegroundInfo(
available=True,
pid=pid_val,
process_name=process_name,
executable_path=executable_path,
window_title=window_title,
window_handle=int(hwnd) if hwnd else None,
is_fullscreen=is_fullscreen,
is_minimized=is_minimized,
monitor_id=monitor_id,
monitor_geometry=monitor_geometry,
window_geometry=window_geometry,
started_at=started_at,
)
def _probe_macos() -> ForegroundInfo:
"""Best-effort probe on macOS via AppKit (PyObjC).
Returns ``available=False`` when PyObjC is not installed — we don't take
a hard dependency on it because the typical macOS install path uses pip
+ the standalone wheel.
"""
try:
from AppKit import NSWorkspace # type: ignore
from Quartz import ( # type: ignore
CGWindowListCopyWindowInfo,
kCGNullWindowID,
kCGWindowListOptionOnScreenOnly,
)
except Exception:
return ForegroundInfo(available=False, error="AppKit/Quartz not available")
try:
ws = NSWorkspace.sharedWorkspace()
app = ws.frontmostApplication()
if app is None:
return ForegroundInfo(available=True, error="no frontmost app")
pid = int(app.processIdentifier())
process_name = str(app.localizedName() or "")
bundle_url = app.bundleURL()
executable_path = str(bundle_url.path()) if bundle_url else None
started_at = None
launch_date = app.launchDate()
if launch_date is not None:
started_at = float(launch_date.timeIntervalSince1970())
# Window title — frontmost on-screen window owned by this PID.
window_title: str | None = None
try:
windows = CGWindowListCopyWindowInfo(
kCGWindowListOptionOnScreenOnly, kCGNullWindowID
)
for w in windows or []:
if int(w.get("kCGWindowOwnerPID", -1)) == pid:
name = w.get("kCGWindowName")
if name:
window_title = str(name)
break
except Exception as e:
logger.debug("CGWindowListCopyWindowInfo failed: %s", e)
return ForegroundInfo(
available=True,
pid=pid,
process_name=process_name,
executable_path=executable_path,
window_title=window_title,
started_at=started_at,
)
except Exception as e:
logger.debug("macOS foreground probe failed: %s", e)
return ForegroundInfo(available=False, error=str(e))
def _probe_linux() -> ForegroundInfo:
"""Best-effort probe on Linux via Xlib (X11 only).
Wayland sessions intentionally hide window/process info from unprivileged
clients, so this returns ``available=False`` on Wayland. The caller still
gets a structured response and can render "unavailable" in the UI.
"""
import os
if os.environ.get("WAYLAND_DISPLAY"):
return ForegroundInfo(
available=False, error="Wayland session — foreground probe unavailable"
)
try:
from Xlib import X, display # type: ignore # noqa: F401
except Exception:
return ForegroundInfo(available=False, error="python-xlib not installed")
try:
d = display.Display()
root = d.screen().root
NET_ACTIVE_WINDOW = d.intern_atom("_NET_ACTIVE_WINDOW")
NET_WM_PID = d.intern_atom("_NET_WM_PID")
NET_WM_NAME = d.intern_atom("_NET_WM_NAME")
UTF8_STRING = d.intern_atom("UTF8_STRING")
active = root.get_full_property(NET_ACTIVE_WINDOW, X.AnyPropertyType)
if not active or not active.value:
return ForegroundInfo(available=True, error="no active window")
win_id = int(active.value[0])
win = d.create_resource_object("window", win_id)
pid_prop = win.get_full_property(NET_WM_PID, X.AnyPropertyType)
pid_val = int(pid_prop.value[0]) if pid_prop and pid_prop.value else None
name_prop = win.get_full_property(NET_WM_NAME, UTF8_STRING)
window_title = (
name_prop.value.decode("utf-8", "replace") if name_prop and name_prop.value else None
)
process_name: str | None = None
executable_path: str | None = None
started_at: float | None = None
if pid_val:
try:
exe = os.readlink(f"/proc/{pid_val}/exe")
executable_path = exe
process_name = os.path.basename(exe)
except OSError as e:
logger.debug("readlink /proc/%d/exe failed: %s", pid_val, e)
try:
started_at = os.stat(f"/proc/{pid_val}").st_ctime
except OSError as e:
logger.debug("stat /proc/%d failed: %s", pid_val, e)
return ForegroundInfo(
available=True,
pid=pid_val,
process_name=process_name,
executable_path=executable_path,
window_title=window_title,
window_handle=win_id,
started_at=started_at,
)
except Exception as e:
logger.debug("Linux foreground probe failed: %s", e)
return ForegroundInfo(available=False, error=str(e))
def _enrich_browser(info: ForegroundInfo) -> ForegroundInfo:
"""If ``info`` describes a focused browser, attach URL + page title.
The UIA lookup is wrapped in its own try/except so a failure here can't
take down the rest of the foreground probe.
"""
try:
from . import browser_url_service as bus
except Exception as e:
logger.debug("browser_url_service unavailable: %s", e)
return info
if not info.available or not bus.is_browser_process(info.process_name):
return info
try:
page = bus.get_browser_page(
hwnd=info.window_handle,
process_name=info.process_name,
window_title=info.window_title,
)
except Exception as e:
logger.debug("Browser URL enrichment failed: %s", e)
return info
# ``dataclasses.replace`` keeps the frozen-dataclass contract.
from dataclasses import replace
return replace(
info,
is_browser=True,
browser_url=page.url,
browser_page_title=page.page_title,
)
def _probe() -> ForegroundInfo:
system = platform.system()
try:
if system == "Windows":
info = _probe_windows()
elif system == "Darwin":
info = _probe_macos()
elif system == "Linux":
info = _probe_linux()
else:
return ForegroundInfo(
available=False, error=f"unsupported platform: {system}"
)
return _enrich_browser(info)
except Exception as e:
logger.warning("Foreground probe crashed: %s", e)
return ForegroundInfo(available=False, error=str(e))
def get_foreground_info(force_refresh: bool = False) -> ForegroundInfo:
"""Return the current foreground window/process snapshot.
Args:
force_refresh: bypass the short TTL cache. WebSocket broadcast loop
should leave this False; the REST endpoint accepts ?refresh=1
for callers that want a fresh probe.
"""
if force_refresh:
_cache.invalidate()
return _cache.get(_CACHE_TTL, _probe)
def reset_cache() -> None:
"""Reset the cache. Useful in tests."""
_cache.invalidate()
@@ -19,6 +19,9 @@ class ConnectionManager:
self._active_connections: set[WebSocket] = set()
self._lock = asyncio.Lock()
self._last_status: dict[str, Any] | None = None
self._last_foreground: dict[str, Any] | None = None
self._foreground_poll_interval: float = 1.0
self._last_foreground_poll: float = 0.0
self._get_status_func: Callable[[], Coroutine[Any, Any, Any]] | None = None
self._broadcast_task: asyncio.Task | None = None
self._poll_interval: float = 0.5 # Internal poll interval for change detection
@@ -54,6 +57,18 @@ class ConnectionManager:
except Exception as e:
logger.debug("Failed to send initial status: %s", e)
# Push a fresh foreground snapshot on connect so the UI can render
# the tile immediately instead of waiting for the next change.
try:
from .foreground_service import get_foreground_info
fg = await asyncio.to_thread(get_foreground_info)
fg_dict = fg.to_dict()
self._last_foreground = fg_dict
await websocket.send_json({"type": "foreground", "data": fg_dict})
except Exception as e:
logger.debug("Failed to send initial foreground snapshot: %s", e)
async def disconnect(self, websocket: WebSocket) -> None:
"""Remove a WebSocket connection. Stops audio capture if last visualizer subscriber."""
should_stop = False
@@ -115,6 +130,35 @@ class ConnectionManager:
await self.broadcast(message)
logger.info("Broadcast sent: links_changed")
def foreground_changed(
self, old: dict[str, Any] | None, new: dict[str, Any]
) -> bool:
"""Detect a meaningful change in the foreground process snapshot.
The probe also returns ``window_geometry`` which jitters on every
pixel of cursor drag — comparing the whole dict would flood clients.
We only diff the fields a user (or HA automation) would actually act
on. ``window_geometry``/``monitor_geometry``/``started_at`` are still
delivered in the payload, but they don't drive broadcast cadence.
"""
if old is None:
return True
diff_fields = (
"pid",
"process_name",
"executable_path",
"window_title",
"is_fullscreen",
"is_minimized",
"monitor_id",
"available",
"error",
)
for f in diff_fields:
if old.get(f) != new.get(f):
return True
return False
async def subscribe_visualizer(self, websocket: WebSocket) -> None:
"""Subscribe a client to audio visualizer data. Starts capture on first subscriber."""
should_start = False
@@ -314,6 +358,10 @@ class ConnectionManager:
get_status_func: Callable[[], Coroutine[Any, Any, Any]],
) -> None:
"""Background loop that polls for status changes and broadcasts."""
# Foreground tracker is imported lazily so unit tests of the WS
# manager don't drag in platform-specific probe code.
from .foreground_service import get_foreground_info
while self._running:
try:
# Only poll if we have connected clients
@@ -340,6 +388,28 @@ class ConnectionManager:
# Update cached status even without broadcast
self._last_status = status_dict
# Foreground process — poll at a coarser interval than media
# status. Broadcasts only fire on a real change, so a quiet
# desktop costs nothing.
now = time.time()
if (
now - self._last_foreground_poll
) >= self._foreground_poll_interval:
self._last_foreground_poll = now
try:
fg = await asyncio.to_thread(get_foreground_info)
fg_dict = fg.to_dict()
if self.foreground_changed(self._last_foreground, fg_dict):
self._last_foreground = fg_dict
await self.broadcast(
{"type": "foreground_update", "data": fg_dict}
)
logger.debug("Broadcast sent: foreground change")
else:
self._last_foreground = fg_dict
except Exception as e:
logger.debug("Foreground poll failed: %s", e)
await asyncio.sleep(self._poll_interval)
except asyncio.CancelledError:
+318
View File
@@ -9321,3 +9321,321 @@ body.is-fullscreen-player .now-playing .vu-meter {
body.is-fullscreen-player .fs-bloom #fs-bloom-art { animation: none !important; }
:root[data-theme="light"] body.is-fullscreen-player .fs-bloom { opacity: 0.22; }
}
/* ════════════════════════════════════════════════════════════════
FOREGROUND container — editorial process plate
════════════════════════════════════════════════════════════════ */
.foreground-container {
background: transparent;
border: 0;
padding: 0;
box-shadow: none;
margin-top: 28px;
}
.foreground-stage {
min-height: 360px;
}
/* Match the inter-section gap used between .settings-section blocks
in the Settings tab — keeps cadence consistent across tabs. */
.display-container > * + * {
margin-top: 16px;
}
.foreground-card {
position: relative;
display: block;
padding: clamp(24px, 3vw, 40px) clamp(24px, 3vw, 40px) 28px;
border: 1px solid var(--rule);
border-top: 2px solid var(--copper);
background:
radial-gradient(120% 80% at 0% 0%, rgba(var(--copper-rgb), 0.05), transparent 60%),
var(--bg-paper);
box-shadow:
0 1px 0 var(--bg-paper),
0 28px 60px -28px rgba(0, 0, 0, 0.45),
0 8px 20px -10px rgba(0, 0, 0, 0.25);
}
.foreground-card[data-fullscreen="1"] {
border-top-color: var(--copper-hi);
box-shadow:
0 1px 0 var(--bg-paper),
0 28px 60px -28px rgba(0, 0, 0, 0.55),
0 0 0 1px rgba(var(--copper-rgb), 0.18),
0 0 60px -12px var(--copper-glow);
}
.foreground-card .fg-kicker {
display: flex;
align-items: center;
gap: 14px;
font-family: var(--mono);
font-size: 10px;
letter-spacing: 0.32em;
text-transform: uppercase;
color: var(--copper);
margin-bottom: 22px;
}
.foreground-card .fg-kicker::before,
.foreground-card .fg-kicker::after {
content: "";
height: 1px;
background: var(--copper);
opacity: 0.6;
flex: 0 0 24px;
}
.foreground-card .fg-kicker::after { flex: 1 0 auto; }
.foreground-card .fg-process {
font-family: var(--serif);
font-weight: 400;
font-size: clamp(34px, 4.4vw, 56px);
line-height: 1.02;
letter-spacing: -0.02em;
font-variation-settings: 'opsz' 144;
color: var(--ink);
margin: 0 0 10px;
word-break: break-word;
overflow-wrap: anywhere;
transition: color 180ms var(--ease, ease);
}
.foreground-card .fg-process:hover {
color: var(--copper-hi);
}
.foreground-card .fg-window-title {
font-family: var(--serif);
font-style: italic;
font-size: 20px;
font-weight: 300;
color: var(--ink-soft);
font-variation-settings: 'opsz' 60;
margin-bottom: 22px;
line-height: 1.35;
overflow: hidden;
text-overflow: ellipsis;
display: -webkit-box;
-webkit-line-clamp: 2;
line-clamp: 2;
-webkit-box-orient: vertical;
word-break: break-word;
}
.foreground-card .fg-window-title:empty { display: none; }
.foreground-card .fg-chips {
display: flex;
flex-wrap: wrap;
gap: 8px;
margin-bottom: 28px;
}
.foreground-card .fg-chips:empty { display: none; }
.fg-chip {
display: inline-flex;
align-items: center;
padding: 5px 11px;
font-family: var(--mono);
font-size: 10px;
letter-spacing: 0.18em;
text-transform: uppercase;
color: var(--ink-soft);
background: transparent;
border: 1px solid var(--rule-strong);
border-radius: 999px;
line-height: 1.2;
white-space: nowrap;
}
.fg-chip.fg-chip-accent {
color: var(--copper);
border-color: var(--copper);
background: rgba(var(--copper-rgb), 0.07);
}
.fg-chip.fg-chip-mute {
color: var(--ink-mute);
border-color: var(--rule);
}
.foreground-card .fg-details {
display: block;
margin: 0;
border-top: 1px solid var(--rule);
}
.foreground-card .fg-row {
display: grid;
grid-template-columns: minmax(160px, 220px) 1fr;
gap: 24px;
padding: 14px 0;
border-bottom: 1px solid var(--rule);
align-items: baseline;
min-width: 0;
}
.foreground-card .fg-row dt {
font-family: var(--mono);
font-size: 9px;
letter-spacing: 0.22em;
text-transform: uppercase;
color: var(--copper);
margin: 0;
}
.foreground-card .fg-row dd {
font-family: var(--serif);
font-style: italic;
font-size: 18px;
color: var(--ink);
font-variation-settings: 'opsz' 30;
margin: 0;
min-width: 0;
overflow: hidden;
text-overflow: ellipsis;
}
.foreground-card .fg-mono {
font-family: var(--mono);
font-style: normal;
font-size: 13px;
letter-spacing: 0.02em;
color: var(--ink-soft);
font-variant-numeric: tabular-nums;
word-break: break-all;
}
.foreground-empty {
padding: 60px 24px;
text-align: center;
color: var(--ink-mute);
}
.foreground-empty svg {
width: 64px;
height: 64px;
margin-bottom: 14px;
opacity: 0.55;
color: var(--ink-faint);
}
.foreground-empty p {
font-family: var(--serif);
font-style: italic;
font-size: 18px;
color: var(--ink-soft);
margin: 0;
}
.foreground-empty .foreground-empty-error {
margin-top: 10px;
font-family: var(--mono);
font-style: normal;
font-size: 11px;
letter-spacing: 0.06em;
color: var(--ink-mute);
word-break: break-word;
}
/* ─── Header status badge ──────────────────────────────────── */
.foreground-status-badge {
display: inline-flex;
align-items: center;
gap: 8px;
height: 32px;
padding: 0 12px 0 10px;
margin-right: 4px;
background: transparent;
border: 1px solid var(--rule-strong);
border-radius: 999px;
color: var(--ink-soft);
font-family: var(--mono);
font-size: 11px;
letter-spacing: 0.04em;
cursor: pointer;
max-width: 240px;
transition: color 180ms ease, border-color 180ms ease, background 180ms ease;
}
.foreground-status-badge:hover {
color: var(--ink);
border-color: var(--copper);
background: rgba(var(--copper-rgb), 0.06);
}
.foreground-status-badge.hidden { display: none !important; }
.foreground-status-badge .fg-badge-mark {
width: 6px;
height: 6px;
border-radius: 50%;
background: var(--ink-mute);
flex-shrink: 0;
}
.foreground-status-badge.is-media .fg-badge-mark,
.foreground-status-badge.is-fullscreen .fg-badge-mark {
background: var(--copper);
box-shadow: 0 0 8px var(--copper-glow);
}
.foreground-status-badge.is-fullscreen {
border-color: var(--copper);
color: var(--ink);
}
.foreground-status-badge .fg-badge-name {
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
max-width: 140px;
}
.foreground-status-badge .fg-badge-tag {
color: var(--copper);
text-transform: uppercase;
letter-spacing: 0.14em;
font-size: 10px;
flex-shrink: 0;
}
.foreground-status-badge .fg-badge-tag.hidden { display: none; }
/* ─── Light theme overrides ──────────────────────────────── */
:root[data-theme="light"] .foreground-card {
background:
radial-gradient(120% 80% at 0% 0%, rgba(var(--copper-rgb), 0.05), transparent 60%),
var(--bg-paper);
box-shadow:
0 1px 0 var(--bg-paper),
0 22px 50px -24px rgba(26, 23, 21, 0.20),
0 6px 16px -8px rgba(26, 23, 21, 0.12);
}
:root[data-theme="light"] .foreground-card[data-fullscreen="1"] {
box-shadow:
0 1px 0 var(--bg-paper),
0 22px 50px -24px rgba(26, 23, 21, 0.28),
0 0 0 1px rgba(var(--copper-rgb), 0.20),
0 0 50px -12px var(--copper-glow);
}
:root[data-theme="light"] .foreground-status-badge {
border-color: rgba(26, 23, 21, 0.18);
}
:root[data-theme="light"] .foreground-status-badge:hover {
background: rgba(var(--copper-rgb), 0.08);
}
/* ─── Mobile breakpoint ──────────────────────────────────── */
@media (max-width: 720px) {
.foreground-card {
padding: 22px 18px 20px;
}
.foreground-card .fg-process {
font-size: 30px;
}
.foreground-card .fg-window-title {
font-size: 16px;
}
.foreground-card .fg-row {
grid-template-columns: 1fr;
gap: 4px;
padding: 12px 0;
}
.foreground-card .fg-row dd {
font-size: 16px;
}
.foreground-status-badge {
max-width: 160px;
}
.foreground-status-badge .fg-badge-name {
max-width: 80px;
}
.foreground-status-badge .fg-badge-tag {
display: none;
}
}
+7 -1
View File
@@ -535,7 +535,7 @@
</details>
</div>
<!-- Display Control Section -->
<!-- Display Control Section (monitors first, foreground overview below) -->
<div class="display-container" data-tab-content="display" role="tabpanel" id="panel-display">
<div class="display-monitors" id="displayMonitors">
<div class="empty-state-illustration">
@@ -543,6 +543,12 @@
<p data-i18n="display.loading">Loading monitors...</p>
</div>
</div>
<div class="foreground-stage" id="foregroundStage">
<div class="empty-state-illustration">
<svg viewBox="0 0 64 64"><rect x="6" y="12" width="52" height="36" rx="3" fill="none" stroke="currentColor" stroke-width="2"/><line x1="6" y1="20" x2="58" y2="20" stroke="currentColor" stroke-width="2"/><circle cx="11" cy="16" r="1.4" fill="currentColor"/><circle cx="15" cy="16" r="1.4" fill="currentColor"/><circle cx="19" cy="16" r="1.4" fill="currentColor"/></svg>
<p data-i18n="foreground.loading">Waiting for foreground signal…</p>
</div>
</div>
</div>
</div>
+6
View File
@@ -74,6 +74,10 @@ import {
toggleDynamicBackground, applyDynamicBackground, updateBackgroundColors,
} from './background.js';
import {
updateForegroundUI, loadForegroundProcess,
} from './foreground.js';
// ============================================================
// Register late-bound callbacks for core's updateAllText()
// ============================================================
@@ -136,6 +140,8 @@ Object.assign(window, {
onAudioDeviceChanged,
// About
showAboutDialog, closeAboutDialog,
// Foreground
loadForegroundProcess,
});
// ============================================================
+188
View File
@@ -0,0 +1,188 @@
// ============================================================
// Foreground: Currently-focused desktop process card (rendered at
// the top of the Display tab)
// ============================================================
import { t } from './core.js';
let latestForeground = null;
let agoTickTimer = null;
function escapeHtml(s) {
if (s === null || s === undefined) return '';
return String(s)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#39;');
}
function formatAgo(epoch) {
if (!epoch) return '';
const now = Date.now() / 1000;
const diff = Math.max(0, now - epoch);
if (diff < 60) {
return t('foreground.ago.seconds', { n: Math.floor(diff) });
}
if (diff < 3600) {
return t('foreground.ago.minutes', { n: Math.floor(diff / 60) });
}
if (diff < 86400) {
const h = Math.floor(diff / 3600);
const m = Math.floor((diff % 3600) / 60);
return t('foreground.ago.hours', { n: h, m: m });
}
return t('foreground.ago.days', { n: Math.floor(diff / 86400) });
}
function formatGeometry(g) {
if (!g) return '—';
const w = g.width ?? (g.right - g.left);
const h = g.height ?? (g.bottom - g.top);
return `${w}×${h} @ (${g.left}, ${g.top})`;
}
function truncatePath(p, max = 64) {
if (!p) return '';
if (p.length <= max) return p;
// Keep the tail (filename) visible — that's the part the user cares about.
return '…' + p.slice(-(max - 1));
}
function renderEmpty(message, errorMsg) {
const stage = document.getElementById('foregroundStage');
if (!stage) return;
stage.innerHTML = `
<div class="empty-state-illustration foreground-empty">
<svg viewBox="0 0 64 64"><rect x="6" y="12" width="52" height="36" rx="3" fill="none" stroke="currentColor" stroke-width="2"/><line x1="6" y1="20" x2="58" y2="20" stroke="currentColor" stroke-width="2"/><circle cx="11" cy="16" r="1.4" fill="currentColor"/><circle cx="15" cy="16" r="1.4" fill="currentColor"/><circle cx="19" cy="16" r="1.4" fill="currentColor"/></svg>
<p>${escapeHtml(message)}</p>
${errorMsg ? `<p class="foreground-empty-error">${escapeHtml(errorMsg)}</p>` : ''}
</div>
`;
}
function renderTile(data) {
const stage = document.getElementById('foregroundStage');
if (!stage) return;
const procName = data.process_name || '—';
const winTitle = data.window_title || '';
const execPath = data.executable_path || '';
const pid = data.pid ?? '—';
const startedEpoch = data.started_at;
const startedAgo = startedEpoch ? formatAgo(startedEpoch) : '—';
const startedAbs = startedEpoch
? new Date(startedEpoch * 1000).toLocaleString()
: '';
const geom = formatGeometry(data.window_geometry);
const platform = data.platform || '—';
const monitorId = data.monitor_id;
// Chips: only render ones that apply
const chips = [];
if (data.is_fullscreen) {
chips.push(`<span class="fg-chip fg-chip-accent">${escapeHtml(t('foreground.fullscreen'))}</span>`);
} else if (!data.is_minimized) {
chips.push(`<span class="fg-chip">${escapeHtml(t('foreground.windowed'))}</span>`);
}
if (data.is_minimized) {
chips.push(`<span class="fg-chip fg-chip-mute">${escapeHtml(t('foreground.minimized'))}</span>`);
}
if (monitorId !== null && monitorId !== undefined) {
chips.push(`<span class="fg-chip">${escapeHtml(t('foreground.monitor', { n: monitorId + 1 }))}</span>`);
}
if (data.is_browser) {
chips.push(`<span class="fg-chip fg-chip-mute">${escapeHtml(t('foreground.browser'))}</span>`);
}
// Optional browser-only detail rows (page title + URL when available)
const browserRows = [];
if (data.is_browser) {
if (data.browser_page_title) {
browserRows.push(`
<div class="fg-row">
<dt>${escapeHtml(t('foreground.page_title'))}</dt>
<dd title="${escapeHtml(data.browser_page_title)}">${escapeHtml(data.browser_page_title)}</dd>
</div>
`);
}
if (data.browser_url) {
browserRows.push(`
<div class="fg-row">
<dt>${escapeHtml(t('foreground.url'))}</dt>
<dd title="${escapeHtml(data.browser_url)}"><span class="fg-mono">${escapeHtml(truncatePath(data.browser_url, 80))}</span></dd>
</div>
`);
}
}
stage.innerHTML = `
<article class="foreground-card" data-fullscreen="${data.is_fullscreen ? '1' : '0'}">
<div class="fg-kicker">
<span data-i18n="foreground.kicker">Foreground</span>
</div>
<h1 class="fg-process" title="${escapeHtml(procName)}">${escapeHtml(procName)}</h1>
<div class="fg-window-title" title="${escapeHtml(winTitle)}">${escapeHtml(winTitle)}</div>
<div class="fg-chips">${chips.join('')}</div>
<dl class="fg-details">
${browserRows.join('')}
<div class="fg-row">
<dt>${escapeHtml(t('foreground.executable'))}</dt>
<dd title="${escapeHtml(execPath)}"><span class="fg-mono">${escapeHtml(truncatePath(execPath))}</span></dd>
</div>
<div class="fg-row">
<dt>${escapeHtml(t('foreground.pid'))}</dt>
<dd><span class="fg-mono">${escapeHtml(String(pid))}</span></dd>
</div>
<div class="fg-row">
<dt>${escapeHtml(t('foreground.started'))}</dt>
<dd title="${escapeHtml(startedAbs)}"><span class="fg-ago" data-started="${startedEpoch ?? ''}">${escapeHtml(startedAgo)}</span></dd>
</div>
<div class="fg-row">
<dt>${escapeHtml(t('foreground.geometry'))}</dt>
<dd><span class="fg-mono">${escapeHtml(geom)}</span></dd>
</div>
<div class="fg-row">
<dt>${escapeHtml(t('foreground.platform'))}</dt>
<dd>${escapeHtml(platform)}</dd>
</div>
</dl>
</article>
`;
}
function startAgoTicker() {
if (agoTickTimer) return;
agoTickTimer = setInterval(() => {
const el = document.querySelector('.fg-ago[data-started]');
if (!el) return;
const epoch = parseFloat(el.getAttribute('data-started'));
if (!epoch) return;
el.textContent = formatAgo(epoch);
}, 15000);
}
export function updateForegroundUI(data) {
latestForeground = data;
if (!data || data.available === false) {
const errMsg = data && data.error ? data.error : '';
renderEmpty(t('foreground.unavailable'), errMsg);
} else if (!data.process_name && !data.pid) {
renderEmpty(t('foreground.no_process'), '');
} else {
renderTile(data);
startAgoTicker();
}
}
export function loadForegroundProcess() {
// Push-only — just render the cached state. If nothing has arrived
// yet, leave the loading placeholder visible.
if (latestForeground !== null) {
updateForegroundUI(latestForeground);
}
}
+2
View File
@@ -13,6 +13,7 @@ import {
} from './core.js';
import { updateBackgroundColors } from './background.js';
import { loadDisplayMonitors } from './links.js';
import { loadForegroundProcess } from './foreground.js';
import { IconSelect } from './icon-select.js';
// Tab management
@@ -75,6 +76,7 @@ export function switchTab(tabName) {
if (tabName === 'display') {
loadDisplayMonitors();
loadForegroundProcess();
}
localStorage.setItem('activeTab', tabName);
+3
View File
@@ -12,6 +12,7 @@ import { updateUI, setFrequencyData, stopPositionInterpolation, loadAudioDevices
import { loadScripts, loadScriptsTable, displayQuickAccess } from './scripts.js';
import { loadCallbacksTable } from './callbacks.js';
import { loadHeaderLinks, loadLinksTable } from './links.js';
import { updateForegroundUI } from './foreground.js';
let reconnectTimeout = null;
let wsReconnectAttempts = 0;
@@ -118,6 +119,8 @@ export function connectWebSocket(token) {
if (msg.type === 'status' || msg.type === 'status_update') {
updateUI(msg.data);
} else if (msg.type === 'foreground' || msg.type === 'foreground_update') {
updateForegroundUI(msg.data);
} else if (msg.type === 'scripts_changed') {
console.log('Scripts changed, reloading...');
loadScripts();
+25 -1
View File
@@ -292,5 +292,29 @@
"about.source_code": "Source Code",
"dialog.close": "Close",
"update.available": "Update available: v{version}",
"update.view_release": "View Release"
"update.view_release": "View Release",
"tab.foreground": "Foreground",
"foreground.kicker": "Foreground",
"foreground.loading": "Waiting for foreground signal…",
"foreground.no_process": "No foreground process",
"foreground.unavailable": "Foreground tracking unavailable on this platform",
"foreground.process": "Process",
"foreground.window_title": "Window title",
"foreground.executable": "Executable",
"foreground.pid": "PID",
"foreground.monitor": "Monitor {n}",
"foreground.started": "Started",
"foreground.geometry": "Geometry",
"foreground.platform": "Platform",
"foreground.fullscreen": "Fullscreen",
"foreground.minimized": "Minimized",
"foreground.windowed": "Windowed",
"foreground.browser": "Browser",
"foreground.page_title": "Page title",
"foreground.url": "URL",
"foreground.badge.title": "View foreground process",
"foreground.ago.seconds": "{n}s ago",
"foreground.ago.minutes": "{n}m ago",
"foreground.ago.hours": "{n}h {m}m ago",
"foreground.ago.days": "{n}d ago"
}
+25 -1
View File
@@ -292,5 +292,29 @@
"about.source_code": "Исходный код",
"dialog.close": "Закрыть",
"update.available": "Доступно обновление: v{version}",
"update.view_release": "Перейти к релизу"
"update.view_release": "Перейти к релизу",
"tab.foreground": "Активное окно",
"foreground.kicker": "Активное окно",
"foreground.loading": "Ожидание сигнала об активном окне…",
"foreground.no_process": "Активное окно не определено",
"foreground.unavailable": "Отслеживание активного окна недоступно",
"foreground.process": "Процесс",
"foreground.window_title": "Заголовок окна",
"foreground.executable": "Путь к программе",
"foreground.pid": "PID",
"foreground.monitor": "Монитор {n}",
"foreground.started": "Запущено",
"foreground.geometry": "Геометрия",
"foreground.platform": "Платформа",
"foreground.fullscreen": "Полноэкранный",
"foreground.minimized": "Свёрнут",
"foreground.windowed": "Оконный",
"foreground.browser": "Браузер",
"foreground.page_title": "Заголовок страницы",
"foreground.url": "URL",
"foreground.badge.title": "Открыть активное окно",
"foreground.ago.seconds": "{n} с назад",
"foreground.ago.minutes": "{n} мин назад",
"foreground.ago.hours": "{n} ч {m} мин назад",
"foreground.ago.days": "{n} дн назад"
}
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "media-server-frontend",
"version": "0.2.5",
"version": "0.2.6",
"lockfileVersion": 2,
"requires": true,
"packages": {
"": {
"name": "media-server-frontend",
"version": "0.2.5",
"version": "0.2.6",
"devDependencies": {
"esbuild": "^0.27.4"
}
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "media-server-frontend",
"version": "0.2.5",
"version": "0.2.6",
"private": true,
"description": "Frontend build tooling for media server WebUI",
"scripts": {
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "media-server"
version = "0.2.5"
version = "0.2.6"
description = "REST API server for controlling system-wide media playback"
readme = "README.md"
license = { text = "MIT" }
+87
View File
@@ -0,0 +1,87 @@
"""Smoke tests for the foreground tracker.
The OS-specific probe code is hard to mock end-to-end inside a CI container,
so these tests focus on the platform-agnostic surface: the dataclass shape,
TTL caching, and graceful fallback when the platform probe raises. The
Windows/Linux/macOS probes themselves are exercised through manual runs.
"""
from __future__ import annotations
import time
import pytest
from media_server.services import foreground_service as fg
def setup_function(_):
fg.reset_cache()
def test_unavailable_default_shape():
info = fg.ForegroundInfo(available=False)
d = info.to_dict()
assert d["available"] is False
assert d["pid"] is None
assert d["process_name"] is None
assert d["is_fullscreen"] is False
assert "platform" in d
def test_cache_returns_same_instance(monkeypatch):
calls = {"n": 0}
def fake_probe():
calls["n"] += 1
return fg.ForegroundInfo(available=True, pid=42, process_name="x.exe")
monkeypatch.setattr(fg, "_probe", fake_probe)
a = fg.get_foreground_info()
b = fg.get_foreground_info()
assert a is b
assert calls["n"] == 1
def test_cache_force_refresh(monkeypatch):
calls = {"n": 0}
def fake_probe():
calls["n"] += 1
return fg.ForegroundInfo(available=True, pid=calls["n"])
monkeypatch.setattr(fg, "_probe", fake_probe)
fg.get_foreground_info()
fg.get_foreground_info(force_refresh=True)
assert calls["n"] == 2
def test_cache_ttl_expiry(monkeypatch):
calls = {"n": 0}
def fake_probe():
calls["n"] += 1
return fg.ForegroundInfo(available=True, pid=calls["n"])
monkeypatch.setattr(fg, "_probe", fake_probe)
monkeypatch.setattr(fg, "_CACHE_TTL", 0.0)
# Re-bind the cache's TTL by exercising it twice with TTL 0.
fg.get_foreground_info()
fg.get_foreground_info()
assert calls["n"] == 2
def test_probe_crash_returns_unavailable(monkeypatch):
def boom():
raise RuntimeError("kaboom")
# Force every platform branch to call our crashing probe.
monkeypatch.setattr(fg, "_probe_windows", boom)
monkeypatch.setattr(fg, "_probe_linux", boom)
monkeypatch.setattr(fg, "_probe_macos", boom)
info = fg._probe()
assert info.available is False
assert info.error and "kaboom" in info.error