"""Which plugins use which font, published for the web interface. The display service and the web interface are separate processes. Plugins run, and register their fonts with ``FontManager.register_manager_font()``, in the display service only; the web interface's Fonts tab lists the files in ``assets/fonts/`` from its own scan and has no FontManager to ask. So the display service publishes a small snapshot to the shared cache directory -- the same channel, and the same file permissions, as ``display_current_state`` and ``plugin_error_snapshot``: files are 0660 and carry the cache directory's group, so root writes and the web user reads. FONT_USAGE_KEY written by the display service only The snapshot:: {"generated_at": "2026-09-23T10:00:00", "fonts": {"PressStart2P-Regular": ["calendar", "hello-world"], "4x6-font": ["calendar"]}} ``fonts`` is keyed the way ``GET /api/v3/fonts/catalog`` keys its rows: the file name in ``assets/fonts/`` without its extension, as it is on disk. A plugin names a font by FontManager family ("press_start", "5x7", "6x13b"), by alias, or by path; :func:`catalog_key_for` resolves each through FontManager's own catalog to the file it loads and keys it by that file's stem. Fonts that live outside ``assets/fonts/`` (a plugin's own ``plugin_id::family`` fonts) and families FontManager cannot resolve are left out: the Fonts tab has no row for them. Only registrations are counted. ``get_font()`` does not know which plugin is calling it, and many plugins open font files directly with PIL, so a font no plugin registered may still be in use; the web interface says so. """ import logging import os import threading import time from datetime import datetime from typing import Any, Callable, Dict, FrozenSet, List, Optional, Tuple logger = logging.getLogger(__name__) FONT_USAGE_KEY = "font_usage_snapshot" #: How often the display service checks for a change. A check compares two #: integers and a set of plugin ids; the snapshot is rebuilt and written only #: when one of them moved, so plugins that register fonts every frame (the #: countdown plugin registers per countdown at render time) cost nothing. TICK_INTERVAL = 10.0 #: Rewrite an unchanged snapshot this often anyway. The cache's disk cleanup #: deletes entries older than its default retention (30 days), and a display #: service that runs longer than that without a font change would otherwise #: turn the Fonts tab back to "unknown". REFRESH_INTERVAL = 24 * 3600.0 FONT_EXTENSIONS = ('.ttf', '.otf', '.bdf') # Bounds on what the web interface accepts from the snapshot file. _MAX_FONTS = 1000 _MAX_PLUGINS_PER_FONT = 200 _MAX_ID_CHARS = 200 def _fonts_dir() -> str: from src.common.font_layout import resolve_asset_path return resolve_asset_path("assets/fonts") def _same_dir(a: str, b: str) -> bool: try: return os.path.normcase(os.path.realpath(a)) == os.path.normcase(os.path.realpath(b)) except (OSError, ValueError): return False def catalog_key_for(family: Any, font_catalog: Dict[str, str], fonts_dir: Optional[str] = None) -> Optional[str]: """The Fonts-tab catalog key for a family a plugin registered, or None. Resolution, in order: 1. ``family`` as a FontManager family -- a scanned file ("5x7", "pressstart2p-regular"), an alias ("press_start", "four_by_six", "five_by_seven", "tom_thumb"), or a plugin's ``plugin_id::family``. FontManager keys scanned files in lower case, so the lookup falls back to ``family.lower()``. 2. ``family`` as a file name or path ("5x7.bdf", "assets/fonts/4x6-font.ttf", an absolute path). The key is the resolved file's name without its extension, and only for a file directly in ``assets/fonts/`` -- anything else has no catalog row. """ if not isinstance(family, str): return None name = family.strip() if not name: return None fonts_dir = fonts_dir or _fonts_dir() path = font_catalog.get(name) or font_catalog.get(name.lower()) if not path: if not name.lower().endswith(FONT_EXTENSIONS): return None path = name directory, filename = os.path.split(str(path)) stem, ext = os.path.splitext(filename) if not stem or ext.lower() not in FONT_EXTENSIONS: return None if directory: if not os.path.isabs(directory): from src.common.font_layout import resolve_asset_path directory = resolve_asset_path(directory) if not _same_dir(directory, fonts_dir): return None return stem def build_font_usage(font_manager: Any, plugin_ids: Optional[FrozenSet[str]] = None, fonts_dir: Optional[str] = None) -> Dict[str, List[str]]: """``{catalog key: sorted plugin ids}`` from FontManager's registrations. ``plugin_ids`` limits it to loaded plugins: a plugin whose constructor registered fonts and which then failed validation is not using them. """ fonts_dir = fonts_dir or _fonts_dir() # list() copies in one step under the GIL; register_manager_font() runs # on the render thread while this runs on the publisher's. registrations = list(getattr(font_manager, 'manager_fonts', {}).items()) font_catalog = dict(getattr(font_manager, 'font_catalog', {})) usage: Dict[str, set] = {} for manager_id, elements in registrations: if not isinstance(manager_id, str): continue if plugin_ids is not None and manager_id not in plugin_ids: continue for spec in list(elements.values()) if isinstance(elements, dict) else (): family = spec.get('family') if isinstance(spec, dict) else None key = catalog_key_for(family, font_catalog, fonts_dir) if key is not None: usage.setdefault(key, set()).add(manager_id) return {key: sorted(ids) for key, ids in sorted(usage.items())} class FontUsagePublisher: """Publishes the display service's font usage to the shared cache. tick() is the whole job; start() calls it from a daemon thread every TICK_INTERVAL seconds. Nothing here raises: a failure is logged at debug and retried on the next tick. """ def __init__(self, cache_manager: Any, font_manager: Any, plugin_manager: Any = None, clock: Callable[[], float] = time.monotonic) -> None: self.cache_manager = cache_manager self.font_manager = font_manager self.plugin_manager = plugin_manager self._clock = clock self._signature: Optional[Tuple[Any, ...]] = None # None forces a first write, replacing whatever a previous run of the # service left in the cache with this run's usage. self._published: Optional[Dict[str, List[str]]] = None self._published_at: Optional[float] = None self._lock = threading.Lock() self._stop = threading.Event() self._thread: Optional[threading.Thread] = None def _loaded_plugin_ids(self) -> Optional[FrozenSet[str]]: plugins = getattr(self.plugin_manager, 'plugins', None) if not isinstance(plugins, dict): return None return frozenset(list(plugins)) def tick(self) -> bool: """Publish if the usage changed. True if a snapshot was written.""" with self._lock: try: plugin_ids = self._loaded_plugin_ids() signature = ( getattr(self.font_manager, 'manager_fonts_version', None), getattr(self.font_manager, 'cache_generation', None), plugin_ids, ) now = self._clock() refresh_due = (self._published_at is not None and now - self._published_at >= REFRESH_INTERVAL) if signature == self._signature and not refresh_due: return False usage = build_font_usage(self.font_manager, plugin_ids) if usage == self._published and not refresh_due: self._signature = signature return False self.cache_manager.set(FONT_USAGE_KEY, { 'generated_at': datetime.now().isoformat(timespec='seconds'), 'fonts': usage, }) self._signature = signature self._published = usage self._published_at = now return True except Exception as err: # never let reporting break the display logger.debug("Could not publish font usage: %s", err, exc_info=True) return False def start(self, interval: float = TICK_INTERVAL) -> None: """Tick now, then from a daemon thread until stop(). No-op while running.""" if self._thread is not None and self._thread.is_alive(): return self._stop.clear() def run() -> None: self.tick() while not self._stop.wait(interval): self.tick() self._thread = threading.Thread(target=run, name="font-usage-publisher", daemon=True) self._thread.start() def stop(self) -> None: self._stop.set() if self._thread is not None: self._thread.join(timeout=2) self._thread = None def start_font_usage_publisher(cache_manager: Any, font_manager: Any, plugin_manager: Any) -> Optional[FontUsagePublisher]: """Start publishing font usage for the web interface. Never raises. Call from the display service, once its plugins are loaded. """ try: if cache_manager is None or font_manager is None or plugin_manager is None: return None publisher = FontUsagePublisher(cache_manager, font_manager, plugin_manager) publisher.start() return publisher except Exception as err: logger.warning("Font usage reporting to the web interface is unavailable: %s", err) return None # --- Reading side (web interface) ------------------------------------------- def read_font_usage(cache_manager: Any) -> Optional[Dict[str, Any]]: """The display service's latest snapshot, or None if it has not published. memory_ttl=0: the display service writes this key, so only the file is current. The contents are checked and bounded -- they arrive from another process through a shared directory. """ try: snapshot = cache_manager.get(FONT_USAGE_KEY, max_age=None, memory_ttl=0) except Exception as err: logger.debug("Could not read font usage: %s", err, exc_info=True) return None if not isinstance(snapshot, dict) or not isinstance(snapshot.get('fonts'), dict): return None fonts: Dict[str, List[str]] = {} for key, ids in list(snapshot['fonts'].items())[:_MAX_FONTS]: if not isinstance(key, str) or not isinstance(ids, list): continue clean = sorted({i[:_MAX_ID_CHARS] for i in ids[:_MAX_PLUGINS_PER_FONT] if isinstance(i, str) and i}) if clean: fonts[key[:_MAX_ID_CHARS]] = clean generated_at = snapshot.get('generated_at') return { 'generated_at': generated_at if isinstance(generated_at, str) else None, 'fonts': fonts, }