6 Commits

Author SHA1 Message Date
Chuck
aae95a1015 refactor(api): resolve sudo/systemctl/reboot/poweroff paths at startup
Use shutil.which() with safe fallbacks for the four privileged binaries
instead of relying on bare names being resolved by the subprocess shell
search. Resolves paths once at module load rather than per-call.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 14:49:34 -04:00
Chuck
246ea54635 fix: address five review findings (Pillow CVEs, daemon exception narrowing, timeout handling, plugin store)
- march-madness/requirements.txt: Pillow>=12.2.0 (patches CVE-2026-42308
  and CVE-2026-42310; previous floor of 10.3.0 was insufficient)

- wifi_monitor_daemon: narrow final except Exception to
  (subprocess.SubprocessError, OSError) so programming errors in the NM
  restart block are no longer silently swallowed

- api_v3/execute_system_action: add explicit subprocess.TimeoutExpired
  handler before the generic Exception catch; returns action-specific
  message with 'status','message','returncode','stdout','stderr' fields
  so the UI receives a precise, actionable payload instead of the generic
  'Failed to execute system action' string

- plugins_manager.js: move searchPluginStore into .finally() so the
  plugin store renders regardless of whether loadInstalledPlugins succeeds
  or fails; .catch() still logs the error

- first_time_install.sh: add safe_plugin_rm.sh NOPASSWD rule to the
  /tmp/ledmatrix_web_sudoers block; configure_web_sudo.sh had this rule
  but the standalone installer never granted it, leaving plugin removal
  broken after first-time install

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 13:56:24 -04:00
Chuck
a0f957be9e fix: address five review findings (NM retry loop, start_display message, code quality)
- wifi_monitor_daemon: reset _consecutive_internet_failures = 0 in both
  NM-restart exception handlers; previously both left the counter at threshold,
  causing an immediate retry on the next iteration instead of waiting another
  full backoff period

- api_v3: fix start_display failure message — when mode is set and systemctl
  returns non-zero, message now includes the failure reason and a hint rather
  than always reporting success phrasing

- wifi_manager: move _redirect_backend from class variable to instance variable
  in __init__ alongside _ap_enabled_at; class-level default shadowed correctly
  in practice (single instance) but was misleading

- wifi_manager: narrow broad except Exception in _check_internet_connectivity
  to (subprocess.SubprocessError, OSError) for ping and OSError for HTTP
  (urllib.error.URLError is an OSError subclass in Python 3)

- wifi_manager: remove redundant local 'import re as _re' in _validate_ap_config;
  re is already imported at module level (line 37)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 13:26:51 -04:00
Chuck
76cd010aab fix: address five valid review findings; skip two
Fixed:
- march-madness/requirements.txt: Pillow>=10.3.0 (patches CVE-2024-28219;
  10.3.0 is the actual fix version — reviewer cited 12.2.0 but that risks
  breaking API changes without test coverage)
- wifi_monitor_daemon.py: add missing `import subprocess`; subprocess.run
  and CalledProcessError would NameError at runtime on the NM restart path
- wifi_manager.py: validate ap_idle_timeout_minutes before arithmetic —
  coerce to int, clamp 1–1440, fall back to 15 on bad config values
- wifi_manager.py: call _remove_nm_dnsmasq_captive_conf() on all three
  rollback paths in _enable_ap_mode_nmcli_hotspot() and in the top-level
  except block so stale dnsmasq drop-ins are never left behind
- api_v3.py: fix wrong_password prefix strip — removeprefix("wrong_password:")
  then lstrip() handles both "wrong_password: msg" and "wrong_password:msg"
- plugins_manager.js: add .catch() to loadInstalledPlugins().then() to
  surface failures instead of silently dropping unhandled rejections

Skipped:
- WiFiManager AP state persistence: architectural overhaul; _is_ap_mode_active()
  already derives from live system state, not in-memory variables
- Absolute subprocess paths in api_v3.py: paths vary by distro (/usr/bin vs
  /bin); web service has a normal PATH; sudoers already use resolved paths

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 13:26:51 -04:00
Chuck
587daa780e revert: restore AP-mode grace period to 90s (3 checks)
The counter reset after NM restart already fully prevents the SSH-lockout
cascade: _disconnected_checks can never accumulate across NM restarts
because it is reset to 0 before the next daemon iteration runs.

The 3→6 increase provided no additional fix for the described problem and
caused a UX regression: fresh Pi devices with no WiFi configured would
wait 3 minutes instead of 90 seconds for the LEDMatrix-Setup hotspot to
appear.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 13:26:31 -04:00
Chuck
c19df29a21 fix: service control buttons and AP-mode SSH lockout post-install
Two user-reported issues after fresh install:

1. All service buttons (Start/Stop/Restart Display, Restart Web Service)
   failed silently — only Reboot worked.

   Root cause: sudoers rules use `ledmatrix.service` (with suffix) but
   api_v3.py called `sudo systemctl start ledmatrix` (no suffix). sudo
   does exact string matching, so every service action was rejected with
   returncode=1. Also missing from sudoers: ledmatrix-web, journalctl,
   and is-active entries.

   Fix:
   - Add `.service` suffix to all 8 sudo systemctl call sites in
     api_v3.py (_ensure_display_service_running, _stop_display_service,
     and all execute_system_action branches).
   - Add timeout=15 to all subprocess.run calls in execute_system_action
     (previously could hang indefinitely).
   - Add missing sudoers rules to first_time_install.sh and
     configure_web_sudo.sh: ledmatrix-web.service start/stop/restart,
     is-active for both name forms, and journalctl -u/-t ledmatrix rules.

2. SSH and web UI became inaccessible after ~1 hour even though the
   display kept running.

   Root cause: wifi_monitor_daemon restarts NetworkManager after 5
   consecutive internet failures (~2.5 min). Each NM restart drops WiFi
   briefly. During that window check_and_manage_ap_mode() increments
   _disconnected_checks but the daemon never reset it after the restart.
   After 3 such NM-restart cycles, _disconnected_checks reached 3 and
   AP mode activated — changing the Pi from WiFi client to hotspot
   (192.168.4.1) and killing SSH on the old IP.

   Fix:
   - Reset wifi_manager._disconnected_checks = 0 in the daemon
     immediately after a successful NM restart so the brief drop it
     causes doesn't count toward AP-mode activation.
   - Increase _disconnected_checks_required from 3 to 6 (90s → 3min)
     as an additional buffer against transient network flaps.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 13:26:31 -04:00
6 changed files with 151 additions and 332 deletions

View File

@@ -118,7 +118,7 @@ total_count=${#ARCHITECTURES[@]}
for arch in "${!ARCHITECTURES[@]}"; do
if download_binary "$arch" "${ARCHITECTURES[$arch]}"; then
success_count=$((success_count + 1))
((success_count++))
fi
done

View File

@@ -3,7 +3,6 @@ if os.getenv("EMULATOR", "false") == "true":
from RGBMatrixEmulator import RGBMatrix, RGBMatrixOptions
else:
from rgbmatrix import RGBMatrix, RGBMatrixOptions
from contextlib import contextmanager
from PIL import Image, ImageDraw, ImageFont
import time
from typing import Dict, Any, List, Tuple
@@ -29,8 +28,6 @@ class DisplayManager:
self.config = config or {}
self._force_fallback = force_fallback
self._suppress_test_pattern = suppress_test_pattern
# When True, update_display() and clear() skip hardware writes (used during off-screen content capture)
self._capture_mode_active = False
# Snapshot settings for web preview integration (service writes, web reads)
self._snapshot_path = "/tmp/led_matrix_preview.png"
self._snapshot_min_interval_sec = 0.2 # max ~5 fps
@@ -258,22 +255,6 @@ class DisplayManager:
except Exception as e:
logger.error(f"Error drawing test pattern: {e}", exc_info=True)
@contextmanager
def capture_mode(self):
"""Suppress hardware output during off-screen content capture.
Plugins call update_display() as part of their normal display() flow.
When fetching content for Vegas mode the render loop is still running,
so any incidental hardware write causes a visible flash on the matrix.
Entering this context prevents those writes without affecting the PIL
image buffer, which the adapter reads to extract content.
"""
self._capture_mode_active = True
try:
yield
finally:
self._capture_mode_active = False
def update_display(self):
"""Update the display using double buffering with proper sync."""
try:
@@ -284,9 +265,6 @@ class DisplayManager:
self._write_snapshot_if_due()
return
if self._capture_mode_active:
return # Skip hardware write — content is being captured off-screen
# Copy the current image to the offscreen canvas
self.offscreen_canvas.SetImage(self.image)
@@ -327,22 +305,20 @@ class DisplayManager:
self.image = Image.new('RGB', (self.matrix.width, self.matrix.height))
self.draw = ImageDraw.Draw(self.image)
if not self._capture_mode_active:
# Clear both canvases and the underlying matrix to ensure no artifacts.
# Failures are non-fatal — the image buffer is already black above, so
# the next update_display() call will push clean content regardless.
try:
self.offscreen_canvas.Clear()
except (RuntimeError, OSError) as e:
logger.error("Failed to clear offscreen canvas: %s", e)
try:
self.current_canvas.Clear()
except (RuntimeError, OSError) as e:
logger.error("Failed to clear current canvas: %s", e)
try:
self.matrix.Clear()
except (RuntimeError, OSError) as e:
logger.error("Failed to clear matrix front buffer: %s", e)
# Clear both canvases and the underlying matrix to ensure no artifacts
try:
self.offscreen_canvas.Clear()
except Exception:
pass
try:
self.current_canvas.Clear()
except Exception:
pass
try:
# Extra safety: clear the matrix front buffer as well
self.matrix.Clear()
except Exception:
pass
# Note: We do NOT call update_display() here to avoid black flashes.
# The caller should call update_display() after drawing new content.

View File

@@ -329,51 +329,50 @@ class PluginAdapter:
# Save display state to restore after
original_image = self.display_manager.image.copy()
with self.display_manager.capture_mode():
# Method 1: Try _create_scrolling_display (stocks pattern)
if hasattr(plugin, '_create_scrolling_display'):
logger.info(
"[%s] Triggering via _create_scrolling_display()",
plugin_id
)
try:
plugin._create_scrolling_display()
cached_image = getattr(scroll_helper, 'cached_image', None)
if cached_image is not None and isinstance(cached_image, Image.Image):
logger.info(
"[%s] _create_scrolling_display() SUCCESS: %dx%d",
plugin_id, cached_image.width, cached_image.height
)
return cached_image
except (AttributeError, TypeError, ValueError, OSError):
logger.exception(
"[%s] _create_scrolling_display() failed", plugin_id
)
# Method 2: Try display(force_clear=True) which typically builds scroll content
if hasattr(plugin, 'display'):
logger.info(
"[%s] Triggering via display(force_clear=True)",
plugin_id
)
try:
self.display_manager.clear()
plugin.display(force_clear=True)
cached_image = getattr(scroll_helper, 'cached_image', None)
if cached_image is not None and isinstance(cached_image, Image.Image):
logger.info(
"[%s] display(force_clear=True) SUCCESS: %dx%d",
plugin_id, cached_image.width, cached_image.height
)
return cached_image
# Method 1: Try _create_scrolling_display (stocks pattern)
if hasattr(plugin, '_create_scrolling_display'):
logger.info(
"[%s] Triggering via _create_scrolling_display()",
plugin_id
)
try:
plugin._create_scrolling_display()
cached_image = getattr(scroll_helper, 'cached_image', None)
if cached_image is not None and isinstance(cached_image, Image.Image):
logger.info(
"[%s] display(force_clear=True) did not populate cached_image",
plugin_id
"[%s] _create_scrolling_display() SUCCESS: %dx%d",
plugin_id, cached_image.width, cached_image.height
)
except (AttributeError, TypeError, ValueError, OSError):
logger.exception(
"[%s] display(force_clear=True) failed", plugin_id
return cached_image
except (AttributeError, TypeError, ValueError, OSError):
logger.exception(
"[%s] _create_scrolling_display() failed", plugin_id
)
# Method 2: Try display(force_clear=True) which typically builds scroll content
if hasattr(plugin, 'display'):
logger.info(
"[%s] Triggering via display(force_clear=True)",
plugin_id
)
try:
self.display_manager.clear()
plugin.display(force_clear=True)
cached_image = getattr(scroll_helper, 'cached_image', None)
if cached_image is not None and isinstance(cached_image, Image.Image):
logger.info(
"[%s] display(force_clear=True) SUCCESS: %dx%d",
plugin_id, cached_image.width, cached_image.height
)
return cached_image
logger.info(
"[%s] display(force_clear=True) did not populate cached_image",
plugin_id
)
except (AttributeError, TypeError, ValueError, OSError):
logger.exception(
"[%s] display(force_clear=True) failed", plugin_id
)
logger.info(
"[%s] Could not trigger scroll content generation",
@@ -409,7 +408,10 @@ class PluginAdapter:
original_image = self.display_manager.image.copy()
logger.info("[%s] Fallback: saved original display state", plugin_id)
# Ensure plugin has fresh data before capturing
# Lightweight in-memory data refresh before capturing.
# Full update() is intentionally skipped here — the background
# update tick in the Vegas coordinator handles periodic API
# refreshes so we don't block the content-fetch thread.
has_update_data = hasattr(plugin, 'update_data')
logger.info("[%s] Fallback: has update_data=%s", plugin_id, has_update_data)
if has_update_data:
@@ -419,24 +421,21 @@ class PluginAdapter:
except (AttributeError, RuntimeError, OSError):
logger.exception("[%s] Fallback: update_data() failed", plugin_id)
# Clear and call plugin display — use capture_mode to suppress hardware writes
# that plugins may trigger internally via update_display().
with self.display_manager.capture_mode():
self.display_manager.clear()
logger.info("[%s] Fallback: display cleared, calling display()", plugin_id)
# Clear and call plugin display
self.display_manager.clear()
logger.info("[%s] Fallback: display cleared, calling display()", plugin_id)
# First try without force_clear (some plugins behave better this way)
try:
plugin.display()
logger.info("[%s] Fallback: display() called successfully", plugin_id)
except TypeError:
# Plugin may require force_clear argument
logger.info("[%s] Fallback: display() failed, trying with force_clear=True", plugin_id)
plugin.display(force_clear=True)
# Capture the result
captured = self.display_manager.image.copy()
# First try without force_clear (some plugins behave better this way)
try:
plugin.display()
logger.info("[%s] Fallback: display() called successfully", plugin_id)
except TypeError:
# Plugin may require force_clear argument
logger.info("[%s] Fallback: display() failed, trying with force_clear=True", plugin_id)
plugin.display(force_clear=True)
# Capture the result
captured = self.display_manager.image.copy()
logger.info(
"[%s] Fallback: captured frame %dx%d, mode=%s",
plugin_id, captured.width, captured.height, captured.mode
@@ -455,10 +454,9 @@ class PluginAdapter:
plugin_id
)
# Try once more with force_clear=True
with self.display_manager.capture_mode():
self.display_manager.clear()
plugin.display(force_clear=True)
captured = self.display_manager.image.copy()
self.display_manager.clear()
plugin.display(force_clear=True)
captured = self.display_manager.image.copy()
is_blank, bright_ratio = self._is_blank_image(captured, return_ratio=True)
logger.info(
@@ -587,6 +585,28 @@ class PluginAdapter:
else:
self._content_cache.clear()
def invalidate_plugin_scroll_cache(self, plugin: 'BasePlugin', plugin_id: str) -> None:
"""
Clear a plugin's scroll_helper cache so Vegas re-fetches fresh visuals.
Uses scroll_helper.clear_cache() to reset all cached state (cached_image,
cached_array, total_scroll_width, scroll_position, etc.) — not just the
image. Without this, plugins that use scroll_helper (stocks, news,
odds-ticker, etc.) would keep serving stale scroll images even after
their data refreshes.
Args:
plugin: Plugin instance
plugin_id: Plugin identifier
"""
scroll_helper = getattr(plugin, 'scroll_helper', None)
if scroll_helper is None:
return
if getattr(scroll_helper, 'cached_image', None) is not None:
scroll_helper.clear_cache()
logger.debug("[%s] Cleared scroll_helper cache", plugin_id)
def get_content_type(self, plugin: 'BasePlugin', plugin_id: str) -> str:
"""
Get the type of content a plugin provides.

View File

@@ -202,25 +202,8 @@ class RenderPipeline:
# Update scroll position
self.scroll_helper.update_scroll_position()
# Determine if the cycle is done.
#
# scroll_helper considers a cycle complete only after
# total_distance_scrolled >= total_scroll_width + display_width.
# That extra display_width of travel causes a "wrap-around" phase
# where scroll_position resets to ~0 and the first plugin's content
# re-enters from the right — the user sees this 2-3 s window as
# "a plugin partially displaying before the next one starts."
#
# We end the cycle as soon as total_distance_scrolled reaches
# total_scroll_width (the wrap-around point), before any second-pass
# content becomes visible. scroll_helper.is_scroll_complete() is
# kept as a fallback for edge-cases where that threshold is skipped.
at_wrap_point = (
not self._cycle_complete and
self.scroll_helper.total_distance_scrolled >= self.scroll_helper.total_scroll_width
)
if at_wrap_point or self.scroll_helper.is_scroll_complete():
# Check if cycle is complete
if self.scroll_helper.is_scroll_complete():
if not self._cycle_complete:
self._cycle_complete = True
self.stats['scroll_cycles'] += 1
@@ -228,20 +211,6 @@ class RenderPipeline:
"Scroll cycle complete after %.1fs",
time.time() - self._cycle_start_time
)
# Push blank immediately so the hardware never shows
# post-wrap content while the coordinator recomposes.
try:
from PIL import Image as _Image
blank = _Image.new('RGB', (self.display_width, self.display_height))
self.display_manager.image = blank
self.display_manager.update_display()
except (ImportError, OSError, RuntimeError, ValueError, TypeError, MemoryError) as exc:
logger.error(
"Failed to push blank frame at cycle end "
"(display=%dx%d): %s",
self.display_width, self.display_height, exc
)
return True # Cycle done; coordinator starts new cycle next frame
# Get visible portion
visible_frame = self.scroll_helper.get_visible_portion()

View File

@@ -226,24 +226,13 @@ def serve_plugin_asset(plugin_id, filename):
'message': 'Internal server error'
}), 500
# Prime psutil CPU measurement once at startup so interval=None returns a real value
try:
import psutil as _psutil_prime
_psutil_prime.cpu_percent(interval=None)
except ImportError:
pass
# Cached AP mode check — avoids creating a WiFiManager per request
_ap_mode_cache = {'value': False, 'timestamp': 0}
_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
_AP_MODE_CACHE_TTL = 5 # seconds
def is_ap_mode_active():
"""
Check if access point mode is currently active (cached, 30s TTL).
Check if access point mode is currently active (cached, 5s TTL).
Uses a direct systemctl check instead of instantiating WiFiManager.
"""
now = time.time()
@@ -455,8 +444,7 @@ def system_status_generator():
# Try to import psutil for system stats
try:
import psutil
# interval=None is non-blocking; primed at module startup above
cpu_percent = round(psutil.cpu_percent(interval=None), 1)
cpu_percent = round(psutil.cpu_percent(interval=1), 1)
memory = psutil.virtual_memory()
memory_used_percent = round(memory.percent, 1)
@@ -473,17 +461,14 @@ def system_status_generator():
memory_used_percent = 0
cpu_temp = 0
# 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:
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):
pass
_ledmatrix_service_cache['timestamp'] = now
service_active = _ledmatrix_service_cache['active']
# Check if display service is running
service_active = False
try:
result = subprocess.run(['systemctl', 'is-active', 'ledmatrix'],
capture_output=True, text=True, timeout=2)
service_active = result.stdout.strip() == 'active'
except (subprocess.SubprocessError, OSError):
pass
status = {
'timestamp': time.time(),
@@ -561,7 +546,7 @@ def display_preview_generator():
except Exception as e:
yield {'error': str(e)}
time.sleep(1.0) # Check once per second — halves PIL encode overhead vs 0.5s
time.sleep(0.5) # Check 2 times per second (reduced frequency for better performance)
# Logs generator for SSE
def logs_generator():

View File

@@ -4277,13 +4277,8 @@ def _set_missing_booleans_to_false(config, schema_props, form_keys, prefix='', c
elif prop_type == 'object' and 'properties' in prop_schema:
# Recurse into nested objects
if config_node is not None:
# Inside an array item — only recurse if the sub-object already exists.
# Never create optional sub-objects that weren't submitted; doing so
# produces e.g. logo:{} on feed items with no logo, which then fails
# schema validation when the object has required fields (id, path).
if prop_name not in node:
continue
if not isinstance(node[prop_name], dict):
# Inside an array item — ensure nested dict exists in item
if prop_name not in node or not isinstance(node[prop_name], dict):
node[prop_name] = {}
_set_missing_booleans_to_false(
config, prop_schema['properties'], form_keys, full_path,
@@ -4423,22 +4418,10 @@ def _filter_config_by_schema(config, schema, prefix=''):
prop_schema = schema_props[key]
# Handle nested objects recursively
item_prefix = f"{prefix}.{key}" if prefix else key
if isinstance(value, dict) and prop_schema.get('type') == 'object' and 'properties' in prop_schema:
filtered[key] = _filter_config_by_schema(value, prop_schema, item_prefix)
elif isinstance(value, list) and prop_schema.get('type') == 'array':
items_schema = prop_schema.get('items', {})
if isinstance(items_schema, dict) and items_schema.get('type') == 'object' and 'properties' in items_schema:
# Filter each item in the array so extra fields are stripped before
# schema validation (important when items has additionalProperties: false)
filtered[key] = [
_filter_config_by_schema(item, items_schema, item_prefix) if isinstance(item, dict) else item
for item in value
]
else:
filtered[key] = value
filtered[key] = _filter_config_by_schema(value, prop_schema, f"{prefix}.{key}" if prefix else key)
else:
# Keep the value as-is for non-object/non-array types
# Keep the value as-is for non-object types
filtered[key] = value
return filtered
@@ -7594,126 +7577,6 @@ _STARLARK_APPS_DIR = PROJECT_ROOT / 'starlark-apps'
_STARLARK_MANIFEST_FILE = _STARLARK_APPS_DIR / 'manifest.json'
def _find_pixlet_binary(explicit_path: Optional[str] = None) -> Optional[str]:
"""Find pixlet binary: explicit path → bundled binary → system PATH."""
import platform
if explicit_path and os.path.isfile(explicit_path) and os.access(explicit_path, os.X_OK):
return explicit_path
bin_dir = PROJECT_ROOT / "bin" / "pixlet"
system = platform.system().lower()
machine = platform.machine().lower()
if system == "linux":
if "aarch64" in machine or "arm64" in machine:
name = "pixlet-linux-arm64"
elif "x86_64" in machine or "amd64" in machine:
name = "pixlet-linux-amd64"
else:
name = None
elif system == "darwin":
name = "pixlet-darwin-arm64" if "arm64" in machine else "pixlet-darwin-amd64"
else:
name = None
if name:
bundled = bin_dir / name
if bundled.is_file():
if os.access(str(bundled), os.X_OK):
return str(bundled)
try:
bundled.chmod(0o755)
except OSError:
logger.warning("Could not make pixlet bundled binary executable (%s); falling back to PATH", bundled)
else:
if os.access(str(bundled), os.X_OK):
return str(bundled)
logger.warning("Pixlet bundled binary still not executable after chmod (%s); falling back to PATH", bundled)
return shutil.which("pixlet")
def _standalone_render_starlark_app(app_id: str) -> Tuple[bool, int, Optional[str]]:
"""Render a Starlark app via pixlet directly (no plugin required).
Reads the .star file and config from starlark-apps/{app_id}/, runs pixlet,
and saves the output to cached_render.webp in the same directory.
This is the web-service fallback when starlark-apps plugin is not loaded.
Returns (success, http_status_code, error_message).
"""
manifest = _read_starlark_manifest()
if not isinstance(manifest, dict):
return False, 400, "Invalid manifest shape: expected object with 'apps' mapping"
apps = manifest.get('apps', {})
if not isinstance(apps, dict):
return False, 400, "Invalid manifest shape: expected object with 'apps' mapping"
app_data = apps.get(app_id)
if not app_data:
return False, 404, f"App not found: {app_id}"
app_dir = _STARLARK_APPS_DIR / app_id
star_file = app_dir / app_data.get('star_file', f'{app_id}.star')
if not star_file.exists():
return False, 404, f"Star file not found: {star_file}"
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
plugin_config = full_config.get('starlark-apps', {})
pixlet_path = _find_pixlet_binary(plugin_config.get('pixlet_path'))
if not pixlet_path:
return False, 503, "Pixlet binary not found — install pixlet first"
magnify = plugin_config.get('magnify')
if magnify is None:
hw = full_config.get('display', {}).get('hardware', {})
cols = hw.get('cols', 64)
chain = hw.get('chain_length', 1)
rows = hw.get('rows', 32)
magnify = max(1, min(8, int(min((cols * chain) / 64, rows / 32))))
else:
try:
magnify = max(1, min(8, int(magnify)))
except (ValueError, TypeError):
magnify = 1
config_file = app_dir / 'config.json'
app_config: Dict[str, Any] = {}
if config_file.exists():
try:
with open(config_file) as f:
app_config = json.load(f)
except json.JSONDecodeError as e:
return False, 400, f"Invalid config.json for {app_id} ({config_file}): {e}"
except OSError as e:
return False, 400, f"Cannot read config.json for {app_id} ({config_file}): {e}"
if not isinstance(app_config, dict):
return False, 400, (
f"config.json for {app_id} must be a JSON object, "
f"got {type(app_config).__name__}"
)
INTERNAL_KEYS = {'render_interval', 'display_duration'}
pixlet_config = {k: v for k, v in app_config.items() if k not in INTERNAL_KEYS}
output_path = str(app_dir / 'cached_render.webp')
cmd = [pixlet_path, 'render', str(star_file)]
for key, value in pixlet_config.items():
if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', key):
continue
value_str = 'true' if value is True else 'false' if value is False else str(value)
if re.search(r'[`$|<>&;\x00]|\$\(', value_str):
continue
cmd.append(f'{key}={value_str}')
cmd.extend(['-o', output_path, '-m', str(magnify)])
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, cwd=str(app_dir))
if result.returncode == 0 and os.path.isfile(output_path):
return True, 200, None
return False, 502, f"Pixlet failed (exit {result.returncode}): {result.stderr.strip()}"
except subprocess.TimeoutExpired:
return False, 504, "Render timed out after 30s"
except Exception as e:
return False, 500, f"Render error: {e}"
def _read_starlark_manifest() -> Dict[str, Any]:
"""Read the starlark-apps manifest.json directly from disk."""
try:
@@ -7833,11 +7696,24 @@ def get_starlark_status():
'display_info': magnify_info
})
# Plugin not loaded - check Pixlet availability via shared resolver
# (respects user-configured pixlet_path, bundled binary, and system PATH)
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
pixlet_path = _find_pixlet_binary(full_config.get('starlark-apps', {}).get('pixlet_path'))
pixlet_available = pixlet_path is not None
# Plugin not loaded - check Pixlet availability directly
import shutil
import platform
system = platform.system().lower()
machine = platform.machine().lower()
bin_dir = PROJECT_ROOT / 'bin' / 'pixlet'
pixlet_binary = None
if system == "linux":
if "aarch64" in machine or "arm64" in machine:
pixlet_binary = bin_dir / "pixlet-linux-arm64"
elif "x86_64" in machine or "amd64" in machine:
pixlet_binary = bin_dir / "pixlet-linux-amd64"
elif system == "darwin":
pixlet_binary = bin_dir / ("pixlet-darwin-arm64" if "arm64" in machine else "pixlet-darwin-amd64")
pixlet_available = (pixlet_binary and pixlet_binary.exists()) or shutil.which('pixlet') is not None
# Read app counts from manifest
manifest = _read_starlark_manifest()
@@ -8290,26 +8166,19 @@ def toggle_starlark_app(app_id):
def render_starlark_app(app_id):
"""Force render a Starlark app."""
try:
is_valid, err = _validate_starlark_app_path(app_id)
if not is_valid:
return jsonify({'status': 'error', 'message': err}), 400
starlark_plugin = _get_starlark_plugin()
if starlark_plugin:
app = starlark_plugin.apps.get(app_id)
if not app:
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
success = starlark_plugin._render_app(app, force=True)
if success:
return jsonify({'status': 'success', 'message': 'App rendered',
'frame_count': len(app.frames) if app.frames else 0})
return jsonify({'status': 'error', 'message': 'Failed to render app'}), 500
if not starlark_plugin:
return jsonify({'status': 'error', 'message': 'Rendering requires the main LEDMatrix service (plugin not loaded in web service)'}), 503
# Web-service context: plugin not loaded, call pixlet directly
success, status_code, error = _standalone_render_starlark_app(app_id)
app = starlark_plugin.apps.get(app_id)
if not app:
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
success = starlark_plugin._render_app(app, force=True)
if success:
return jsonify({'status': 'success', 'message': 'App rendered successfully', 'frame_count': 0}), status_code
return jsonify({'status': 'error', 'message': error or 'Render failed', 'frame_count': 0}), status_code
return jsonify({'status': 'success', 'message': 'App rendered', 'frame_count': len(app.frames) if app.frames else 0})
else:
return jsonify({'status': 'error', 'message': 'Failed to render app'}), 500
except Exception as e:
logger.exception("[Starlark] render_starlark_app failed")