Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 35 additions & 8 deletions src/display_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,8 @@
from contextlib import contextmanager
from PIL import Image, ImageDraw, ImageFont
import time
from typing import Dict, Any, List, Optional
from collections import OrderedDict
from typing import Dict, Any, List, Optional, Tuple
import logging
import math
import freetype
Expand Down Expand Up @@ -180,14 +181,25 @@ def __init__(self, config: Dict[str, Any] = None, force_fallback: bool = False,
# the logical image is blitted to the matrix unchanged.
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
self._physical_image = None # full-chain buffer reused each frame when tiling
# Text-width measurement cache: (text, id(font)) -> pixel_width
# Text-width measurement cache: (text, id(font)) -> (width, font_ref)
# Avoids re-measuring the same string+font on every display() call.
# LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
# live score) would otherwise grow it forever on a 24/7 service.
# Entries hold a strong reference to the font so its id() can't be
# recycled by a different font object — an id-keyed cache without
# the reference can return the WRONG width after garbage collection.
# Cleared on _load_fonts() so stale entries don't survive a font reload.
self._text_width_cache: Dict[tuple, int] = {}
self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
self._TEXT_WIDTH_CACHE_MAX = 1024
# Snapshot settings for web preview integration (service writes, web reads)
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
self._snapshot_min_interval_sec = 0.2 # max ~5 fps
self._last_snapshot_ts = 0.0
# Snapshot failures are logged as warnings, rate-limited so a
# persistent failure (e.g. an unwritable file) can't spam the log —
# but is never silent: the snapshot's mtime doubles as the web UI's
# hardware-liveness signal, so a quiet failure makes health checks lie.
self._snapshot_fail_log_ts = 0.0

# Scrolling state tracking for graceful updates
self._scrolling_state = {
Expand Down Expand Up @@ -699,12 +711,15 @@ def get_text_width(self, text, font):

Results are cached by (text, font identity) so plugins that measure
the same string every frame (e.g. to centre a score) pay only one
measurement per unique (text, font) pair.
measurement per unique (text, font) pair. The entry keeps the font
alive so its id() can't be recycled, and the cache is LRU-bounded so
ever-changing text (clocks, tickers) can't grow it without limit.
"""
cache_key = (text, id(font))
cached = self._text_width_cache.get(cache_key)
if cached is not None:
return cached
self._text_width_cache.move_to_end(cache_key)
return cached[0]

try:
if isinstance(font, freetype.Face):
Expand All @@ -719,7 +734,9 @@ def get_text_width(self, text, font):
logger.error("Error getting text width: %s", e)
return 0

self._text_width_cache[cache_key] = width
self._text_width_cache[cache_key] = (width, font)
while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
self._text_width_cache.popitem(last=False)
return width

def get_font_height(self, font):
Expand Down Expand Up @@ -1164,5 +1181,15 @@ def _write_snapshot_if_due(self) -> None:
pass
self._last_snapshot_ts = now
except Exception as e:
# Snapshot failures should never break display; log at debug to avoid noise
logger.debug(f"Snapshot write skipped: {e}")
# Snapshot failures must never break display — but they must not
# be silent either: the snapshot's mtime is the web UI's display
# mirror AND its hardware-liveness proxy, so a quietly failing
# write freezes the mirror and makes health checks lie (seen in
# the field: a stale root-owned /tmp file froze it for a day).
# Warn at most once per 5 minutes to avoid log spam.
if (now - self._snapshot_fail_log_ts) > 300:
self._snapshot_fail_log_ts = now
logger.warning("Snapshot write failing (web preview/health "
"mirror is stale): %s", e)
else:
logger.debug(f"Snapshot write skipped: {e}")
23 changes: 18 additions & 5 deletions src/font_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import zipfile
import tempfile
import time
from collections import OrderedDict
from pathlib import Path
from PIL import ImageFont
from typing import Dict, Tuple, Optional, Union, Any, List
Expand All @@ -58,7 +59,13 @@ def __init__(self, config: Dict[str, Any]):
# Font discovery and catalog
self.font_catalog: Dict[str, str] = {} # family_name -> file_path
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
self.metrics_cache: Dict[str, Tuple[int, int, int]] = {} # (text, font_id) -> (width, height, baseline)
# (text, id(font)) -> ((width, height, baseline), font_ref).
# LRU-bounded — keys embed the measured TEXT, so changing strings
# (clocks, live scores) would otherwise grow it forever. Entries
# keep the font alive so its id() can't be recycled by a different
# font object (which would silently return wrong metrics).
self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
self._METRICS_CACHE_MAX = 1024

# Plugin font management
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
Expand Down Expand Up @@ -507,10 +514,14 @@ def measure_text(self, text: str, font: Union[ImageFont.FreeTypeFont, freetype.F
Returns:
Tuple of (width, height, baseline_offset)
"""
cache_key = f"{hash(text)}_{id(font)}"
# Key on the text itself (hash(text) could collide) + font identity;
# the entry below keeps the font referenced so the id stays valid.
cache_key = (text, id(font))

if cache_key in self.metrics_cache:
return self.metrics_cache[cache_key]
cached = self.metrics_cache.get(cache_key)
if cached is not None:
self.metrics_cache.move_to_end(cache_key)
return cached[0]

try:
if isinstance(font, freetype.Face):
Expand Down Expand Up @@ -547,7 +558,9 @@ def measure_text(self, text: str, font: Union[ImageFont.FreeTypeFont, freetype.F
baseline = 10

result = (width, height, baseline)
self.metrics_cache[cache_key] = result
self.metrics_cache[cache_key] = (result, font)
while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
self.metrics_cache.popitem(last=False)
return result

def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
Expand Down
6 changes: 5 additions & 1 deletion test/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,11 @@ def mock_cache_manager():
mock._memory_cache_timestamps = {}
mock.cache_dir = "/tmp/test_cache"

def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
def mock_get(key: str, max_age: Optional[int] = 300,
memory_ttl: Optional[int] = None) -> Optional[Dict]:
# Signature mirrors CacheManager.get — keep in sync or callers
# passing keyword args (health tracker, resource monitor) break
# only in tests, hiding real-API compatibility.
return mock._memory_cache.get(key)

def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
Expand Down
Loading