mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-02 17:28:05 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
22f0a96cbb | ||
|
|
2a99cf6e64 | ||
|
|
b89653c836 | ||
|
|
d59a66133b | ||
|
|
d86dc5914b |
+8
-35
@@ -33,8 +33,7 @@ else:
|
|||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
from PIL import Image, ImageDraw, ImageFont
|
from PIL import Image, ImageDraw, ImageFont
|
||||||
import time
|
import time
|
||||||
from collections import OrderedDict
|
from typing import Dict, Any, List, Optional
|
||||||
from typing import Dict, Any, List, Optional, Tuple
|
|
||||||
import logging
|
import logging
|
||||||
import math
|
import math
|
||||||
import freetype
|
import freetype
|
||||||
@@ -181,25 +180,14 @@ class DisplayManager:
|
|||||||
# the logical image is blitted to the matrix unchanged.
|
# the logical image is blitted to the matrix unchanged.
|
||||||
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
|
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
|
self._physical_image = None # full-chain buffer reused each frame when tiling
|
||||||
# Text-width measurement cache: (text, id(font)) -> (width, font_ref)
|
# Text-width measurement cache: (text, id(font)) -> pixel_width
|
||||||
# Avoids re-measuring the same string+font on every display() call.
|
# 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.
|
# Cleared on _load_fonts() so stale entries don't survive a font reload.
|
||||||
self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
|
self._text_width_cache: Dict[tuple, int] = {}
|
||||||
self._TEXT_WIDTH_CACHE_MAX = 1024
|
|
||||||
# Snapshot settings for web preview integration (service writes, web reads)
|
# 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_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._snapshot_min_interval_sec = 0.2 # max ~5 fps
|
||||||
self._last_snapshot_ts = 0.0
|
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
|
# Scrolling state tracking for graceful updates
|
||||||
self._scrolling_state = {
|
self._scrolling_state = {
|
||||||
@@ -711,15 +699,12 @@ class DisplayManager:
|
|||||||
|
|
||||||
Results are cached by (text, font identity) so plugins that measure
|
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
|
the same string every frame (e.g. to centre a score) pay only one
|
||||||
measurement per unique (text, font) pair. The entry keeps the font
|
measurement per unique (text, font) pair.
|
||||||
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))
|
cache_key = (text, id(font))
|
||||||
cached = self._text_width_cache.get(cache_key)
|
cached = self._text_width_cache.get(cache_key)
|
||||||
if cached is not None:
|
if cached is not None:
|
||||||
self._text_width_cache.move_to_end(cache_key)
|
return cached
|
||||||
return cached[0]
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if isinstance(font, freetype.Face):
|
if isinstance(font, freetype.Face):
|
||||||
@@ -734,9 +719,7 @@ class DisplayManager:
|
|||||||
logger.error("Error getting text width: %s", e)
|
logger.error("Error getting text width: %s", e)
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
self._text_width_cache[cache_key] = (width, font)
|
self._text_width_cache[cache_key] = width
|
||||||
while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
|
|
||||||
self._text_width_cache.popitem(last=False)
|
|
||||||
return width
|
return width
|
||||||
|
|
||||||
def get_font_height(self, font):
|
def get_font_height(self, font):
|
||||||
@@ -1181,15 +1164,5 @@ class DisplayManager:
|
|||||||
pass
|
pass
|
||||||
self._last_snapshot_ts = now
|
self._last_snapshot_ts = now
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# Snapshot failures must never break display — but they must not
|
# Snapshot failures should never break display; log at debug to avoid noise
|
||||||
# be silent either: the snapshot's mtime is the web UI's display
|
logger.debug(f"Snapshot write skipped: {e}")
|
||||||
# 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}")
|
|
||||||
+5
-18
@@ -35,7 +35,6 @@ import urllib.request
|
|||||||
import zipfile
|
import zipfile
|
||||||
import tempfile
|
import tempfile
|
||||||
import time
|
import time
|
||||||
from collections import OrderedDict
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from PIL import ImageFont
|
from PIL import ImageFont
|
||||||
from typing import Dict, Tuple, Optional, Union, Any, List
|
from typing import Dict, Tuple, Optional, Union, Any, List
|
||||||
@@ -59,13 +58,7 @@ class FontManager:
|
|||||||
# Font discovery and catalog
|
# Font discovery and catalog
|
||||||
self.font_catalog: Dict[str, str] = {} # family_name -> file_path
|
self.font_catalog: Dict[str, str] = {} # family_name -> file_path
|
||||||
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
|
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
|
||||||
# (text, id(font)) -> ((width, height, baseline), font_ref).
|
self.metrics_cache: Dict[str, Tuple[int, int, int]] = {} # (text, font_id) -> (width, height, baseline)
|
||||||
# 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
|
# Plugin font management
|
||||||
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
|
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
|
||||||
@@ -514,14 +507,10 @@ class FontManager:
|
|||||||
Returns:
|
Returns:
|
||||||
Tuple of (width, height, baseline_offset)
|
Tuple of (width, height, baseline_offset)
|
||||||
"""
|
"""
|
||||||
# Key on the text itself (hash(text) could collide) + font identity;
|
cache_key = f"{hash(text)}_{id(font)}"
|
||||||
# the entry below keeps the font referenced so the id stays valid.
|
|
||||||
cache_key = (text, id(font))
|
|
||||||
|
|
||||||
cached = self.metrics_cache.get(cache_key)
|
if cache_key in self.metrics_cache:
|
||||||
if cached is not None:
|
return self.metrics_cache[cache_key]
|
||||||
self.metrics_cache.move_to_end(cache_key)
|
|
||||||
return cached[0]
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if isinstance(font, freetype.Face):
|
if isinstance(font, freetype.Face):
|
||||||
@@ -558,9 +547,7 @@ class FontManager:
|
|||||||
baseline = 10
|
baseline = 10
|
||||||
|
|
||||||
result = (width, height, baseline)
|
result = (width, height, baseline)
|
||||||
self.metrics_cache[cache_key] = (result, font)
|
self.metrics_cache[cache_key] = result
|
||||||
while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
|
|
||||||
self.metrics_cache.popitem(last=False)
|
|
||||||
return result
|
return result
|
||||||
|
|
||||||
def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
|
def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ import json
|
|||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
import threading
|
import threading
|
||||||
import types
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, List, Optional, Any
|
from typing import Dict, List, Optional, Any
|
||||||
import logging
|
import logging
|
||||||
@@ -744,18 +743,8 @@ class PluginManager:
|
|||||||
# If resource monitor exists, wrap the call
|
# If resource monitor exists, wrap the call
|
||||||
def monitored_update():
|
def monitored_update():
|
||||||
self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
|
self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
|
||||||
# SimpleNamespace stores `update` as an *instance*
|
|
||||||
# attribute, so attribute lookup returns the plain
|
|
||||||
# function object as-is. A dynamically-built class
|
|
||||||
# (`type(..., {'update': monitored_update})`) instead
|
|
||||||
# stores it as a *class* attribute, which the
|
|
||||||
# descriptor protocol turns into a bound method on
|
|
||||||
# access -- silently prepending the instance as an
|
|
||||||
# implicit first argument to a function that takes
|
|
||||||
# none, raising "monitored_update() takes 0
|
|
||||||
# positional arguments but 1 was given" on every call.
|
|
||||||
success = self.plugin_executor.execute_update(
|
success = self.plugin_executor.execute_update(
|
||||||
types.SimpleNamespace(update=monitored_update),
|
type('obj', (object,), {'update': monitored_update})(),
|
||||||
plugin_id
|
plugin_id
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -454,18 +454,6 @@ class VisualTestDisplayManager:
|
|||||||
"""Check if display is currently scrolling."""
|
"""Check if display is currently scrolling."""
|
||||||
return self._scrolling_state['is_scrolling']
|
return self._scrolling_state['is_scrolling']
|
||||||
|
|
||||||
def process_deferred_updates(self):
|
|
||||||
"""Process any deferred updates (no-op for testing).
|
|
||||||
|
|
||||||
Several ticker-style plugins (news, odds-ticker, leaderboard,
|
|
||||||
stock-news, stocks) call this unconditionally between
|
|
||||||
set_scrolling_state() and their scroll-position update, mirroring the
|
|
||||||
real display_manager's deferred-update queue. This double has no such
|
|
||||||
queue, so there is nothing to process — the no-op just lets those
|
|
||||||
plugins render under the harness instead of raising AttributeError.
|
|
||||||
"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
# Utility methods
|
# Utility methods
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|||||||
+1
-5
@@ -38,11 +38,7 @@ def mock_cache_manager():
|
|||||||
mock._memory_cache_timestamps = {}
|
mock._memory_cache_timestamps = {}
|
||||||
mock.cache_dir = "/tmp/test_cache"
|
mock.cache_dir = "/tmp/test_cache"
|
||||||
|
|
||||||
def mock_get(key: str, max_age: Optional[int] = 300,
|
def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
|
||||||
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)
|
return mock._memory_cache.get(key)
|
||||||
|
|
||||||
def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
|
def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
|
||||||
|
|||||||
@@ -172,16 +172,6 @@ class TestVisualDisplayManager:
|
|||||||
vdm.set_scrolling_state(False)
|
vdm.set_scrolling_state(False)
|
||||||
assert vdm.is_currently_scrolling() is False
|
assert vdm.is_currently_scrolling() is False
|
||||||
|
|
||||||
def test_process_deferred_updates_is_noop(self):
|
|
||||||
# Ticker-style plugins (news, odds-ticker, leaderboard, stock-news,
|
|
||||||
# stocks) call this unconditionally alongside set_scrolling_state();
|
|
||||||
# it must exist and be harmless so those plugins render under the
|
|
||||||
# harness instead of raising AttributeError.
|
|
||||||
vdm = VisualTestDisplayManager(width=128, height=32)
|
|
||||||
vdm.set_scrolling_state(True)
|
|
||||||
vdm.process_deferred_updates() # should not raise
|
|
||||||
assert vdm.is_currently_scrolling() is True
|
|
||||||
|
|
||||||
def test_format_date_with_ordinal(self):
|
def test_format_date_with_ordinal(self):
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
vdm = VisualTestDisplayManager(width=128, height=32)
|
vdm = VisualTestDisplayManager(width=128, height=32)
|
||||||
|
|||||||
@@ -3,7 +3,6 @@ from unittest.mock import MagicMock, patch
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from src.plugin_system.plugin_manager import PluginManager
|
from src.plugin_system.plugin_manager import PluginManager
|
||||||
from src.plugin_system.plugin_state import PluginState
|
from src.plugin_system.plugin_state import PluginState
|
||||||
from src.plugin_system.resource_monitor import PluginResourceMonitor
|
|
||||||
|
|
||||||
class TestPluginManager:
|
class TestPluginManager:
|
||||||
"""Test PluginManager functionality."""
|
"""Test PluginManager functionality."""
|
||||||
@@ -75,58 +74,10 @@ class TestPluginManager:
|
|||||||
|
|
||||||
# No manifest in pm.plugin_manifests
|
# No manifest in pm.plugin_manifests
|
||||||
result = pm.load_plugin("non_existent_plugin")
|
result = pm.load_plugin("non_existent_plugin")
|
||||||
|
|
||||||
assert result is False
|
assert result is False
|
||||||
assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR
|
assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR
|
||||||
|
|
||||||
def test_run_scheduled_updates_calls_update_with_resource_monitor(
|
|
||||||
self, mock_config_manager, mock_display_manager, mock_cache_manager
|
|
||||||
):
|
|
||||||
"""Regression test: run_scheduled_updates() must actually call a
|
|
||||||
plugin's update() when self.resource_monitor is set (as it is in
|
|
||||||
every real deployment -- display_controller.py and web_interface/
|
|
||||||
app.py both assign a real PluginResourceMonitor after construction).
|
|
||||||
|
|
||||||
Previously, the resource_monitor branch wrapped the call in a
|
|
||||||
function stored as a *class* attribute on a dynamically-built type
|
|
||||||
(`type('obj', (object,), {'update': monitored_update})()`), which
|
|
||||||
the descriptor protocol turns into a bound method on access --
|
|
||||||
silently passing the synthetic instance as an implicit first
|
|
||||||
argument to monitored_update(), which takes none. Every plugin's
|
|
||||||
scheduled update failed with "monitored_update() takes 0 positional
|
|
||||||
arguments but 1 was given" and was silently swallowed into a
|
|
||||||
circuit-breaker retry loop that never succeeded, so plugin data
|
|
||||||
(scores, odds, etc.) never refreshed.
|
|
||||||
"""
|
|
||||||
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
|
|
||||||
pm = PluginManager(
|
|
||||||
plugins_dir="plugins",
|
|
||||||
config_manager=mock_config_manager,
|
|
||||||
display_manager=mock_display_manager,
|
|
||||||
cache_manager=mock_cache_manager
|
|
||||||
)
|
|
||||||
|
|
||||||
plugin_instance = MagicMock()
|
|
||||||
plugin_instance.enabled = True
|
|
||||||
plugin_instance.update = MagicMock()
|
|
||||||
|
|
||||||
pm.plugins["test_plugin"] = plugin_instance
|
|
||||||
pm.plugin_manifests["test_plugin"] = {"update_interval": 10}
|
|
||||||
pm.state_manager.set_state("test_plugin", PluginState.ENABLED)
|
|
||||||
# Plain MagicMock, not the mock_cache_manager fixture: this test
|
|
||||||
# is about run_scheduled_updates() actually invoking update()
|
|
||||||
# through the resource-monitor wrapper, not about
|
|
||||||
# PluginResourceMonitor's own cache-backed metrics persistence
|
|
||||||
# (which calls cache_manager.get(..., memory_ttl=...) --
|
|
||||||
# a kwarg the fixture's mock_get() doesn't accept).
|
|
||||||
pm.resource_monitor = PluginResourceMonitor(MagicMock())
|
|
||||||
|
|
||||||
pm.run_scheduled_updates(current_time=time.time())
|
|
||||||
|
|
||||||
plugin_instance.update.assert_called_once()
|
|
||||||
assert "test_plugin" in pm.plugin_last_update
|
|
||||||
assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
|
|
||||||
|
|
||||||
|
|
||||||
class TestPluginLoader:
|
class TestPluginLoader:
|
||||||
"""Test PluginLoader functionality."""
|
"""Test PluginLoader functionality."""
|
||||||
|
|||||||
Reference in New Issue
Block a user