Merge remote-tracking branch 'origin/main' into claude/hdpi-scroll-performance-antialiasing-4ae609

web_interface/app.py

@@ -16,6 +16,17 @@ from datetime import datetime, timedelta
 # Add parent directory to path for imports
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+# Configure logging before anything below logs: the same setup as the display
+# service (run.py), so this process's journal lines carry their real syslog
+# priority too (`journalctl -p err -u ledmatrix-web`). LEDMATRIX_DEBUG=true
+# turns on DEBUG, which includes the routine per-request lines.
+from src.logging_config import setup_logging
+setup_logging(format_type=(
+    'json' if os.environ.get('LEDMATRIX_JSON_LOGGING', 'false').lower() == 'true'
+    else 'readable'))
+logging.getLogger('werkzeug').setLevel(logging.WARNING)  # request_logging covers requests
+logging.getLogger('urllib3').setLevel(logging.WARNING)
+
 from src.config_manager import ConfigManager
 from src.web_interface.error_handler import describe_exception
 from src.common.path_safety import (
@@ -284,34 +295,47 @@ try:
 except ImportError:
     pass
 
-# Cached AP mode check — avoids creating a WiFiManager per request
-_ap_mode_cache = {'value': False, 'timestamp': 0}
+# systemctl answers, memoised so they are not a subprocess fork per request
+# (AP mode) or per SSE tick (display service). A failed check keeps the last
+# known answer for the same TTL rather than retrying on every request.
+from web_interface.cache import TTLCache
+_service_status_cache = TTLCache()
 _AP_MODE_CACHE_TTL = 30  # seconds — AP mode is user-initiated; 30s is fine
-
-# Cached ledmatrix service status for SSE stats stream
-_ledmatrix_service_cache = {'active': False, 'timestamp': 0}
 _LEDMATRIX_SERVICE_CACHE_TTL = 15  # seconds
 
+# The only units _unit_is_active() may ask systemctl about: its argv is built
+# from these literals, never from request data.
+_CHECKABLE_UNITS = frozenset({'hostapd', 'ledmatrix'})
+
+def _unit_is_active(unit, ttl):
+    """`systemctl is-active <unit>`, cached for ``ttl`` seconds.
+
+    False where there is no systemctl (a dev machine); on a failed check, the
+    last known answer.
+    """
+    if unit not in _CHECKABLE_UNITS:
+        raise ValueError(f"not a checkable unit: {unit!r}")
+    active = _service_status_cache.get(unit)
+    if active is not None:
+        return active
+    active = _service_status_cache.peek(unit, False)
+    if _SYSTEMCTL:
+        try:
+            result = subprocess.run([_SYSTEMCTL, 'is-active', unit],  # nosec B603 - list argv, unit is from _CHECKABLE_UNITS  # nosemgrep
+                                    capture_output=True, text=True, timeout=2)
+            active = result.stdout.strip() == 'active'
+        except (subprocess.SubprocessError, OSError) as e:
+            logging.getLogger('web_interface').warning(
+                "systemctl is-active %s failed: %s", unit, e)
+    _service_status_cache.set(unit, active, ttl=ttl)
+    return active
+
 def is_ap_mode_active():
     """
     Check if access point mode is currently active (cached, 30s TTL).
     Uses a direct systemctl check instead of instantiating WiFiManager.
     """
-    now = time.time()
-    if (now - _ap_mode_cache['timestamp']) < _AP_MODE_CACHE_TTL:
-        return _ap_mode_cache['value']
-    try:
-        result = subprocess.run(
-            ['systemctl', 'is-active', 'hostapd'],
-            capture_output=True, text=True, timeout=2
-        )
-        active = result.stdout.strip() == 'active'
-        _ap_mode_cache['value'] = active
-        _ap_mode_cache['timestamp'] = now
-        return active
-    except (subprocess.SubprocessError, OSError) as e:
-        logging.getLogger('web_interface').error(f"AP mode check failed: {e}")
-        return _ap_mode_cache['value']
+    return _unit_is_active('hostapd', _AP_MODE_CACHE_TTL)
 
 # Captive portal detection endpoints
 # When AP mode is active, return responses that TRIGGER the captive portal popup.
@@ -346,41 +370,9 @@ def success_txt():
         return redirect(url_for('pages_v3.captive_setup'), code=302)
     return 'success', 200
 
-# Initialize logging
-try:
-    from web_interface.logging_config import setup_web_interface_logging, log_api_request
-    # Use JSON logging in production, readable logs in development
-    use_json_logging = os.environ.get('LEDMATRIX_JSON_LOGGING', 'false').lower() == 'true'
-    setup_web_interface_logging(level='INFO', use_json=use_json_logging)
-except ImportError:
-    # Logging config not available, use default
-    log_api_request = None
-
-# Request timing and logging middleware
-@app.before_request
-def before_request():
-    """Track request start time for logging."""
-    from flask import request
-    request.start_time = time.time()
-
-@app.after_request
-def after_request_logging(response):
-    """Log API requests after response."""
-    if log_api_request:
-        try:
-            from flask import request
-            duration_ms = (time.time() - getattr(request, 'start_time', time.time())) * 1000
-            ip_address = request.remote_addr if hasattr(request, 'remote_addr') else None
-            log_api_request(
-                method=request.method,
-                path=request.path,
-                status_code=response.status_code,
-                duration_ms=duration_ms,
-                ip_address=ip_address
-            )
-        except Exception:  # nosec B110 - request logging must never interrupt a live HTTP response
-            pass  # Don't break response if logging fails
-    return response
+# Request timing and logging (routine reads at DEBUG; see request_logging)
+from web_interface import request_logging
+request_logging.init_app(app)
 
 # Global error handlers
 @app.errorhandler(404)
@@ -693,17 +685,7 @@ def system_status_generator():
             cpu_temp = metrics['cpu_temp']
 
             # Check if display service is running (cached to avoid per-client subprocess forks)
-            now = time.time()
-            if (now - _ledmatrix_service_cache['timestamp']) >= _LEDMATRIX_SERVICE_CACHE_TTL:
-                if _SYSTEMCTL:
-                    try:
-                        result = subprocess.run([_SYSTEMCTL, 'is-active', 'ledmatrix'],
-                                                capture_output=True, text=True, timeout=2)
-                        _ledmatrix_service_cache['active'] = result.stdout.strip() == 'active'
-                    except (subprocess.SubprocessError, OSError) as e:
-                        app.logger.warning("systemctl status check failed: %s", e)
-                _ledmatrix_service_cache['timestamp'] = now
-            service_active = _ledmatrix_service_cache['active']
+            service_active = _unit_is_active('ledmatrix', _LEDMATRIX_SERVICE_CACHE_TTL)
             
             status = {
                 'timestamp': time.time(),

web_interface/auto_update.py

@@ -172,9 +172,10 @@ def _plugin_fingerprint(store_manager, plugin_dir):
     try:
         with open(plugin_dir / 'manifest.json', 'r', encoding='utf-8') as f:
             manifest = json.load(f)
-        if manifest.get('local_only'):
-            return None
-        version = manifest.get('version')
+        if isinstance(manifest, dict):  # valid JSON need not be an object
+            if manifest.get('local_only'):
+                return None
+            version = manifest.get('version')
     except (OSError, ValueError):
         pass
     sha = None
@@ -190,8 +191,12 @@ def update_plugins(store_manager, operation_history=None):
     """Update every installed plugin that has an update. Returns (updated, failed)."""
     updated, failed = [], []
     plugins_dir = Path(store_manager.plugins_dir)
+    # list_installed_plugins() reports manifest ids, and a plugin's directory
+    # may be named differently (ledmatrix-stocks/ holding id "stocks"), so
+    # the directory comes from the store's own lookup, not a join.
+    find_dir = getattr(store_manager, '_find_plugin_path', None)
     for plugin_id in sorted(store_manager.list_installed_plugins()):
-        plugin_dir = plugins_dir / plugin_id
+        plugin_dir = (find_dir(plugin_id) if find_dir else None) or plugins_dir / plugin_id
         before = _plugin_fingerprint(store_manager, plugin_dir)
         if before is None:
             continue  # local_only: managed by hand, never from the registry

web_interface/blueprints/api_v3/__init__.py

@@ -35,13 +35,16 @@ from typing import Dict, Any, Optional, Tuple, Type
 from urllib.parse import urlparse, urlunparse
 logger = logging.getLogger(__name__)
 # Import new infrastructure
-from src.web_interface.api_helpers import success_response, error_response, validate_request_json
+from src.web_interface.api_helpers import (success_response, error_response,
+                                           exception_error_response, validate_request_json)
 from src.web_interface.errors import ErrorCode
 from src.web_interface.secret_helpers import (find_secret_fields, mask_all_secret_values,
                                               merge_secrets, remove_empty_secrets,
                                               separate_secrets,
                                               strip_masked_values)
-from src.web_interface.error_handler import describe_exception, redact_text
+from src.web_interface.error_handler import (describe_exception, http_exception_payload,
+                                             redact_text, unhandled_exception_payload)
+from werkzeug.exceptions import HTTPException
 from src.plugin_system.operation_types import OperationType
 from src.web_interface.validators import (
     validate_file_upload
@@ -114,6 +117,43 @@ SYSTEM_FONTS = frozenset([
     'clr6x12', 'helvr12', 'texgyre-27'
 ])
 api_v3 = Blueprint('api_v3', __name__)
+
+
+@api_v3.errorhandler(Exception)
+def _api_v3_unhandled_exception(error):
+    """The answer for any exception an api_v3 route does not handle itself.
+
+    Fifty-odd routes used to end in the same four lines -- log the traceback,
+    return {status, message: "An error occurred; see logs for details",
+    details: describe_exception(e)} with a 500. This is those four lines, once.
+    A route still catches for itself when its failure needs something else: a
+    specific message, extra keys, an operation-history record, or cleanup.
+
+    It is registered on the blueprint, not left to web_interface/app.py's
+    global handler, because the two answers differ: the global one adds
+    `error_code: UNKNOWN_ERROR`, and the plugin API client treats a body with
+    an error_code differently from one without (see api_client.js). Tests that
+    mount this blueprint on a bare Flask app get the same answer as the real
+    app does, too.
+
+    `details` is describe_exception(), which redacts credentials and caps the
+    length. CodeQL reads returning it as stack-trace exposure; it is the
+    project's deliberate trade-off, because a device whose storage is failing
+    otherwise answers "see logs for details" from the log viewer too
+    (test_web_error_detail.py).
+
+    Werkzeug's HTTPExceptions subclass Exception, so a 400/405/413/415 raised
+    inside a route lands here as well; it goes back as itself, in the global
+    handler's shape. A 404 or explicit 500 never arrives: Flask prefers the
+    app's code-specific handlers over a blueprint's class-based one.
+    """
+    if isinstance(error, HTTPException):
+        return jsonify(http_exception_payload(error)), error.code or 500
+    logger.error("Unhandled exception in %s", request.endpoint or request.path,
+                 exc_info=error)
+    return jsonify(unhandled_exception_payload(error)), 500
+
+
 def _get_plugin_version(plugin_id: str) -> str:
     """Read the installed version from a plugin's manifest.json.
 
@@ -1712,7 +1752,7 @@ def _get_starlark_device_location() -> DeviceLocationResolver:
     global _starlark_device_location
     if _starlark_device_location is None:
         _starlark_device_location = DeviceLocationResolver(
-            getattr(api_v3, 'cache_manager', None) or _ensure_cache_manager(), logger)
+            getattr(api_v3, 'cache_manager', None), logger)
     return _starlark_device_location
 
 

web_interface/blueprints/api_v3/config.py

@@ -47,15 +47,11 @@ def _day_setting(data, day, flat_key, nested_key):
 @api_v3.route('/config/main', methods=['GET'])
 def get_main_config():
     """Get main configuration, with credentials redacted."""
-    try:
-        if not api_v3.config_manager:
-            return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
+    if not api_v3.config_manager:
+        return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        config = api_v3.config_manager.load_config()
-        return jsonify({'status': 'success', 'data': _redact_credentials(config)})
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    config = api_v3.config_manager.load_config()
+    return jsonify({'status': 'success', 'data': _redact_credentials(config)})
 @api_v3.route('/config/schedule', methods=['GET'])
 def get_schedule_config():
     """Get current schedule configuration"""
@@ -1165,20 +1161,16 @@ def save_main_config():
 @api_v3.route('/config/secrets', methods=['GET'])
 def get_secrets_config():
     """Get secrets configuration"""
-    try:
-        if not api_v3.config_manager:
-            return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
+    if not api_v3.config_manager:
+        return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        config = api_v3.config_manager.get_raw_file_content('secrets')
-        # This interface has no authentication, and this file is nothing but
-        # credentials. It was handing all of them to anyone who could reach
-        # the port. Values are masked; empty and YOUR_* placeholders are left
-        # alone so a client can still tell "set" from "not set".
-        return jsonify({'status': 'success',
-                        'data': mask_all_secret_values(config)})
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    config = api_v3.config_manager.get_raw_file_content('secrets')
+    # This interface has no authentication, and this file is nothing but
+    # credentials. It was handing all of them to anyone who could reach
+    # the port. Values are masked; empty and YOUR_* placeholders are left
+    # alone so a client can still tell "set" from "not set".
+    return jsonify({'status': 'success',
+                    'data': mask_all_secret_values(config)})
 @api_v3.route('/config/raw/main', methods=['POST'])
 def save_raw_main_config():
     """Save raw main configuration JSON"""

web_interface/blueprints/api_v3/display.py

@@ -6,7 +6,7 @@ endpoint names are unchanged by living here.
 from web_interface.blueprints.api_v3 import (
     _ensure_display_service_running,
     _get_display_service_status, _stop_display_service, api_v3,
-    describe_exception, jsonify, logger, os, request, uuid,
+    jsonify, logger, os, request, uuid,
 )
 import web_interface.blueprints.api_v3 as _pkg
 # Read through the module rather than bound by value: tests patch these
@@ -31,45 +31,41 @@ def _cache_manager():
 @api_v3.route('/display/current', methods=['GET'])
 def get_display_current():
     """Get current display state"""
+    import base64
+    from PIL import Image
+    import io
+
+    snapshot_path = "/tmp/led_matrix_preview.png"
+
+    # Get display dimensions from config: the logical size DisplayManager
+    # renders at, so double-sided setups preview one screen
+    from src.display_geometry import logical_size
     try:
-        import base64
-        from PIL import Image
-        import io
+        config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
+        width, height = logical_size(config)
+    except Exception:
+        width, height = logical_size({})
 
-        snapshot_path = "/tmp/led_matrix_preview.png"
-
-        # Get display dimensions from config: the logical size DisplayManager
-        # renders at, so double-sided setups preview one screen
-        from src.display_geometry import logical_size
+    # Try to read snapshot file
+    image_data = None
+    if os.path.exists(snapshot_path):
         try:
-            config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
-            width, height = logical_size(config)
-        except Exception:
-            width, height = logical_size({})
+            with Image.open(snapshot_path) as img:
+                # Convert to PNG and encode as base64
+                buffer = io.BytesIO()
+                img.save(buffer, format='PNG')
+                image_data = base64.b64encode(buffer.getvalue()).decode('utf-8')
+        except Exception as img_err:
+            # File might be being written or corrupted, return None
+            pass
 
-        # Try to read snapshot file
-        image_data = None
-        if os.path.exists(snapshot_path):
-            try:
-                with Image.open(snapshot_path) as img:
-                    # Convert to PNG and encode as base64
-                    buffer = io.BytesIO()
-                    img.save(buffer, format='PNG')
-                    image_data = base64.b64encode(buffer.getvalue()).decode('utf-8')
-            except Exception as img_err:
-                # File might be being written or corrupted, return None
-                pass
-
-        display_data = {
-            'timestamp': _pkg.time.time(),
-            'width': width,
-            'height': height,
-            'image': image_data  # Base64 encoded image data or None if unavailable
-        }
-        return jsonify({'status': 'success', 'data': display_data})
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    display_data = {
+        'timestamp': _pkg.time.time(),
+        'width': width,
+        'height': height,
+        'image': image_data  # Base64 encoded image data or None if unavailable
+    }
+    return jsonify({'status': 'success', 'data': display_data})
 @api_v3.route('/display/modes', methods=['GET'])
 def get_display_modes():
     """Every display mode that can be requested on-demand, with its plugin.
@@ -92,230 +88,208 @@ def get_display_modes():
             for the duration -- so they are reported with enabled: false
             rather than omitted.
     """
-    try:
-        if not api_v3.plugin_manager:
-            return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
+    if not api_v3.plugin_manager:
+        return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
 
-        # Discovery is lazy and normally triggered by whichever endpoint runs
-        # first, which is a person opening the dashboard. A caller that never
-        # visits it would otherwise see an empty list.
-        api_v3.plugin_manager.discover_plugins()
+    # Discovery is lazy and normally triggered by whichever endpoint runs
+    # first, which is a person opening the dashboard. A caller that never
+    # visits it would otherwise see an empty list.
+    api_v3.plugin_manager.discover_plugins()
 
-        include_disabled = request.args.get('include_disabled') in ('1', 'true', 'True')
-        full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
+    include_disabled = request.args.get('include_disabled') in ('1', 'true', 'True')
+    full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
 
-        modes = []
-        for plugin_id, manifest in sorted(api_v3.plugin_manager.plugin_manifests.items()):
-            # A hand-edited or migrated config.json can hold a non-dict under a
-            # plugin id; DisplayController._reconcile guards the same shape, so
-            # it happens in practice. Without this, .get() raises AttributeError,
-            # the loop aborts and the endpoint answers 500 with no modes at all
-            # -- one bad section would blank every entity the MQTT bridge builds
-            # from this list.
-            plugin_config = full_config.get(plugin_id)
-            if not isinstance(plugin_config, dict):
-                if plugin_config is not None:
-                    logger.warning(
-                        "Config for plugin %r is %s, not an object; treating it as disabled",
-                        plugin_id, type(plugin_config).__name__)
-                plugin_config = {}
-            enabled = bool(plugin_config.get('enabled', False))
-            if not enabled and not include_disabled:
-                continue
-            plugin_name = (manifest or {}).get('name') or plugin_id
-            plugin_modes = api_v3.plugin_manager.get_plugin_display_modes(plugin_id) or [plugin_id]
-            for mode in plugin_modes:
-                # A single-mode plugin's mode is the plugin, so its own name is
-                # the readable label. Multi-mode plugins have no per-mode name
-                # anywhere, so the raw mode string is the only thing to show.
-                modes.append({
-                    'mode': mode,
-                    'plugin_id': plugin_id,
-                    'plugin_name': plugin_name,
-                    'name': plugin_name if len(plugin_modes) == 1 else mode,
-                    'enabled': enabled,
-                })
+    modes = []
+    for plugin_id, manifest in sorted(api_v3.plugin_manager.plugin_manifests.items()):
+        # A hand-edited or migrated config.json can hold a non-dict under a
+        # plugin id; DisplayController._reconcile guards the same shape, so
+        # it happens in practice. Without this, .get() raises AttributeError,
+        # the loop aborts and the endpoint answers 500 with no modes at all
+        # -- one bad section would blank every entity the MQTT bridge builds
+        # from this list.
+        plugin_config = full_config.get(plugin_id)
+        if not isinstance(plugin_config, dict):
+            if plugin_config is not None:
+                logger.warning(
+                    "Config for plugin %r is %s, not an object; treating it as disabled",
+                    plugin_id, type(plugin_config).__name__)
+            plugin_config = {}
+        enabled = bool(plugin_config.get('enabled', False))
+        if not enabled and not include_disabled:
+            continue
+        plugin_name = (manifest or {}).get('name') or plugin_id
+        plugin_modes = api_v3.plugin_manager.get_plugin_display_modes(plugin_id) or [plugin_id]
+        for mode in plugin_modes:
+            # A single-mode plugin's mode is the plugin, so its own name is
+            # the readable label. Multi-mode plugins have no per-mode name
+            # anywhere, so the raw mode string is the only thing to show.
+            modes.append({
+                'mode': mode,
+                'plugin_id': plugin_id,
+                'plugin_name': plugin_name,
+                'name': plugin_name if len(plugin_modes) == 1 else mode,
+                'enabled': enabled,
+            })
 
-        return jsonify({'status': 'success', 'data': {'modes': modes}})
-    except Exception as exc:
-        # describe_exception, not a bare message: test_web_error_detail.py
-        # enforces that every handler here returns it, because a device whose
-        # storage is failing otherwise answers "see logs for details" from the
-        # log viewer too. It redacts credentials out of the exception text.
-        # CodeQL flags this as stack-trace exposure across all ~75 handlers;
-        # it is the project's deliberate, reviewed trade-off.
-        logger.error('Error in get_display_modes', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
+    return jsonify({'status': 'success', 'data': {'modes': modes}})
 @api_v3.route('/display/on-demand/status', methods=['GET'])
 def get_on_demand_status():
     """Return the current on-demand display state."""
-    try:
-        cache = _cache_manager()
-        # memory_ttl=0: the display service writes this key, so only the file
-        # is current. This process's memory tier would keep serving the first
-        # copy it read for the full max_age -- "active" for two minutes after
-        # the display had already stopped.
-        state = cache.get('display_on_demand_state', max_age=120, memory_ttl=0)
-        if state is None:
-            state = {
-                'active': False,
-                'status': 'idle',
-                'last_updated': None
-            }
-        service_status = _get_display_service_status()
-        return jsonify({
-            'status': 'success',
-            'data': {
-                'state': state,
-                'service': service_status
-            }
-        })
-    except Exception as exc:
-        logger.error('Error in get_on_demand_status', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
+    cache = _cache_manager()
+    # memory_ttl=0: the display service writes this key, so only the file
+    # is current. This process's memory tier would keep serving the first
+    # copy it read for the full max_age -- "active" for two minutes after
+    # the display had already stopped.
+    state = cache.get('display_on_demand_state', max_age=120, memory_ttl=0)
+    if state is None:
+        state = {
+            'active': False,
+            'status': 'idle',
+            'last_updated': None
+        }
+    service_status = _get_display_service_status()
+    return jsonify({
+        'status': 'success',
+        'data': {
+            'state': state,
+            'service': service_status
+        }
+    })
 @api_v3.route('/display/on-demand/start', methods=['POST'])
 def start_on_demand_display():
     """Request the display controller to run a specific plugin on-demand."""
-    try:
-        data = request.get_json(silent=True) or {}
-        plugin_id = data.get('plugin_id')
-        mode = data.get('mode')
-        duration = data.get('duration')
-        pinned = bool(data.get('pinned', False))
-        start_service = data.get('start_service', True)
+    data = request.get_json(silent=True) or {}
+    plugin_id = data.get('plugin_id')
+    mode = data.get('mode')
+    duration = data.get('duration')
+    pinned = bool(data.get('pinned', False))
+    start_service = data.get('start_service', True)
 
-        if not plugin_id and not mode:
-            return jsonify({'status': 'error', 'message': 'plugin_id or mode is required'}), 400
+    if not plugin_id and not mode:
+        return jsonify({'status': 'error', 'message': 'plugin_id or mode is required'}), 400
 
-        resolved_plugin = plugin_id
-        resolved_mode = mode
+    resolved_plugin = plugin_id
+    resolved_mode = mode
 
-        if api_v3.plugin_manager:
-            if resolved_plugin and resolved_plugin not in _pkg._discovered_plugin_manifests(resolved_plugin):
-                return jsonify({'status': 'error', 'message': f'Plugin {resolved_plugin} not found'}), 404
+    if api_v3.plugin_manager:
+        if resolved_plugin and resolved_plugin not in _pkg._discovered_plugin_manifests(resolved_plugin):
+            return jsonify({'status': 'error', 'message': f'Plugin {resolved_plugin} not found'}), 404
 
-            if resolved_plugin and not resolved_mode:
-                modes = api_v3.plugin_manager.get_plugin_display_modes(resolved_plugin)
-                resolved_mode = modes[0] if modes else resolved_plugin
-            elif resolved_mode and not resolved_plugin:
-                _pkg._discovered_plugin_manifests()
+        if resolved_plugin and not resolved_mode:
+            modes = api_v3.plugin_manager.get_plugin_display_modes(resolved_plugin)
+            resolved_mode = modes[0] if modes else resolved_plugin
+        elif resolved_mode and not resolved_plugin:
+            _pkg._discovered_plugin_manifests()
+            resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
+            if not resolved_plugin:
+                # Not among what was discovered: the plugin that declares
+                # it may have been installed since. Scan once more.
+                _pkg._discovered_plugin_manifests(rescan=True)
                 resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
-                if not resolved_plugin:
-                    # Not among what was discovered: the plugin that declares
-                    # it may have been installed since. Scan once more.
-                    _pkg._discovered_plugin_manifests(rescan=True)
-                    resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
-                if not resolved_plugin:
-                    return jsonify({'status': 'error', 'message': f'Mode {resolved_mode} not found'}), 404
+            if not resolved_plugin:
+                return jsonify({'status': 'error', 'message': f'Mode {resolved_mode} not found'}), 404
 
-        # Note: On-demand can work with disabled plugins - the display controller
-        # will temporarily enable them during initialization if needed
-        # We don't block the request here, but log it for debugging
-        if api_v3.config_manager and resolved_plugin:
-            config = api_v3.config_manager.load_config()
-            plugin_config = config.get(resolved_plugin, {})
-            if 'enabled' in plugin_config and not plugin_config.get('enabled', False):
-                logger.info(
-                    "On-demand request for disabled plugin '%s' - will be temporarily enabled",
-                    resolved_plugin,
-                )
+    # Note: On-demand can work with disabled plugins - the display controller
+    # will temporarily enable them during initialization if needed
+    # We don't block the request here, but log it for debugging
+    if api_v3.config_manager and resolved_plugin:
+        config = api_v3.config_manager.load_config()
+        plugin_config = config.get(resolved_plugin, {})
+        if 'enabled' in plugin_config and not plugin_config.get('enabled', False):
+            logger.info(
+                "On-demand request for disabled plugin '%s' - will be temporarily enabled",
+                resolved_plugin,
+            )
 
-        # Set the on-demand request in cache FIRST (before starting service)
-        # This ensures the request is available when the service starts/restarts
-        cache = _cache_manager()
-        request_id = data.get('request_id') or str(uuid.uuid4())
-        request_payload = {
-            'request_id': request_id,
-            'action': 'start',
-            'plugin_id': resolved_plugin,
-            'mode': resolved_mode,
-            'duration': duration,
-            'pinned': pinned,
-            'timestamp': _pkg.time.time()
-        }
-        cache.set('display_on_demand_request', request_payload)
+    # Set the on-demand request in cache FIRST (before starting service)
+    # This ensures the request is available when the service starts/restarts
+    cache = _cache_manager()
+    request_id = data.get('request_id') or str(uuid.uuid4())
+    request_payload = {
+        'request_id': request_id,
+        'action': 'start',
+        'plugin_id': resolved_plugin,
+        'mode': resolved_mode,
+        'duration': duration,
+        'pinned': pinned,
+        'timestamp': _pkg.time.time()
+    }
+    cache.set('display_on_demand_request', request_payload)
 
-        # Check if display service is running (or will be started)
-        service_status = _get_display_service_status()
-        service_was_running = service_status.get('active', False)
+    # Check if display service is running (or will be started)
+    service_status = _get_display_service_status()
+    service_was_running = service_status.get('active', False)
         
-        # Stop the display service first to ensure clean state when we will restart it
-        if service_was_running and start_service:
-            import time as time_module
-            logger.debug("Stopping display service before starting on-demand mode")
-            _stop_display_service()
-            # Wait a brief moment for the service to fully stop
-            time_module.sleep(1.5)
-            logger.debug("Display service stopped, now starting with on-demand request")
+    # Stop the display service first to ensure clean state when we will restart it
+    if service_was_running and start_service:
+        import time as time_module
+        logger.debug("Stopping display service before starting on-demand mode")
+        _stop_display_service()
+        # Wait a brief moment for the service to fully stop
+        time_module.sleep(1.5)
+        logger.debug("Display service stopped, now starting with on-demand request")
 
-        if not service_status.get('active') and not start_service:
+    if not service_status.get('active') and not start_service:
+        return jsonify({
+            'status': 'error',
+            'message': 'Display service is not running. Please start the display service or enable "Start Service" option.',
+            'service_status': service_status
+        }), 400
+
+    service_result = None
+    if start_service:
+        service_result = _ensure_display_service_running()
+        # Check if service actually started
+        if service_result and not service_result.get('active'):
             return jsonify({
                 'status': 'error',
-                'message': 'Display service is not running. Please start the display service or enable "Start Service" option.',
-                'service_status': service_status
-            }), 400
-
-        service_result = None
-        if start_service:
-            service_result = _ensure_display_service_running()
-            # Check if service actually started
-            if service_result and not service_result.get('active'):
-                return jsonify({
-                    'status': 'error',
-                    'message': 'Failed to start display service. Please check service logs or start it manually.',
-                    'service_result': service_result
-                }), 500
+                'message': 'Failed to start display service. Please check service logs or start it manually.',
+                'service_result': service_result
+            }), 500
             
-            # Service was restarted (or started fresh) with on-demand request in cache
-            # The display controller will read the request during initialization or when it polls
+        # Service was restarted (or started fresh) with on-demand request in cache
+        # The display controller will read the request during initialization or when it polls
 
-        response_data = {
-            'request_id': request_id,
-            'plugin_id': resolved_plugin,
-            'mode': resolved_mode,
-            'duration': duration,
-            'pinned': pinned,
-            'service': service_result
-        }
-        return jsonify({'status': 'success', 'data': response_data})
-    except Exception as exc:
-        logger.error('Error in start_on_demand_display', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
+    response_data = {
+        'request_id': request_id,
+        'plugin_id': resolved_plugin,
+        'mode': resolved_mode,
+        'duration': duration,
+        'pinned': pinned,
+        'service': service_result
+    }
+    return jsonify({'status': 'success', 'data': response_data})
 @api_v3.route('/display/on-demand/stop', methods=['POST'])
 def stop_on_demand_display():
     """Request the display controller to stop on-demand mode."""
-    try:
-        data = request.get_json(silent=True) or {}
-        stop_service = data.get('stop_service', False)
+    data = request.get_json(silent=True) or {}
+    stop_service = data.get('stop_service', False)
 
-        # Set the stop request in cache FIRST
-        # The display controller will poll this and restart without the on-demand filter
-        cache = _cache_manager()
-        request_id = data.get('request_id') or str(uuid.uuid4())
-        request_payload = {
+    # Set the stop request in cache FIRST
+    # The display controller will poll this and restart without the on-demand filter
+    cache = _cache_manager()
+    request_id = data.get('request_id') or str(uuid.uuid4())
+    request_payload = {
+        'request_id': request_id,
+        'action': 'stop',
+        'timestamp': _pkg.time.time()
+    }
+    cache.set('display_on_demand_request', request_payload)
+        
+    # Note: The display controller's _clear_on_demand() will handle the restart
+    # to restore normal operation with all plugins
+        
+    service_result = None
+    if stop_service:
+        service_result = _stop_display_service()
+
+    return jsonify({
+        'status': 'success',
+        'data': {
             'request_id': request_id,
-            'action': 'stop',
-            'timestamp': _pkg.time.time()
+            'service': service_result
         }
-        cache.set('display_on_demand_request', request_payload)
-        
-        # Note: The display controller's _clear_on_demand() will handle the restart
-        # to restore normal operation with all plugins
-        
-        service_result = None
-        if stop_service:
-            service_result = _stop_display_service()
-
-        return jsonify({
-            'status': 'success',
-            'data': {
-                'request_id': request_id,
-                'service': service_result
-            }
-        })
-    except Exception as exc:
-        logger.error('Error in stop_on_demand_display', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
+    })
 @api_v3.route('/display/current-status', methods=['GET'])
 def get_current_display_status():
     """Return the display mode/plugin currently intended to be shown.
@@ -325,17 +299,13 @@ def get_current_display_status():
     System Logs page) can show what's on screen without querying the display
     process directly.
     """
-    try:
-        cache = _cache_manager()
-        # memory_ttl=0: written by the display service; see get_on_demand_status.
-        state = cache.get('display_current_state', max_age=120, memory_ttl=0)
-        if state is None:
-            state = {
-                'mode': None,
-                'plugin_id': None,
-                'last_updated': None,
-            }
-        return jsonify({'status': 'success', 'data': state})
-    except Exception as e:
-        logger.error('Error in get_current_display_status', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    cache = _cache_manager()
+    # memory_ttl=0: written by the display service; see get_on_demand_status.
+    state = cache.get('display_current_state', max_age=120, memory_ttl=0)
+    if state is None:
+        state = {
+            'mode': None,
+            'plugin_id': None,
+            'last_updated': None,
+        }
+    return jsonify({'status': 'success', 'data': state})

web_interface/blueprints/api_v3/fonts.py

@@ -4,7 +4,7 @@ Routes decorate the shared `api_v3` Blueprint from ._common, so their
 endpoint names are unchanged by living here.
 """
 from web_interface.blueprints.api_v3 import (
-    PROJECT_ROOT, Path, Response, SYSTEM_FONTS, api_v3, describe_exception,
+    PROJECT_ROOT, Path, Response, SYSTEM_FONTS, api_v3,
     jsonify, logger, os, re, request, validate_file_upload,
 )
 
@@ -52,439 +52,417 @@ def _catalog_response(catalog):
 @api_v3.route('/fonts/catalog', methods=['GET'])
 def get_fonts_catalog():
     """Get fonts catalog"""
+    # Check cache first (5 minute TTL)
     try:
-        # Check cache first (5 minute TTL)
+        from web_interface.cache import get_cached, set_cached
+        cached_result = get_cached('fonts_catalog', ttl_seconds=300)
+        if cached_result is not None:
+            return _catalog_response(cached_result)
+    except ImportError:
+        # Cache not available, continue without caching
+        get_cached = None
+        set_cached = None
+
+    # Try to import freetype, but continue without it if unavailable
+    try:
+        import freetype
+        freetype_available = True
+    except ImportError:
+        freetype_available = False
+
+    # Scan assets/fonts directory for actual font files
+    fonts_dir = PROJECT_ROOT / "assets" / "fonts"
+    catalog = {}
+
+    if fonts_dir.exists() and fonts_dir.is_dir():
+        for filename in os.listdir(fonts_dir):
+            if filename.endswith(('.ttf', '.otf', '.bdf')):
+                filepath = fonts_dir / filename
+                # Generate family name from filename (without extension)
+                family_name = os.path.splitext(filename)[0]
+
+                # Try to get font metadata using freetype (for TTF/OTF)
+                metadata = {}
+                if filename.endswith(('.ttf', '.otf')) and freetype_available:
+                    try:
+                        face = freetype.Face(str(filepath))
+                        if face.valid:
+                            # Get font family name from font file
+                            family_name_from_font = face.family_name.decode('utf-8') if face.family_name else family_name
+                            metadata = {
+                                'family': family_name_from_font,
+                                'style': face.style_name.decode('utf-8') if face.style_name else 'Regular',
+                                'num_glyphs': face.num_glyphs,
+                                'units_per_em': face.units_per_EM
+                            }
+                            # Use font's family name if available
+                            if family_name_from_font:
+                                family_name = family_name_from_font
+                    except Exception:
+                        # If freetype fails, use filename-based name
+                        pass
+
+                # Store relative path from project root
+                relative_path = str(filepath.relative_to(PROJECT_ROOT))
+                font_type = 'ttf' if filename.endswith('.ttf') else 'otf' if filename.endswith('.otf') else 'bdf'
+
+                # Generate human-readable display name from family_name
+                display_name = family_name.replace('-', ' ').replace('_', ' ')
+                # Add space before capital letters for camelCase names
+                display_name = re.sub(r'([a-z])([A-Z])', r'\1 \2', display_name)
+                # Add space before numbers that follow letters
+                display_name = re.sub(r'([a-zA-Z])(\d)', r'\1 \2', display_name)
+                # Clean up multiple spaces
+                display_name = ' '.join(display_name.split())
+
+                # Use filename (without extension) as unique key to avoid collisions
+                # when multiple files share the same family_name from font metadata
+                catalog_key = os.path.splitext(filename)[0]
+
+                # Check if this is a system font (cannot be deleted)
+                is_system = catalog_key.lower() in SYSTEM_FONTS
+
+                # BDF files are fixed-size bitmap strikes: FreeType
+                # accepts only the pixel size baked into the file. The
+                # UI needs to know that before offering a size control,
+                # or it offers a number that cannot take effect.
+                native_size = None
+                if font_type == 'bdf':
+                    try:
+                        from src.element_style import _read_bdf_native_size
+                        native_size = _read_bdf_native_size(str(filepath))
+                    except Exception as e:
+                        logger.debug("Could not read native size for BDF font %s: %s",
+                                     filepath, e)
+                        native_size = None
+
+                catalog[catalog_key] = {
+                    'filename': filename,
+                    'family_name': family_name,
+                    'display_name': display_name,
+                    'path': relative_path,
+                    'type': font_type,
+                    'is_system': is_system,
+                    'scalable': font_type != 'bdf',
+                    'native_size': native_size,
+                    'metadata': metadata if metadata else None
+                }
+
+    # Cache the result (5 minute TTL) if available
+    if set_cached:
         try:
-            from web_interface.cache import get_cached, set_cached
-            cached_result = get_cached('fonts_catalog', ttl_seconds=300)
-            if cached_result is not None:
-                return _catalog_response(cached_result)
-        except ImportError:
-            # Cache not available, continue without caching
-            get_cached = None
-            set_cached = None
+            set_cached('fonts_catalog', catalog, ttl_seconds=300)
+        except Exception:
+            logger.error("[FontCatalog] Failed to cache fonts_catalog", exc_info=True)
 
-        # Try to import freetype, but continue without it if unavailable
-        try:
-            import freetype
-            freetype_available = True
-        except ImportError:
-            freetype_available = False
-
-        # Scan assets/fonts directory for actual font files
-        fonts_dir = PROJECT_ROOT / "assets" / "fonts"
-        catalog = {}
-
-        if fonts_dir.exists() and fonts_dir.is_dir():
-            for filename in os.listdir(fonts_dir):
-                if filename.endswith(('.ttf', '.otf', '.bdf')):
-                    filepath = fonts_dir / filename
-                    # Generate family name from filename (without extension)
-                    family_name = os.path.splitext(filename)[0]
-
-                    # Try to get font metadata using freetype (for TTF/OTF)
-                    metadata = {}
-                    if filename.endswith(('.ttf', '.otf')) and freetype_available:
-                        try:
-                            face = freetype.Face(str(filepath))
-                            if face.valid:
-                                # Get font family name from font file
-                                family_name_from_font = face.family_name.decode('utf-8') if face.family_name else family_name
-                                metadata = {
-                                    'family': family_name_from_font,
-                                    'style': face.style_name.decode('utf-8') if face.style_name else 'Regular',
-                                    'num_glyphs': face.num_glyphs,
-                                    'units_per_em': face.units_per_EM
-                                }
-                                # Use font's family name if available
-                                if family_name_from_font:
-                                    family_name = family_name_from_font
-                        except Exception:
-                            # If freetype fails, use filename-based name
-                            pass
-
-                    # Store relative path from project root
-                    relative_path = str(filepath.relative_to(PROJECT_ROOT))
-                    font_type = 'ttf' if filename.endswith('.ttf') else 'otf' if filename.endswith('.otf') else 'bdf'
-
-                    # Generate human-readable display name from family_name
-                    display_name = family_name.replace('-', ' ').replace('_', ' ')
-                    # Add space before capital letters for camelCase names
-                    display_name = re.sub(r'([a-z])([A-Z])', r'\1 \2', display_name)
-                    # Add space before numbers that follow letters
-                    display_name = re.sub(r'([a-zA-Z])(\d)', r'\1 \2', display_name)
-                    # Clean up multiple spaces
-                    display_name = ' '.join(display_name.split())
-
-                    # Use filename (without extension) as unique key to avoid collisions
-                    # when multiple files share the same family_name from font metadata
-                    catalog_key = os.path.splitext(filename)[0]
-
-                    # Check if this is a system font (cannot be deleted)
-                    is_system = catalog_key.lower() in SYSTEM_FONTS
-
-                    # BDF files are fixed-size bitmap strikes: FreeType
-                    # accepts only the pixel size baked into the file. The
-                    # UI needs to know that before offering a size control,
-                    # or it offers a number that cannot take effect.
-                    native_size = None
-                    if font_type == 'bdf':
-                        try:
-                            from src.element_style import _read_bdf_native_size
-                            native_size = _read_bdf_native_size(str(filepath))
-                        except Exception as e:
-                            logger.debug("Could not read native size for BDF font %s: %s",
-                                         filepath, e)
-                            native_size = None
-
-                    catalog[catalog_key] = {
-                        'filename': filename,
-                        'family_name': family_name,
-                        'display_name': display_name,
-                        'path': relative_path,
-                        'type': font_type,
-                        'is_system': is_system,
-                        'scalable': font_type != 'bdf',
-                        'native_size': native_size,
-                        'metadata': metadata if metadata else None
-                    }
-
-        # Cache the result (5 minute TTL) if available
-        if set_cached:
-            try:
-                set_cached('fonts_catalog', catalog, ttl_seconds=300)
-            except Exception:
-                logger.error("[FontCatalog] Failed to cache fonts_catalog", exc_info=True)
-
-        return _catalog_response(catalog)
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
-        return jsonify({'status': 'error',
-                        'message': 'An error occurred; see logs for details',
-                        'details': describe_exception(e)}), 500
+    return _catalog_response(catalog)
 @api_v3.route('/fonts/tokens', methods=['GET'])
 def get_font_tokens():
     """Get font size tokens"""
-    try:
-        # This would integrate with the actual font system
-        # For now, return sample tokens
-        tokens = {
-            'xs': 6,
-            'sm': 8,
-            'md': 10,
-            'lg': 12,
-            'xl': 14,
-            'xxl': 16
-        }
-        return jsonify({'status': 'success', 'data': {'tokens': tokens}})
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    # This would integrate with the actual font system
+    # For now, return sample tokens
+    tokens = {
+        'xs': 6,
+        'sm': 8,
+        'md': 10,
+        'lg': 12,
+        'xl': 14,
+        'xxl': 16
+    }
+    return jsonify({'status': 'success', 'data': {'tokens': tokens}})
 @api_v3.route('/fonts/upload', methods=['POST'])
 def upload_font():
     """Upload font file"""
-    try:
-        if 'font_file' not in request.files:
-            return jsonify({'status': 'error', 'message': 'No font file provided'}), 400
+    if 'font_file' not in request.files:
+        return jsonify({'status': 'error', 'message': 'No font file provided'}), 400
 
-        font_file = request.files['font_file']
-        if font_file.filename == '':
-            return jsonify({'status': 'error', 'message': 'No file selected'}), 400
+    font_file = request.files['font_file']
+    if font_file.filename == '':
+        return jsonify({'status': 'error', 'message': 'No file selected'}), 400
 
-        # Validate filename. validate_file_upload takes max_size_mb but only
-        # checks the filename/extension with it -- it never looks at the
-        # actual upload size, so the size limit below is enforced separately
-        # before the file is saved (same pattern as the .star upload above).
-        MAX_FONT_SIZE_MB = 10
-        is_valid, error_msg = validate_file_upload(
-            font_file.filename,
-            max_size_mb=MAX_FONT_SIZE_MB,
-            allowed_extensions=['.ttf', '.otf', '.bdf']
-        )
-        if not is_valid:
-            return jsonify({'status': 'error', 'message': error_msg}), 400
-
-        # Check file size (stated limit is MAX_FONT_SIZE_MB)
-        font_file.seek(0, 2)  # Seek to end
-        file_size = font_file.tell()
-        font_file.seek(0)  # Reset to beginning
-        max_font_size_bytes = MAX_FONT_SIZE_MB * 1024 * 1024
-        if file_size > max_font_size_bytes:
-            return jsonify({
-                'status': 'error',
-                'message': f'File too large (max {MAX_FONT_SIZE_MB}MB, got {file_size / 1024 / 1024:.1f}MB)'
-            }), 400
-
-        font_family = request.form.get('font_family', '')
-
-        if not font_family:
-            return jsonify({'status': 'error', 'message': 'Font file and family name required'}), 400
-
-        # Validate font family name
-        if not font_family.replace('_', '').replace('-', '').isalnum():
-            return jsonify({'status': 'error', 'message': 'Font family name must contain only letters, numbers, underscores, and hyphens'}), 400
-
-        # Save the font file to assets/fonts directory
-        fonts_dir = PROJECT_ROOT / "assets" / "fonts"
-        fonts_dir.mkdir(parents=True, exist_ok=True)
-
-        # Create filename from family name
-        original_ext = os.path.splitext(font_file.filename)[1].lower()
-        safe_filename = f"{font_family}{original_ext}"
-        filepath = fonts_dir / safe_filename
-
-        # Check if file already exists
-        if filepath.exists():
-            return jsonify({'status': 'error', 'message': f'Font with name {font_family} already exists'}), 400
-
-        # Save the file
-        font_file.save(str(filepath))
-
-        # Clear font catalog cache
-        try:
-            from web_interface.cache import delete_cached
-            delete_cached('fonts_catalog')
-        except ImportError as e:
-            logger.warning("[FontUpload] Cache module not available: %s", e)
-        except Exception:
-            logger.error("[FontUpload] Failed to clear fonts_catalog cache", exc_info=True)
+    # Validate filename. validate_file_upload takes max_size_mb but only
+    # checks the filename/extension with it -- it never looks at the
+    # actual upload size, so the size limit below is enforced separately
+    # before the file is saved (same pattern as the .star upload above).
+    MAX_FONT_SIZE_MB = 10
+    is_valid, error_msg = validate_file_upload(
+        font_file.filename,
+        max_size_mb=MAX_FONT_SIZE_MB,
+        allowed_extensions=['.ttf', '.otf', '.bdf']
+    )
+    if not is_valid:
+        return jsonify({'status': 'error', 'message': error_msg}), 400
 
+    # Check file size (stated limit is MAX_FONT_SIZE_MB)
+    font_file.seek(0, 2)  # Seek to end
+    file_size = font_file.tell()
+    font_file.seek(0)  # Reset to beginning
+    max_font_size_bytes = MAX_FONT_SIZE_MB * 1024 * 1024
+    if file_size > max_font_size_bytes:
         return jsonify({
-            'status': 'success',
-            'message': f'Font {font_family} uploaded successfully',
-            'font_family': font_family,
-            'filename': safe_filename,
-            'path': f'assets/fonts/{safe_filename}'
-        })
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+            'status': 'error',
+            'message': f'File too large (max {MAX_FONT_SIZE_MB}MB, got {file_size / 1024 / 1024:.1f}MB)'
+        }), 400
+
+    font_family = request.form.get('font_family', '')
+
+    if not font_family:
+        return jsonify({'status': 'error', 'message': 'Font file and family name required'}), 400
+
+    # Validate font family name
+    if not font_family.replace('_', '').replace('-', '').isalnum():
+        return jsonify({'status': 'error', 'message': 'Font family name must contain only letters, numbers, underscores, and hyphens'}), 400
+
+    # Save the font file to assets/fonts directory
+    fonts_dir = PROJECT_ROOT / "assets" / "fonts"
+    fonts_dir.mkdir(parents=True, exist_ok=True)
+
+    # Create filename from family name
+    original_ext = os.path.splitext(font_file.filename)[1].lower()
+    safe_filename = f"{font_family}{original_ext}"
+    filepath = fonts_dir / safe_filename
+
+    # Check if file already exists
+    if filepath.exists():
+        return jsonify({'status': 'error', 'message': f'Font with name {font_family} already exists'}), 400
+
+    # Save the file
+    font_file.save(str(filepath))
+
+    # Clear font catalog cache
+    try:
+        from web_interface.cache import delete_cached
+        delete_cached('fonts_catalog')
+    except ImportError as e:
+        logger.warning("[FontUpload] Cache module not available: %s", e)
+    except Exception:
+        logger.error("[FontUpload] Failed to clear fonts_catalog cache", exc_info=True)
+
+    return jsonify({
+        'status': 'success',
+        'message': f'Font {font_family} uploaded successfully',
+        'font_family': font_family,
+        'filename': safe_filename,
+        'path': f'assets/fonts/{safe_filename}'
+    })
 @api_v3.route('/fonts/preview', methods=['GET'])
 def get_font_preview() -> tuple[Response, int] | Response:
     """Generate a preview image of text rendered with a specific font"""
+    from PIL import Image, ImageDraw, ImageFont
+    import io
+    import base64
+
+    # Limits to prevent DoS via large image generation on constrained devices
+    MAX_TEXT_CHARS = 100
+    MAX_TEXT_LINES = 3
+    MAX_DIM = 1024  # Max width or height in pixels
+    MAX_PIXELS = 500000  # Max total pixels (e.g., ~700x700)
+
+    font_filename = request.args.get('font', '')
+    text = request.args.get('text', 'Sample Text 123')
+    bg_color = request.args.get('bg', '000000')
+    fg_color = request.args.get('fg', 'ffffff')
+
+    # Validate text length and line count early
+    if len(text) > MAX_TEXT_CHARS:
+        return jsonify({'status': 'error', 'message': f'Text exceeds maximum length of {MAX_TEXT_CHARS} characters'}), 400
+    if text.count('\n') >= MAX_TEXT_LINES:
+        return jsonify({'status': 'error', 'message': f'Text exceeds maximum of {MAX_TEXT_LINES} lines'}), 400
+
+    # Safe integer parsing for size
     try:
-        from PIL import Image, ImageDraw, ImageFont
-        import io
-        import base64
+        size = int(request.args.get('size', 12))
+    except (ValueError, TypeError, OverflowError):
+        return jsonify({'status': 'error', 'message': 'Invalid font size'}), 400
 
-        # Limits to prevent DoS via large image generation on constrained devices
-        MAX_TEXT_CHARS = 100
-        MAX_TEXT_LINES = 3
-        MAX_DIM = 1024  # Max width or height in pixels
-        MAX_PIXELS = 500000  # Max total pixels (e.g., ~700x700)
+    if not font_filename:
+        return jsonify({'status': 'error', 'message': 'Font filename required'}), 400
 
-        font_filename = request.args.get('font', '')
-        text = request.args.get('text', 'Sample Text 123')
-        bg_color = request.args.get('bg', '000000')
-        fg_color = request.args.get('fg', 'ffffff')
+    # Validate size
+    if size < 4 or size > 72:
+        return jsonify({'status': 'error', 'message': 'Font size must be between 4 and 72'}), 400
 
-        # Validate text length and line count early
-        if len(text) > MAX_TEXT_CHARS:
-            return jsonify({'status': 'error', 'message': f'Text exceeds maximum length of {MAX_TEXT_CHARS} characters'}), 400
-        if text.count('\n') >= MAX_TEXT_LINES:
-            return jsonify({'status': 'error', 'message': f'Text exceeds maximum of {MAX_TEXT_LINES} lines'}), 400
+    # Security: Validate font_filename to prevent path traversal
+    # Only allow alphanumeric, hyphen, underscore, and dot (for extension)
+    safe_name = Path(font_filename).name  # Strip any directory components
+    if safe_name != font_filename or '..' in font_filename:
+        return jsonify({'status': 'error', 'message': 'Invalid font filename'}), 400
 
-        # Safe integer parsing for size
-        try:
-            size = int(request.args.get('size', 12))
-        except (ValueError, TypeError, OverflowError):
-            return jsonify({'status': 'error', 'message': 'Invalid font size'}), 400
+    # Validate extension
+    allowed_extensions = ['.ttf', '.otf', '.bdf']
+    has_valid_ext = any(safe_name.lower().endswith(ext) for ext in allowed_extensions)
+    name_without_ext = safe_name.rsplit('.', 1)[0] if '.' in safe_name else safe_name
 
-        if not font_filename:
-            return jsonify({'status': 'error', 'message': 'Font filename required'}), 400
+    # Find the font file
+    fonts_dir = PROJECT_ROOT / "assets" / "fonts"
+    if not fonts_dir.exists():
+        return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
 
-        # Validate size
-        if size < 4 or size > 72:
-            return jsonify({'status': 'error', 'message': 'Font size must be between 4 and 72'}), 400
+    font_path = fonts_dir / safe_name
 
-        # Security: Validate font_filename to prevent path traversal
-        # Only allow alphanumeric, hyphen, underscore, and dot (for extension)
-        safe_name = Path(font_filename).name  # Strip any directory components
-        if safe_name != font_filename or '..' in font_filename:
-            return jsonify({'status': 'error', 'message': 'Invalid font filename'}), 400
+    if not font_path.exists() and not has_valid_ext:
+        # Try finding by family name (without extension)
+        for ext in allowed_extensions:
+            potential_path = fonts_dir / f"{name_without_ext}{ext}"
+            if potential_path.exists():
+                font_path = potential_path
+                break
 
-        # Validate extension
-        allowed_extensions = ['.ttf', '.otf', '.bdf']
-        has_valid_ext = any(safe_name.lower().endswith(ext) for ext in allowed_extensions)
-        name_without_ext = safe_name.rsplit('.', 1)[0] if '.' in safe_name else safe_name
+    # Final security check: ensure path is within fonts_dir
+    try:
+        font_path.resolve().relative_to(fonts_dir.resolve())
+    except ValueError:
+        return jsonify({'status': 'error', 'message': 'Invalid font path'}), 400
 
-        # Find the font file
-        fonts_dir = PROJECT_ROOT / "assets" / "fonts"
-        if not fonts_dir.exists():
-            return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
+    if not font_path.exists():
+        return jsonify({'status': 'error', 'message': f'Font file not found: {font_filename}'}), 404
 
-        font_path = fonts_dir / safe_name
-
-        if not font_path.exists() and not has_valid_ext:
-            # Try finding by family name (without extension)
-            for ext in allowed_extensions:
-                potential_path = fonts_dir / f"{name_without_ext}{ext}"
-                if potential_path.exists():
-                    font_path = potential_path
-                    break
-
-        # Final security check: ensure path is within fonts_dir
-        try:
-            font_path.resolve().relative_to(fonts_dir.resolve())
-        except ValueError:
-            return jsonify({'status': 'error', 'message': 'Invalid font path'}), 400
-
-        if not font_path.exists():
-            return jsonify({'status': 'error', 'message': f'Font file not found: {font_filename}'}), 404
-
-        # Parse colors
-        try:
-            bg_rgb = tuple(int(bg_color[i:i+2], 16) for i in (0, 2, 4))
-            fg_rgb = tuple(int(fg_color[i:i+2], 16) for i in (0, 2, 4))
-        except (ValueError, IndexError):
-            bg_rgb = (0, 0, 0)
-            fg_rgb = (255, 255, 255)
-
-        # Load font
-        font = None
-        if str(font_path).endswith('.bdf'):
-            # BDF fonts require complex per-glyph rendering via freetype
-            # Return explicit error rather than showing misleading preview with default font
-            return jsonify({
-                'status': 'error',
-                'message': 'BDF font preview not supported. BDF fonts will render correctly on the LED matrix.'
-            }), 400
-        else:
-            # TTF/OTF fonts
-            try:
-                font = ImageFont.truetype(str(font_path), size)
-            except (IOError, OSError) as e:
-                # IOError/OSError raised for invalid/corrupt font files
-                logger.warning("[FontPreview] Failed to load font %s: %s", font_path, e)
-                font = ImageFont.load_default()
-
-        # Calculate text size
-        temp_img = Image.new('RGB', (1, 1))
-        temp_draw = ImageDraw.Draw(temp_img)
-        bbox = temp_draw.textbbox((0, 0), text, font=font)
-        text_width = bbox[2] - bbox[0]
-        text_height = bbox[3] - bbox[1]
-
-        # Create image with padding
-        padding = 10
-        img_width = max(text_width + padding * 2, 100)
-        img_height = max(text_height + padding * 2, 30)
-
-        # Validate resulting image size to prevent memory/CPU spikes
-        if img_width > MAX_DIM or img_height > MAX_DIM:
-            return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
-        if img_width * img_height > MAX_PIXELS:
-            return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
-
-        img = Image.new('RGB', (img_width, img_height), bg_rgb)
-        draw = ImageDraw.Draw(img)
-
-        # Center text
-        x = (img_width - text_width) // 2
-        y = (img_height - text_height) // 2
-
-        draw.text((x, y), text, font=font, fill=fg_rgb)
-
-        # Convert to base64
-        buffer = io.BytesIO()
-        img.save(buffer, format='PNG')
-        buffer.seek(0)
-        img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
+    # Parse colors
+    try:
+        bg_rgb = tuple(int(bg_color[i:i+2], 16) for i in (0, 2, 4))
+        fg_rgb = tuple(int(fg_color[i:i+2], 16) for i in (0, 2, 4))
+    except (ValueError, IndexError):
+        bg_rgb = (0, 0, 0)
+        fg_rgb = (255, 255, 255)
 
+    # Load font
+    font = None
+    if str(font_path).endswith('.bdf'):
+        # BDF fonts require complex per-glyph rendering via freetype
+        # Return explicit error rather than showing misleading preview with default font
         return jsonify({
-            'status': 'success',
-            'data': {
-                'image': f'data:image/png;base64,{img_base64}',
-                'width': img_width,
-                'height': img_height
-            }
-        })
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+            'status': 'error',
+            'message': 'BDF font preview not supported. BDF fonts will render correctly on the LED matrix.'
+        }), 400
+    else:
+        # TTF/OTF fonts
+        try:
+            font = ImageFont.truetype(str(font_path), size)
+        except (IOError, OSError) as e:
+            # IOError/OSError raised for invalid/corrupt font files
+            logger.warning("[FontPreview] Failed to load font %s: %s", font_path, e)
+            font = ImageFont.load_default()
+
+    # Calculate text size
+    temp_img = Image.new('RGB', (1, 1))
+    temp_draw = ImageDraw.Draw(temp_img)
+    bbox = temp_draw.textbbox((0, 0), text, font=font)
+    text_width = bbox[2] - bbox[0]
+    text_height = bbox[3] - bbox[1]
+
+    # Create image with padding
+    padding = 10
+    img_width = max(text_width + padding * 2, 100)
+    img_height = max(text_height + padding * 2, 30)
+
+    # Validate resulting image size to prevent memory/CPU spikes
+    if img_width > MAX_DIM or img_height > MAX_DIM:
+        return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
+    if img_width * img_height > MAX_PIXELS:
+        return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
+
+    img = Image.new('RGB', (img_width, img_height), bg_rgb)
+    draw = ImageDraw.Draw(img)
+
+    # Center text
+    x = (img_width - text_width) // 2
+    y = (img_height - text_height) // 2
+
+    draw.text((x, y), text, font=font, fill=fg_rgb)
+
+    # Convert to base64
+    buffer = io.BytesIO()
+    img.save(buffer, format='PNG')
+    buffer.seek(0)
+    img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
+
+    return jsonify({
+        'status': 'success',
+        'data': {
+            'image': f'data:image/png;base64,{img_base64}',
+            'width': img_width,
+            'height': img_height
+        }
+    })
 @api_v3.route('/fonts/<font_family>', methods=['DELETE'])
 def delete_font(font_family: str) -> tuple[Response, int] | Response:
     """Delete a user-uploaded font file"""
-    try:
-        # Security: Validate font_family to prevent path traversal
-        # Reject if it contains path separators or ..
-        if '..' in font_family or '/' in font_family or '\\' in font_family:
-            return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
+    # Security: Validate font_family to prevent path traversal
+    # Reject if it contains path separators or ..
+    if '..' in font_family or '/' in font_family or '\\' in font_family:
+        return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
 
-        # Only allow safe characters: alphanumeric, hyphen, underscore, dot
-        if not re.match(r'^[a-zA-Z0-9_\-\.]+$', font_family):
-            return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
+    # Only allow safe characters: alphanumeric, hyphen, underscore, dot
+    if not re.match(r'^[a-zA-Z0-9_\-\.]+$', font_family):
+        return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
 
-        # Check if this is a system font (uses module-level SYSTEM_FONTS frozenset)
-        if font_family.lower() in SYSTEM_FONTS:
-            return jsonify({'status': 'error', 'message': 'Cannot delete system fonts'}), 403
+    # Check if this is a system font (uses module-level SYSTEM_FONTS frozenset)
+    if font_family.lower() in SYSTEM_FONTS:
+        return jsonify({'status': 'error', 'message': 'Cannot delete system fonts'}), 403
 
-        # Find and delete the font file
-        fonts_dir = PROJECT_ROOT / "assets" / "fonts"
+    # Find and delete the font file
+    fonts_dir = PROJECT_ROOT / "assets" / "fonts"
 
-        # Ensure fonts directory exists
-        if not fonts_dir.exists() or not fonts_dir.is_dir():
-            return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
+    # Ensure fonts directory exists
+    if not fonts_dir.exists() or not fonts_dir.is_dir():
+        return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
 
-        deleted = False
-        deleted_filename = None
+    deleted = False
+    deleted_filename = None
 
-        # Only try valid font extensions (no empty string to avoid matching directories)
-        for ext in ['.ttf', '.otf', '.bdf']:
-            potential_path = fonts_dir / f"{font_family}{ext}"
+    # Only try valid font extensions (no empty string to avoid matching directories)
+    for ext in ['.ttf', '.otf', '.bdf']:
+        potential_path = fonts_dir / f"{font_family}{ext}"
 
-            # Security: Verify path is within fonts_dir
-            try:
-                potential_path.resolve().relative_to(fonts_dir.resolve())
-            except ValueError:
-                continue  # Path escapes fonts_dir, skip
-
-            if potential_path.exists() and potential_path.is_file():
-                potential_path.unlink()
-                deleted = True
-                deleted_filename = f"{font_family}{ext}"
-                break
-
-        if not deleted:
-            # Try case-insensitive match within fonts directory
-            font_family_lower = font_family.lower()
-            for filename in os.listdir(fonts_dir):
-                # Only consider files with valid font extensions
-                if not any(filename.lower().endswith(ext) for ext in ['.ttf', '.otf', '.bdf']):
-                    continue
-
-                name_without_ext = os.path.splitext(filename)[0]
-                if name_without_ext.lower() == font_family_lower:
-                    filepath = fonts_dir / filename
-
-                    # Security: Verify path is within fonts_dir
-                    try:
-                        filepath.resolve().relative_to(fonts_dir.resolve())
-                    except ValueError:
-                        continue  # Path escapes fonts_dir, skip
-
-                    if filepath.is_file():
-                        filepath.unlink()
-                        deleted = True
-                        deleted_filename = filename
-                        break
-
-        if not deleted:
-            return jsonify({'status': 'error', 'message': f'Font not found: {font_family}'}), 404
-
-        # Clear font catalog cache
+        # Security: Verify path is within fonts_dir
         try:
-            from web_interface.cache import delete_cached
-            delete_cached('fonts_catalog')
-        except ImportError as e:
-            logger.warning("[FontDelete] Cache module not available: %s", e)
-        except Exception:
-            logger.error("[FontDelete] Failed to clear fonts_catalog cache", exc_info=True)
+            potential_path.resolve().relative_to(fonts_dir.resolve())
+        except ValueError:
+            continue  # Path escapes fonts_dir, skip
 
-        return jsonify({
-            'status': 'success',
-            'message': f'Font {deleted_filename} deleted successfully'
-        })
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+        if potential_path.exists() and potential_path.is_file():
+            potential_path.unlink()
+            deleted = True
+            deleted_filename = f"{font_family}{ext}"
+            break
+
+    if not deleted:
+        # Try case-insensitive match within fonts directory
+        font_family_lower = font_family.lower()
+        for filename in os.listdir(fonts_dir):
+            # Only consider files with valid font extensions
+            if not any(filename.lower().endswith(ext) for ext in ['.ttf', '.otf', '.bdf']):
+                continue
+
+            name_without_ext = os.path.splitext(filename)[0]
+            if name_without_ext.lower() == font_family_lower:
+                filepath = fonts_dir / filename
+
+                # Security: Verify path is within fonts_dir
+                try:
+                    filepath.resolve().relative_to(fonts_dir.resolve())
+                except ValueError:
+                    continue  # Path escapes fonts_dir, skip
+
+                if filepath.is_file():
+                    filepath.unlink()
+                    deleted = True
+                    deleted_filename = filename
+                    break
+
+    if not deleted:
+        return jsonify({'status': 'error', 'message': f'Font not found: {font_family}'}), 404
+
+    # Clear font catalog cache
+    try:
+        from web_interface.cache import delete_cached
+        delete_cached('fonts_catalog')
+    except ImportError as e:
+        logger.warning("[FontDelete] Cache module not available: %s", e)
+    except Exception:
+        logger.error("[FontDelete] Failed to clear fonts_catalog cache", exc_info=True)
+
+    return jsonify({
+        'status': 'success',
+        'message': f'Font {deleted_filename} deleted successfully'
+    })

web_interface/blueprints/api_v3/misc.py

@@ -186,13 +186,6 @@ def get_logs():
             'status': 'error',
             'message': 'Timeout while fetching logs'
         }), 500
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
-        return jsonify({
-            'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
 # Multi-Display Sync Endpoints
 @api_v3.route('/sync/status', methods=['GET'])
 def get_sync_status():
@@ -231,57 +224,49 @@ def get_sync_status():
 @api_v3.route('/cache/list', methods=['GET'])
 def list_cache_files():
     """List all cache files with metadata"""
-    try:
-        if not api_v3.cache_manager:
-            # Initialize cache manager if not already initialized
-            from src.cache_manager import CacheManager
-            api_v3.cache_manager = CacheManager()
+    if not api_v3.cache_manager:
+        # Initialize cache manager if not already initialized
+        from src.cache_manager import CacheManager
+        api_v3.cache_manager = CacheManager()
 
-        cache_files = api_v3.cache_manager.list_cache_files()
-        cache_dir = api_v3.cache_manager.get_cache_dir()
+    cache_files = api_v3.cache_manager.list_cache_files()
+    cache_dir = api_v3.cache_manager.get_cache_dir()
 
-        return jsonify({
-            'status': 'success',
-            'data': {
-                'cache_files': cache_files,
-                'cache_dir': cache_dir,
-                'total_files': len(cache_files)
-            }
-        })
-    except Exception as e:
-        logger.error('Error in list_cache_files', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    return jsonify({
+        'status': 'success',
+        'data': {
+            'cache_files': cache_files,
+            'cache_dir': cache_dir,
+            'total_files': len(cache_files)
+        }
+    })
 @api_v3.route('/cache/delete', methods=['POST'])
 def delete_cache_file():
     """Delete a specific cache file by key"""
-    try:
-        if not api_v3.cache_manager:
-            # Initialize cache manager if not already initialized
-            from src.cache_manager import CacheManager
-            api_v3.cache_manager = CacheManager()
+    if not api_v3.cache_manager:
+        # Initialize cache manager if not already initialized
+        from src.cache_manager import CacheManager
+        api_v3.cache_manager = CacheManager()
 
-        data = request.get_json(silent=True)
-        if not data or 'key' not in data:
-            return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
+    data = request.get_json(silent=True)
+    if not data or 'key' not in data:
+        return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
 
-        cache_key = data['key']
+    cache_key = data['key']
 
-        # The key names the file about to be removed. DiskCache refuses an
-        # unusable key on its own, but silently: say so here instead of
-        # reporting a deletion that never happened.
-        if safe_path_component(cache_key) is None:
-            return jsonify({'status': 'error', 'message': 'Invalid cache key'}), 400
+    # The key names the file about to be removed. DiskCache refuses an
+    # unusable key on its own, but silently: say so here instead of
+    # reporting a deletion that never happened.
+    if safe_path_component(cache_key) is None:
+        return jsonify({'status': 'error', 'message': 'Invalid cache key'}), 400
 
-        # Delete the cache file
-        api_v3.cache_manager.clear_cache(cache_key)
+    # Delete the cache file
+    api_v3.cache_manager.clear_cache(cache_key)
 
-        return jsonify({
-            'status': 'success',
-            'message': f'Cache file for key "{cache_key}" deleted successfully'
-        })
-    except Exception as e:
-        logger.error('Error in delete_cache_file', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    return jsonify({
+        'status': 'success',
+        'message': f'Cache file for key "{cache_key}" deleted successfully'
+    })
 def _errors_cache():
     """The shared cache the display service publishes its errors to."""
     if not api_v3.cache_manager:

web_interface/blueprints/api_v3/starlark.py

@@ -11,6 +11,7 @@ from web_interface.blueprints.api_v3 import (
     _PIXLET_EDITOR_DEFAULT_TIMEOUT, _PIXLET_EDITOR_MAX_TIMEOUT,
     _PIXLET_EDITOR_SCRIPT, _PIXLET_EDITOR_STATE, _clear_pixlet_editor_state,
     _find_pixlet_binary, _install_star_file, _pixlet_editor_alive,
+    _run_systemctl_command,
     _pixlet_editor_status, _read_pixlet_editor_state,
     _STARLARK_APPS_DIR, _standalone_render_starlark_app,
     _starlark_github_token, _starlark_manifest_lock,

web_interface/blueprints/api_v3/system.py

@@ -24,95 +24,91 @@ import web_interface.blueprints.api_v3 as _pkg
 @api_v3.route('/system/status', methods=['GET'])
 def get_system_status():
     """Get system status"""
+    # Check cache first (10 second TTL for system status)
     try:
-        # Check cache first (10 second TTL for system status)
-        try:
-            from web_interface.cache import get_cached, set_cached
-            cached_result = get_cached('system_status', ttl_seconds=10)
-            if cached_result is not None:
-                return jsonify({'status': 'success', 'data': cached_result})
-        except ImportError:
-            # Cache not available, continue without caching
-            get_cached = None
-            set_cached = None
+        from web_interface.cache import get_cached, set_cached
+        cached_result = get_cached('system_status', ttl_seconds=10)
+        if cached_result is not None:
+            return jsonify({'status': 'success', 'data': cached_result})
+    except ImportError:
+        # Cache not available, continue without caching
+        get_cached = None
+        set_cached = None
 
-        # Import psutil for system monitoring
-        try:
-            import psutil
-        except ImportError:
-            # Fallback if psutil not available
-            return jsonify({
-                'status': 'error',
-                'message': 'psutil not available for system monitoring'
-            }), 503
+    # Import psutil for system monitoring
+    try:
+        import psutil
+    except ImportError:
+        # Fallback if psutil not available
+        return jsonify({
+            'status': 'error',
+            'message': 'psutil not available for system monitoring'
+        }), 503
 
-        # Get system metrics using psutil
-        cpu_percent = psutil.cpu_percent(interval=0.1)  # Short interval for responsiveness
-        memory = psutil.virtual_memory()
-        memory_percent = memory.percent
-        disk = psutil.disk_usage('/')
-        disk_percent = disk.percent
+    # Get system metrics using psutil
+    cpu_percent = psutil.cpu_percent(interval=0.1)  # Short interval for responsiveness
+    memory = psutil.virtual_memory()
+    memory_percent = memory.percent
+    disk = psutil.disk_usage('/')
+    disk_percent = disk.percent
 
-        # Calculate uptime
-        boot_time = psutil.boot_time()
-        uptime_seconds = _pkg.time.time() - boot_time
-        uptime_hours = uptime_seconds / 3600
-        uptime_days = uptime_hours / 24
+    # Calculate uptime
+    boot_time = psutil.boot_time()
+    uptime_seconds = _pkg.time.time() - boot_time
+    uptime_hours = uptime_seconds / 3600
+    uptime_days = uptime_hours / 24
 
-        # Format uptime string
-        if uptime_days >= 1:
-            uptime_str = f"{int(uptime_days)}d {int(uptime_hours % 24)}h"
-        elif uptime_hours >= 1:
-            uptime_str = f"{int(uptime_hours)}h {int((uptime_seconds % 3600) / 60)}m"
-        else:
-            uptime_str = f"{int(uptime_seconds / 60)}m"
+    # Format uptime string
+    if uptime_days >= 1:
+        uptime_str = f"{int(uptime_days)}d {int(uptime_hours % 24)}h"
+    elif uptime_hours >= 1:
+        uptime_str = f"{int(uptime_hours)}h {int((uptime_seconds % 3600) / 60)}m"
+    else:
+        uptime_str = f"{int(uptime_seconds / 60)}m"
 
-        # Get CPU temperature (Raspberry Pi)
+    # Get CPU temperature (Raspberry Pi)
+    cpu_temp = None
+    try:
+        temp_file = '/sys/class/thermal/thermal_zone0/temp'
+        if os.path.exists(temp_file):
+            with open(temp_file, 'r') as f:
+                temp_millidegrees = int(f.read().strip())
+                cpu_temp = temp_millidegrees / 1000.0  # Convert to Celsius
+    except (IOError, ValueError, OSError):
+        # Temperature sensor not available or error reading
         cpu_temp = None
+
+    # Get display service status
+    service_status = _get_display_service_status()
+
+    status = {
+        'timestamp': _pkg.time.time(),
+        'uptime': uptime_str,
+        'uptime_seconds': int(uptime_seconds),
+        'service_active': service_status.get('active', False),
+        'cpu_percent': round(cpu_percent, 1),
+        'memory_used_percent': round(memory_percent, 1),
+        'memory_total_mb': round(memory.total / (1024 * 1024), 1),
+        'memory_used_mb': round(memory.used / (1024 * 1024), 1),
+        # MemAvailable, not total-minus-used: it accounts for reclaimable
+        # page cache, so it is what actually predicts memory trouble. A
+        # board can read 70% "used" and be fine, or read the same and be
+        # about to fail fork(), and only this number tells them apart.
+        'memory_available_mb': round(memory.available / (1024 * 1024), 1),
+        'cpu_temp': round(cpu_temp, 1) if cpu_temp is not None else None,
+        'disk_used_percent': round(disk_percent, 1),
+        'disk_total_gb': round(disk.total / (1024 * 1024 * 1024), 1),
+        'disk_used_gb': round(disk.used / (1024 * 1024 * 1024), 1)
+    }
+
+    # Cache the result if available
+    if set_cached:
         try:
-            temp_file = '/sys/class/thermal/thermal_zone0/temp'
-            if os.path.exists(temp_file):
-                with open(temp_file, 'r') as f:
-                    temp_millidegrees = int(f.read().strip())
-                    cpu_temp = temp_millidegrees / 1000.0  # Convert to Celsius
-        except (IOError, ValueError, OSError):
-            # Temperature sensor not available or error reading
-            cpu_temp = None
+            set_cached('system_status', status, ttl_seconds=10)
+        except Exception:
+            pass  # Cache write failed, but continue
 
-        # Get display service status
-        service_status = _get_display_service_status()
-
-        status = {
-            'timestamp': _pkg.time.time(),
-            'uptime': uptime_str,
-            'uptime_seconds': int(uptime_seconds),
-            'service_active': service_status.get('active', False),
-            'cpu_percent': round(cpu_percent, 1),
-            'memory_used_percent': round(memory_percent, 1),
-            'memory_total_mb': round(memory.total / (1024 * 1024), 1),
-            'memory_used_mb': round(memory.used / (1024 * 1024), 1),
-            # MemAvailable, not total-minus-used: it accounts for reclaimable
-            # page cache, so it is what actually predicts memory trouble. A
-            # board can read 70% "used" and be fine, or read the same and be
-            # about to fail fork(), and only this number tells them apart.
-            'memory_available_mb': round(memory.available / (1024 * 1024), 1),
-            'cpu_temp': round(cpu_temp, 1) if cpu_temp is not None else None,
-            'disk_used_percent': round(disk_percent, 1),
-            'disk_total_gb': round(disk.total / (1024 * 1024 * 1024), 1),
-            'disk_used_gb': round(disk.used / (1024 * 1024 * 1024), 1)
-        }
-
-        # Cache the result if available
-        if set_cached:
-            try:
-                set_cached('system_status', status, ttl_seconds=10)
-            except Exception:
-                pass  # Cache write failed, but continue
-
-        return jsonify({'status': 'success', 'data': status})
-    except Exception as e:
-        logger.error('Unhandled exception', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
+    return jsonify({'status': 'success', 'data': status})
 @api_v3.route('/system/version', methods=['GET'])
 def get_system_version():
     """Get LEDMatrix repository version"""

web_interface/blueprints/api_v3/wifi.py

@@ -111,34 +111,26 @@ def _parse_bool_ish(value):
 @api_v3.route('/wifi/status', methods=['GET'])
 def get_wifi_status():
     """Get current WiFi connection status"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        status = wifi_manager.get_wifi_status()
+    wifi_manager = WiFiManager()
+    status = wifi_manager.get_wifi_status()
 
-        # Get auto-enable setting from config
-        auto_enable_ap = wifi_manager.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
+    # Get auto-enable setting from config
+    auto_enable_ap = wifi_manager.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
 
-        return jsonify({
-            'status': 'success',
-            'data': {
-                'connected': status.connected,
-                'ssid': status.ssid,
-                'ip_address': status.ip_address,
-                'signal': status.signal,
-                'ap_mode_active': status.ap_mode_active,
-                'auto_enable_ap_mode': auto_enable_ap,
-                'last_connect_attempt': _last_connect_snapshot(),
-            }
-        })
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
-        return jsonify({
-            'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
+    return jsonify({
+        'status': 'success',
+        'data': {
+            'connected': status.connected,
+            'ssid': status.ssid,
+            'ip_address': status.ip_address,
+            'signal': status.signal,
+            'ap_mode_active': status.ap_mode_active,
+            'auto_enable_ap_mode': auto_enable_ap,
+            'last_connect_attempt': _last_connect_snapshot(),
+        }
+    })
 @api_v3.route('/wifi/scan', methods=['GET'])
 def scan_wifi_networks():
     """Scan for available WiFi networks
@@ -219,241 +211,188 @@ def connect_wifi():
     background (see _last_connect_attempt); otherwise it waits for the result.
     """
     global _last_connect_attempt
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        data = request.get_json(silent=True)
-        if not data:
-            return jsonify({
-                'status': 'error',
-                'message': 'Request body is required'
-            }), 400
-
-        if 'ssid' not in data:
-            return jsonify({
-                'status': 'error',
-                'message': 'SSID is required'
-            }), 400
-
-        ssid = data['ssid']
-        if not ssid or not ssid.strip():
-            return jsonify({
-                'status': 'error',
-                'message': 'SSID cannot be empty'
-            }), 400
-
-        ssid = ssid.strip()
-        password = data.get('password', '') or ''
-
-        wifi_manager = WiFiManager()
-        ap_mode_active = wifi_manager._is_ap_mode_active()
-
-        # One attempt at a time on either path: concurrent connects fight over
-        # the radio, and the first to finish would clear the in-progress flag
-        # the monitor daemon still needs for the other. The check can't depend
-        # on AP state either -- a background attempt takes the AP down long
-        # before it finishes.
-        with _connect_lock:
-            if _last_connect_attempt and _last_connect_attempt['state'] == 'pending':
-                return jsonify({
-                    'status': 'error',
-                    'message': f"Already connecting to {_last_connect_attempt['ssid']}"
-                }), 409
-            _last_connect_attempt = {
-                'ssid': ssid, 'state': 'pending', 'message': None,
-                'error_type': None, 'finished_at': None,
-            }
-
-        if ap_mode_active:
-            try:
-                _spawn(lambda: _run_background_connect(ssid, password))
-            except Exception:
-                # Nothing will ever finish this attempt; don't leave every
-                # later request refused.
-                with _connect_lock:
-                    _last_connect_attempt = None
-                raise
-            return jsonify({
-                'status': 'pending',
-                'message': (
-                    f'Connecting to {ssid}. The LEDMatrix-Setup network will turn off, '
-                    'so this page will lose its connection.'
-                ),
-                'data': {'ssid': ssid},
-            }), 202
-
-        try:
-            success, message = wifi_manager.connect_to_network(ssid, password)
-        except Exception as e:
-            _record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
-            raise
-        payload = _connect_result_payload(ssid, success, message)
-        _record_connect_result(ssid, payload)
-        return jsonify(payload), (200 if success else 400)
-    except Exception as e:
-        logger.error("Error connecting to WiFi", exc_info=True)
+    data = request.get_json(silent=True)
+    if not data:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
-        }), 500
+            'message': 'Request body is required'
+        }), 400
+
+    if 'ssid' not in data:
+        return jsonify({
+            'status': 'error',
+            'message': 'SSID is required'
+        }), 400
+
+    ssid = data['ssid']
+    if not ssid or not ssid.strip():
+        return jsonify({
+            'status': 'error',
+            'message': 'SSID cannot be empty'
+        }), 400
+
+    ssid = ssid.strip()
+    password = data.get('password', '') or ''
+
+    wifi_manager = WiFiManager()
+    ap_mode_active = wifi_manager._is_ap_mode_active()
+
+    # One attempt at a time on either path: concurrent connects fight over
+    # the radio, and the first to finish would clear the in-progress flag
+    # the monitor daemon still needs for the other. The check can't depend
+    # on AP state either -- a background attempt takes the AP down long
+    # before it finishes.
+    with _connect_lock:
+        if _last_connect_attempt and _last_connect_attempt['state'] == 'pending':
+            return jsonify({
+                'status': 'error',
+                'message': f"Already connecting to {_last_connect_attempt['ssid']}"
+            }), 409
+        _last_connect_attempt = {
+            'ssid': ssid, 'state': 'pending', 'message': None,
+            'error_type': None, 'finished_at': None,
+        }
+
+    if ap_mode_active:
+        try:
+            _spawn(lambda: _run_background_connect(ssid, password))
+        except Exception:
+            # Nothing will ever finish this attempt; don't leave every
+            # later request refused.
+            with _connect_lock:
+                _last_connect_attempt = None
+            raise
+        return jsonify({
+            'status': 'pending',
+            'message': (
+                f'Connecting to {ssid}. The LEDMatrix-Setup network will turn off, '
+                'so this page will lose its connection.'
+            ),
+            'data': {'ssid': ssid},
+        }), 202
+
+    try:
+        success, message = wifi_manager.connect_to_network(ssid, password)
+    except Exception as e:
+        _record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
+        raise
+    payload = _connect_result_payload(ssid, success, message)
+    _record_connect_result(ssid, payload)
+    return jsonify(payload), (200 if success else 400)
 @api_v3.route('/wifi/disconnect', methods=['POST'])
 def disconnect_wifi():
     """Disconnect from the current WiFi network"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        success, message = wifi_manager.disconnect_from_network()
+    wifi_manager = WiFiManager()
+    success, message = wifi_manager.disconnect_from_network()
 
-        if success:
-            return jsonify({
-                'status': 'success',
-                'message': message
-            })
-        else:
-            return jsonify({
-                'status': 'error',
-                'message': message or 'Failed to disconnect from network'
-            }), 400
-    except Exception as e:
-        logger.error("Error disconnecting from WiFi", exc_info=True)
+    if success:
+        return jsonify({
+            'status': 'success',
+            'message': message
+        })
+    else:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
-        }), 500
+            'message': message or 'Failed to disconnect from network'
+        }), 400
 @api_v3.route('/wifi/ap/enable', methods=['POST'])
 def enable_ap_mode():
     """Enable access point mode"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        _force_raw = (request.get_json(silent=True) or {}).get('force', False)
-        force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1'))
-        success, message = wifi_manager.enable_ap_mode(force=force)
+    wifi_manager = WiFiManager()
+    _force_raw = (request.get_json(silent=True) or {}).get('force', False)
+    force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1'))
+    success, message = wifi_manager.enable_ap_mode(force=force)
 
-        if success:
-            return jsonify({
-                'status': 'success',
-                'message': message
-            })
-        else:
-            return jsonify({
-                'status': 'error',
-                'message': message
-            }), 400
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
+    if success:
+        return jsonify({
+            'status': 'success',
+            'message': message
+        })
+    else:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
+            'message': message
+        }), 400
 @api_v3.route('/wifi/ap/disable', methods=['POST'])
 def disable_ap_mode():
     """Disable access point mode"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        success, message = wifi_manager.disable_ap_mode()
+    wifi_manager = WiFiManager()
+    success, message = wifi_manager.disable_ap_mode()
 
-        if success:
-            return jsonify({
-                'status': 'success',
-                'message': message
-            })
-        else:
-            return jsonify({
-                'status': 'error',
-                'message': message
-            }), 400
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
+    if success:
+        return jsonify({
+            'status': 'success',
+            'message': message
+        })
+    else:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
+            'message': message
+        }), 400
 @api_v3.route('/wifi/ap/auto-enable', methods=['GET'])
 def get_auto_enable_ap_mode():
     """Get auto-enable AP mode setting"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        auto_enable = wifi_manager.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
+    wifi_manager = WiFiManager()
+    auto_enable = wifi_manager.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
 
-        return jsonify({
-            'status': 'success',
-            'data': {
-                'auto_enable_ap_mode': auto_enable
-            }
-        })
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
-        return jsonify({
-            'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
+    return jsonify({
+        'status': 'success',
+        'data': {
+            'auto_enable_ap_mode': auto_enable
+        }
+    })
 @api_v3.route('/wifi/ap/auto-enable', methods=['POST'])
 def set_auto_enable_ap_mode():
     """Set auto-enable AP mode setting"""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        data = request.get_json(silent=True)
-        if data is None or 'auto_enable_ap_mode' not in data:
-            return jsonify({
-                'status': 'error',
-                'message': 'auto_enable_ap_mode is required'
-            }), 400
-
-        auto_enable = _parse_bool_ish(data['auto_enable_ap_mode'])
-        if auto_enable is None:
-            return jsonify({
-                'status': 'error',
-                'message': 'auto_enable_ap_mode must be a boolean'
-            }), 400
-
-        wifi_manager = WiFiManager()
-        wifi_manager.config["auto_enable_ap_mode"] = auto_enable
-        wifi_manager._save_config()
-
-        return jsonify({
-            'status': 'success',
-            'message': f'Auto-enable AP mode set to {auto_enable}',
-            'data': {
-                'auto_enable_ap_mode': auto_enable
-            }
-        })
-    except Exception as e:
-        logger.error("%s failed", request.path, exc_info=True)
+    data = request.get_json(silent=True)
+    if data is None or 'auto_enable_ap_mode' not in data:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details',
-            'details': describe_exception(e)
-        }), 500
+            'message': 'auto_enable_ap_mode is required'
+        }), 400
+
+    auto_enable = _parse_bool_ish(data['auto_enable_ap_mode'])
+    if auto_enable is None:
+        return jsonify({
+            'status': 'error',
+            'message': 'auto_enable_ap_mode must be a boolean'
+        }), 400
+
+    wifi_manager = WiFiManager()
+    wifi_manager.config["auto_enable_ap_mode"] = auto_enable
+    wifi_manager._save_config()
+
+    return jsonify({
+        'status': 'success',
+        'message': f'Auto-enable AP mode set to {auto_enable}',
+        'data': {
+            'auto_enable_ap_mode': auto_enable
+        }
+    })
 @api_v3.route('/wifi/radio', methods=['GET'])
 def get_wifi_radio():
     """Get current WiFi radio state (enabled/disabled) and wired-fallback status."""
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        wifi_manager = WiFiManager()
-        state = wifi_manager.get_wifi_radio_state()
+    wifi_manager = WiFiManager()
+    state = wifi_manager.get_wifi_radio_state()
 
-        return jsonify({
-            'status': 'success',
-            'data': state
-        })
-    except Exception as e:
-        logger.error("Error getting WiFi radio state", exc_info=True)
-        return jsonify({
-            'status': 'error',
-            'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
-        }), 500
+    return jsonify({
+        'status': 'success',
+        'data': state
+    })
 @api_v3.route('/wifi/radio', methods=['POST'])
 def set_wifi_radio():
     """Turn the WiFi radio on or off.
@@ -462,53 +401,46 @@ def set_wifi_radio():
     unless Ethernet is connected or force=True, to avoid locking the user out
     of this web interface.
     """
-    try:
-        from src.wifi_manager import WiFiManager
+    from src.wifi_manager import WiFiManager
 
-        data = request.get_json(silent=True) or {}
-        if 'enabled' not in data:
-            return jsonify({
-                'status': 'error',
-                'message': 'enabled is required'
-            }), 400
-
-        # Parse defensively: bool("false") is True and a plain int never
-        # matches `is True`, so `_parse_bool_ish` handles bool, string and
-        # int 1/0 — the endpoint is a public contract, not just the shipped
-        # UI (which always sends real JSON booleans). An unrecognized value
-        # must be rejected, not silently disable the radio: this is the
-        # route that can drop the caller's own connection to this interface.
-        enabled = _parse_bool_ish(data['enabled'])
-        if enabled is None:
-            return jsonify({
-                'status': 'error',
-                'message': 'enabled must be a boolean'
-            }), 400
-        force = _parse_bool_ish(data.get('force', False))
-        if force is None:
-            return jsonify({
-                'status': 'error',
-                'message': 'force must be a boolean'
-            }), 400
-
-        wifi_manager = WiFiManager()
-        success, message, reason = wifi_manager.set_wifi_radio(enabled, force=force)
-
-        if success:
-            return jsonify({
-                'status': 'success',
-                'message': message,
-                'data': wifi_manager.get_wifi_radio_state()
-            })
-        else:
-            return jsonify({
-                'status': 'error',
-                'message': message,
-                'reason': reason
-            }), 400
-    except Exception as e:
-        logger.error("Error setting WiFi radio state", exc_info=True)
+    data = request.get_json(silent=True) or {}
+    if 'enabled' not in data:
         return jsonify({
             'status': 'error',
-            'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
-        }), 500
+            'message': 'enabled is required'
+        }), 400
+
+    # Parse defensively: bool("false") is True and a plain int never
+    # matches `is True`, so `_parse_bool_ish` handles bool, string and
+    # int 1/0 — the endpoint is a public contract, not just the shipped
+    # UI (which always sends real JSON booleans). An unrecognized value
+    # must be rejected, not silently disable the radio: this is the
+    # route that can drop the caller's own connection to this interface.
+    enabled = _parse_bool_ish(data['enabled'])
+    if enabled is None:
+        return jsonify({
+            'status': 'error',
+            'message': 'enabled must be a boolean'
+        }), 400
+    force = _parse_bool_ish(data.get('force', False))
+    if force is None:
+        return jsonify({
+            'status': 'error',
+            'message': 'force must be a boolean'
+        }), 400
+
+    wifi_manager = WiFiManager()
+    success, message, reason = wifi_manager.set_wifi_radio(enabled, force=force)
+
+    if success:
+        return jsonify({
+            'status': 'success',
+            'message': message,
+            'data': wifi_manager.get_wifi_radio_state()
+        })
+    else:
+        return jsonify({
+            'status': 'error',
+            'message': message,
+            'reason': reason
+        }), 400

web_interface/cache.py

@@ -1,48 +1,107 @@
 """
-Simple in-memory cache for expensive operations.
-Separated from app.py to avoid circular import issues.
+In-process TTL cache for the web interface.
+
+The one place the web process memoises cheap-to-recompute values for a few
+seconds or minutes (the font catalog, the system-status snapshot, systemctl
+checks). It is per-process and in-memory only; data shared with the display
+service goes through ``src.cache_manager.CacheManager`` instead.
+
+Separated from app.py to avoid circular imports: blueprints import the
+module-level helpers below lazily, inside their request handlers.
 """
+import threading
 import time
-from typing import Any, Optional
+from typing import Any, Callable, Dict, Optional, Tuple
 
 
-# Simple in-memory cache for expensive operations
-_cache = {}
-_cache_timestamps = {}
+class TTLCache:
+    """A small thread-safe key/value store whose entries expire.
+
+    Each entry keeps the TTL it was stored with. A reader may additionally
+    pass ``max_age`` to ask for something fresher than that; an entry is only
+    returned while it is younger than both.
+
+    Expired entries are not dropped on read: :meth:`peek` still returns them,
+    which is what a "keep the last known answer if the refresh fails" caller
+    needs. They are replaced by the next :meth:`set` of the same key, so this
+    is meant for a small, fixed set of keys, not an unbounded key space.
+
+    Ages are measured with ``time.monotonic`` so a wall-clock jump (NTP sync
+    on a Pi that booted without an RTC) neither expires nor immortalises
+    everything at once.
+    """
+
+    def __init__(self, default_ttl: float = 60,
+                 clock: Callable[[], float] = time.monotonic):
+        self._default_ttl = default_ttl
+        self._clock = clock
+        self._lock = threading.Lock()
+        # key -> (value, stored_at, ttl)
+        self._entries: Dict[str, Tuple[Any, float, float]] = {}
+
+    def get(self, key: str, default: Any = None,
+            max_age: Optional[float] = None) -> Any:
+        """The value for ``key`` if it is still fresh, else ``default``."""
+        with self._lock:
+            entry = self._entries.get(key)
+        if entry is None:
+            return default
+        value, stored_at, ttl = entry
+        age = self._clock() - stored_at
+        if age >= ttl or (max_age is not None and age >= max_age):
+            return default
+        return value
+
+    def peek(self, key: str, default: Any = None) -> Any:
+        """The last value stored for ``key``, fresh or not."""
+        with self._lock:
+            entry = self._entries.get(key)
+        return default if entry is None else entry[0]
+
+    def set(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
+        """Store ``value`` for ``ttl`` seconds (the cache default if None)."""
+        ttl = self._default_ttl if ttl is None else ttl
+        with self._lock:
+            self._entries[key] = (value, self._clock(), ttl)
+
+    def delete(self, key: str) -> None:
+        """Remove ``key`` if present."""
+        with self._lock:
+            self._entries.pop(key, None)
+
+    def clear(self, pattern: Optional[str] = None) -> None:
+        """Remove every entry, or only those whose key contains ``pattern``."""
+        with self._lock:
+            if pattern is None:
+                self._entries.clear()
+            else:
+                for key in [k for k in self._entries if pattern in k]:
+                    del self._entries[key]
 
 
-def get_cached(key: str, ttl_seconds: int = 60) -> Optional[Any]:
-    """Get value from cache if not expired."""
-    if key in _cache:
-        if time.time() - _cache_timestamps[key] < ttl_seconds:
-            return _cache[key]
-        else:
-            # Expired, remove
-            del _cache[key]
-            del _cache_timestamps[key]
-    return None
+# The shared cache behind the functional helpers the blueprints use.
+_default_cache = TTLCache(default_ttl=60)
 
 
-def set_cached(key: str, value: Any, ttl_seconds: int = 60) -> None:
-    """Set value in cache with TTL."""
-    _cache[key] = value
-    _cache_timestamps[key] = time.time()
+def get_cached(key: str, ttl_seconds: Optional[float] = None) -> Optional[Any]:
+    """Get a value from the cache if it has not expired.
+
+    The entry expires after the TTL it was stored with; ``ttl_seconds``, when
+    given, is an extra upper bound on its age for this read.
+    """
+    return _default_cache.get(key, max_age=ttl_seconds)
+
+
+def set_cached(key: str, value: Any, ttl_seconds: float = 60) -> None:
+    """Store a value in the cache for ``ttl_seconds``."""
+    _default_cache.set(key, value, ttl=ttl_seconds)
 
 
 def delete_cached(key: str) -> None:
     """Remove a single key from the cache if present."""
-    _cache.pop(key, None)
-    _cache_timestamps.pop(key, None)
+    _default_cache.delete(key)
 
 
 def invalidate_cache(pattern: Optional[str] = None) -> None:
     """Invalidate cache entries matching pattern, or all if pattern is None."""
-    if pattern is None:
-        _cache.clear()
-        _cache_timestamps.clear()
-    else:
-        keys_to_remove = [k for k in _cache.keys() if pattern in k]
-        for key in keys_to_remove:
-            del _cache[key]
-            del _cache_timestamps[key]
-
+    _default_cache.clear(pattern)

web_interface/logging_config.py

@@ -1,110 +0,0 @@
-"""
-Structured logging configuration for the web interface.
-Provides JSON-formatted logs for production and readable logs for development.
-"""
-import logging
-import json
-import sys
-from datetime import datetime
-from typing import Optional
-
-
-class JSONFormatter(logging.Formatter):
-    """Formatter that outputs logs as JSON for structured logging."""
-    
-    def format(self, record: logging.LogRecord) -> str:
-        """Format log record as JSON."""
-        log_data = {
-            'timestamp': datetime.utcnow().isoformat(),
-            'level': record.levelname,
-            'logger': record.name,
-            'message': record.getMessage(),
-            'module': record.module,
-            'function': record.funcName,
-            'line': record.lineno,
-        }
-        
-        # Add exception info if present
-        if record.exc_info:
-            log_data['exception'] = self.formatException(record.exc_info)
-        
-        # Add extra fields if present
-        if hasattr(record, 'request_id'):
-            log_data['request_id'] = record.request_id
-        if hasattr(record, 'user_id'):
-            log_data['user_id'] = record.user_id
-        if hasattr(record, 'ip_address'):
-            log_data['ip_address'] = record.ip_address
-        if hasattr(record, 'duration_ms'):
-            log_data['duration_ms'] = record.duration_ms
-        
-        return json.dumps(log_data)
-
-
-def setup_web_interface_logging(level: str = 'INFO', use_json: bool = False):
-    """
-    Set up logging for the web interface.
-    
-    Args:
-        level: Log level (DEBUG, INFO, WARNING, ERROR)
-        use_json: If True, use JSON formatting (for production)
-    """
-    # Get root logger
-    logger = logging.getLogger()
-    logger.setLevel(getattr(logging, level.upper()))
-    
-    # Remove existing handlers
-    logger.handlers.clear()
-    
-    # Create console handler
-    console_handler = logging.StreamHandler(sys.stdout)
-    console_handler.setLevel(getattr(logging, level.upper()))
-    
-    # Set formatter
-    if use_json:
-        formatter = JSONFormatter()
-    else:
-        formatter = logging.Formatter(
-            '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-            datefmt='%Y-%m-%d %H:%M:%S'
-        )
-    
-    console_handler.setFormatter(formatter)
-    logger.addHandler(console_handler)
-    
-    # Set levels for specific loggers
-    logging.getLogger('werkzeug').setLevel(logging.WARNING)  # Reduce Flask noise
-    logging.getLogger('urllib3').setLevel(logging.WARNING)  # Reduce HTTP noise
-
-
-def log_api_request(method: str, path: str, status_code: int, duration_ms: float,
-                   ip_address: Optional[str] = None, **kwargs):
-    """
-    Log an API request with structured data.
-    
-    Args:
-        method: HTTP method
-        path: Request path
-        status_code: HTTP status code
-        duration_ms: Request duration in milliseconds
-        ip_address: Client IP address
-        **kwargs: Additional context
-    """
-    logger = logging.getLogger('web_interface.api')
-    
-    extra = {
-        'method': method,
-        'path': path,
-        'status_code': status_code,
-        'duration_ms': round(duration_ms, 2),
-        'ip_address': ip_address,
-        **kwargs
-    }
-    
-    # Log at appropriate level based on status code
-    if status_code >= 500:
-        logger.error(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)
-    elif status_code >= 400:
-        logger.warning(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)
-    else:
-        logger.info(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)

web_interface/request_logging.py

@@ -0,0 +1,62 @@
+"""
+Per-request logging for the web interface.
+
+Logging itself is configured by ``src.logging_config.setup_logging`` (the same
+formatter and journald priorities as the display service); this module only
+decides what one HTTP request is worth logging, and at which level.
+
+The UI polls: the error summary, system status, display preview and log
+streams are fetched every few seconds by every open tab. Logging each of those
+at INFO buried everything else in the journal (``GET /api/v3/errors/summary -
+200`` once a minute per tab, forever). So a request that only read something
+and succeeded is DEBUG; one that changed something, or failed, is logged at a
+level that shows up by default.
+"""
+import logging
+import time
+
+from flask import Flask, request
+
+logger = logging.getLogger('web_interface.api')
+
+#: Methods that do not change server state. A successful one is routine.
+_READ_ONLY_METHODS = frozenset({'GET', 'HEAD', 'OPTIONS'})
+
+
+def request_log_level(method: str, status_code: int) -> int:
+    """The level a finished request is logged at."""
+    if status_code >= 500:
+        return logging.ERROR
+    if status_code >= 400:
+        return logging.WARNING
+    if method.upper() in _READ_ONLY_METHODS:
+        return logging.DEBUG
+    return logging.INFO
+
+
+def log_request(method: str, path: str, status_code: int,
+                duration_ms: float) -> None:
+    """Log one finished request."""
+    level = request_log_level(method, status_code)
+    if logger.isEnabledFor(level):
+        logger.log(level, "%s %s - %d (%.1fms)",
+                   method, path, status_code, duration_ms)
+
+
+def init_app(app: Flask) -> None:
+    """Time every request and log it when its response is ready."""
+
+    @app.before_request
+    def _start_request_timer():
+        request.start_time = time.perf_counter()
+
+    @app.after_request
+    def _log_finished_request(response):
+        try:
+            started = getattr(request, 'start_time', None)
+            duration_ms = 0.0 if started is None else (time.perf_counter() - started) * 1000
+            log_request(request.method, request.path, response.status_code,
+                        duration_ms)
+        except Exception:  # nosec B110 - request logging must never interrupt a live HTTP response
+            pass
+        return response