Compare commits

..
Author SHA1 Message Date
ChuckandClaude Sonnet 4.6 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
ChuckandClaude Sonnet 4.6 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
ChuckandClaude Sonnet 4.6 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
ChuckandClaude Sonnet 4.6 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
ChuckandClaude Sonnet 4.6 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
ChuckandClaude Sonnet 4.6 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
278 changed files with 7385 additions and 25102 deletions
-7
View File
@@ -1,7 +0,0 @@
---
exclude_paths:
- "plugin-repos/**"
- "plugins/**"
- "assets/**"
- "test/**"
- "scripts/debug/**"
-44
View File
@@ -1,44 +0,0 @@
name: Claude Code Review
on:
pull_request:
types: [opened, synchronize, ready_for_review, reopened]
# Optional: Only run on specific file changes
# paths:
# - "src/**/*.ts"
# - "src/**/*.tsx"
# - "src/**/*.js"
# - "src/**/*.jsx"
jobs:
claude-review:
# Optional: Filter by PR author
# if: |
# github.event.pull_request.user.login == 'external-contributor' ||
# github.event.pull_request.user.login == 'new-developer' ||
# github.event.pull_request.author_association == 'FIRST_TIME_CONTRIBUTOR'
runs-on: ubuntu-latest
permissions:
contents: read
pull-requests: read
issues: read
id-token: write
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 1
- name: Run Claude Code Review
id: claude-review
uses: anthropics/claude-code-action@v1
with:
claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
plugin_marketplaces: 'https://github.com/anthropics/claude-code.git'
plugins: 'code-review@claude-code-plugins'
prompt: '/code-review:code-review ${{ github.repository }}/pull/${{ github.event.pull_request.number }}'
# See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
# or https://code.claude.com/docs/en/cli-reference for available options
-50
View File
@@ -1,50 +0,0 @@
name: Claude Code
on:
issue_comment:
types: [created]
pull_request_review_comment:
types: [created]
issues:
types: [opened, assigned]
pull_request_review:
types: [submitted]
jobs:
claude:
if: |
(github.event_name == 'issue_comment' && contains(github.event.comment.body, '@claude')) ||
(github.event_name == 'pull_request_review_comment' && contains(github.event.comment.body, '@claude')) ||
(github.event_name == 'pull_request_review' && contains(github.event.review.body, '@claude')) ||
(github.event_name == 'issues' && (contains(github.event.issue.body, '@claude') || contains(github.event.issue.title, '@claude')))
runs-on: ubuntu-latest
permissions:
contents: read
pull-requests: read
issues: read
id-token: write
actions: read # Required for Claude to read CI results on PRs
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 1
- name: Run Claude Code
id: claude
uses: anthropics/claude-code-action@v1
with:
claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
# This is an optional setting that allows Claude to read CI results on PRs
additional_permissions: |
actions: read
# Optional: Give a custom prompt to Claude. If this is not specified, Claude will perform the instructions specified in the comment that tagged it.
# prompt: 'Update the pull request description to include a summary of changes.'
# Optional: Add claude_args to customize behavior and configuration
# See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
# or https://code.claude.com/docs/en/cli-reference for available options
# claude_args: '--allowed-tools Bash(gh pr *)'
-33
View File
@@ -1,33 +0,0 @@
name: Tests
on:
pull_request:
push:
branches: [main]
jobs:
plugin-safety:
name: Plugin safety harness + unit tests
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
with:
persist-credentials: false
- uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0
with:
python-version: "3.12"
cache: pip
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt -r requirements-test.txt
pip install RGBMatrixEmulator
- name: Run harness + visual rendering tests
run: |
pytest --no-cov \
test/plugins/test_harness.py \
test/plugins/test_visual_rendering.py \
test/plugins/test_plugin_matrix.py
-1
View File
@@ -8,7 +8,6 @@ config/config_secrets.json
config/config.json
config/config.json.backup
config/wifi_config.json
config/uninstalled_plugins.json
credentials.json
token.pickle
-1
View File
@@ -1,4 +1,3 @@
[submodule "rpi-rgb-led-matrix-master"]
path = rpi-rgb-led-matrix-master
url = https://github.com/hzeller/rpi-rgb-led-matrix.git
branch = master
+3 -13
View File
@@ -1,10 +1,5 @@
# LEDMatrix
[![License](https://img.shields.io/badge/license-GPL--3.0-green)](LICENSE)
[![Discord](https://img.shields.io/badge/Discord-community-5865F2?logo=discord&logoColor=white)](https://discord.gg/RdrC37rEag)
[![GitHub Stars](https://img.shields.io/github/stars/ChuckBuilds/ledmatrix?style=flat&color=yellow)](https://github.com/ChuckBuilds/ledmatrix)
[![Codacy Badge](https://app.codacy.com/project/badge/Grade/77fc9b446a5948e5b0aed7a7aaeb1bab)](https://app.codacy.com/gh/ChuckBuilds/LEDMatrix/dashboard?utm_source=gh&utm_medium=referral&utm_content=&utm_campaign=Badge_grade)
## Welcome to LEDMatrix!
Welcome to the LEDMatrix Project! This open-source project enables you to run an information-rich display on a Raspberry Pi connected to an LED RGB Matrix panel. Whether you want to see your calendar, weather forecasts, sports scores, stock prices, or any other information at a glance, LEDMatrix brings it all together.
@@ -132,15 +127,10 @@ The system supports live, recent, and upcoming game information for multiple spo
| This project can be finnicky! RGB LED Matrix displays are not built the same or to a high-quality standard. We have seen many displays arrive dead or partially working in our discord. Please purchase from a reputable vendor. |
### Raspberry Pi
- Raspberry Pi Zero's don't have enough processing power for this project.
- **Raspberry Pi 3B, 4, or 5**
- Raspberry Pi Zero's don't have enough processing power for this project and the Pi 5 is unsupported due to new GPIO output.
- **Raspberry Pi 3B or 4 (NOT RPi 5!)**
[Amazon Affiliate Link Raspberry Pi 4 4GB RAM](https://amzn.to/4dJixuX)
[Amazon Affiliate Link Raspberry Pi 4 8GB RAM](https://amzn.to/4qbqY7F)
- **Pi 5 users**: the installer automatically detects Pi 5 and builds the `rpi-rgb-led-matrix` library with RP1 support. If you previously installed on a Pi 4 and migrated the SD card, or if you see `mmap` errors in the logs, force a fresh library build:
```bash
sudo RPI_RGB_FORCE_REBUILD=1 ./first_time_install.sh
```
- Pi 5 config: leave `rp1_rio` at `0` (PIO mode, default) and set `gpio_slowdown` to `1` or `2`.
### RGB Matrix Bonnet / HAT
@@ -592,7 +582,7 @@ These settings control runtime behavior and GPIO timing:
- **Critical setting**: Must match your Raspberry Pi model for stability
- **Raspberry Pi 3**: Use 3
- **Raspberry Pi 4**: Use 4
- **Raspberry Pi 5**: Use 12 in PIO mode (`rp1_rio: 0`, the default); start with `1` and increase if you see flickering
- **Raspberry Pi 5**: Use 5 (or higher if needed)
- **Raspberry Pi Zero/1**: Use 1-2
- Incorrect values can cause display corruption, flickering, or system instability
- If you experience issues, try adjusting this value up or down by 1
Binary file not shown.

Before

Width:  |  Height:  |  Size: 102 KiB

After

Width:  |  Height:  |  Size: 48 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 111 KiB

After

Width:  |  Height:  |  Size: 90 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 96 KiB

After

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 109 KiB

After

Width:  |  Height:  |  Size: 52 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 98 KiB

After

Width:  |  Height:  |  Size: 43 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 93 KiB

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 120 KiB

After

Width:  |  Height:  |  Size: 69 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 55 KiB

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 66 KiB

After

Width:  |  Height:  |  Size: 77 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 70 KiB

After

Width:  |  Height:  |  Size: 40 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 58 KiB

After

Width:  |  Height:  |  Size: 20 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 105 KiB

After

Width:  |  Height:  |  Size: 90 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 50 KiB

After

Width:  |  Height:  |  Size: 27 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 87 KiB

After

Width:  |  Height:  |  Size: 42 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 25 KiB

After

Width:  |  Height:  |  Size: 68 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 140 KiB

After

Width:  |  Height:  |  Size: 96 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 102 KiB

After

Width:  |  Height:  |  Size: 91 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 64 KiB

After

Width:  |  Height:  |  Size: 153 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 91 KiB

After

Width:  |  Height:  |  Size: 89 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 54 KiB

After

Width:  |  Height:  |  Size: 101 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 60 KiB

After

Width:  |  Height:  |  Size: 55 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 29 KiB

After

Width:  |  Height:  |  Size: 9.8 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 30 KiB

After

Width:  |  Height:  |  Size: 18 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 126 KiB

After

Width:  |  Height:  |  Size: 103 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 54 KiB

After

Width:  |  Height:  |  Size: 94 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 93 KiB

After

Width:  |  Height:  |  Size: 92 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 60 KiB

After

Width:  |  Height:  |  Size: 59 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 41 KiB

After

Width:  |  Height:  |  Size: 80 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 77 KiB

After

Width:  |  Height:  |  Size: 38 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 140 KiB

After

Width:  |  Height:  |  Size: 111 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 32 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 33 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 20 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 16 KiB

-29
View File
@@ -1,29 +0,0 @@
# bandit.yaml — LEDMatrix bandit configuration
# https://bandit.readthedocs.io/en/latest/config.html
#
# Skips are justified by the specific codebase context documented below.
# Do not remove skips without updating the justification comment.
skips:
# B104: Binding to all interfaces (0.0.0.0)
# Intentional — the Flask server binds 0.0.0.0 for LAN access on a Raspberry Pi.
# This is not internet-facing and is documented in web_interface/app.py.
- B104
# B603: subprocess call without shell=True
# All subprocess.run() calls in this codebase use list arguments (confirmed by
# grep — zero uses of shell=True in src/ or web_interface/). List args prevent
# shell injection. See src/common/permission_utils.py for the primary usage.
- B603
# B607: Starting a process with a partial executable path
# The subprocess calls invoke system utilities (systemctl, sudo, git) by name.
# These are fixed-list invocations, not user-controlled, and rely on PATH.
- B607
exclude_dirs:
- tests
- test
- venv
- .venv
- rpi-rgb-led-matrix-master
+19 -30
View File
@@ -1,43 +1,43 @@
{
"web_display_autostart": true,
"schedule": {
"enabled": false,
"enabled": true,
"mode": "per-day",
"start_time": "07:00",
"end_time": "23:00",
"days": {
"monday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"tuesday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"wednesday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"thursday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"friday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"saturday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
},
"sunday": {
"enabled": false,
"enabled": true,
"start_time": "07:00",
"end_time": "23:00"
}
@@ -51,46 +51,46 @@
"end_time": "07:00",
"days": {
"monday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"tuesday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"wednesday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"thursday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"friday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"saturday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
},
"sunday": {
"enabled": false,
"enabled": true,
"start_time": "20:00",
"end_time": "07:00"
}
}
},
"timezone": "America/New_York",
"timezone": "America/Chicago",
"location": {
"city": "Tampa",
"state": "Florida",
"city": "Dallas",
"state": "Texas",
"country": "US"
},
"display": {
@@ -112,13 +112,7 @@
"limit_refresh_rate_hz": 100
},
"runtime": {
"gpio_slowdown": 3,
"rp1_rio": 0
},
"double_sided": {
"enabled": false,
"copies": 2,
"axis": "horizontal"
"gpio_slowdown": 3
},
"display_durations": {},
"use_short_date_format": true,
@@ -132,11 +126,6 @@
"buffer_ahead": 2
}
},
"sync": {
"role": "standalone",
"port": 5765,
"follower_position": "left"
},
"plugin_system": {
"plugins_directory": "plugin-repos",
"auto_discover": true,
-234
View File
@@ -1,234 +0,0 @@
# Adaptive Layout & Font Scaling
`src/adaptive_layout.py` lets a plugin render legibly on **any** panel size
(64x32, 128x32, 96x48, 128x64, 256x64, ...) without hand-tuned per-display
layouts. It is **opt-in**: nothing changes for plugins that don't use it.
It generalizes three patterns proven in the plugin ecosystem:
| Pattern | Origin | Core API |
|---|---|---|
| Geometry scale factor vs. a design size | f1-scoreboard | `ctx.px(base)` / `ctx.scale` |
| Breakpoint tiers | masters-tournament | `ctx.tier` / `ctx.by_tier({...})` |
| "Largest crisp font that fits" ladder | baseball-scoreboard | `ctx.fit_text(...)` and friends |
## Quick start
Every `BasePlugin` has a lazy `self.layout` (a `LayoutContext` for the
current logical display size, rebuilt automatically if the size changes)
and a one-liner `self.draw_fit(...)`:
```python
def display(self, force_clear=False):
from src.adaptive_layout import LADDER_ARCADE
b = self.layout.bounds.inset(1) # Region(0,0,W,H) minus 1px margin
rows = b.split_v(3, 1, 1, gap=1) # 3/5 for time, 1/5 each for the rest
self.draw_fit(self.time_str, rows[0], ladder=LADDER_ARCADE)
self.draw_fit(self.weekday, rows[1]) # default LADDER_GRID
self.draw_fit(self.date_str, rows[2])
self.display_manager.update_display()
```
On 128x64 the time renders at press_start 24px; on 64x32 it steps down to
8px. The rows partition the height, so bands can never overlap — no more
`y = height - 7` magic numbers.
## Region — rect algebra
`Region(x, y, w, h)` is a frozen dataclass. All carving clamps to
non-negative dimensions, so degenerate panels behave.
- Carving: `inset(dx, dy)`, `top_band(h)`, `bottom_band(h)`,
`middle(top_h, bottom_h)`, `left_col(w)`, `right_col(w)`,
`split_h(*weights, gap=0)`, `split_v(*weights, gap=0)`
- Placement: `align_xy(w, h, align, valign)`, `center_xy(w, h)`,
`contains(w, h)`, `.center`, `.right`, `.bottom`
Scoreboard-style layout:
```python
b = self.layout.bounds
status = b.top_band(self.layout.px(7))
detail = b.bottom_band(self.layout.px(7))
score_area = b.middle(status.h, detail.h)
away_slot, home_slot = b.left_col(b.h), b.right_col(b.h)
```
## Font ladders — discrete, never fractional
Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes, so
fonts are never scaled continuously. A `FontLadder` is an ordered tuple of
`FontStep(family, size_px)` rungs, largest first; fitting walks down until
the measured text fits.
- `LADDER_GRID` (default): X11 BDFs at native sizes — 10x20 → 9x18 → 9x15 →
8x13 → 7x13 → 6x13 → 6x12 → 6x10 → 6x9 → 5x8 → 5x7 → 4x6 → tom-thumb.
Body text, labels, multi-row content.
- `LADDER_ARCADE`: PressStart2P at 32/24/16/8 (integer multiples of its 8px
grid). Headline text: clocks, scores.
Custom ladders are just tuples — e.g. to add your plugin's registered font
on top: `(FontStep("myplugin::digits", 16),) + LADDER_GRID`.
## LayoutContext
Built per (width, height); exposes facts and fit queries:
- `bounds`, `width`, `height`, `aspect`
- `tier` by height (`xs`≤16, `sm`≤32, `md`≤48, `lg`≤64, `xl`) and
`width_tier` (`narrow`≤64, `normal`≤128, `wide`≤256, `ultrawide`)
- `is_wide_short` — aspect ≥ 2.5 and height ≤ 32 (the classic 128x32 shape)
- `scale``min(w/design_w, h/design_h)` vs. your manifest's
`display.design_size` (default 128x32). **Geometry only** — gaps, icon
and logo sizes via `px(base, minimum, maximum)`; fonts use ladders.
- `by_tier({"sm": 10, "lg": 18})` — value for the nearest defined tier
at-or-below the panel's tier.
- `fit_text(text, box, ladder, ellipsis=True)``FitResult` — largest rung
that fits; ellipsizes as a last resort. Cached per (text, box, ladder).
- `fit_text_proportional(text, box, base_size_px, ladder, ellipsis=True, scale=None)`
rung closest to (not exceeding) `base_size_px * scale`, still capped to
what fits the box. Use this instead of `fit_text` when several
independently-fitted elements need to stay visually harmonious as the
panel grows — `fit_text` maximizes *each one* within its own region,
which can make one element (e.g. a score with a generous box) balloon
out of proportion to a neighbor that scales by geometry (e.g. logos
sized via `px()`), even though each individual pick is "correct" in
isolation. `base_size_px` is normally the element's existing classic/
fixed font size. `scale` defaults to `self.scale` (the conservative
min-of-both-axes factor `px()` uses); pass an axis-specific value when
the surrounding composition already scales that way — e.g. a scoreboard
whose logo slots track height alone (`min(height, width // 2)`) should
size its text by `height / design_height` too, or the text reads as
under-scaled next to bigger logos on a panel that only grew taller.
- `fit_lines(lines, box, ladder, spacing)` — every line fits the width and
the stack fits the height (measures the actual strings).
- `font_for_rows(rows, box_h, ladder)` — largest rung whose line height
fits `rows` rows.
`FitResult` carries the ready-to-use `font` (drops straight into
`display_manager.draw_text(font=...)`), the possibly-ellipsized `text`,
ink `width`/`height`, `baseline`, `y_offset`, `line_height`, and `fits`.
## Adaptive images
`src/adaptive_images.py` is the image counterpart to `fit_text`, exposed as
`self.layout.fit_image(...)` (cached per panel size) and the one-liner
`self.draw_image(...)`:
```python
# Team logo: trim its transparent padding, fill the slot height (the
# football/hockey pattern), cached across frames by a stable key
self.draw_image(logo, regs.away_slot, mode="fill_height",
crop_to_ink=True, cache_key=f"logo:{abbr}")
# Album art: cover-crop a square, faces kept by the top anchor
self.draw_image(art, row.art, mode="cover", anchor="top")
# Pixel flags / sprite icons: NEAREST keeps hard edges
from src.adaptive_images import RESAMPLE_NEAREST
self.draw_image(flag, box, resample=RESAMPLE_NEAREST)
```
Modes: `contain` (letterbox, default), `cover` (crop-to-fill),
`fill_height` (logo-style), `stretch`. Unlike PIL's `thumbnail()`
(downscale-only — why imagery stays tiny on big panels) fitting **upscales
by default**; pass `upscale=False` for the legacy behavior. Results are
cached per (image, box size, options) with a bounded LRU — always pass a
stable `cache_key` (e.g. `"logo:KC"`) for images you reload. The module
also exports the Pillow-compat `RESAMPLE_LANCZOS`/`RESAMPLE_NEAREST`
constants so plugins can drop their local shims.
## Composite layouts
Pre-carved Region arrangements for the layouts plugins keep rebuilding:
```python
from src.adaptive_layout import scoreboard_regions, media_row
regs = scoreboard_regions(self.layout.bounds, ctx=self.layout)
# regs.away_slot / home_slot — logo slots (logo_slot = min(H, W // 2),
# capped so a center reserve always exists —
# see below)
# regs.status_band — top band (replaces the magic y = 1)
# regs.score_area — center gap, plus a controlled bleed into
# each logo slot (replaces y = H//2 - 3)
# regs.detail_band — bottom band (replaces y = H - 7)
# regs.bottom_left / bottom_right — record/timeout corners
row = media_row(self.layout.bounds, ctx=self.layout) # art left, text right
```
Both work on the full panel or on a scroll-mode card Region. They return
Regions and never draw — compose them with `draw_fit`/`draw_image`.
**`scoreboard_regions`'s center reserve.** The raw `logo_slot = min(H, W//2)`
formula has a blind spot: at exactly 2:1 aspect ratio (width = 2×height —
two, four, or more square modules stacked into a taller panel, e.g.
96x48, 128x64, 256x128) the two logo slots mathematically claim the
*entire* width, leaving zero pixels for a center column no matter how
big the panel gets. Wide panels (the 128x32 design baseline, 192x48,
256x32) never hit this, since height is already the tighter constraint
there. Two parameters fix it without any plugin-side code:
`min_center_fraction`/`min_center_design_px` guarantee a real minimum
center reserve at any aspect ratio, and `score_bleed_fraction` lets the
score's *fit box* extend a controlled amount into each logo slot — the
same way a real broadcast scoreboard's numbers cross slightly into the
team marks flanking them — so a short score string never has to truncate
even on the tightest aspect ratios. All three have sane defaults; override
them per call if a plugin's card proportions genuinely differ.
## Preserving user customization
Adaptive layout supplies *defaults*; explicit user configuration wins:
- **User-set fonts win.** If the plugin's config has an explicit
`font`/`font_size` for an element, load it as before and skip the ladder —
fit only when the user hasn't overridden (see the football-scoreboard
`_resolve_element_fit` pattern).
- **Offsets apply on top.** `customization.layout.<element>.{x_offset,y_offset}`
style knobs translate the *computed* region as a final step:
`region.offset(user_dx, user_dy)`. `draw_image(..., offset=(dx, dy))`
does the same for images.
- **Colors pass through.** `draw_fit`/`draw_fitted_text` take explicit
`color=` params; adaptive mode never repaints semantic or user-chosen
colors.
## Manifest declaration
Declare the size your layout was authored against so `ctx.scale` means
something:
```json
"display": { "design_size": { "width": 128, "height": 32 } }
```
Also available under `requires.display_size`: `min_width`, `min_height`,
`max_width`, `max_height`.
## Performance notes (Pi)
Fit queries are cached, so cost is O(unique strings). For per-second text
(clocks, live scores), fit on a **shape placeholder** and reuse the font:
```python
fit = self.layout.fit_text("00:00", box, ladder=LADDER_ARCADE) # cached once
self.display_manager.draw_text(current_time, font=fit.font, ...)
```
## Testing across sizes
The harness already renders every plugin at a spread of sizes (now
including 96x48):
```bash
python scripts/check_plugin.py <plugin-dir> --sizes 64x32,128x32,96x48,128x64,256x64
python scripts/render_plugin.py <plugin-dir> --width 96 --height 48
```
`BoundsCheckingDisplayManager` flags right/bottom overflow and now records
mediated draw calls with negative coordinates in
`negative_coordinate_calls` (raw-PIL draws remain uncovered).
Reference migration: the **text-display** plugin's `font_mode: "auto"`.
-6
View File
@@ -2,12 +2,6 @@
Advanced patterns, examples, and best practices for developing LEDMatrix plugins.
> **Adaptive layout:** for plugins that should render legibly on any panel
> size (fonts that grow on big panels, layouts that degrade gracefully on
> small ones), use the adaptive layout system — `self.layout`, `draw_fit`,
> `draw_image`, `scoreboard_regions` — documented in
> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
## Table of Contents
- [Using Weather Icons](#using-weather-icons)
-6
View File
@@ -48,12 +48,6 @@ display_manager.draw_text("Centered", centered=True) # Auto-center
width = display_manager.get_text_width("Text", font)
height = display_manager.get_font_height(font)
# Adaptive layout (recommended for multi-size support — text and images
# that scale to any panel; see docs/ADAPTIVE_LAYOUT.md)
rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
self.draw_fit("12:34", rows[0]) # largest crisp font that fits
self.draw_image(logo, rows[1], mode="fill_height", crop_to_ink=True)
# Weather icons
display_manager.draw_weather_icon("rain", x=10, y=10, size=16)
-6
View File
@@ -6,12 +6,6 @@ Tools for rapid plugin development without deploying to the RPi.
Interactive web UI for tweaking plugin configs and seeing the rendered display in real time.
The size inputs have a preset dropdown with the harness's standard panel
sizes, and the **All Sizes** button renders the current config at every
harness size in a side-by-side gallery (`POST /api/render-matrix`) — the
quickest way to eyeball adaptive-layout behavior across panels
(see [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md)).
### Quick Start
```bash
-7
View File
@@ -1,12 +1,5 @@
# FontManager Usage Guide
> **Picking a size automatically:** if you want the *largest font that fits
> a given area* rather than a fixed size, use the adaptive layout system's
> font ladders, which resolve through this FontManager. `BasePlugin`
> subclasses get this as `self.layout.fit_text(...)`; other code can build
> a `LayoutContext(width, height, font_manager)` directly — see
> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
## Overview
The enhanced FontManager provides comprehensive font management for the LEDMatrix application with support for:
+1
View File
@@ -248,6 +248,7 @@ test/
├── test_config_service.py # Config service tests
├── test_config_validation_edge_cases.py # Config edge cases
├── test_font_manager.py # Font manager tests
├── test_layout_manager.py # Layout manager tests
├── test_text_helper.py # Text helper tests
├── test_error_handling.py # Error handling tests
├── test_error_aggregator.py # Error aggregation tests
-5
View File
@@ -2,11 +2,6 @@
Complete API reference for plugin developers. This document describes all methods and properties available to plugins through the Display Manager, Cache Manager, and Plugin Manager.
> **Adaptive layout:** every `BasePlugin` also exposes `self.layout`,
> `self.draw_fit(text, region)` and `self.draw_image(img, region, ...)`
> the recommended way to render text and images that scale to any panel
> size. See [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
## Table of Contents
- [BasePlugin](#baseplugin)
+10 -10
View File
@@ -34,16 +34,16 @@ This document outlines the transformation of the LEDMatrix project into a modula
## Table of Contents
1. [Current Architecture Analysis](#1-current-architecture-analysis)
2. [Plugin System Design](#2-plugin-system-design)
3. [Plugin Store & Discovery](#3-plugin-store--discovery)
4. [Web UI Transformation](#4-web-ui-transformation)
5. [Migration Strategy](#5-migration-strategy)
6. [Plugin Developer Guidelines](#6-plugin-developer-guidelines)
7. [Technical Implementation Details](#7-technical-implementation-details)
8. [Best Practices & Standards](#8-best-practices--standards)
9. [Security Considerations](#9-security-considerations)
10. [Implementation Roadmap](#10-implementation-roadmap)
1. [Current Architecture Analysis](#current-architecture-analysis)
2. [Plugin System Design](#plugin-system-design)
3. [Plugin Store & Discovery](#plugin-store--discovery)
4. [Web UI Transformation](#web-ui-transformation)
5. [Migration Strategy](#migration-strategy)
6. [Plugin Developer Guidelines](#plugin-developer-guidelines)
7. [Technical Implementation Details](#technical-implementation-details)
8. [Best Practices & Standards](#best-practices--standards)
9. [Security Considerations](#security-considerations)
10. [Implementation Roadmap](#implementation-roadmap)
---
-8
View File
@@ -2,14 +2,6 @@
This guide explains how to set up a development workflow for plugins that are maintained in separate Git repositories while still being able to test them within the LEDMatrix project.
> **Rendering guidance:** plugins should read the display size dynamically
> (`self.display_manager.matrix.width/height`) rather than hardcoding one
> panel. For plugins that want to *scale* their layout to any panel, the
> opt-in adaptive layout system ([ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md))
> provides the shared helpers — fonts, images, and composite layouts that
> scale. Existing plugins keep their classic rendering unless they adopt
> those APIs; nothing migrates automatically.
## Overview
When developing plugins in separate repositories, you need a way to:
-136
View File
@@ -1,136 +0,0 @@
# Plugin Safety Harness
Renders a plugin across **every declared screen (mode)** and **a spread of
matrix sizes**, and fails if any combination crashes, draws past the panel edge,
or — for plugins that ship golden images — drifts visually. The goal: change a
plugin without breaking a size or screen you didn't think to test.
## Sizes: a sample, not a fixed list
There is **no fixed set of supported panel sizes** — an RGB matrix build can be
any width/height and configuration (square, rectangle, 2×2, 4×4, 8×2, long
strips, tall stacks). Plugins are expected to read dimensions dynamically
(`self.display_manager.matrix.width/height`) and lay themselves out
accordingly, so a hardcoded coordinate or unscaled font shows up as a failure
here.
The harness therefore renders against a **representative sample** that spans the
axes of variation (`DEFAULT_TEST_SIZES` in `src/plugin_system/testing/sizes.py`),
not an authoritative list:
Each module is 64×32; entries are real panel-grid arrangements (cols × rows):
| Size | Grid | Why it's in the sample |
|---------|------|--------------------------------------------|
| 64×32 | 1×1 | single panel — tightest common rectangle |
| 128×32 | 2×1 | the baseline most plugins are tuned for |
| 64×64 | 1×2 | stacked — tall-narrow centering |
| 128×64 | 2×2 | block — icon scaling / vertical centering |
| 256×32 | 4×1 | long strip — wide horizontal layout |
| 128×96 | 2×3 | tall — vertical overflow |
| 256×128 | 4×4 | large block — both dimensions big at once |
**Override the sizes entirely** to test your actual hardware (or any shape):
```bash
# CLI — one-off:
python scripts/check_plugin.py --plugin clock-simple --sizes 8x16,64x64,256x32
# pytest — force every plugin onto your panel(s):
LEDMATRIX_TEST_SIZES="8x16,128x128" pytest test/plugins/test_plugin_matrix.py
# Per-plugin — declare the shapes a plugin targets in its test/harness.json:
# { "sizes": [[8, 16], [64, 64]] }
```
Precedence: `LEDMATRIX_TEST_SIZES` env (global) → per-plugin `harness.json`
`sizes` → the default sample. Bounds checking adapts to whatever sizes a run
uses — the backing canvas is padded out to the **largest** panel in the run, so
a coordinate meant for a big build is still caught when rendering a small one.
## Quick start
```bash
# Functional + bounds check across all sizes/screens:
python scripts/check_plugin.py --plugin clock-simple
# Every discovered plugin:
python scripts/check_plugin.py --all
# Dump PNGs to eyeball each size/screen:
python scripts/check_plugin.py --plugin ledmatrix-weather --out-dir /tmp/preview
```
Exit code is non-zero if any `(plugin, size, screen)` fails. Plugins are
discovered in `plugin-repos/` and `plugins/` (override with `--plugin-dir`).
## What it checks (Phase 1 — always on)
1. **Loads** and builds its mode list.
2. **Renders every screen** at every size without raising. `update()` may fail
(no network in CI) and is tolerated; a crash in `display()` is a failure —
`display()` must handle the no-data state.
3. **Bounds**: nothing is drawn past the right/bottom edge. Implemented by
`BoundsCheckingDisplayManager`, which backs the declared panel with an
oversized canvas and flags any pixels that land in the margin. (Left/top
overflow at negative coordinates and BDF text are not flagged — golden images
cover those.)
## Golden images (Phase 2 — opt-in per plugin)
A plugin opts in by committing reference PNGs and (usually) a small harness spec:
```
plugins/<id>/test/harness.json # how to render deterministically
plugins/<id>/test/fixtures/mock.json # optional cached data
plugins/<id>/test/golden/<WxH>/<mode>.png
```
`test/harness.json` keys (all optional):
```json
{
"config": { "timezone": "UTC" },
"mock_data": "fixtures/mock.json",
"freeze_time": "2025-08-01 15:25:00",
"skip_update": false,
"sizes": [[128, 32], [128, 64]]
}
```
Generate / refresh goldens after an intentional visual change, then review the
diff before committing:
```bash
python scripts/check_plugin.py --plugin clock-simple --update-golden \
--config '{"timezone":"UTC"}' --freeze-time "2025-08-01 15:25:00"
```
Comparison is exact by default (`compare_images` in `harness.py` accepts a
tolerance for known anti-aliasing noise). Determinism requires a pinned Pillow
and the bundled fonts — keep both stable when regenerating goldens.
## Tests & CI
- `test/plugins/test_harness.py` — unit tests for bounds detection, image
comparison, and mode enumeration (run anywhere).
- `test/plugins/test_plugin_matrix.py` — parametrized over discovered plugins ×
sizes × screens; honors each plugin's `test/harness.json` and goldens. Skips
when no plugins are present (e.g. a fresh core checkout); set
`LEDMATRIX_REQUIRE_PLUGINS=1` in a pipeline where plugins must be present to
turn an empty discovery into a hard failure instead. Point it at the monorepo
with `LEDMATRIX_PLUGINS_DIR=/path/to/ledmatrix-plugins/plugins`.
- `.github/workflows/test.yml` — runs the harness + visual tests on every PR.
The plugin monorepo has its own `Plugin Safety` workflow that runs this harness
against changed plugins on every PR.
## Developer workflow
1. Change the plugin on a branch.
2. `python scripts/check_plugin.py --plugin <id> --out-dir /tmp/preview` and
eyeball the PNGs.
3. Intentional visual change? `--update-golden`, review diffs, commit goldens.
4. (Monorepo) bump `manifest.json` version and let the pre-commit hook sync
`plugins.json`.
5. Push — CI re-runs the harness across all sizes and gates the PR.
-92
View File
@@ -10,98 +10,6 @@ The LEDMatrix Widget Registry system allows plugins to use reusable UI component
## Available Core Widgets
### Plugin File Manager Widget (`plugin-file-manager`)
Full inline file management UI for plugins that manage files via the `web_ui_actions` system. Renders a card grid, upload zone, create/delete modals, and an entry table editor — entirely inline, no iframe.
`plugin_id` is **automatically injected** from template context. File operations call `/api/v3/plugins/action` immediately on user action; no Save Configuration needed.
**Schema Configuration:**
```json
{
"file_manager": {
"type": "null",
"title": "Data Files",
"x-widget": "plugin-file-manager",
"x-widget-config": {
"actions": {
"list": "list-files",
"get": "get-file",
"save": "save-file",
"upload": "upload-file",
"delete": "delete-file",
"create": "create-file",
"toggle": "toggle-category"
},
"upload_hint": "JSON files with day numbers 1365 as keys",
"directory_label": "my_data/",
"create_fields": [
{ "key": "category_name", "label": "Category Name",
"placeholder": "e.g., my_words", "pattern": "^[a-z0-9_]+$",
"hint": "Lowercase letters, numbers, underscores" },
{ "key": "display_name", "label": "Display Name",
"placeholder": "e.g., My Words", "hint": "Optional" }
]
}
}
}
```
**`list` is required** — the widget calls it on render to populate the file grid; omitting it leaves the widget stuck in a loading state. All other actions are optional — omit any key to hide its UI element (e.g., no `create` = no New File button, no `toggle` = no enable/disable switch).
The edit view auto-detects whether file content is tabular (object-of-objects with uniform keys) and shows a paginated table editor with inline cells. Otherwise falls back to a JSON textarea.
**Used by:** of-the-day
---
### Time Picker Widget (`time-picker`)
Single time selection using the browser's native time input. Returns a string in `HH:MM` (24-hour) format. Generic — works in any plugin without configuration.
**Schema Configuration:**
```json
{
"target_time": {
"type": "string",
"x-widget": "time-picker",
"default": "00:00",
"x-options": {
"placeholder": "Select time",
"clearable": true
}
}
}
```
**Used by:** countdown
---
### File Upload Single Widget (`file-upload-single`)
Single-image upload for string fields. Uploads to the plugin's asset folder (`assets/plugins/<plugin_id>/uploads/`) and sets the string field value to the returned relative path. Shows a thumbnail preview and a clear button. The `plugin_id` is **automatically injected** from the template context — no need to specify it in the schema.
**Schema Configuration:**
```json
{
"image_path": {
"type": "string",
"x-widget": "file-upload-single",
"x-upload-config": {
"allowed_types": ["image/png", "image/jpeg", "image/bmp", "image/gif"],
"max_size_mb": 5
}
}
}
```
Note: Unlike `file-upload` (array-level), this widget is for a single `string` field. It is ideal for per-item images inside `array-table` rows.
**Used by:** countdown
---
### File Upload Widget (`file-upload`)
Upload and manage image files with drag-and-drop support, preview, delete, and scheduling.
+40 -137
View File
@@ -15,8 +15,8 @@ on_error() {
echo "✗ An error occurred during: $CURRENT_STEP (line $line_no, exit $exit_code)" >&2
if [ -n "${LOG_FILE:-}" ]; then
echo "See the log for details: $LOG_FILE" >&2
echo "-- Last 100 lines from log --" >&2
tail -n 100 "$LOG_FILE" >&2 || true
echo "-- Last 50 lines from log --" >&2
tail -n 50 "$LOG_FILE" >&2 || true
fi
echo "\nCommon fixes:" >&2
echo "- Ensure the Pi is online (try: ping -c1 8.8.8.8)." >&2
@@ -36,17 +36,9 @@ if [ -r /proc/device-tree/model ]; then
DEVICE_MODEL=$(tr -d '\0' </proc/device-tree/model)
echo "Detected device: $DEVICE_MODEL"
else
DEVICE_MODEL=""
echo "⚠ Could not detect Raspberry Pi model (continuing anyway)"
fi
# Detect Pi 5 for hardware-specific install decisions (RP1 library verification)
IS_PI5=0
if echo "${DEVICE_MODEL:-}" | grep -qi "Raspberry Pi 5"; then
IS_PI5=1
echo "Raspberry Pi 5 detected — will verify RP1 library support."
fi
# Check OS version - must be Raspberry Pi OS Lite (Trixie)
echo ""
echo "Checking operating system requirements..."
@@ -202,33 +194,8 @@ retry() {
done
}
# Wait for another apt/dpkg process (commonly unattended-upgrades running
# shortly after first boot) to release its lock before we try apt ourselves.
# Without this, apt_update/apt_install can fail outright in the first couple
# minutes after a fresh Pi OS boot with a generic "Command failed after 3
# attempts" error.
wait_for_apt_lock() {
command -v flock >/dev/null 2>&1 || return 0
local lock_file="/var/lib/dpkg/lock-frontend"
local max_wait=180
local waited=0
local printed=0
while ! flock -n "$lock_file" -c true 2>/dev/null; do
if [ "$printed" -eq 0 ]; then
echo "⚠ Waiting for another apt/dpkg process to finish (e.g. unattended-upgrades on first boot)..."
printed=1
fi
if [ "$waited" -ge "$max_wait" ]; then
echo "⚠ Still waiting after ${max_wait}s; proceeding anyway."
break
fi
sleep 5
waited=$((waited+5))
done
}
apt_update() { wait_for_apt_lock; retry apt-get -o DPkg::Lock::Timeout=180 update; }
apt_install() { wait_for_apt_lock; retry apt-get -o DPkg::Lock::Timeout=180 install -y "$@"; }
apt_update() { retry apt update; }
apt_install() { retry apt install -y "$@"; }
apt_remove() { apt-get remove -y "$@" || true; }
check_network() {
@@ -247,22 +214,6 @@ check_network() {
exit 1
}
check_disk_space() {
command -v df >/dev/null 2>&1 || return 0
local available_mb
available_mb=$(df -m "$PROJECT_ROOT_DIR" | awk 'NR==2{print $4}')
available_mb=${available_mb:-0}
if [ "$available_mb" -lt 500 ]; then
echo "✗ ERROR: Insufficient disk space: ${available_mb}MB available (need at least 500MB)"
echo " Free up space first, e.g.: sudo apt clean && sudo apt autoremove"
exit 1
elif [ "$available_mb" -lt 1024 ]; then
echo "⚠ Limited disk space: ${available_mb}MB available (recommend at least 1GB for the rpi-rgb-led-matrix build in Step 6)"
else
echo "✓ Disk space sufficient: ${available_mb}MB available"
fi
}
echo ""
echo "This script will perform the following steps:"
echo "1. Install system dependencies"
@@ -308,20 +259,21 @@ else
fi
echo ""
CLEAR='
'
CURRENT_STEP="Install system dependencies"
echo "Step 1: Installing system dependencies..."
echo "----------------------------------------"
# Pre-flight checks before APT operations
# Ensure network is available before APT operations
check_network
check_disk_space
# Update package list
apt_update
# Install required system packages
echo "Installing Python packages and dependencies..."
apt_install python3-pip python3-venv python-dev-is-python3 python3-pil python3-pil.imagetk build-essential python3-setuptools python3-wheel cmake ninja-build
apt_install python3-pip python3-venv python3-dev python3-pil python3-pil.imagetk build-essential python3-setuptools python3-wheel cython3 scons cmake ninja-build
# Install additional system dependencies that might be needed
echo "Installing additional system dependencies..."
@@ -719,6 +671,8 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
echo "[$PACKAGE_NUM/$TOTAL_PACKAGES] Installing: $line"
# Check if package is already installed (basic check - may not catch all cases)
PACKAGE_NAME=$(echo "$line" | sed -E 's/[<>=!].*$//' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')
# Try installing with verbose output and timeout (if available)
# Use --no-cache-dir to avoid cache issues, --verbose for diagnostics
INSTALL_OUTPUT=$(mktemp)
@@ -726,11 +680,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
if command -v timeout >/dev/null 2>&1; then
# Use timeout if available (10 minutes = 600 seconds)
# --ignore-installed: apt-managed packages (e.g. python3-requests)
# ship no pip RECORD file, so upgrading them would otherwise abort
# with "uninstall-no-record-file"; this lays the new version down
# alongside instead of trying to uninstall the apt copy first.
if timeout 600 python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
if timeout 600 python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
INSTALL_SUCCESS=true
else
EXIT_CODE=$?
@@ -738,7 +688,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
echo "✗ Timeout (10 minutes) installing: $line"
echo " This package may require building from source, which can be slow on Raspberry Pi."
echo " You can try installing it manually later with:"
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose '$line'"
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose '$line'"
else
echo "✗ Failed to install: $line (exit code: $EXIT_CODE)"
fi
@@ -746,7 +696,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
else
# No timeout command available, install without timeout
echo " Note: timeout command not available, installation may take a while..."
if python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
if python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
INSTALL_SUCCESS=true
else
EXIT_CODE=$?
@@ -798,7 +748,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
echo " 1. Ensure you have enough disk space: df -h"
echo " 2. Check available memory: free -h"
echo " 3. Try installing failed packages individually with verbose output:"
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose <package>"
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose <package>"
echo " 4. For packages that build from source (like numpy), consider:"
echo " - Installing pre-built wheels: python3 -m pip install --only-binary :all: <package>"
echo " - Or installing via apt if available: sudo apt install python3-<package>"
@@ -820,10 +770,7 @@ echo ""
# Install web interface dependencies
echo "Installing web interface dependencies..."
if [ -f "$PROJECT_ROOT_DIR/web_interface/requirements.txt" ]; then
# --ignore-installed: apt-managed packages (e.g. python3-requests) ship no
# pip RECORD file, so upgrading them to the version pinned here would
# otherwise abort the whole install with "uninstall-no-record-file".
if python3 -m pip install --break-system-packages --prefer-binary --ignore-installed -r "$PROJECT_ROOT_DIR/web_interface/requirements.txt"; then
if python3 -m pip install --break-system-packages --prefer-binary -r "$PROJECT_ROOT_DIR/web_interface/requirements.txt"; then
echo "✓ Web interface dependencies installed"
# Create marker file to indicate dependencies are installed
touch "$PROJECT_ROOT_DIR/.web_deps_installed"
@@ -840,54 +787,29 @@ CURRENT_STEP="Build and install rpi-rgb-led-matrix"
echo "Step 6: Building and installing rpi-rgb-led-matrix..."
echo "-----------------------------------------------------"
# On Pi 5, also check that the installed library has rp1_rio support.
# A library built before Pi 5 support was added imports fine but maps to the
# Pi 3 peripheral bus address (0x3f000000) instead of the RP1 chip at runtime.
_HAS_RP1=0
if python3 -c 'from rgbmatrix import RGBMatrixOptions; assert hasattr(RGBMatrixOptions(), "rp1_rio")' >/dev/null 2>&1; then
_HAS_RP1=1
fi
_SKIP_BUILD=0
# If already installed and not forcing rebuild, skip expensive build
if python3 -c 'from rgbmatrix import RGBMatrix, RGBMatrixOptions' >/dev/null 2>&1 && [ "${RPI_RGB_FORCE_REBUILD:-0}" != "1" ]; then
if [ "$IS_PI5" = "1" ] && [ "$_HAS_RP1" = "0" ]; then
echo "⚠ Pi 5 detected: installed rgbmatrix lacks rp1_rio support (older build)."
echo " Forcing rebuild to get Pi 5 RP1 support..."
else
_SKIP_BUILD=1
fi
fi
if [ "$_SKIP_BUILD" = "1" ]; then
_skip_suffix=""
if [ "$IS_PI5" = "1" ]; then _skip_suffix=" with Pi 5 RP1 support"; fi
echo "rgbmatrix already installed${_skip_suffix}; skipping build (set RPI_RGB_FORCE_REBUILD=1 to force rebuild)."
echo "rgbmatrix Python package already available; skipping build (set RPI_RGB_FORCE_REBUILD=1 to force rebuild)."
else
# Ensure rpi-rgb-led-matrix submodule is initialized
# Wrapper used with retry(): removes any partial clone dir before each attempt
# so git clone doesn't fail with "destination path already exists".
_clone_rpi_rgb() {
rm -rf "$PROJECT_ROOT_DIR/rpi-rgb-led-matrix-master"
git clone https://github.com/hzeller/rpi-rgb-led-matrix.git rpi-rgb-led-matrix-master
}
if [ ! -d "$PROJECT_ROOT_DIR/rpi-rgb-led-matrix-master" ]; then
echo "rpi-rgb-led-matrix-master not found. Initializing git submodule..."
cd "$PROJECT_ROOT_DIR"
# Try to initialize submodule if .gitmodules exists
if [ -f "$PROJECT_ROOT_DIR/.gitmodules" ] && grep -q "rpi-rgb-led-matrix" "$PROJECT_ROOT_DIR/.gitmodules"; then
echo "Initializing rpi-rgb-led-matrix submodule..."
if ! retry git submodule update --init --recursive rpi-rgb-led-matrix-master; then
if ! git submodule update --init --recursive rpi-rgb-led-matrix-master 2>&1; then
echo "⚠ Submodule init failed, cloning directly from GitHub..."
retry _clone_rpi_rgb
git clone https://github.com/hzeller/rpi-rgb-led-matrix.git rpi-rgb-led-matrix-master
fi
else
# Fallback: clone directly if submodule not configured
echo "Submodule not configured, cloning directly from GitHub..."
retry _clone_rpi_rgb
git clone https://github.com/hzeller/rpi-rgb-led-matrix.git rpi-rgb-led-matrix-master
fi
fi
# Build and install rpi-rgb-led-matrix Python bindings
if [ -d "$PROJECT_ROOT_DIR/rpi-rgb-led-matrix-master" ]; then
# Check if submodule is properly initialized (not empty)
@@ -896,34 +818,30 @@ else
cd "$PROJECT_ROOT_DIR"
rm -rf rpi-rgb-led-matrix-master
if [ -f "$PROJECT_ROOT_DIR/.gitmodules" ] && grep -q "rpi-rgb-led-matrix" "$PROJECT_ROOT_DIR/.gitmodules"; then
retry git submodule update --init --recursive rpi-rgb-led-matrix-master
git submodule update --init --recursive rpi-rgb-led-matrix-master
else
retry _clone_rpi_rgb
git clone https://github.com/hzeller/rpi-rgb-led-matrix.git rpi-rgb-led-matrix-master
fi
fi
pushd "$PROJECT_ROOT_DIR/rpi-rgb-led-matrix-master" >/dev/null
echo "Installing rpi-rgb-led-matrix Python package (scikit-build-core + cmake)..."
echo " Build deps required: python-dev-is-python3 cmake"
echo " This compiles C++ — may take 2-5 minutes on Pi 4/5..."
BUILD_OUTPUT=$(mktemp)
BUILD_SUCCESS=false
if python3 -m pip install --break-system-packages . > "$BUILD_OUTPUT" 2>&1; then
BUILD_SUCCESS=true
echo "Building rpi-rgb-led-matrix Python bindings..."
# Build the library first, then Python bindings
# The build-python target depends on the library being built
if ! make build-python; then
echo "✗ Failed to build rpi-rgb-led-matrix Python bindings"
echo " Make sure you have the required build tools installed:"
echo " sudo apt install -y build-essential python3-dev cython3 scons"
popd >/dev/null
exit 1
fi
cat "$BUILD_OUTPUT" >> "$LOG_FILE"
if [ "$BUILD_SUCCESS" != true ]; then
cd bindings/python
echo "Installing rpi-rgb-led-matrix Python package via pip..."
if ! python3 -m pip install --break-system-packages .; then
echo "✗ Failed to install rpi-rgb-led-matrix Python package"
echo " Ensure build tools are installed:"
echo " sudo apt install -y python-dev-is-python3 cmake build-essential"
echo ""
echo "-- Last 50 lines of build output --"
tail -n 50 "$BUILD_OUTPUT"
rm -f "$BUILD_OUTPUT"
popd >/dev/null
exit 1
fi
rm -f "$BUILD_OUTPUT"
popd >/dev/null
else
echo "✗ rpi-rgb-led-matrix-master directory not found at $PROJECT_ROOT_DIR"
@@ -945,17 +863,6 @@ except Exception as e:
PY
then
echo "✓ rpi-rgb-led-matrix installed and verified"
# Pi 5: confirm the freshly-built library has rp1_rio support
if [ "$IS_PI5" = "1" ]; then
if python3 -c 'from rgbmatrix import RGBMatrixOptions; assert hasattr(RGBMatrixOptions(), "rp1_rio")' >/dev/null 2>&1; then
echo "✓ Pi 5 RP1 (rp1_rio) support confirmed"
else
echo "⚠ rp1_rio not found after rebuild — the submodule may be an older version."
echo " Try updating the submodule and rebuilding:"
echo " git submodule update --remote rpi-rgb-led-matrix-master"
echo " sudo RPI_RGB_FORCE_REBUILD=1 ./first_time_install.sh"
fi
fi
else
echo "✗ rpi-rgb-led-matrix import test failed"
exit 1
@@ -978,15 +885,11 @@ else
# Try to install dependencies using the smart installer if available
if [ -f "$PROJECT_ROOT_DIR/scripts/install_dependencies_apt.py" ]; then
echo "Using smart dependency installer..."
# -u: unbuffered stdout/stderr so output is captured in $LOG_FILE in
# real time and in order relative to this script's own echo statements
python3 -u "$PROJECT_ROOT_DIR/scripts/install_dependencies_apt.py"
python3 "$PROJECT_ROOT_DIR/scripts/install_dependencies_apt.py"
else
echo "Using pip to install dependencies..."
if [ -f "$PROJECT_ROOT_DIR/requirements_web_v2.txt" ]; then
# --ignore-installed: see the Step 5 web_interface/requirements.txt
# install above — same apt/pip RECORD-file conflict applies here.
python3 -m pip install --break-system-packages --prefer-binary --ignore-installed -r requirements_web_v2.txt
python3 -m pip install --break-system-packages --prefer-binary -r requirements_web_v2.txt
else
echo "⚠ requirements_web_v2.txt not found; skipping web dependency install"
fi
@@ -1576,7 +1479,7 @@ echo "WiFi Connection Status:"
if command -v nmcli >/dev/null 2>&1; then
WIFI_STATUS=$(nmcli -t -f DEVICE,TYPE,STATE device status 2>/dev/null | grep -i wifi || echo "")
if [ -n "$WIFI_STATUS" ]; then
echo "$WIFI_STATUS" | while IFS=':' read -r _ _ state; do
echo "$WIFI_STATUS" | while IFS=':' read -r device type state; do
if [ "$state" = "connected" ]; then
SSID=$(nmcli -t -f active,ssid device wifi 2>/dev/null | grep "^yes:" | cut -d: -f2 | head -1)
if [ -n "$SSID" ]; then
@@ -0,0 +1,138 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "March Madness Plugin Configuration",
"type": "object",
"properties": {
"enabled": {
"type": "boolean",
"default": false,
"description": "Enable the March Madness tournament display"
},
"leagues": {
"type": "object",
"title": "Tournament Leagues",
"description": "Which NCAA tournaments to display",
"properties": {
"ncaam": {
"type": "boolean",
"default": true,
"description": "Show NCAA Men's Tournament games"
},
"ncaaw": {
"type": "boolean",
"default": true,
"description": "Show NCAA Women's Tournament games"
}
},
"additionalProperties": false
},
"favorite_teams": {
"type": "array",
"title": "Favorite Teams",
"description": "Team abbreviations to highlight (e.g., DUKE, UNC). Leave empty to show all teams equally.",
"items": {
"type": "string"
},
"uniqueItems": true,
"default": []
},
"display_options": {
"type": "object",
"title": "Display Options",
"x-collapsed": true,
"properties": {
"show_seeds": {
"type": "boolean",
"default": true,
"description": "Show tournament seeds (1-16) next to team names"
},
"show_round_logos": {
"type": "boolean",
"default": true,
"description": "Show round logo separators between game groups"
},
"highlight_upsets": {
"type": "boolean",
"default": true,
"description": "Highlight upset winners (higher seed beating lower seed) in gold"
},
"show_bracket_progress": {
"type": "boolean",
"default": true,
"description": "Show which teams are still alive in each region"
},
"scroll_speed": {
"type": "number",
"default": 1.0,
"minimum": 0.5,
"maximum": 5.0,
"description": "Scroll speed (pixels per frame)"
},
"scroll_delay": {
"type": "number",
"default": 0.02,
"minimum": 0.001,
"maximum": 0.1,
"description": "Delay between scroll frames (seconds)"
},
"target_fps": {
"type": "integer",
"default": 120,
"minimum": 30,
"maximum": 200,
"description": "Target frames per second"
},
"loop": {
"type": "boolean",
"default": true,
"description": "Loop the scroll continuously"
},
"dynamic_duration": {
"type": "boolean",
"default": true,
"description": "Automatically adjust display duration based on content width"
},
"min_duration": {
"type": "integer",
"default": 30,
"minimum": 10,
"maximum": 300,
"description": "Minimum display duration in seconds"
},
"max_duration": {
"type": "integer",
"default": 300,
"minimum": 30,
"maximum": 600,
"description": "Maximum display duration in seconds"
}
},
"additionalProperties": false
},
"data_settings": {
"type": "object",
"title": "Data Settings",
"x-collapsed": true,
"properties": {
"update_interval": {
"type": "integer",
"default": 300,
"minimum": 60,
"maximum": 3600,
"description": "How often to refresh tournament data (seconds). Automatically shortens to 60s when live games are detected."
},
"request_timeout": {
"type": "integer",
"default": 30,
"minimum": 5,
"maximum": 60,
"description": "API request timeout in seconds"
}
},
"additionalProperties": false
}
},
"required": ["enabled"],
"additionalProperties": false,
"x-propertyOrder": ["enabled", "leagues", "favorite_teams", "display_options", "data_settings"]
}
+910
View File
@@ -0,0 +1,910 @@
"""March Madness Plugin — NCAA Tournament bracket tracker for LED Matrix.
Displays a horizontally-scrolling ticker of NCAA Tournament games grouped by
round, with seeds, round logos, live scores, and upset highlighting.
"""
import re
import threading
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
import numpy as np
import pytz
import requests
from PIL import Image, ImageDraw, ImageFont
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from src.plugin_system.base_plugin import BasePlugin
try:
from src.common.scroll_helper import ScrollHelper
except ImportError:
ScrollHelper = None
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
SCOREBOARD_URLS = {
"ncaam": "https://site.api.espn.com/apis/site/v2/sports/basketball/mens-college-basketball/scoreboard",
"ncaaw": "https://site.api.espn.com/apis/site/v2/sports/basketball/womens-college-basketball/scoreboard",
}
ROUND_ORDER = {"NCG": 0, "F4": 1, "E8": 2, "S16": 3, "R32": 4, "R64": 5, "": 6}
ROUND_DISPLAY_NAMES = {
"NCG": "Championship",
"F4": "Final Four",
"E8": "Elite Eight",
"S16": "Sweet Sixteen",
"R32": "Round of 32",
"R64": "Round of 64",
}
ROUND_LOGO_FILES = {
"NCG": "CHAMPIONSHIP.png",
"F4": "FINAL_4.png",
"E8": "ELITE_8.png",
"S16": "SWEET_16.png",
"R32": "ROUND_32.png",
"R64": "ROUND_64.png",
}
REGION_ORDER = {"E": 0, "W": 1, "S": 2, "MW": 3, "": 4}
# Colors
COLOR_WHITE = (255, 255, 255)
COLOR_GOLD = (255, 215, 0)
COLOR_GRAY = (160, 160, 160)
COLOR_DIM = (100, 100, 100)
COLOR_RED = (255, 60, 60)
COLOR_GREEN = (60, 200, 60)
COLOR_BLACK = (0, 0, 0)
COLOR_DARK_BG = (20, 20, 20)
# ---------------------------------------------------------------------------
# Plugin Class
# ---------------------------------------------------------------------------
class MarchMadnessPlugin(BasePlugin):
"""NCAA March Madness tournament bracket tracker."""
def __init__(
self,
plugin_id: str,
config: Dict[str, Any],
display_manager: Any,
cache_manager: Any,
plugin_manager: Any,
):
super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
# Config
leagues_config = config.get("leagues", {})
self.show_ncaam: bool = leagues_config.get("ncaam", True)
self.show_ncaaw: bool = leagues_config.get("ncaaw", True)
self.favorite_teams: List[str] = [t.upper() for t in config.get("favorite_teams", [])]
display_options = config.get("display_options", {})
self.show_seeds: bool = display_options.get("show_seeds", True)
self.show_round_logos: bool = display_options.get("show_round_logos", True)
self.highlight_upsets: bool = display_options.get("highlight_upsets", True)
self.show_bracket_progress: bool = display_options.get("show_bracket_progress", True)
self.scroll_speed: float = display_options.get("scroll_speed", 1.0)
self.scroll_delay: float = display_options.get("scroll_delay", 0.02)
self.target_fps: int = display_options.get("target_fps", 120)
self.loop: bool = display_options.get("loop", True)
self.dynamic_duration_enabled: bool = display_options.get("dynamic_duration", True)
self.min_duration: int = display_options.get("min_duration", 30)
self.max_duration: int = display_options.get("max_duration", 300)
if self.min_duration > self.max_duration:
self.logger.warning(
f"min_duration ({self.min_duration}) > max_duration ({self.max_duration}); swapping values"
)
self.min_duration, self.max_duration = self.max_duration, self.min_duration
data_settings = config.get("data_settings", {})
self.update_interval: int = data_settings.get("update_interval", 300)
self.request_timeout: int = data_settings.get("request_timeout", 30)
# Scrolling flag for display controller
self.enable_scrolling = True
# State
self.games_data: List[Dict] = []
self.ticker_image: Optional[Image.Image] = None
self.last_update: float = 0
self.dynamic_duration: float = 60
self.total_scroll_width: int = 0
self._display_start_time: Optional[float] = None
self._end_reached_logged: bool = False
self._update_lock = threading.Lock()
self._has_live_games: bool = False
self._cached_dynamic_duration: Optional[float] = None
self._duration_cache_time: float = 0
# Display dimensions
self.display_width: int = self.display_manager.matrix.width
self.display_height: int = self.display_manager.matrix.height
# HTTP session with retry
self.session = requests.Session()
retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
self.session.mount("https://", HTTPAdapter(max_retries=retry))
self.headers = {"User-Agent": "LEDMatrix/2.0"}
# ScrollHelper
if ScrollHelper:
self.scroll_helper = ScrollHelper(self.display_width, self.display_height, logger=self.logger)
if hasattr(self.scroll_helper, "set_frame_based_scrolling"):
self.scroll_helper.set_frame_based_scrolling(True)
self.scroll_helper.set_scroll_speed(self.scroll_speed)
self.scroll_helper.set_scroll_delay(self.scroll_delay)
if hasattr(self.scroll_helper, "set_target_fps"):
self.scroll_helper.set_target_fps(self.target_fps)
self.scroll_helper.set_dynamic_duration_settings(
enabled=self.dynamic_duration_enabled,
min_duration=self.min_duration,
max_duration=self.max_duration,
buffer=0.1,
)
else:
self.scroll_helper = None
self.logger.warning("ScrollHelper not available")
# Fonts
self.fonts = self._load_fonts()
# Logos
self._round_logos: Dict[str, Image.Image] = {}
self._team_logo_cache: Dict[str, Optional[Image.Image]] = {}
self._march_madness_logo: Optional[Image.Image] = None
self._load_round_logos()
self.logger.info(
f"MarchMadnessPlugin initialized — NCAAM: {self.show_ncaam}, "
f"NCAAW: {self.show_ncaaw}, favorites: {self.favorite_teams}"
)
# ------------------------------------------------------------------
# Fonts
# ------------------------------------------------------------------
def _load_fonts(self) -> Dict[str, ImageFont.FreeTypeFont]:
fonts = {}
try:
fonts["score"] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 10)
except IOError:
fonts["score"] = ImageFont.load_default()
try:
fonts["time"] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
except IOError:
fonts["time"] = ImageFont.load_default()
try:
fonts["detail"] = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6)
except IOError:
fonts["detail"] = ImageFont.load_default()
return fonts
# ------------------------------------------------------------------
# Logo loading
# ------------------------------------------------------------------
def _load_round_logos(self) -> None:
logo_dir = Path("assets/sports/ncaa_logos")
for round_key, filename in ROUND_LOGO_FILES.items():
path = logo_dir / filename
try:
img = Image.open(path).convert("RGBA")
# Resize to fit display height
target_h = self.display_height - 4
ratio = target_h / img.height
target_w = int(img.width * ratio)
self._round_logos[round_key] = img.resize((target_w, target_h), Image.Resampling.LANCZOS)
except (OSError, ValueError) as e:
self.logger.warning(f"Could not load round logo {filename}: {e}")
except Exception:
self.logger.exception(f"Unexpected error loading round logo {filename}")
# March Madness logo
mm_path = logo_dir / "MARCH_MADNESS.png"
try:
img = Image.open(mm_path).convert("RGBA")
target_h = self.display_height - 4
ratio = target_h / img.height
target_w = int(img.width * ratio)
self._march_madness_logo = img.resize((target_w, target_h), Image.Resampling.LANCZOS)
except (OSError, ValueError) as e:
self.logger.warning(f"Could not load March Madness logo: {e}")
except Exception:
self.logger.exception("Unexpected error loading March Madness logo")
def _get_team_logo(self, abbr: str) -> Optional[Image.Image]:
if abbr in self._team_logo_cache:
return self._team_logo_cache[abbr]
logo_dir = Path("assets/sports/ncaa_logos")
path = logo_dir / f"{abbr}.png"
try:
img = Image.open(path).convert("RGBA")
target_h = self.display_height - 6
ratio = target_h / img.height
target_w = int(img.width * ratio)
img = img.resize((target_w, target_h), Image.Resampling.LANCZOS)
self._team_logo_cache[abbr] = img
return img
except (FileNotFoundError, OSError, ValueError):
self._team_logo_cache[abbr] = None
return None
except Exception:
self.logger.exception(f"Unexpected error loading team logo for {abbr}")
self._team_logo_cache[abbr] = None
return None
# ------------------------------------------------------------------
# Data fetching
# ------------------------------------------------------------------
def _is_tournament_window(self) -> bool:
today = datetime.now(pytz.utc)
return (3, 10) <= (today.month, today.day) <= (4, 10)
def _fetch_tournament_data(self) -> List[Dict]:
"""Fetch tournament games from ESPN scoreboard API."""
all_games: List[Dict] = []
leagues = []
if self.show_ncaam:
leagues.append("ncaam")
if self.show_ncaaw:
leagues.append("ncaaw")
for league_key in leagues:
url = SCOREBOARD_URLS.get(league_key)
if not url:
continue
cache_key = f"march_madness_{league_key}_scoreboard"
cache_max_age = 60 if self._has_live_games else self.update_interval
cached = self.cache_manager.get(cache_key, max_age=cache_max_age)
if cached:
all_games.extend(cached)
continue
try:
# NCAA basketball scoreboard without dates param returns current games
params = {"limit": 1000, "groups": 100}
resp = self.session.get(url, params=params, headers=self.headers, timeout=self.request_timeout)
resp.raise_for_status()
data = resp.json()
events = data.get("events", [])
league_games = []
for event in events:
game = self._parse_event(event, league_key)
if game:
league_games.append(game)
self.cache_manager.set(cache_key, league_games)
self.logger.info(f"Fetched {len(league_games)} {league_key} tournament games")
all_games.extend(league_games)
except Exception:
self.logger.exception(f"Error fetching {league_key} tournament data")
return all_games
def _parse_event(self, event: Dict, league_key: str) -> Optional[Dict]:
"""Parse an ESPN event into a game dict."""
competitions = event.get("competitions", [])
if not competitions:
return None
comp = competitions[0]
# Confirm tournament game
comp_type = comp.get("type", {})
is_tournament = comp_type.get("abbreviation") == "TRNMNT"
notes = comp.get("notes", [])
headline = ""
if notes:
headline = notes[0].get("headline", "")
if not is_tournament and "Championship" in headline:
is_tournament = True
if not is_tournament:
return None
# Status
status = comp.get("status", {}).get("type", {})
state = status.get("state", "pre")
status_detail = status.get("shortDetail", "")
# Teams
competitors = comp.get("competitors", [])
home_team = next((c for c in competitors if c.get("homeAway") == "home"), None)
away_team = next((c for c in competitors if c.get("homeAway") == "away"), None)
if not home_team or not away_team:
return None
home_abbr = home_team.get("team", {}).get("abbreviation", "???")
away_abbr = away_team.get("team", {}).get("abbreviation", "???")
home_score = home_team.get("score", "0")
away_score = away_team.get("score", "0")
# Seeds
home_seed = home_team.get("curatedRank", {}).get("current", 0)
away_seed = away_team.get("curatedRank", {}).get("current", 0)
if home_seed >= 99:
home_seed = 0
if away_seed >= 99:
away_seed = 0
# Round and region
tournament_round = self._parse_round(headline)
tournament_region = self._parse_region(headline)
# Date/time
date_str = event.get("date", "")
start_time_utc = None
game_date = ""
game_time = ""
try:
if date_str.endswith("Z"):
date_str = date_str.replace("Z", "+00:00")
dt = datetime.fromisoformat(date_str)
if dt.tzinfo is None:
start_time_utc = dt.replace(tzinfo=pytz.UTC)
else:
start_time_utc = dt.astimezone(pytz.UTC)
local = start_time_utc.astimezone(pytz.timezone("US/Eastern"))
game_date = local.strftime("%-m/%-d")
game_time = local.strftime("%-I:%M%p").replace("AM", "am").replace("PM", "pm")
except (ValueError, AttributeError):
pass
# Period / clock for live games
period = 0
clock = ""
period_text = ""
is_halftime = False
if state == "in":
status_obj = comp.get("status", {})
period = status_obj.get("period", 0)
clock = status_obj.get("displayClock", "")
detail_lower = status_detail.lower()
uses_quarters = league_key == "ncaaw" or "quarter" in detail_lower or detail_lower.startswith("q")
if period <= (4 if uses_quarters else 2):
period_text = f"Q{period}" if uses_quarters else f"H{period}"
else:
ot_num = period - (4 if uses_quarters else 2)
period_text = f"OT{ot_num}" if ot_num > 1 else "OT"
if "halftime" in detail_lower:
is_halftime = True
elif state == "post":
period_text = status.get("shortDetail", "Final")
if "Final" not in period_text:
period_text = "Final"
# Determine winner and upset
is_final = state == "post"
is_upset = False
winner_side = ""
if is_final:
try:
h = int(float(home_score))
a = int(float(away_score))
if h > a:
winner_side = "home"
if home_seed > away_seed > 0:
is_upset = True
elif a > h:
winner_side = "away"
if away_seed > home_seed > 0:
is_upset = True
except (ValueError, TypeError):
pass
return {
"id": event.get("id", ""),
"league": league_key,
"home_abbr": home_abbr,
"away_abbr": away_abbr,
"home_score": str(home_score),
"away_score": str(away_score),
"home_seed": home_seed,
"away_seed": away_seed,
"tournament_round": tournament_round,
"tournament_region": tournament_region,
"state": state,
"is_final": is_final,
"is_live": state == "in",
"is_upcoming": state == "pre",
"is_halftime": is_halftime,
"period": period,
"period_text": period_text,
"clock": clock,
"status_detail": status_detail,
"game_date": game_date,
"game_time": game_time,
"start_time_utc": start_time_utc,
"is_upset": is_upset,
"winner_side": winner_side,
"headline": headline,
}
@staticmethod
def _parse_round(headline: str) -> str:
hl = headline.lower()
if "national championship" in hl:
return "NCG"
if "final four" in hl:
return "F4"
if "elite 8" in hl or "elite eight" in hl:
return "E8"
if "sweet 16" in hl or "sweet sixteen" in hl:
return "S16"
if "2nd round" in hl or "second round" in hl:
return "R32"
if "1st round" in hl or "first round" in hl:
return "R64"
return ""
@staticmethod
def _parse_region(headline: str) -> str:
if "East Region" in headline:
return "E"
if "West Region" in headline:
return "W"
if "South Region" in headline:
return "S"
if "Midwest Region" in headline:
return "MW"
m = re.search(r"Regional (\d+)", headline)
if m:
return f"R{m.group(1)}"
return ""
# ------------------------------------------------------------------
# Game processing
# ------------------------------------------------------------------
def _process_games(self, games: List[Dict]) -> Dict[str, List[Dict]]:
"""Group games by round, sorted by round significance then region/seed."""
grouped: Dict[str, List[Dict]] = {}
for game in games:
rnd = game.get("tournament_round", "")
grouped.setdefault(rnd, []).append(game)
# Sort each round's games by region then seed matchup
for rnd, round_games in grouped.items():
round_games.sort(
key=lambda g: (
REGION_ORDER.get(g.get("tournament_region", ""), 4),
min(g.get("away_seed", 99), g.get("home_seed", 99)),
)
)
return grouped
# ------------------------------------------------------------------
# Rendering
# ------------------------------------------------------------------
def _draw_text_with_outline(
self,
draw: ImageDraw.Draw,
text: str,
xy: tuple,
font: ImageFont.FreeTypeFont,
fill: tuple = COLOR_WHITE,
outline: tuple = COLOR_BLACK,
) -> None:
x, y = xy
for dx in (-1, 0, 1):
for dy in (-1, 0, 1):
if dx or dy:
draw.text((x + dx, y + dy), text, font=font, fill=outline)
draw.text((x, y), text, font=font, fill=fill)
def _create_round_separator(self, round_key: str) -> Image.Image:
"""Create a separator tile for a tournament round."""
height = self.display_height
name = ROUND_DISPLAY_NAMES.get(round_key, round_key)
font = self.fonts["time"]
# Measure text
tmp = Image.new("RGB", (1, 1))
tmp_draw = ImageDraw.Draw(tmp)
text_width = int(tmp_draw.textlength(name, font=font))
# Logo on each side
logo = self._round_logos.get(round_key, self._march_madness_logo)
logo_w = logo.width if logo else 0
padding = 6
total_w = padding + logo_w + padding + text_width + padding + logo_w + padding
total_w = max(total_w, 80)
img = Image.new("RGB", (total_w, height), COLOR_DARK_BG)
draw = ImageDraw.Draw(img)
# Draw logos
x = padding
if logo:
logo_y = (height - logo.height) // 2
img.paste(logo, (x, logo_y), logo)
x += logo_w + padding
# Draw round name
text_y = (height - 8) // 2 # 8px font
self._draw_text_with_outline(draw, name, (x, text_y), font, fill=COLOR_GOLD)
x += text_width + padding
if logo:
logo_y = (height - logo.height) // 2
img.paste(logo, (x, logo_y), logo)
return img
def _create_game_tile(self, game: Dict) -> Image.Image:
"""Create a single game tile for the scrolling ticker."""
height = self.display_height
font_score = self.fonts["score"]
font_time = self.fonts["time"]
font_detail = self.fonts["detail"]
# Load team logos
away_logo = self._get_team_logo(game["away_abbr"])
home_logo = self._get_team_logo(game["home_abbr"])
logo_w = 0
if away_logo:
logo_w = max(logo_w, away_logo.width)
if home_logo:
logo_w = max(logo_w, home_logo.width)
if logo_w == 0:
logo_w = 24
# Build text elements
away_seed_str = f"({game['away_seed']})" if self.show_seeds and game.get("away_seed", 0) > 0 else ""
home_seed_str = f"({game['home_seed']})" if self.show_seeds and game.get("home_seed", 0) > 0 else ""
away_text = f"{away_seed_str}{game['away_abbr']}"
home_text = f"{game['home_abbr']}{home_seed_str}"
# Measure text widths
tmp = Image.new("RGB", (1, 1))
tmp_draw = ImageDraw.Draw(tmp)
away_text_w = int(tmp_draw.textlength(away_text, font=font_detail))
home_text_w = int(tmp_draw.textlength(home_text, font=font_detail))
# Center content: status line
if game["is_live"]:
if game["is_halftime"]:
status_text = "Halftime"
else:
status_text = f"{game['period_text']} {game['clock']}".strip()
elif game["is_final"]:
status_text = game.get("period_text", "Final")
else:
status_text = f"{game['game_date']} {game['game_time']}".strip()
status_w = int(tmp_draw.textlength(status_text, font=font_time))
# Score line (for live/final)
score_text = ""
if game["is_live"] or game["is_final"]:
score_text = f"{game['away_score']}-{game['home_score']}"
score_w = int(tmp_draw.textlength(score_text, font=font_score)) if score_text else 0
# Calculate tile width
h_pad = 4
center_w = max(status_w, score_w, 40)
tile_w = h_pad + logo_w + h_pad + away_text_w + h_pad + center_w + h_pad + home_text_w + h_pad + logo_w + h_pad
img = Image.new("RGB", (tile_w, height), COLOR_BLACK)
draw = ImageDraw.Draw(img)
# Paste away logo
x = h_pad
if away_logo:
logo_y = (height - away_logo.height) // 2
img.paste(away_logo, (x, logo_y), away_logo)
x += logo_w + h_pad
# Away team text (seed + abbr)
is_fav_away = game["away_abbr"] in self.favorite_teams if self.favorite_teams else False
away_color = COLOR_GOLD if is_fav_away else COLOR_WHITE
if game["is_final"] and game["winner_side"] == "away" and self.highlight_upsets and game["is_upset"]:
away_color = COLOR_GOLD
team_text_y = (height - 6) // 2 - 5 # Upper half
self._draw_text_with_outline(draw, away_text, (x, team_text_y), font_detail, fill=away_color)
x += away_text_w + h_pad
# Center block
center_x = x
center_mid = center_x + center_w // 2
# Status text (top center of center block)
status_x = center_mid - status_w // 2
status_y = 2
status_color = COLOR_GREEN if game["is_live"] else COLOR_GRAY
self._draw_text_with_outline(draw, status_text, (status_x, status_y), font_time, fill=status_color)
# Score (bottom center of center block, for live/final)
if score_text:
score_x = center_mid - score_w // 2
score_y = height - 13
# Upset highlighting
if game["is_final"] and game["is_upset"] and self.highlight_upsets:
score_color = COLOR_GOLD
elif game["is_live"]:
score_color = COLOR_WHITE
else:
score_color = COLOR_WHITE
self._draw_text_with_outline(draw, score_text, (score_x, score_y), font_score, fill=score_color)
# Date for final games (below score)
if game["is_final"] and game.get("game_date"):
date_w = int(draw.textlength(game["game_date"], font=font_detail))
date_x = center_mid - date_w // 2
date_y = height - 6
self._draw_text_with_outline(draw, game["game_date"], (date_x, date_y), font_detail, fill=COLOR_DIM)
x = center_x + center_w + h_pad
# Home team text
is_fav_home = game["home_abbr"] in self.favorite_teams if self.favorite_teams else False
home_color = COLOR_GOLD if is_fav_home else COLOR_WHITE
if game["is_final"] and game["winner_side"] == "home" and self.highlight_upsets and game["is_upset"]:
home_color = COLOR_GOLD
self._draw_text_with_outline(draw, home_text, (x, team_text_y), font_detail, fill=home_color)
x += home_text_w + h_pad
# Paste home logo
if home_logo:
logo_y = (height - home_logo.height) // 2
img.paste(home_logo, (x, logo_y), home_logo)
return img
def _create_ticker_image(self) -> None:
"""Build the full scrolling ticker image from game tiles."""
if not self.games_data:
self.ticker_image = None
if self.scroll_helper:
self.scroll_helper.clear_cache()
return
grouped = self._process_games(self.games_data)
content_items: List[Image.Image] = []
# Order rounds by significance (most important first)
sorted_rounds = sorted(grouped.keys(), key=lambda r: ROUND_ORDER.get(r, 6))
for rnd in sorted_rounds:
games = grouped[rnd]
if not games:
continue
# Add round separator
if self.show_round_logos and rnd:
separator = self._create_round_separator(rnd)
content_items.append(separator)
# Add game tiles
for game in games:
tile = self._create_game_tile(game)
content_items.append(tile)
if not content_items:
self.ticker_image = None
if self.scroll_helper:
self.scroll_helper.clear_cache()
return
if not self.scroll_helper:
self.ticker_image = None
return
gap_width = 16
# Use ScrollHelper to create the scrolling image
self.ticker_image = self.scroll_helper.create_scrolling_image(
content_items=content_items,
item_gap=gap_width,
element_gap=0,
)
self.total_scroll_width = self.scroll_helper.total_scroll_width
self.dynamic_duration = self.scroll_helper.get_dynamic_duration()
self.logger.info(
f"Ticker image created: {self.ticker_image.width}px wide, "
f"{len(self.games_data)} games, dynamic_duration={self.dynamic_duration:.0f}s"
)
# ------------------------------------------------------------------
# Plugin lifecycle
# ------------------------------------------------------------------
def update(self) -> None:
"""Fetch and process tournament data."""
if not self.enabled:
return
current_time = time.time()
# Use shorter interval if live games detected
interval = 60 if self._has_live_games else self.update_interval
if current_time - self.last_update < interval:
return
with self._update_lock:
self.last_update = current_time
if not self._is_tournament_window():
self.logger.debug("Outside tournament window, skipping fetch")
self.games_data = []
self.ticker_image = None
if self.scroll_helper:
self.scroll_helper.clear_cache()
return
try:
games = self._fetch_tournament_data()
self._has_live_games = any(g["is_live"] for g in games)
self.games_data = games
self._create_ticker_image()
self.logger.info(
f"Updated: {len(games)} games, "
f"live={self._has_live_games}"
)
except Exception as e:
self.logger.error(f"Update error: {e}", exc_info=True)
def display(self, force_clear: bool = False) -> None:
"""Render one scroll frame."""
if not self.enabled:
return
if force_clear or self._display_start_time is None:
self._display_start_time = time.time()
if self.scroll_helper:
self.scroll_helper.reset_scroll()
self._end_reached_logged = False
if not self.games_data or self.ticker_image is None:
self._display_fallback()
return
if not self.scroll_helper:
self._display_fallback()
return
try:
if self.loop or not self.scroll_helper.is_scroll_complete():
self.scroll_helper.update_scroll_position()
elif not self._end_reached_logged:
self.logger.info("Scroll complete")
self._end_reached_logged = True
visible = self.scroll_helper.get_visible_portion()
if visible is None:
self._display_fallback()
return
self.dynamic_duration = self.scroll_helper.get_dynamic_duration()
matrix_w = self.display_manager.matrix.width
matrix_h = self.display_manager.matrix.height
if not hasattr(self.display_manager, "image") or self.display_manager.image is None:
self.display_manager.image = Image.new("RGB", (matrix_w, matrix_h), COLOR_BLACK)
self.display_manager.image.paste(visible, (0, 0))
self.display_manager.update_display()
self.scroll_helper.log_frame_rate()
except Exception as e:
self.logger.error(f"Display error: {e}", exc_info=True)
self._display_fallback()
def _display_fallback(self) -> None:
w = self.display_manager.matrix.width
h = self.display_manager.matrix.height
img = Image.new("RGB", (w, h), COLOR_BLACK)
draw = ImageDraw.Draw(img)
if self._is_tournament_window():
text = "No games"
else:
text = "Off-season"
text_w = int(draw.textlength(text, font=self.fonts["time"]))
text_x = (w - text_w) // 2
text_y = (h - 8) // 2
draw.text((text_x, text_y), text, font=self.fonts["time"], fill=COLOR_GRAY)
# Show March Madness logo if available
if self._march_madness_logo:
logo_y = (h - self._march_madness_logo.height) // 2
img.paste(self._march_madness_logo, (2, logo_y), self._march_madness_logo)
self.display_manager.image = img
self.display_manager.update_display()
# ------------------------------------------------------------------
# Duration / cycle management
# ------------------------------------------------------------------
def get_display_duration(self) -> float:
current_time = time.time()
if self._cached_dynamic_duration is not None:
cache_age = current_time - self._duration_cache_time
if cache_age < 5.0:
return self._cached_dynamic_duration
self._cached_dynamic_duration = self.dynamic_duration
self._duration_cache_time = current_time
return self.dynamic_duration
def supports_dynamic_duration(self) -> bool:
if not self.enabled:
return False
return self.dynamic_duration_enabled
def is_cycle_complete(self) -> bool:
if not self.supports_dynamic_duration():
return True
if self._display_start_time is not None and self.dynamic_duration > 0:
elapsed = time.time() - self._display_start_time
if elapsed >= self.dynamic_duration:
return True
if not self.loop and self.scroll_helper and self.scroll_helper.is_scroll_complete():
return True
return False
def reset_cycle_state(self) -> None:
super().reset_cycle_state()
self._display_start_time = None
self._end_reached_logged = False
if self.scroll_helper:
self.scroll_helper.reset_scroll()
# ------------------------------------------------------------------
# Vegas mode
# ------------------------------------------------------------------
def get_vegas_content(self):
if not self.games_data:
return None
tiles = []
for game in self.games_data:
tiles.append(self._create_game_tile(game))
return tiles if tiles else None
def get_vegas_content_type(self) -> str:
return "multi"
# ------------------------------------------------------------------
# Info / cleanup
# ------------------------------------------------------------------
def get_info(self) -> Dict:
info = super().get_info()
info["total_games"] = len(self.games_data)
info["has_live_games"] = self._has_live_games
info["dynamic_duration"] = self.dynamic_duration
info["tournament_window"] = self._is_tournament_window()
return info
def cleanup(self) -> None:
self.games_data = []
self.ticker_image = None
if self.scroll_helper:
self.scroll_helper.clear_cache()
self._team_logo_cache.clear()
if self.session:
self.session.close()
self.session = None
super().cleanup()
+37
View File
@@ -0,0 +1,37 @@
{
"id": "march-madness",
"name": "March Madness",
"version": "1.0.0",
"description": "NCAA March Madness tournament bracket tracker with round branding, seeded matchups, live scores, and upset highlighting",
"author": "ChuckBuilds",
"category": "sports",
"tags": [
"ncaa",
"basketball",
"march-madness",
"tournament",
"bracket",
"scrolling"
],
"repo": "https://github.com/ChuckBuilds/ledmatrix-plugins",
"branch": "main",
"plugin_path": "plugins/march-madness",
"versions": [
{
"version": "1.0.0",
"ledmatrix_min": "2.0.0",
"released": "2026-02-16"
}
],
"stars": 0,
"downloads": 0,
"last_updated": "2026-02-16",
"verified": true,
"screenshot": "",
"display_modes": [
"march_madness"
],
"dependencies": {},
"entry_point": "manager.py",
"class_name": "MarchMadnessPlugin"
}
@@ -0,0 +1,4 @@
requests>=2.28.0
Pillow>=12.2.0
pytz>=2022.1
numpy>=1.24.0
+1 -2
View File
@@ -22,6 +22,5 @@
"Pillow>=10.0.0",
"PyYAML>=6.0",
"requests>=2.31.0"
],
"local_only": true
]
}
+2 -2
View File
@@ -1,3 +1,3 @@
Pillow>=12.2.0
Pillow>=10.4.0
PyYAML>=6.0.2
requests>=2.33.0
requests>=2.32.0
-9
View File
@@ -1,9 +0,0 @@
# Test-only dependencies for the plugin safety harness and pytest suite.
# Install alongside requirements.txt: pip install -r requirements.txt -r requirements-test.txt
#
# pytest, pytest-cov, pytest-mock, and jsonschema are already pinned (with
# major-version caps) in requirements.txt, so they are intentionally NOT
# repeated here — re-pinning pytest to <9 collided with requirements.txt's
# pytest>=9.0.3,<10 and made the two files impossible to install together.
# Only declare what requirements.txt doesn't already provide.
freezegun>=1.2,<2 # deterministic time for golden-image tests
+13 -9
View File
@@ -3,32 +3,39 @@
# Tested on Raspbian OS 12 (Bookworm) and 13 (Trixie)
# Image processing
Pillow>=12.2.0,<13.0.0
Pillow>=10.4.0,<12.0.0
numpy>=1.24.0 # For fast array operations in ScrollHelper (compatible with 2.x)
# Timezone handling
pytz>=2024.2,<2025.0 # Updated for latest timezone data
timezonefinder>=6.5.0,<7.0.0 # Updated for better performance and accuracy
geopy>=2.4.1,<3.0.0
# HTTP requests
requests>=2.33.0,<3.0.0
requests>=2.32.0,<3.0.0
# Google API integration
google-auth-oauthlib>=1.2.0,<2.0.0
google-auth-httplib2>=0.2.0,<1.0.0
google-api-python-client>=2.147.0,<3.0.0
# Font rendering
freetype-py>=2.5.1,<3.0.0
# Spotify integration
spotipy>=2.25.2,<3.0.0
spotipy>=2.24.0,<3.0.0
# Flask web framework
Flask>=3.1.3,<4.0.0
Flask>=3.0.0,<4.0.0
# Text processing
unidecode>=1.3.8,<2.0.0
# Calendar integration
icalevents>=0.1.27,<1.0.0
# WebSocket support
python-socketio>=5.14.0,<6.0.0
python-socketio>=5.11.0,<6.0.0
python-engineio>=4.9.0,<5.0.0
websockets>=12.0,<14.0
websocket-client>=1.8.0,<2.0.0
@@ -36,11 +43,8 @@ websocket-client>=1.8.0,<2.0.0
# JSON Schema validation
jsonschema>=4.20.0,<5.0.0
# Requirement specifier parsing (plugin dependency satisfaction checks)
packaging>=23.0,<27.0
# Testing dependencies
pytest>=9.0.3,<10.0.0
pytest>=7.4.0,<8.0.0
pytest-cov>=4.1.0,<5.0.0
pytest-mock>=3.11.0,<4.0.0
mypy>=1.5.0,<2.0.0
+1
View File
@@ -51,6 +51,7 @@ if debug_mode:
# Try to import the plugin system directly to get better error info
print("DEBUG: Attempting to import src.plugin_system...", flush=True)
from src.plugin_system import PluginManager
print("DEBUG: Plugin system import successful", flush=True)
except ImportError as e:
print(f"DEBUG: Plugin system import failed: {e}", flush=True)
-29
View File
@@ -90,40 +90,11 @@
"min_height": {
"type": "integer",
"minimum": 1
},
"max_width": {
"type": "integer",
"minimum": 1
},
"max_height": {
"type": "integer",
"minimum": 1
}
}
}
}
},
"display": {
"type": "object",
"properties": {
"design_size": {
"type": "object",
"properties": {
"width": {
"type": "integer",
"minimum": 8
},
"height": {
"type": "integer",
"minimum": 8
}
},
"required": ["width", "height"],
"description": "Panel size the plugin's layout was authored against; core derives the adaptive-layout scale factor from it. Defaults to 128x32 when omitted."
}
},
"description": "Display/layout hints for the adaptive layout system"
},
"config_schema": {
"type": "string",
"description": "Path to configuration schema file"
+1 -1
View File
@@ -9,7 +9,7 @@ and preventing validation errors.
import json
import sys
from pathlib import Path
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional
def get_default_for_field(prop: Dict[str, Any]) -> Any:
+2 -1
View File
@@ -9,8 +9,9 @@ Analyze all plugin config schemas to identify issues:
"""
import json
import os
from pathlib import Path
from typing import Dict, List, Any
from typing import Dict, List, Set, Any
import jsonschema
from jsonschema import Draft7Validator
-344
View File
@@ -1,344 +0,0 @@
#!/usr/bin/env python3
"""
LEDMatrix Plugin Security Auditor
Performs AST-based security analysis of all Python files in plugin directories.
Designed to run in CI exits non-zero on CRITICAL findings only.
Usage:
python scripts/audit_plugins.py
python scripts/audit_plugins.py --verbose
python scripts/audit_plugins.py --plugin hello-world
python scripts/audit_plugins.py --output results.json
"""
import ast
import argparse
import json
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from datetime import datetime, timezone
PROJECT_ROOT = Path(__file__).resolve().parent.parent
PLUGIN_BASE_DIRS = [
PROJECT_ROOT / "plugins",
PROJECT_ROOT / "plugin-repos",
]
# ─────────────────────────────────────────────────────────────────────────────
# Finding dataclass
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Finding:
plugin_id: str
file: str
line: int
severity: str # CRITICAL | WARNING | INFO
rule: str
message: str
def to_dict(self) -> dict:
return asdict(self)
# ─────────────────────────────────────────────────────────────────────────────
# AST visitor
# ─────────────────────────────────────────────────────────────────────────────
class _PluginVisitor(ast.NodeVisitor):
"""Collect security findings from a single plugin Python file."""
def __init__(self, filepath: Path, plugin_id: str):
self.filepath = filepath
self.plugin_id = plugin_id
self.findings: list[Finding] = []
# Local name -> real dotted path, so aliased imports and from-imports
# of dangerous APIs (import subprocess as sp; from builtins import
# eval as e) are still recognized in visit_Call below.
self._aliases: dict[str, str] = {}
def _add(self, node: ast.AST, severity: str, rule: str, message: str) -> None:
self.findings.append(Finding(
plugin_id=self.plugin_id,
file=str(self.filepath.relative_to(PROJECT_ROOT)),
line=getattr(node, "lineno", 0),
severity=severity,
rule=rule,
message=message,
))
def _resolve(self, local_name: str) -> str:
"""Resolve a local name through recorded import aliases to its real
dotted path (e.g. "sp" -> "subprocess"); unresolved names pass through
unchanged."""
return self._aliases.get(local_name, local_name)
def _resolve_call_target(self, func: ast.expr) -> str | None:
"""Resolve a Call's func node to a fully-qualified dotted target,
covering a direct name (bare builtin, aliased import, or
from-import: from builtins import eval as e; from subprocess
import run; from os import system as s) and module-attribute
access (subprocess.run, sp.run, os.system, o.system) uniformly.
Returns None for call shapes this doesn't attempt to resolve."""
if isinstance(func, ast.Name):
return self._resolve(func.id)
if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
base = self._resolve(func.value.id)
return f"{base}.{func.attr}"
return None
def visit_Call(self, node: ast.Call) -> None:
target = self._resolve_call_target(node.func)
if target is None:
self.generic_visit(node)
return
leaf = target.rsplit(".", 1)[-1]
# eval() / exec() / compile() — arbitrary code execution, whether a
# bare call, an aliased import, or a from-import
# (from builtins import eval as e; e(...))
if leaf == "eval":
self._add(node, "CRITICAL", "PLUGIN-001",
"eval() call — arbitrary code execution risk")
elif leaf == "exec":
self._add(node, "CRITICAL", "PLUGIN-002",
"exec() call — arbitrary code execution risk")
elif leaf == "compile":
self._add(node, "WARNING", "PLUGIN-003",
"compile() call — dynamic code compilation")
# subprocess.*(shell=True), whether subprocess.run(...), sp.run(...),
# or a from-import (from subprocess import run; run(..., shell=True))
if target in {
"subprocess.run", "subprocess.call", "subprocess.Popen",
"subprocess.check_call", "subprocess.check_output",
}:
for kw in node.keywords:
if (kw.arg == "shell" and
isinstance(kw.value, ast.Constant) and
kw.value.value is True):
self._add(node, "WARNING", "PLUGIN-004",
f"subprocess.{leaf}(shell=True) — "
f"shell injection risk if args include user input")
# os.system(), whether os.system(...), o.system(...), or a
# from-import (from os import system as s; s(...))
if target == "os.system":
self._add(node, "WARNING", "PLUGIN-005",
"os.system() call — prefer subprocess with list args")
self.generic_visit(node)
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
if alias.asname:
local, real = alias.asname, alias.name
else:
# `import os.path` binds the top-level name `os`, not `os.path`
local = real = alias.name.split(".")[0]
self._aliases[local] = real
self._check_import(node, alias.name)
self.generic_visit(node)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
if node.module:
for alias in node.names:
local = alias.asname or alias.name
self._aliases[local] = f"{node.module}.{alias.name}"
self._check_import(node, node.module)
self.generic_visit(node)
def _check_import(self, node: ast.AST, module_name: str) -> None:
dangerous = {
"ctypes": ("WARNING", "PLUGIN-010", "ctypes import — native code execution"),
"cffi": ("WARNING", "PLUGIN-011", "cffi import — native code execution"),
"pickle": ("WARNING", "PLUGIN-012",
"pickle import — deserialization can execute arbitrary code"),
"marshal": ("WARNING", "PLUGIN-013",
"marshal import — deserialization risk"),
}
for mod, (severity, rule, msg) in dangerous.items():
if module_name == mod or module_name.startswith(mod + "."):
self._add(node, severity, rule, msg)
# ─────────────────────────────────────────────────────────────────────────────
# Per-plugin audit
# ─────────────────────────────────────────────────────────────────────────────
def audit_plugin(plugin_dir: Path) -> list[Finding]:
"""Audit a single plugin directory. Returns all findings."""
findings: list[Finding] = []
plugin_id = plugin_dir.name
# Check for required files
for required_file, rule, msg in [
("manifest.json", "PLUGIN-020",
"manifest.json missing — plugin may be incomplete"),
("config_schema.json", "PLUGIN-021",
"config_schema.json missing — no input validation schema declared"),
]:
if not (plugin_dir / required_file).exists():
findings.append(Finding(
plugin_id=plugin_id,
file=str((plugin_dir / required_file).relative_to(PROJECT_ROOT)),
line=0,
severity="WARNING",
rule=rule,
message=msg,
))
# AST analysis of all Python files
for py_file in sorted(plugin_dir.rglob("*.py")):
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source, filename=str(py_file))
visitor = _PluginVisitor(py_file, plugin_id)
visitor.visit(tree)
findings.extend(visitor.findings)
except SyntaxError as exc:
# A file the visitor can't even parse is a file we can't verify
# is safe -- this must block the audit, not just warn.
findings.append(Finding(
plugin_id=plugin_id,
file=str(py_file.relative_to(PROJECT_ROOT)),
line=getattr(exc, "lineno", 0) or 0,
severity="CRITICAL",
rule="PLUGIN-030",
message=f"Python syntax error — cannot be parsed: {exc}",
))
except OSError as exc:
# Same reasoning as SyntaxError: an unreadable file was never
# actually scanned, so it must block rather than pass silently.
findings.append(Finding(
plugin_id=plugin_id,
file=str(py_file.relative_to(PROJECT_ROOT)),
line=0,
severity="CRITICAL",
rule="PLUGIN-031",
message=f"Could not read file: {exc}",
))
return findings
# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
def main() -> int:
parser = argparse.ArgumentParser(
description="LEDMatrix plugin security auditor",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--plugin", "-p", default=None,
help="Audit a specific plugin ID only")
parser.add_argument("--output", "-o", default=None,
help="Write JSON results to this file")
parser.add_argument("--verbose", "-v", action="store_true",
help="Show all findings, not just summary")
args = parser.parse_args()
print("=" * 60)
print("LEDMatrix Plugin Security Audit")
print(f"Project root: {PROJECT_ROOT}")
print("=" * 60)
all_findings: list[Finding] = []
plugins_scanned = 0
plugin_found = args.plugin is None
for base_dir in PLUGIN_BASE_DIRS:
if not base_dir.exists():
if args.verbose:
print(f" ⏭️ Skipping {base_dir.name}/ (directory not found)")
continue
base_label = base_dir.relative_to(PROJECT_ROOT)
print(f"\n Scanning {base_label}/")
for plugin_dir in sorted(base_dir.iterdir()):
if not plugin_dir.is_dir():
continue
if plugin_dir.name.startswith((".", "_")):
continue
if args.plugin and plugin_dir.name != args.plugin:
continue
if args.plugin:
plugin_found = True
findings = audit_plugin(plugin_dir)
all_findings.extend(findings)
plugins_scanned += 1
critical = [f for f in findings if f.severity == "CRITICAL"]
warnings = [f for f in findings if f.severity == "WARNING"]
if critical:
icon, label = "🚨", "CRITICAL"
elif warnings:
icon, label = "⚠️ ", "WARN "
else:
icon, label = "", "PASS "
print(f" {icon} [{label}] {plugin_dir.name}"
f"{len(critical)} critical, {len(warnings)} warnings")
if args.verbose:
for f in findings:
severity_icon = {"CRITICAL": "🚨", "WARNING": "⚠️ ", "INFO": ""}.get(
f.severity, " "
)
print(f" {severity_icon} {f.rule} {f.file}:{f.line}{f.message}")
if args.plugin and not plugin_found:
print(f"\n 🚨 Plugin '{args.plugin}' not found in any of "
f"{[str(d.relative_to(PROJECT_ROOT)) for d in PLUGIN_BASE_DIRS]}"
f"nothing was audited")
return 1
# Summary
critical_findings = [f for f in all_findings if f.severity == "CRITICAL"]
warning_findings = [f for f in all_findings if f.severity == "WARNING"]
print(f"\n{'=' * 60}")
print(f" Plugins scanned : {plugins_scanned}")
print(f" CRITICAL : {len(critical_findings)}")
print(f" WARNING : {len(warning_findings)}")
if critical_findings:
print("\n 🚨 CRITICAL findings:")
for f in critical_findings:
print(f" {f.plugin_id} | {Path(f.file).name}:{f.line} | {f.message}")
# Write JSON output
if args.output:
output_data = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"plugins_scanned": plugins_scanned,
"summary": {
"critical": len(critical_findings),
"warnings": len(warning_findings),
},
"findings": [f.to_dict() for f in all_findings],
}
Path(args.output).write_text(
json.dumps(output_data, indent=2), encoding="utf-8"
)
print(f"\n Results written to: {args.output}")
if critical_findings:
print("\n 🚨 Blocking — CRITICAL issues must be resolved")
return 1
print("\n ✅ No critical issues found")
return 0
if __name__ == "__main__":
sys.exit(main())
-267
View File
@@ -1,267 +0,0 @@
#!/usr/bin/env python3
"""
Plugin safety checker.
Renders a plugin across every declared screen (mode) and every supported matrix
size, and fails if any screen crashes, overflows the panel, or (for plugins with
committed golden images) drifts visually.
Usage:
# Functional + bounds check across all sizes/modes:
python scripts/check_plugin.py --plugin clock-simple
# Every discovered plugin:
python scripts/check_plugin.py --all
# Dump PNGs for each size/mode so you can eyeball them:
python scripts/check_plugin.py --plugin ledmatrix-weather --out-dir /tmp/preview
# Refresh committed golden images after an intentional visual change:
python scripts/check_plugin.py --plugin clock-simple --update-golden \
--mock-data plugins/clock-simple/test/fixtures/mock.json
Exit code is non-zero if any (plugin, size, mode) fails.
"""
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Dict, List, Optional
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
os.environ['EMULATOR'] = 'true'
from src.logging_config import get_logger # noqa: E402
from src.plugin_system.testing.loading import ( # noqa: E402
build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
)
from src.plugin_system.testing.harness import ( # noqa: E402
RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
check_scale_up,
)
from src.plugin_system.testing.sizes import ( # noqa: E402
parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
)
logger = get_logger("[Check Plugin]")
DEFAULT_SEARCH_DIRS = [
str(PROJECT_ROOT / 'plugins'),
str(PROJECT_ROOT / 'plugin-repos'),
]
def discover_plugins(search_dirs: List[str]) -> List[str]:
"""All plugin ids found across the search dirs (dirs containing manifest.json)."""
found = []
for d in search_dirs:
base = Path(d)
if not base.exists():
continue
for child in sorted(base.iterdir()):
if (child / 'manifest.json').exists() and child.name not in found:
found.append(child.name)
return found
def parse_sizes(spec: Optional[str]):
if not spec:
return None
sizes = []
for token in spec.split(','):
if not token.strip():
continue
try:
sizes.append(parse_size_token(token))
except ValueError as exc:
raise SystemExit(str(exc)) from exc
return sizes
def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
config: Dict, run_update: bool, out_dir: Optional[Path],
update_golden: bool, golden_dir_override: Optional[Path],
freeze_time: Optional[str]) -> List[RenderResult]:
plugin_dir = find_plugin_dir(plugin_id, search_dirs)
if not plugin_dir:
logger.error("Plugin '%s' not found in: %s", plugin_id, search_dirs)
return [RenderResult(plugin_id, 0, 0, "<not-found>", error="plugin directory not found")]
# Per-plugin test/harness.json holds the deterministic settings the committed
# goldens were generated with (config, mock data, frozen time, sizes). Load
# them so the CLI/CI render reproduces the golden the same way the pytest
# matrix path does; explicit CLI flags still override the file.
spec = load_harness_spec(plugin_dir)
# config_schema defaults (real-install behavior, with enabled forced True
# so a plugin's own enabled:false default can't accidentally disable
# testing), then harness.json config, then CLI --config — most specific
# wins.
full_config = build_full_config(plugin_dir, spec, config)
# Precedence: CLI flag > LEDMATRIX_TEST_SIZES env > harness.json > default.
effective_sizes = sizes if sizes else resolve_test_sizes(spec.get("sizes"))
# CLI value wins when provided, else fall back to the harness.json setting.
effective_mock_data = mock_data or spec.get("mock_data_contents", {})
effective_freeze = freeze_time or spec.get("freeze_time")
effective_run_update = run_update and not spec.get("skip_update", False)
# The plugin's declared design size drives the scale-up fill check
# (panels >= 2x the design size must not be left mostly empty).
declared = load_manifest(plugin_dir).get("display", {}).get("design_size", {})
design_size = (int(declared.get("width", 128)), int(declared.get("height", 32)))
fill_strict = spec.get("fill_check") == "strict"
# Every run: the base config, plus one per harness.json "variant" —
# a config overlay with its own golden dir (e.g. adaptive layout mode
# tested alongside the classic default).
runs = [(None, {}, golden_dir_override or (plugin_dir / 'test' / 'golden'))]
for variant in spec.get("variants", []):
name = variant.get("name") or "variant"
vdir = plugin_dir / variant.get("golden_dir", f"test/golden-{name}")
runs.append((name, variant.get("config", {}), vdir))
all_run_results: List[RenderResult] = []
for variant_name, overlay, golden_dir in runs:
run_config = {**full_config, **overlay}
results = render_plugin_matrix(
plugin_id=plugin_id, plugin_dir=plugin_dir, config=run_config,
mock_data=effective_mock_data, sizes=effective_sizes,
run_update=effective_run_update, freeze_time=effective_freeze,
)
if update_golden:
written = write_goldens(results, golden_dir)
logger.info("Wrote %d golden image(s) for %s%s to %s", written, plugin_id,
f" [{variant_name}]" if variant_name else "", golden_dir)
else:
compare_to_goldens(results, golden_dir)
check_scale_up(results, design_size=design_size, strict=fill_strict)
# Tag variant runs so the report and PNG dumps stay distinguishable.
if variant_name:
for r in results:
r.mode = f"{r.mode}@{variant_name}"
if out_dir:
for r in results:
if r.image is None:
continue
dest = out_dir / plugin_id / size_label(r.width, r.height)
dest.mkdir(parents=True, exist_ok=True)
r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")
all_run_results.extend(results)
return all_run_results
def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
"""Print a per-plugin grid. Returns True if everything passed."""
everything_ok = True
for plugin_id, results in all_results.items():
print(f"\n=== {plugin_id} ===")
for r in results:
if r.ok:
status = "PASS"
detail = ""
if r.golden_checked:
detail = " (golden ✓)"
if r.update_error is not None:
detail += f" (update warn: {r.update_error})"
if r.fill_checked and r.fill_ok is None and r.fill_extent:
# warn-only underfill: big panel left mostly empty
ex, ey = r.fill_extent
detail += f" (fill warn: extent {ex:.0%}x{ey:.0%})"
else:
everything_ok = False
if r.error is not None:
status, detail = "FAIL", f" error={r.error}"
elif r.overflow is not None:
status, detail = "FAIL", f" overflow bbox={r.overflow}"
elif r.golden_ok is False:
status = "FAIL"
detail = f" golden drift: {r.golden_diff_pixels}px (max Δ={r.golden_max_delta})"
elif r.fill_ok is False:
ex, ey = r.fill_extent or (0.0, 0.0)
status = "FAIL"
detail = f" fill: extent {ex:.0%}x{ey:.0%} below required coverage"
else:
status, detail = "FAIL", ""
print(f" [{status}] {r.size_label:>7} {r.mode}{detail}")
print()
return everything_ok
def main() -> int:
parser = argparse.ArgumentParser(description="Check a plugin renders safely across sizes & screens")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--plugin', '-p', help='Plugin id to check')
group.add_argument('--all', action='store_true', help='Check every discovered plugin')
parser.add_argument('--plugin-dir', '-d', default=None, help='Directory to search for plugins')
parser.add_argument('--sizes', default=None, help='Comma-separated WxH list (default: all supported)')
parser.add_argument('--config', '-c', default='{}', help='Plugin config overrides as JSON')
parser.add_argument('--mock-data', '-m', default=None, help='Path to JSON file with mock cache data')
parser.add_argument('--out-dir', '-o', default=None, help='Also dump rendered PNGs here')
parser.add_argument('--skip-update', action='store_true', help='Skip calling update()')
parser.add_argument('--update-golden', action='store_true', help='Write/refresh golden images')
parser.add_argument('--golden-dir', default=None, help='Override golden dir (default: <plugin>/test/golden)')
parser.add_argument('--freeze-time', default=None,
help='Freeze wall clock, e.g. "2025-08-01 15:25:00" (for time-dependent plugins)')
args = parser.parse_args()
search_dirs = [args.plugin_dir] if args.plugin_dir else DEFAULT_SEARCH_DIRS
sizes = parse_sizes(args.sizes)
try:
config = json.loads(args.config)
except json.JSONDecodeError as e:
logger.error("Invalid --config JSON: %s", e)
return 2
if not isinstance(config, dict):
logger.error("--config must be a JSON object, got %s", type(config).__name__)
return 2
mock_data = {}
if args.mock_data:
mock_path = Path(args.mock_data)
if not mock_path.exists():
logger.error("Mock data file not found: %s", args.mock_data)
return 2
with open(mock_path) as f:
mock_data = json.load(f)
if not isinstance(mock_data, dict):
logger.error("--mock-data must be a JSON object (key -> cache value), got %s",
type(mock_data).__name__)
return 2
plugin_ids = discover_plugins(search_dirs) if args.all else [args.plugin]
if not plugin_ids:
logger.error("No plugins found in: %s", search_dirs)
return 2
out_dir = Path(args.out_dir) if args.out_dir else None
golden_dir_override = Path(args.golden_dir) if args.golden_dir else None
all_results: Dict[str, List[RenderResult]] = {}
for plugin_id in plugin_ids:
all_results[plugin_id] = check_one(
plugin_id=plugin_id, search_dirs=search_dirs, sizes=sizes,
mock_data=mock_data, config=config, run_update=not args.skip_update,
out_dir=out_dir, update_golden=args.update_golden,
golden_dir_override=golden_dir_override, freeze_time=args.freeze_time,
)
# When refreshing goldens we skip drift comparison, but a crash or overflow
# still means the plugin is broken — never let --update-golden mask that.
ok = print_report(all_results)
return 0 if ok else 1
if __name__ == '__main__':
sys.exit(main())
+29
View File
@@ -0,0 +1,29 @@
#!/bin/bash
# Clear all plugin dependency markers to force fresh dependency check
# Useful after updating plugins or troubleshooting dependency issues
echo "Clearing plugin dependency markers..."
# Check both possible cache locations
CACHE_DIRS=(
"/var/cache/ledmatrix"
"$HOME/.cache/ledmatrix"
)
for CACHE_DIR in "${CACHE_DIRS[@]}"; do
if [ -d "$CACHE_DIR" ]; then
echo "Checking $CACHE_DIR..."
marker_count=$(find "$CACHE_DIR" -name "plugin_*_deps_installed" 2>/dev/null | wc -l)
if [ "$marker_count" -gt 0 ]; then
echo "Found $marker_count dependency marker(s) in $CACHE_DIR"
find "$CACHE_DIR" -name "plugin_*_deps_installed" -delete
echo "Cleared $marker_count marker(s)"
else
echo "No dependency markers found in $CACHE_DIR"
fi
fi
done
echo "Done! Dependency markers cleared."
echo "Next startup will check and install dependencies as needed."
+2
View File
@@ -3,6 +3,8 @@
Check what imports are actually in the app.py file on the Pi
"""
import sys
import os
from pathlib import Path
# Read the app.py file and check the import lines
+2 -3
View File
@@ -67,9 +67,8 @@ def main():
print(" 📍 Will run on: http://0.0.0.0:5000")
print(" ⏹️ Press Ctrl+C to stop")
# Run the app (debug mode controlled by env var to satisfy security scanners)
_debug = os.environ.get('LEDMATRIX_FLASK_DEBUG', '0') == '1'
app.run(host='0.0.0.0', port=5000, debug=_debug)
# Run the app (this should start the server)
app.run(host='0.0.0.0', port=5000, debug=True)
except KeyboardInterrupt:
print("\n ⏹️ Server stopped by user")
+1 -1
View File
@@ -203,7 +203,7 @@ link_github_plugin() {
log_info "Repository already exists at $target_dir"
if [[ -d "$target_dir/.git" ]]; then
log_info "Updating repository..."
(cd "$target_dir" && git pull --rebase) || true
(cd "$target_dir" && git pull --rebase || true)
fi
else
# Clone the repository
-95
View File
@@ -1,95 +0,0 @@
#!/usr/bin/env python3
"""
Pillow compatibility smoke test.
Exercises the Pillow APIs used throughout LEDMatrix to verify a new
Pillow version doesn't break image rendering, font handling, or resize ops.
Run after upgrading Pillow:
python3 scripts/dev/test_pillow_compat.py
"""
import sys
def check(label, fn):
try:
result = fn()
print(f"{label}" + (f"{result}" if result is not None else ""))
return True
except Exception as e:
print(f"{label}{type(e).__name__}: {e}", file=sys.stderr)
return False
def main():
from PIL import Image, ImageDraw, ImageFont
import PIL
print(f"Pillow {PIL.__version__} on Python {sys.version.split()[0]}\n")
failures = 0
print("Image creation:")
failures += not check("Image.new RGB",
lambda: Image.new('RGB', (128, 32), (0, 0, 0)).size)
failures += not check("Image.new RGBA",
lambda: Image.new('RGBA', (64, 64), (255, 0, 0, 128)).size)
failures += not check("Image.new 1-bit",
lambda: Image.new('1', (16, 16)).size)
print("\nDraw operations:")
img = Image.new('RGB', (128, 32), (0, 0, 0))
draw = ImageDraw.Draw(img)
font = ImageFont.load_default()
failures += not check("draw.rectangle",
lambda: draw.rectangle([0, 0, 127, 31], outline=(255, 0, 0)))
failures += not check("draw.text",
lambda: draw.text((2, 2), "Hello", fill=(255, 255, 255), font=font))
failures += not check("draw.line",
lambda: draw.line([0, 0, 127, 31], fill=(0, 255, 0)))
print("\nFont metrics (used in text_helper, scroll_helper):")
failures += not check("draw.textlength",
lambda: f"{draw.textlength('Test', font=font):.1f}px")
failures += not check("draw.textbbox",
lambda: draw.textbbox((0, 0), "Test", font=font))
print("\nResampling (used in logo_helper, sports base):")
logo = Image.new('RGBA', (200, 200), (255, 128, 0, 200))
failures += not check("Image.Resampling.LANCZOS exists",
lambda: str(Image.Resampling.LANCZOS))
failures += not check("thumbnail with LANCZOS",
lambda: (logo.thumbnail((64, 32), Image.Resampling.LANCZOS), logo.size)[1])
big = Image.new('RGB', (300, 300), (0, 128, 255))
failures += not check("resize with LANCZOS",
lambda: big.resize((128, 32), Image.Resampling.LANCZOS).size)
print("\nComposite / paste (used in display rendering):")
base = Image.new('RGB', (128, 32), (0, 0, 0))
overlay = Image.new('RGBA', (32, 32), (255, 0, 0, 128))
failures += not check("paste RGBA onto RGB",
lambda: (base.paste(overlay.convert('RGB'), (0, 0)), base.size)[1])
failures += not check("Image.alpha_composite",
lambda: Image.alpha_composite(
Image.new('RGBA', (32, 32)), overlay).size)
print("\nImage I/O:")
import io
buf = io.BytesIO()
img.save(buf, format='PNG')
buf.seek(0)
failures += not check("save/load PNG roundtrip",
lambda: Image.open(buf).size)
print()
if failures == 0:
print(f"All checks passed. Pillow {PIL.__version__} is compatible.")
return 0
else:
print(f"{failures} check(s) failed — review output above.", file=sys.stderr)
return 1
if __name__ == '__main__':
sys.exit(main())
+1
View File
@@ -15,6 +15,7 @@ Usage: python tools/validate_python.py <python_file>
import ast
import sys
import os
from pathlib import Path
def validate_file(filepath: str) -> bool:
"""Validate a Python file for common issues."""
+72 -198
View File
@@ -16,7 +16,6 @@ Opens at http://localhost:5001
import sys
import os
import json
import re
import time
import argparse
import logging
@@ -45,10 +44,6 @@ MAX_HEIGHT = 512
MIN_WIDTH = 1
MIN_HEIGHT = 1
# plugin_id arrives in request input and is used to build filesystem paths —
# allowlist it (same pattern the web UI's pages_v3 uses)
_SAFE_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
# --------------------------------------------------------------------------
# Plugin discovery
@@ -111,30 +106,15 @@ def discover_plugins() -> List[Dict[str, Any]]:
def find_plugin_dir(plugin_id: str) -> Optional[Path]:
"""Find a plugin directory by ID.
plugin_id comes from request input: it must pass an allowlist match,
and the resulting directory is normalized and required to live inside
one of the plugin search dirs, so a crafted id can never name a path
outside them.
"""
if not isinstance(plugin_id, str) or not _SAFE_PLUGIN_ID_RE.match(plugin_id):
return None
"""Find a plugin directory by ID."""
from src.plugin_system.plugin_loader import PluginLoader
loader = PluginLoader()
for search_dir in get_search_dirs():
if not search_dir.exists():
continue
result = loader.find_plugin_directory(plugin_id, search_dir)
if not result:
continue
# Normalize WITHOUT following symlinks (dev plugins are often
# symlinked into plugins/) and require lexical containment in the
# search dir, so no id can ever name a path outside it.
result_abs = os.path.abspath(str(result))
root_abs = os.path.abspath(str(search_dir))
if os.path.commonpath([result_abs, root_abs]) == root_abs:
return Path(result_abs)
if result:
return Path(result)
return None
@@ -196,118 +176,6 @@ def api_plugin_defaults(plugin_id):
return jsonify({'defaults': defaults})
def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, height,
skip_update):
"""Render one plugin at one size. Returns the /api/render response dict.
A fresh plugin instance per call, mirroring the safety harness, so sizes
never share state.
"""
from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
from src.plugin_system.plugin_loader import PluginLoader
display_manager = VisualTestDisplayManager(width=width, height=height)
cache_manager = MockCacheManager()
plugin_manager = MockPluginManager()
# Pre-populate cache with mock data
for key, value in mock_data.items():
cache_manager.set(key, value)
loader = PluginLoader()
errors = []
warnings = []
plugin_instance, _module = loader.load_plugin(
plugin_id=plugin_id,
manifest=manifest,
plugin_dir=plugin_dir,
config=config,
display_manager=display_manager,
cache_manager=cache_manager,
plugin_manager=plugin_manager,
install_deps=False,
)
start_time = time.time()
# Run update()
if not skip_update:
try:
plugin_instance.update()
except Exception as e:
logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
warnings.append(f"update() raised: {type(e).__name__} — see server log")
# Run display()
try:
plugin_instance.display(force_clear=True)
except Exception as e:
logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
errors.append(f"display() raised: {type(e).__name__} — see server log")
render_time_ms = round((time.time() - start_time) * 1000, 1)
return {
'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
'width': width,
'height': height,
'render_time_ms': render_time_ms,
'errors': errors,
'warnings': warnings,
}
def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
"""Re-derive a plugin directory from the search dirs' own listings.
Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
returned Path is built from a trusted root plus a name the filesystem
itself produced under that root via scandir request-derived strings
never enter its construction so a crafted plugin id can never make
downstream file access leave the plugin search dirs. Comparison is by
name, deliberately without symlink resolution (dev plugins are
commonly symlinked into plugins/).
"""
wanted_name = Path(os.path.normpath(str(plugin_dir))).name
for search_dir in get_search_dirs():
search_dir_str = str(search_dir)
try:
with os.scandir(search_dir_str) as entries:
for entry in entries:
if entry.name == wanted_name and entry.is_dir():
return Path(search_dir_str) / entry.name
except OSError:
continue
return None
def _parse_render_request(data):
"""Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
mock_data, skip_update) or raises ValueError with a client message."""
plugin_id = data['plugin_id']
candidate_dir = find_plugin_dir(plugin_id)
# Never reuse `candidate_dir` past this point: it's built from
# request-derived input, and a variable reassigned only on some paths
# isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
# sole name used below, always the scandir-sourced result.
trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
if not trusted_dir:
raise LookupError(f'Plugin not found: {plugin_id}')
manifest_path = trusted_dir / 'manifest.json'
with open(manifest_path, 'r') as f:
manifest = json.load(f)
# Build config: schema defaults + user overrides
config = {'enabled': True}
config.update(load_config_defaults(trusted_dir))
config.update(data.get('config', {}))
return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
@app.route('/api/render', methods=['POST'])
def api_render():
"""Render a plugin and return the display as base64 PNG."""
@@ -315,6 +183,11 @@ def api_render():
if not data or 'plugin_id' not in data:
return jsonify({'error': 'plugin_id is required'}), 400
plugin_id = data['plugin_id']
user_config = data.get('config', {})
mock_data = data.get('mock_data', {})
skip_update = data.get('skip_update', False)
try:
width = int(data.get('width', 128))
height = int(data.get('height', 32))
@@ -326,77 +199,78 @@ def api_render():
if not (MIN_HEIGHT <= height <= MAX_HEIGHT):
return jsonify({'error': f'height must be between {MIN_HEIGHT} and {MAX_HEIGHT}'}), 400
try:
plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
except LookupError:
return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
except Exception:
# Bad manifest.json / schema / fixture — details go to the dev's
# console, not the HTTP response
app.logger.exception('render request preparation failed')
return jsonify({'error': 'Could not prepare render request; see server log'}), 400
# Find plugin
plugin_dir = find_plugin_dir(plugin_id)
if not plugin_dir:
return jsonify({'error': f'Plugin not found: {plugin_id}'}), 404
# Load manifest
manifest_path = plugin_dir / 'manifest.json'
with open(manifest_path, 'r') as f:
manifest = json.load(f)
# Build config: schema defaults + user overrides
config_defaults = load_config_defaults(plugin_dir)
config = {'enabled': True}
config.update(config_defaults)
config.update(user_config)
# Create display manager and mocks
from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
from src.plugin_system.plugin_loader import PluginLoader
display_manager = VisualTestDisplayManager(width=width, height=height)
cache_manager = MockCacheManager()
plugin_manager = MockPluginManager()
# Pre-populate cache with mock data
for key, value in mock_data.items():
cache_manager.set(key, value)
# Load plugin
loader = PluginLoader()
errors = []
warnings = []
try:
result = _render_once(data['plugin_id'], plugin_dir, manifest, config,
mock_data, width, height, skip_update)
except Exception:
app.logger.exception('plugin load failed during render')
return jsonify({'error': 'Failed to load plugin; see server log'}), 500
return jsonify(result)
plugin_instance, module = loader.load_plugin(
plugin_id=plugin_id,
manifest=manifest,
plugin_dir=plugin_dir,
config=config,
display_manager=display_manager,
cache_manager=cache_manager,
plugin_manager=plugin_manager,
install_deps=False,
)
except Exception as e:
return jsonify({'error': f'Failed to load plugin: {e}'}), 500
start_time = time.time()
@app.route('/api/sizes')
def api_sizes():
"""The representative panel-size sample the safety harness renders at."""
from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
return jsonify({'sizes': [list(s) for s in DEFAULT_TEST_SIZES]})
MAX_MATRIX_SIZES = 12
@app.route('/api/render-matrix', methods=['POST'])
def api_render_matrix():
"""Render a plugin at a list of sizes (default: the harness sample) so the
UI can show a side-by-side multi-resolution gallery."""
data = request.get_json()
if not data or 'plugin_id' not in data:
return jsonify({'error': 'plugin_id is required'}), 400
from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
sizes = data.get('sizes') or [list(s) for s in DEFAULT_TEST_SIZES]
if len(sizes) > MAX_MATRIX_SIZES:
return jsonify({'error': f'at most {MAX_MATRIX_SIZES} sizes per request'}), 400
parsed_sizes = []
for pair in sizes:
# Run update()
if not skip_update:
try:
w, h = int(pair[0]), int(pair[1])
except (TypeError, ValueError, IndexError):
return jsonify({'error': f'invalid size entry {pair!r} (expected [w, h])'}), 400
if not (MIN_WIDTH <= w <= MAX_WIDTH and MIN_HEIGHT <= h <= MAX_HEIGHT):
return jsonify({'error': f'size {w}x{h} out of bounds'}), 400
parsed_sizes.append((w, h))
plugin_instance.update()
except Exception as e:
warnings.append(f"update() raised: {e}")
# Run display()
try:
plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
except LookupError:
return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
except Exception:
app.logger.exception('render request preparation failed')
return jsonify({'error': 'Could not prepare render request; see server log'}), 400
plugin_instance.display(force_clear=True)
except Exception as e:
errors.append(f"display() raised: {e}")
results = []
for w, h in parsed_sizes:
try:
results.append(_render_once(data['plugin_id'], plugin_dir, manifest,
config, mock_data, w, h, skip_update))
except Exception:
app.logger.exception('plugin load failed during %dx%d render', w, h)
results.append({'image': None, 'width': w, 'height': h,
'render_time_ms': 0,
'errors': ['Failed to load plugin; see server log'],
'warnings': []})
return jsonify({'results': results})
render_time_ms = round((time.time() - start_time) * 1000, 1)
return jsonify({
'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
'width': width,
'height': height,
'render_time_ms': render_time_ms,
'errors': errors,
'warnings': warnings,
})
# --------------------------------------------------------------------------
+1
View File
@@ -13,6 +13,7 @@ echo ""
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# Get the actual user
+1 -1
View File
@@ -41,7 +41,7 @@ if [ -f "$PROJECT_DIR/config/config.json" ]; then
echo -e "${GREEN}✓ Config file found${NC}"
# Check web_display_autostart setting
AUTOSTART=$(grep -o '"web_display_autostart"[[:space:]]*:[[:space:]]*[a-z]*' "$PROJECT_DIR/config/config.json" | grep -o '[a-z]*$')
AUTOSTART=$(cat "$PROJECT_DIR/config/config.json" | grep -o '"web_display_autostart"[[:space:]]*:[[:space:]]*[a-z]*' | grep -o '[a-z]*$')
if [ "$AUTOSTART" == "true" ]; then
echo -e "${GREEN}✓ web_display_autostart: true${NC}"
+4 -1
View File
@@ -16,8 +16,11 @@ YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# Check if running as root or with sudo
if [ "$EUID" -ne 0 ]; then
if [ "$EUID" -ne 0 ]; then
echo -e "${YELLOW}Warning: Some checks require sudo. Running what we can...${NC}"
SUDO=""
else
SUDO=""
fi
PROJECT_DIR="${HOME}/LEDMatrix"
+1 -1
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
@@ -7,6 +7,12 @@ echo "Fixing LEDMatrix assets directory permissions..."
# Get the real user (not root when running with sudo)
REAL_USER=${SUDO_USER:-$USER}
# Resolve the home directory of the real user robustly
if command -v getent >/dev/null 2>&1; then
REAL_HOME=$(getent passwd "$REAL_USER" | cut -d: -f6)
else
REAL_HOME=$(eval echo ~"$REAL_USER")
fi
REAL_GROUP=$(id -gn "$REAL_USER")
# Get the project directory
-74
View File
@@ -1,74 +0,0 @@
#!/bin/bash
# safe_pip_install.sh — Install a requirements.txt as root after validating
# that the resolved path is the project's own requirements.txt or a plugin's
# requirements.txt under plugin-repos/ or plugins/.
#
# This script is intended to be called via sudo from the web interface, so
# that packages a plugin declares end up visible to ledmatrix.service (which
# runs as root) rather than only to whichever non-root user runs the web
# interface. Plugin code already runs as root once loaded, so installing its
# declared dependencies as root is not a new trust boundary.
#
# Usage: safe_pip_install.sh <requirements_txt_path>
set -euo pipefail
if [ $# -ne 1 ]; then
echo "Usage: $0 <requirements_txt_path>" >&2
exit 1
fi
TARGET="$1"
# Determine the project root (parent of scripts/fix_perms/)
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
# Allowed locations (resolved, no trailing slash):
# - the project's own requirements.txt
# - any requirements.txt under plugin-repos/ or plugins/
ALLOWED_EXACT="$(realpath --canonicalize-missing "$PROJECT_ROOT/requirements.txt")"
ALLOWED_BASES=(
"$(realpath --canonicalize-missing "$PROJECT_ROOT/plugin-repos")"
"$(realpath --canonicalize-missing "$PROJECT_ROOT/plugins")"
)
# Resolve the target path (follow symlinks); works even if it doesn't exist.
RESOLVED_TARGET="$(realpath --canonicalize-missing "$TARGET")"
# Must be named requirements.txt — never install from an arbitrary file.
if [ "$(basename "$RESOLVED_TARGET")" != "requirements.txt" ]; then
echo "DENIED: $RESOLVED_TARGET is not a requirements.txt file" >&2
exit 2
fi
ALLOWED=false
if [ "$RESOLVED_TARGET" = "$ALLOWED_EXACT" ]; then
ALLOWED=true
else
for BASE in "${ALLOWED_BASES[@]}"; do
if [[ "$RESOLVED_TARGET" == "$BASE/"* ]]; then
ALLOWED=true
break
fi
done
fi
if [ "$ALLOWED" = false ]; then
echo "DENIED: $RESOLVED_TARGET is not an allowed requirements.txt location" >&2
echo "Allowed: $ALLOWED_EXACT, or any requirements.txt under: ${ALLOWED_BASES[*]}" >&2
exit 2
fi
if [ ! -f "$RESOLVED_TARGET" ]; then
echo "ERROR: $RESOLVED_TARGET does not exist" >&2
exit 3
fi
PYTHON_PATH="$(command -v python3)"
# --ignore-installed: root's site-packages often has apt/dpkg-managed copies
# of common libraries (requests, urllib3, ...) with no pip RECORD file, which
# pip refuses to uninstall in place ("Cannot uninstall: no RECORD file was
# found"). This tells pip to install the newer version alongside rather than
# aborting the whole requirements.txt install over one such conflict.
exec "$PYTHON_PATH" -m pip install --break-system-packages --ignore-installed -r "$RESOLVED_TARGET"
-356
View File
@@ -1,356 +0,0 @@
#!/usr/bin/env python3
"""
Security Report Generator
Aggregates JSON output from all CI security audit jobs into a single
Markdown report suitable for PR comments and artifact storage.
Expected artifact layout (from actions/download-artifact@v4):
<artifact-dir>/
sast-results/
bandit-results.json
semgrep-results.json
dependency-audit-results/
pip-audit-results.json
safety-results.json
secrets-scan-results/
gitleaks-results.json
security-proofs-results/
security-proofs-results.json
plugin-audit-results/
plugin-audit-results.json
Usage:
python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md
python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md --verbose
"""
import argparse
import json
import sys
from pathlib import Path
from datetime import datetime, timezone
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# Gitleaks matches exactly equal to one of these (not a substring match -- a
# real secret that merely contains one of these words as part of its actual
# value must still be reported) are known template placeholders.
_GITLEAKS_SUPPRESS_EXACT_VALUES = {
"YOUR_YOUTUBE_API_KEY",
"YOUR_YOUTUBE_CHANNEL_ID",
"YOUR_GITHUB_PERSONAL_ACCESS_TOKEN",
}
# Findings in these files are suppressed regardless of value -- they are
# template/example files that are expected to only ever contain placeholders.
_GITLEAKS_SUPPRESS_PATHS = [
"config_secrets.template.json",
"config.template.json",
]
# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────
def _load(path: Path) -> tuple[dict | list | None, str | None]:
"""Load a JSON artifact file.
Returns (data, error): error is None on success (data is whatever was
parsed, which may legitimately be an empty list/dict for a clean scan);
otherwise error is a human-readable reason the artifact is unavailable,
distinguishing "missing/malformed artifact" from "valid empty result" so
callers don't silently treat a broken CI job as a clean pass.
"""
if not path.exists():
return None, f"artifact not found: {path}"
try:
return json.loads(path.read_text(encoding="utf-8")), None
except (json.JSONDecodeError, OSError) as exc:
return None, f"could not read/parse {path}: {exc}"
def _md_sanitize_cell(value: object) -> str:
"""Escape/normalize a value so scanner-controlled content (a matched
secret, a bandit issue_text, a file path) can't alter the Markdown
table's structure: pipes would add bogus columns, newlines would break
out of the row (or forge a fake header/separator line)."""
text = str(value)
text = text.replace("\\", "\\\\").replace("|", "\\|")
text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
return text
def _md_table_row(*cells: str) -> str:
return "| " + " | ".join(_md_sanitize_cell(c) for c in cells) + " |"
# ─────────────────────────────────────────────────────────────────────────────
# Per-tool summarizers
# Returns: (markdown_lines: list[str], critical_count: int, available: bool)
# `available=False` means the artifact was missing or malformed -- distinct
# from a valid scan that simply found nothing -- so the caller can report
# INCOMPLETE instead of silently counting it as a clean pass.
# ─────────────────────────────────────────────────────────────────────────────
def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int, bool]:
data, error = _load(artifact_dir / "sast-results" / "bandit-results.json")
if error:
return [f"_bandit results unavailable: {error}_"], 0, False
results = data.get("results", [])
high = [r for r in results if r.get("issue_severity") == "HIGH"]
medium = [r for r in results if r.get("issue_severity") == "MEDIUM"]
low = [r for r in results if r.get("issue_severity") == "LOW"]
lines = [
f"**Bandit**: {len(high)} HIGH · {len(medium)} MEDIUM · {len(low)} LOW"
]
if high:
lines += [
"",
"| Severity | File | Line | Issue |",
"| --- | --- | --- | --- |",
]
for r in high[:10]:
fname = Path(r.get("filename", "")).name
lines.append(_md_table_row(
"HIGH", f"`{fname}`",
str(r.get("line_number", "?")),
r.get("issue_text", "")
))
if len(high) > 10:
lines.append(f"_… and {len(high) - 10} more HIGH findings_")
return lines, len(high), True
def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
data, error = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json")
if error:
return [f"_pip-audit results unavailable: {error}_"], 0, False
# pip-audit JSON format: {"dependencies": [{"name": ..., "vulns": [...]}]}
vulns: list[dict] = []
for dep in data.get("dependencies", []):
for v in dep.get("vulns", []):
vulns.append({"package": dep.get("name", "?"), **v})
lines = [f"**pip-audit**: {len(vulns)} vulnerabilities found"]
if vulns:
lines += ["", "| Package | ID | Fix |", "| --- | --- | --- |"]
for v in vulns[:10]:
fix = v.get("fix_versions", ["none"])
fix_str = ", ".join(fix) if fix else "none"
lines.append(_md_table_row(
v.get("package", "?"),
v.get("id", "?"),
fix_str,
))
# Treat known vulnerabilities as warnings, not critical (they may be unavoidable)
return lines, 0, True
def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int, bool]:
data, error = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json")
if error:
return [f"_gitleaks results unavailable: {error}_"], 0, False
if not isinstance(data, list):
data = []
real_findings = []
suppressed = 0
for finding in data:
secret_val = str(finding.get("Secret", "") or finding.get("Match", ""))
file_name = Path(finding.get("File", "")).name
if (secret_val in _GITLEAKS_SUPPRESS_EXACT_VALUES
or file_name in _GITLEAKS_SUPPRESS_PATHS):
suppressed += 1
else:
real_findings.append(finding)
lines = [
f"**Gitleaks**: {len(real_findings)} finding(s) "
f"({suppressed} suppressed as template placeholders)"
]
if real_findings:
lines += ["", "| Rule | File | Line | Description |", "| --- | --- | --- | --- |"]
for f in real_findings[:10]:
fname = Path(f.get("File", "")).name
lines.append(_md_table_row(
f.get("RuleID", "?"),
f"`{fname}`",
str(f.get("StartLine", "?")),
f.get("Description", ""),
))
critical = len(real_findings) # any real secret is critical
return lines, critical, True
def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int, bool]:
data, error = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json")
if error:
return [f"_security proofs results unavailable: {error}_"], 0, False
if not isinstance(data, list):
data = []
critical = [r for r in data if r.get("severity") == "CRITICAL"]
warnings = [r for r in data if r.get("severity") == "WARNING"]
passed = [r for r in data if r.get("severity") == "PASS"]
skipped = [r for r in data if r.get("severity") == "SKIP"]
lines = [
f"**Security Proofs**: "
f"{len(passed)} PASS · {len(warnings)} WARN · "
f"{len(critical)} CRITICAL · {len(skipped)} SKIP",
"",
]
_icon = {"PASS": "", "INFO": "", "WARNING": "⚠️", # nosec B105 - severity labels, not credentials
"CRITICAL": "🚨", "SKIP": "⏭️"}
for r in data:
icon = _icon.get(r.get("severity", ""), "")
lines.append(
f"- {icon} **{r.get('test_id', '?')}**: {r.get('message', '')}"
)
if r.get("details") and r.get("severity") in ("CRITICAL", "WARNING"):
lines.append(f" - _{r['details']}_")
return lines, len(critical), True
def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
data, error = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json")
if error:
return [f"_plugin audit results unavailable: {error}_"], 0, False
summary = data.get("summary", {})
findings = data.get("findings", [])
critical_findings = [f for f in findings if f.get("severity") == "CRITICAL"]
warning_findings = [f for f in findings if f.get("severity") == "WARNING"]
lines = [
f"**Plugin Audit**: {data.get('plugins_scanned', '?')} plugins scanned — "
f"{summary.get('critical', 0)} CRITICAL · {summary.get('warnings', 0)} WARNINGS"
]
if critical_findings:
lines += ["", "| Plugin | File | Line | Rule | Message |",
"| --- | --- | --- | --- | --- |"]
for f in critical_findings[:10]:
fname = Path(f.get("file", "")).name
lines.append(_md_table_row(
f.get("plugin_id", "?"),
f"`{fname}`",
str(f.get("line", "?")),
f.get("rule", "?"),
f.get("message", ""),
))
if warning_findings and not critical_findings:
lines.append(f"\n_{len(warning_findings)} warning(s) found — see artifact for details_")
return lines, summary.get("critical", 0), True
# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
def main() -> int:
parser = argparse.ArgumentParser(
description="Generate consolidated security audit report",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--artifact-dir", required=True,
help="Directory containing downloaded CI artifacts")
parser.add_argument("--output", "-o", required=True,
help="Output Markdown file path")
parser.add_argument("--verbose", "-v", action="store_true")
args = parser.parse_args()
artifact_dir = Path(args.artifact_dir)
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
bandit_lines, bandit_crit, bandit_ok = _summarize_bandit(artifact_dir)
pip_audit_lines, pip_audit_crit, pip_audit_ok = _summarize_pip_audit(artifact_dir)
gitleaks_lines, gitleaks_crit, gitleaks_ok = _summarize_gitleaks(artifact_dir)
proofs_lines, proofs_crit, proofs_ok = _summarize_security_proofs(artifact_dir)
plugins_lines, plugins_crit, plugins_ok = _summarize_plugin_audit(artifact_dir)
unavailable_tools = [
name for name, ok in [
("bandit", bandit_ok), ("pip-audit", pip_audit_ok),
("gitleaks", gitleaks_ok), ("security-proofs", proofs_ok),
("plugin-audit", plugins_ok),
] if not ok
]
total_critical = bandit_crit + pip_audit_crit + gitleaks_crit + proofs_crit + plugins_crit
if unavailable_tools:
# A missing/malformed artifact means that tool's checks never
# actually ran -- this must not be reported as a clean PASS just
# because the *artifacts that did load* found nothing.
overall = "INCOMPLETE ⚠️"
elif total_critical > 0:
overall = "ACTION REQUIRED 🚨"
else:
overall = "PASSED ✅"
def section(title: str, lines: list[str]) -> str:
return f"### {title}\n\n" + "\n".join(lines) + "\n"
incomplete_note = (
f"\n_⚠️ Incomplete: results unavailable for {', '.join(unavailable_tools)} "
f"— see the corresponding section(s) below for details_\n"
if unavailable_tools else ""
)
report = f"""## 🔒 Security Audit — {overall}
_Generated: {timestamp}_
{incomplete_note}
| Critical | High/Warn | Overall |
| :---: | :---: | :---: |
| {'🚨 ' + str(total_critical) if total_critical else '✅ 0'} | see below | {overall} |
---
{section('SAST — Bandit', bandit_lines)}
{section('Dependencies — pip-audit', pip_audit_lines)}
{section('Secrets — Gitleaks', gitleaks_lines)}
{section('LEDMatrix Security Proofs', proofs_lines)}
{section('Plugin Security Audit', plugins_lines)}
---
_Total critical findings: **{total_critical}**_
"""
output_path = Path(args.output)
output_path.write_text(report, encoding="utf-8")
if args.verbose:
print(f" Report written to: {output_path}")
print(f" Status: {overall}")
print(f" Critical findings: {total_critical}")
print(f" bandit={bandit_crit} pip-audit={pip_audit_crit} "
f"gitleaks={gitleaks_crit} proofs={proofs_crit} plugins={plugins_crit}")
if unavailable_tools:
print(f" Unavailable: {', '.join(unavailable_tools)}")
if unavailable_tools:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
+5 -28
View File
@@ -33,7 +33,6 @@ POWEROFF_PATH=$(command -v poweroff) || true
BASH_PATH=$(command -v bash) || true
JOURNALCTL_PATH=$(command -v journalctl) || true
SAFE_RM_PATH="$PROJECT_ROOT/scripts/fix_perms/safe_plugin_rm.sh"
SAFE_PIP_INSTALL_PATH="$PROJECT_ROOT/scripts/fix_perms/safe_pip_install.sh"
# Validate required commands (systemctl, bash, python3 are essential)
for CMD_NAME in SYSTEMCTL_PATH BASH_PATH PYTHON_PATH; do
@@ -49,15 +48,11 @@ if [ ${#MISSING_CMDS[@]} -gt 0 ]; then
exit 1
fi
# Validate helper scripts exist
# Validate helper script exists
if [ ! -f "$SAFE_RM_PATH" ]; then
echo "Error: Safe plugin removal helper not found: $SAFE_RM_PATH" >&2
exit 1
fi
if [ ! -f "$SAFE_PIP_INSTALL_PATH" ]; then
echo "Error: Safe pip install helper not found: $SAFE_PIP_INSTALL_PATH" >&2
exit 1
fi
echo "Command paths:"
echo " Python: $PYTHON_PATH"
@@ -67,7 +62,6 @@ echo " Poweroff: ${POWEROFF_PATH:-(not found, skipping)}"
echo " Bash: $BASH_PATH"
echo " Journalctl: ${JOURNALCTL_PATH:-(not found, skipping)}"
echo " Safe plugin rm: $SAFE_RM_PATH"
echo " Safe pip install: $SAFE_PIP_INSTALL_PATH"
# Create a temporary sudoers file
TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$"
@@ -107,22 +101,13 @@ TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$"
fi
# Required: python3, bash
# NOTE: display_controller.py/start_display.sh/stop_display.sh live at the
# project root, not under scripts/install/ (where this script lives) —
# must use PROJECT_ROOT here, not PROJECT_DIR.
echo "$WEB_USER ALL=(ALL) NOPASSWD: $PYTHON_PATH $PROJECT_ROOT/display_controller.py"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT/start_display.sh"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT/stop_display.sh"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $PYTHON_PATH $PROJECT_DIR/display_controller.py"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_DIR/start_display.sh"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_DIR/stop_display.sh"
echo ""
echo "# Allow web user to remove plugin directories via vetted helper script"
echo "# The helper validates that the target path resolves inside plugin-repos/ or plugins/"
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_RM_PATH *"
echo ""
echo "# Allow web user to install a plugin's requirements.txt as root via vetted"
echo "# helper script, so packages are visible to root-run ledmatrix.service"
echo "# (not just the web interface's own user). The helper validates the target"
echo "# is requirements.txt at the project root or under plugin-repos/ or plugins/."
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_PIP_INSTALL_PATH *"
} > "$TEMP_SUDOERS"
echo ""
@@ -141,7 +126,6 @@ echo "- Run display_controller.py directly"
echo "- Execute start_display.sh and stop_display.sh"
echo "- Reboot and shutdown the system"
echo "- Remove plugin directories (for update/uninstall when root-owned files block deletion)"
echo "- Install plugin/base requirements.txt as root (so ledmatrix.service can see them)"
echo ""
# Ask for confirmation
@@ -163,13 +147,6 @@ fi
if ! sudo chmod 755 "$SAFE_RM_PATH"; then
echo "Warning: Could not set permissions on $SAFE_RM_PATH"
fi
echo "Hardening safe_pip_install.sh ownership..."
if ! sudo chown root:root "$SAFE_PIP_INSTALL_PATH"; then
echo "Warning: Could not set ownership on $SAFE_PIP_INSTALL_PATH"
fi
if ! sudo chmod 755 "$SAFE_PIP_INSTALL_PATH"; then
echo "Warning: Could not set permissions on $SAFE_PIP_INSTALL_PATH"
fi
if sudo cp "$TEMP_SUDOERS" /etc/sudoers.d/ledmatrix_web; then
echo "Configuration applied successfully!"
@@ -183,7 +160,7 @@ if sudo cp "$TEMP_SUDOERS" /etc/sudoers.d/ledmatrix_web; then
echo "✗ systemctl status ledmatrix.service - Failed"
fi
if sudo -n test -f "$PROJECT_ROOT/start_display.sh"; then
if sudo -n test -f "$PROJECT_DIR/start_display.sh"; then
echo "✓ File access test - OK"
else
echo "✗ File access test - Failed"
+4 -2
View File
@@ -14,6 +14,9 @@ else
ACTUAL_USER=$(whoami)
fi
# Get the home directory of the actual user
USER_HOME=$(eval echo ~$ACTUAL_USER)
# Determine the Project Root Directory (parent of scripts/install/)
PROJECT_ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
@@ -31,8 +34,7 @@ echo "Generating service file with dynamic paths..."
WEB_SERVICE_FILE_CONTENT=$(cat <<EOF
[Unit]
Description=LED Matrix Web Interface Service
After=network-online.target
Wants=network-online.target
After=network.target
[Service]
Type=simple
+2 -8
View File
@@ -340,14 +340,9 @@ main() {
echo ""
# Execute with proper error handling and non-interactive mode
# Temporarily disable errexit AND the ERR trap to capture exit code instead of
# exiting immediately. `set +e` alone does not suppress the ERR trap, so without
# `trap '' ERR` a non-zero exit from first_time_install.sh would trigger on_error
# here with the generic "Main installation" message instead of the detailed
# if/else handling below.
# Temporarily disable errexit to capture exit code instead of exiting immediately
set +e
trap '' ERR
# Check /tmp permissions - only fix if actually wrong (common in automated scenarios)
# When running manually, /tmp usually has correct permissions (1777)
TMP_PERMS=$(stat -c '%a' /tmp 2>/dev/null || echo "unknown")
@@ -375,7 +370,6 @@ main() {
sudo -E env TMPDIR=/tmp LEDMATRIX_ASSUME_YES=1 bash ./first_time_install.sh -y </dev/null
fi
INSTALL_EXIT_CODE=$?
trap 'on_error $LINENO' ERR # Re-enable ERR trap
set -e # Re-enable errexit
if [ $INSTALL_EXIT_CODE -eq 0 ]; then
+82 -149
View File
@@ -6,143 +6,82 @@ then falls back to pip with --break-system-packages
import subprocess
import sys
import tempfile
import warnings
from collections import deque
from pathlib import Path
from typing import List, Tuple
# How many trailing lines of a failed command's output to keep for the
# end-of-run failure summary. Keeps the root cause near the end of the log,
# which is where first_time_install.sh's error handler tails from.
ERROR_TAIL_LINES = 15
def _run(cmd: List[str]) -> Tuple[bool, str]:
"""Run a command, streaming combined stdout/stderr to a temp file.
Returns (success, output) instead of raising, so callers can report
*why* a command failed rather than just that it failed. `output` is
bounded to the last ERROR_TAIL_LINES lines so failures from very
chatty commands (e.g. pip build logs) don't get buffered in memory.
"""
with tempfile.TemporaryFile(mode='w+b') as f:
result = subprocess.run(cmd, stdout=f, stderr=subprocess.STDOUT) # nosec B603 B607 - hardcoded apt/pip args # nosemgrep
f.seek(0)
# Stream line-by-line so only the last ERROR_TAIL_LINES are ever held
# in memory, regardless of how much output the command produced.
tail = deque(
(line.decode('utf-8', errors='replace').rstrip('\n') for line in f),
maxlen=ERROR_TAIL_LINES,
)
return result.returncode == 0, '\n'.join(tail)
def install_via_apt(package_name: str) -> Tuple[bool, str]:
"""Try to install a package via apt. Returns (success, output)."""
# Map pip package names to apt package names
apt_package_map = {
'flask': 'python3-flask',
'PIL': 'python3-pil',
'freetype': 'python3-freetype',
'psutil': 'python3-psutil',
'werkzeug': 'python3-werkzeug',
'numpy': 'python3-numpy',
'requests': 'python3-requests',
'python-dateutil': 'python3-dateutil',
'pytz': 'python3-tz',
'geopy': 'python3-geopy',
'unidecode': 'python3-unidecode',
'websockets': 'python3-websockets',
'websocket-client': 'python3-websocket-client'
}
apt_package = apt_package_map.get(package_name, f'python3-{package_name}')
print(f"Trying to install {apt_package} via apt...")
success, output = _run(['sudo', 'apt-get', '-o', 'DPkg::Lock::Timeout=180', 'install', '-y', apt_package])
if success:
def install_via_apt(package_name):
"""Try to install a package via apt."""
try:
# Map pip package names to apt package names
apt_package_map = {
'flask': 'python3-flask',
'PIL': 'python3-pil',
'freetype': 'python3-freetype',
'psutil': 'python3-psutil',
'werkzeug': 'python3-werkzeug',
'numpy': 'python3-numpy',
'requests': 'python3-requests',
'python-dateutil': 'python3-dateutil',
'pytz': 'python3-tz',
'geopy': 'python3-geopy',
'unidecode': 'python3-unidecode',
'websockets': 'python3-websockets',
'websocket-client': 'python3-websocket-client'
}
apt_package = apt_package_map.get(package_name, f'python3-{package_name}')
print(f"Trying to install {apt_package} via apt...")
subprocess.check_call([
'sudo', 'apt', 'update'
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
subprocess.check_call([
'sudo', 'apt', 'install', '-y', apt_package
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
print(f"Successfully installed {apt_package} via apt")
return True, ""
return True
except subprocess.CalledProcessError:
print(f"Failed to install {package_name} via apt, will try pip")
return False
print(f"Failed to install {apt_package} via apt, will try pip")
return False, output
def install_via_pip(package_name: str) -> Tuple[bool, str]:
def install_via_pip(package_name):
"""Install a package via pip with --break-system-packages and --prefer-binary.
--break-system-packages allows pip to install into the system Python on
Debian/Ubuntu-based systems without a virtual environment.
--prefer-binary prefers pre-built wheels over source distributions to avoid
exhausting /tmp space during compilation.
--ignore-installed stops pip from trying to *uninstall* packages that were
installed by apt (e.g. python3-requests). Those Debian packages ship no
pip RECORD file, so an uninstall attempt fails with "uninstall-no-record-file"
and aborts the whole install. With --ignore-installed, pip lays the new
version down in /usr/local where it shadows the apt copy instead of removing
it. This matters when a pip dependency (google-api-python-client pulls a
newer requests) needs to upgrade an apt-managed package.
Returns (success, output).
"""
print(f"Installing {package_name} via pip...")
success, output = _run([
sys.executable, '-m', 'pip', 'install',
'--break-system-packages', '--prefer-binary', '--ignore-installed', package_name
])
if success:
try:
print(f"Installing {package_name} via pip...")
subprocess.check_call([
sys.executable, '-m', 'pip', 'install', '--break-system-packages', '--prefer-binary', package_name
])
print(f"Successfully installed {package_name} via pip")
return True, ""
return True
except subprocess.CalledProcessError as e:
print(f"Failed to install {package_name} via pip: {e}")
return False
print(f"Failed to install {package_name} via pip (see failure summary at end of log)")
return False, output
# Distribution (pip/apt) names whose importable module name differs.
IMPORT_NAME_MAP = {
'python-dateutil': 'dateutil',
'websocket-client': 'websocket',
}
def check_package_installed(package_name: str) -> bool:
def check_package_installed(package_name):
"""Check if a package is already installed."""
import_name = IMPORT_NAME_MAP.get(package_name, package_name)
# Suppress deprecation warnings when checking if packages are installed
# (we're just checking, not using them)
with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=DeprecationWarning)
try:
__import__(import_name)
__import__(package_name)
return True
except ImportError:
return False
def print_failure_summary(failed_packages: List[str], failure_details: dict) -> None:
print("\n" + "=" * 60)
print("DEPENDENCY INSTALLATION FAILURES - DETAILS")
print("=" * 60)
for package in failed_packages:
print(f"\nPackage: {package}")
print("-" * 40)
output = failure_details.get(package, "").strip()
if not output:
print(" (no output captured)")
continue
for line in output.splitlines()[-ERROR_TAIL_LINES:]:
print(f" {line}")
print("=" * 60)
def main():
"""Main installation function."""
print("Installing dependencies for LED Matrix Web Interface V2...")
print("Refreshing apt package index...")
_run(['sudo', 'apt', 'update']) # best-effort; individual installs surface their own errors
# List of required packages
required_packages = [
'flask',
@@ -159,23 +98,19 @@ def main():
'websockets',
'websocket-client'
]
failed_packages = []
failure_details = {}
for package in required_packages:
if check_package_installed(package):
print(f"{package} is already installed")
continue
# Try apt first, then pip
ok, apt_output = install_via_apt(package)
if not ok:
ok, pip_output = install_via_pip(package)
if not ok:
if not install_via_apt(package):
if not install_via_pip(package):
failed_packages.append(package)
failure_details[package] = pip_output or apt_output
# Install packages that don't have apt equivalents
special_packages = [
'timezonefinder>=6.5.0,<7.0.0',
@@ -187,49 +122,47 @@ def main():
'python-socketio>=5.11.0,<6.0.0',
'python-engineio>=4.9.0,<5.0.0'
]
for package in special_packages:
ok, pip_output = install_via_pip(package)
if not ok:
if not install_via_pip(package):
failed_packages.append(package)
failure_details[package] = pip_output
# Install rgbmatrix module from local source (optional - may already be installed in Step 6)
# Check if already installed first
if check_package_installed('rgbmatrix'):
print("rgbmatrix module already installed, skipping...")
else:
print("Installing rgbmatrix module from local source...")
# Get project root (parent of scripts directory)
PROJECT_ROOT = Path(__file__).parent.parent
rgbmatrix_path = PROJECT_ROOT / 'rpi-rgb-led-matrix-master' / 'bindings' / 'python'
if rgbmatrix_path.exists():
# Check if the module has been built (look for setup.py)
setup_py = rgbmatrix_path / 'setup.py'
if setup_py.exists():
# Try installing - use regular install, not editable mode
# This is optional for web interface and should already be installed in Step 6
ok, output = _run([sys.executable, '-m', 'pip', 'install', '--break-system-packages', str(rgbmatrix_path)])
if ok:
try:
# Get project root (parent of scripts directory)
PROJECT_ROOT = Path(__file__).parent.parent
rgbmatrix_path = PROJECT_ROOT / 'rpi-rgb-led-matrix-master' / 'bindings' / 'python'
if rgbmatrix_path.exists():
# Check if the module has been built (look for setup.py)
setup_py = rgbmatrix_path / 'setup.py'
if setup_py.exists():
# Try installing - use regular install, not editable mode
# This is optional for web interface and should already be installed in Step 6
subprocess.check_call([
sys.executable, '-m', 'pip', 'install', '--break-system-packages', str(rgbmatrix_path)
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
print("rgbmatrix module installed successfully")
else:
# Don't fail the whole installation - rgbmatrix is optional for web interface
# and should be installed in Step 6 of first_time_install.sh
print("Warning: Failed to install rgbmatrix module:")
for line in output.strip().splitlines()[-ERROR_TAIL_LINES:]:
print(f" {line}")
print(" This is normal if rgbmatrix hasn't been built yet (Step 6).")
print(" The web interface will work without it.")
print("Warning: rgbmatrix setup.py not found, module may need to be built first")
print(" This is normal if Step 6 hasn't completed yet.")
else:
print("Warning: rgbmatrix setup.py not found, module may need to be built first")
print(" This is normal if Step 6 hasn't completed yet.")
else:
print("Warning: rgbmatrix source not found (this is normal if Step 6 hasn't run yet)")
print("Warning: rgbmatrix source not found (this is normal if Step 6 hasn't run yet)")
except subprocess.CalledProcessError as e:
# Don't fail the whole installation - rgbmatrix is optional for web interface
# and should be installed in Step 6 of first_time_install.sh
print(f"Warning: Failed to install rgbmatrix module: {e}")
print(" This is normal if rgbmatrix hasn't been built yet (Step 6).")
print(" The web interface will work without it.")
# Don't add to failed_packages since it's optional
if failed_packages:
print(f"\nFailed to install the following packages: {failed_packages}")
print("You may need to install them manually or check your system configuration.")
print_failure_summary(failed_packages, failure_details)
return False
else:
print("\nAll dependencies installed successfully!")
-593
View File
@@ -1,593 +0,0 @@
#!/usr/bin/env python3
"""
LEDMatrix Security Proof Tests
Automated proofs that run in CI to verify security properties hold on every
commit. Inspired by the Huntarr security review approach of using standard
tooling to confirm specific vulnerability classes are absent.
Usage:
python scripts/prove_security.py
python scripts/prove_security.py --verbose
python scripts/prove_security.py --output results.json
Exit code: 1 only if CRITICAL findings are detected. Warnings are reported
but do not block CI.
"""
import ast
import argparse
import hashlib
import json
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# ─────────────────────────────────────────────────────────────────────────────
# Result dataclass
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class TestResult:
test_id: str
severity: str # PASS | INFO | WARNING | CRITICAL | SKIP
message: str
details: str = ""
def to_dict(self) -> dict:
return asdict(self)
@property
def icon(self) -> str:
return {
"PASS": "", # nosec B105 - severity label, not a credential
"INFO": "",
"WARNING": "⚠️ ",
"CRITICAL": "🚨",
"SKIP": "⏭️ ",
}.get(self.severity, "")
# ─────────────────────────────────────────────────────────────────────────────
# T1: Plugin Loading / Zip Slip
# ─────────────────────────────────────────────────────────────────────────────
def test_t1a_zip_slip_protection() -> TestResult:
"""
Verify that zip-slip protection actually guards zip extraction in
store_manager.py.
A whole-file substring check for "is_relative_to"/"Zip-slip detected"
would pass even if the guard existed somewhere unrelated, or covered
only one of several extract()/extractall() call sites. Instead, this
walks the AST: for every extract()/extractall() call, it confirms an
is_relative_to() check (and the "Zip-slip detected" log) appears
earlier in that same enclosing function -- validate-then-bulk-extract
(validate every member, then call extractall() only after all passed)
counts as protecting the call, since it covers the same member list.
"""
store_manager = PROJECT_ROOT / "src" / "plugin_system" / "store_manager.py"
if not store_manager.exists():
return TestResult("T1a", "CRITICAL",
"store_manager.py not found",
f"Expected at {store_manager}")
content = store_manager.read_text(encoding="utf-8")
try:
tree = ast.parse(content, filename=str(store_manager))
except SyntaxError as exc:
return TestResult("T1a", "CRITICAL",
"store_manager.py could not be parsed",
str(exc))
extraction_sites = 0
unprotected: list[str] = []
for func in ast.walk(tree):
if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
extract_calls = [
node for node in ast.walk(func)
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
and node.func.attr in ("extract", "extractall")
]
if not extract_calls:
continue
extraction_sites += len(extract_calls)
guard_lines = [
n.lineno for n in ast.walk(func)
if isinstance(n, ast.Attribute) and n.attr == "is_relative_to"
]
has_zip_slip_log = any(
isinstance(n, ast.Constant) and isinstance(n.value, str)
and "Zip-slip detected" in n.value
for n in ast.walk(func)
)
for call in extract_calls:
guarded = has_zip_slip_log and any(g < call.lineno for g in guard_lines)
if not guarded:
unprotected.append(
f"{func.name}() line {call.lineno}: {call.func.attr}() call not "
f"clearly preceded by an is_relative_to() guard + Zip-slip log "
f"in the same function"
)
if extraction_sites == 0:
return TestResult("T1a", "WARNING",
"No zipfile extract()/extractall() calls found in store_manager.py",
"Verify plugin installation no longer extracts zip archives, "
"or that this check still targets the right file")
if unprotected:
return TestResult("T1a", "CRITICAL",
f"{len(unprotected)} of {extraction_sites} zip extraction "
f"call(s) not clearly guarded",
"; ".join(unprotected))
return TestResult("T1a", "PASS",
"Zip-slip protection verified",
f"All {extraction_sites} extract()/extractall() call(s) in "
f"store_manager.py are preceded by an is_relative_to() guard "
f"with a Zip-slip log in the same function")
def test_t1b_dangerous_plugin_calls() -> list[TestResult]:
"""
Scan plugin directories for dangerous function calls (eval, exec).
These represent arbitrary code execution risks in plugin code.
"""
results = []
plugin_dirs = [
PROJECT_ROOT / "plugins",
PROJECT_ROOT / "plugin-repos",
]
violations: list[str] = []
files_scanned = 0
scan_errors: list[str] = []
for base in plugin_dirs:
if not base.exists():
continue
for plugin_dir in sorted(base.iterdir()):
if not plugin_dir.is_dir() or plugin_dir.name.startswith(('.', '_')):
continue
for py_file in plugin_dir.rglob("*.py"):
files_scanned += 1
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source, filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
if node.func.id in ("eval", "exec"):
rel = py_file.relative_to(PROJECT_ROOT)
violations.append(
f"{rel}:{node.lineno}{node.func.id}() call")
except (SyntaxError, OSError) as exc:
# A file we couldn't parse/read was never actually
# scanned for eval()/exec() -- that must block this
# test, not silently pass as if it were clean.
rel = py_file.relative_to(PROJECT_ROOT)
scan_errors.append(f"{rel}{type(exc).__name__}: {exc}")
if scan_errors:
results.append(TestResult(
"T1b", "CRITICAL",
f"{len(scan_errors)} plugin file(s) could not be scanned for eval()/exec()",
"; ".join(scan_errors[:10])
))
if violations:
results.append(TestResult(
"T1b", "CRITICAL",
f"Dangerous function calls found in plugins ({len(violations)} instance(s))",
"; ".join(violations[:10])
))
elif not scan_errors:
results.append(TestResult(
"T1b", "PASS",
"No eval()/exec() calls found in plugins",
f"{files_scanned} plugin Python files scanned"
))
return results
# ─────────────────────────────────────────────────────────────────────────────
# T2: API Surface Inventory
# ─────────────────────────────────────────────────────────────────────────────
def test_t2a_api_surface_inventory() -> TestResult:
"""
Document the API surface area.
This app intentionally has no authentication (local-only Raspberry Pi
design, documented in web_interface/app.py). This test produces an
inventory for audit purposes and warns only if the design-intent comment
is removed from app.py (which would indicate someone deleted the rationale
without adding auth, rather than a deliberate undocumented change).
"""
api_file = PROJECT_ROOT / "web_interface" / "blueprints" / "api_v3.py"
app_file = PROJECT_ROOT / "web_interface" / "app.py"
if not api_file.exists():
return TestResult("T2a", "WARNING", "api_v3.py not found", str(api_file))
api_content = api_file.read_text(encoding="utf-8")
routes = re.findall(r"@api_v3\.route\('([^']+)'", api_content)
csrf_documented = False
if app_file.exists():
app_content = app_file.read_text(encoding="utf-8")
csrf_documented = "CSRF protection disabled for local-only" in app_content
summary = (
f"{len(routes)} API routes in api_v3.py. "
f"No auth decorators (intentional local-only design). "
f"CSRF disabled: {'YES — design intent documented in app.py' if csrf_documented else 'YES — but design intent comment NOT found in app.py'}. "
f"Rate limiting: 1000/min."
)
if not csrf_documented:
return TestResult(
"T2a", "WARNING",
"CSRF is disabled but the design-intent comment is missing from app.py",
"Add the rationale comment back, or add proper CSRF protection if "
"the app is now internet-facing"
)
# There is currently no config mechanism that actually enforces the
# local-only boundary the design-intent comment describes -- app.py
# hardcodes host='0.0.0.0' unconditionally, so nothing here can confirm
# this deployment is in fact LAN-only. Reporting this as mere INFO
# understates that: an unauthenticated, CSRF-disabled API surface is a
# real risk the moment this ever runs somewhere other than a home LAN,
# documented rationale or not.
return TestResult(
"T2a", "WARNING",
"API surface has no auth and CSRF disabled; enforcement of the "
"documented local-only boundary cannot be confirmed",
summary
)
# ─────────────────────────────────────────────────────────────────────────────
# T3: Secrets & Credential Handling
# ─────────────────────────────────────────────────────────────────────────────
# Patterns that suggest real credentials (must be >8 chars, not placeholders)
_SECRET_PATTERNS = [
(r'(?i)password\s*=\s*["\'](?!none|empty|placeholder|example|test|default|""|'')[^"\']{8,}["\']', "WARNING", "password"),
(r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "api_key"),
(r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "secret"),
# Real GitHub token pattern
(r'ghp_[a-zA-Z0-9]{36}', "CRITICAL", "github_token"),
# Generic long bearer tokens
(r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING", "bearer_token"),
]
_TEMPLATE_SKIP_STRINGS = [
"YOUR_", "PLACEHOLDER", "_HERE", "example.com", "config_secrets.template",
"prove_security", # this file itself
]
_SCAN_DIRS = ["src", "web_interface", "scripts"]
def test_t3a_hardcoded_secrets() -> TestResult:
"""Scan source code for hardcoded credentials."""
violations: list[str] = []
for dir_name in _SCAN_DIRS:
scan_dir = PROJECT_ROOT / dir_name
if not scan_dir.exists():
continue
for py_file in scan_dir.rglob("*.py"):
# Skip test files and this script
if "test" in str(py_file).lower() or "prove_security" in str(py_file):
continue
try:
content = py_file.read_text(encoding="utf-8")
except OSError:
continue
for pattern, severity, pattern_type in _SECRET_PATTERNS:
for match in re.finditer(pattern, content):
line_content = match.group(0)
# Skip lines containing template placeholder strings.
# line_content is only used for this in-memory check --
# it must never be stored or included in output below.
if any(skip in line_content for skip in _TEMPLATE_SKIP_STRINGS):
continue
rel = py_file.relative_to(PROJECT_ROOT)
line_no = content[: match.start()].count("\n") + 1
# Redacted fingerprint lets the same finding be recognized
# across scans without ever reporting the matched
# credential itself (which would otherwise get published
# into CI logs, JSON artifacts, and PR comments -- wider
# exposure than the original leak).
fingerprint = hashlib.sha256(line_content.encode()).hexdigest()[:12]
violations.append(
f"[{severity}] {rel}:{line_no}{pattern_type} "
f"(fingerprint {fingerprint})"
)
critical_violations = [v for v in violations if "[CRITICAL]" in v]
if critical_violations:
return TestResult(
"T3a", "CRITICAL",
f"Hardcoded secrets found ({len(critical_violations)} critical)",
"; ".join(critical_violations[:5])
)
if violations:
return TestResult(
"T3a", "WARNING",
f"Potential hardcoded secrets found ({len(violations)} instance(s))",
"; ".join(violations[:5])
)
return TestResult("T3a", "PASS", "No hardcoded secrets detected",
f"Scanned {', '.join(_SCAN_DIRS)}")
def test_t3b_plaintext_password_storage() -> TestResult:
"""
Check for user account password storage without hashing.
The LEDMatrix app has no user account system, so this should produce INFO.
It would only CRITICAL if someone added user auth and stored passwords without hashing.
We require all three of: a password *variable assignment or DB operation*,
a clear storage call (INSERT / db commit / ORM save), and no hashing lib present
to avoid false positives from files that contain 'password' for WiFi handling
and '.save()' for image/file saving in unrelated functions.
"""
hashing_libs = ["bcrypt", "argon2", "pbkdf2", "scrypt",
"generate_password_hash", "hashpw", "make_password"]
# Patterns that indicate password being stored in a database / ORM context.
# Must be specific enough to avoid matching set.add(), file.save(), etc.
db_storage_patterns = ["INSERT INTO", "db.session", "session.add(", "session.commit(", "orm.save"]
password_storage_found = False
for dir_name in _SCAN_DIRS:
scan_dir = PROJECT_ROOT / dir_name
if not scan_dir.exists():
continue
for py_file in scan_dir.rglob("*.py"):
try:
content = py_file.read_text(encoding="utf-8")
except OSError:
continue
# Require DB/ORM context specifically — not just any .save() call
if ("password" in content.lower() and
any(store in content for store in db_storage_patterns) and
not any(h in content for h in hashing_libs)):
password_storage_found = True
if password_storage_found:
return TestResult(
"T3b", "CRITICAL",
"Potential plaintext password storage in database/ORM detected",
"Found password + database storage operations without a recognized hashing library"
)
return TestResult("T3b", "INFO",
"No plaintext password storage detected",
"App has no user account system — expected result")
# ─────────────────────────────────────────────────────────────────────────────
# T4: Path Traversal
# ─────────────────────────────────────────────────────────────────────────────
def test_t4a_path_traversal() -> TestResult:
"""
Verify static file serving uses send_from_directory (safe) rather than
open() with user-supplied paths. Also checks for extractall() calls that
lack the is_relative_to() guard.
"""
issues: list[str] = []
app_file = PROJECT_ROOT / "web_interface" / "app.py"
if app_file.exists():
content = app_file.read_text(encoding="utf-8")
# The file-serve route should use send_from_directory or commonpath
if "send_from_directory" not in content and "commonpath" not in content:
issues.append("app.py: file-serve routes may not use send_from_directory/commonpath")
# Check all extractall() calls have a preceding is_relative_to guard
for py_file in (PROJECT_ROOT / "src").rglob("*.py"):
try:
content = py_file.read_text(encoding="utf-8")
except OSError:
continue
if "extractall(" in content and "is_relative_to" not in content:
rel = py_file.relative_to(PROJECT_ROOT)
issues.append(f"{rel}: extractall() without is_relative_to() guard")
if issues:
return TestResult(
"T4a", "WARNING",
f"Potential path traversal patterns found ({len(issues)})",
"; ".join(issues)
)
return TestResult("T4a", "PASS",
"Path traversal mitigations verified",
"send_from_directory/commonpath used for file serving; "
"extractall() calls have is_relative_to() guards")
# ─────────────────────────────────────────────────────────────────────────────
# T5: Auth Bypass Patterns
# ─────────────────────────────────────────────────────────────────────────────
def test_t5a_auth_bypass_patterns() -> TestResult:
"""
Look for broken auth bypass patterns not the intentional no-auth design
(T2a covers that), but patterns that suggest auth was INTENDED to exist
but has an exploitable bypass: broad substring matching, debug-mode skips,
or if-True conditions.
"""
bypass_signals = [
(r'if\s+True\s*:', "if True: bypass"),
(r'if\s+debug\s*:', "debug-mode auth skip"),
(r'request\.path\s+in\s+', "substring path matching in auth (Huntarr pattern)"),
(r'EXEMPT_ROUTES\s*=', "exempt routes list"),
]
findings: list[str] = []
for dir_name in ["src", "web_interface"]:
scan_dir = PROJECT_ROOT / dir_name
if not scan_dir.exists():
continue
for py_file in scan_dir.rglob("*.py"):
try:
content = py_file.read_text(encoding="utf-8")
except OSError:
continue
for pattern, label in bypass_signals:
if re.search(pattern, content):
# Only flag if the file also contains auth-related terms
if any(auth in content.lower() for auth in
["auth", "login", "authenticate", "token", "permission"]):
rel = py_file.relative_to(PROJECT_ROOT)
findings.append(f"{rel}: {label}")
if findings:
return TestResult(
"T5a", "WARNING",
f"Potential auth bypass patterns found ({len(findings)})",
"; ".join(findings[:5])
)
return TestResult("T5a", "PASS",
"No auth bypass patterns detected",
"Checked src/ and web_interface/ for bypass signals")
# ─────────────────────────────────────────────────────────────────────────────
# T6: Docker / Container Hardening
# ─────────────────────────────────────────────────────────────────────────────
def test_t6_docker_hardening() -> TestResult:
"""Container security — skipped if no Dockerfile exists."""
dockerfile = PROJECT_ROOT / "Dockerfile"
if not dockerfile.exists():
return TestResult("T6", "SKIP",
"No Dockerfile found — container security scan not applicable",
"If Docker support is added in future, enable hadolint/trivy scanning "
"in .github/workflows/security-audit.yml")
content = dockerfile.read_text(encoding="utf-8")
issues: list[str] = []
# Check for non-root USER directive
user_lines = [l for l in content.splitlines() if l.strip().startswith("USER")]
if not user_lines or user_lines[-1].strip() == "USER root":
issues.append("Container runs as root — use USER directive to drop privileges")
# Check for pinned base image tags. A tag (even a specific version, not
# just :latest) is mutable -- the same tag can point to a different
# image later. Only a @sha256 digest is truly immutable/reproducible.
from_lines = [line for line in content.splitlines() if line.strip().startswith("FROM")]
for from_line in from_lines:
parts = from_line.split()
# FROM [--platform=<platform>] <image> [AS <name>] -- skip an
# optional --platform= flag so it's never mistaken for the image
# token itself (which would falsely report it as unpinned).
image_parts = [p for p in parts[1:] if not p.startswith("--platform=")]
if image_parts:
image = image_parts[0]
if "@sha256:" not in image:
issues.append(f"Base image not pinned to a digest: {image}")
if issues:
return TestResult("T6", "WARNING",
f"Dockerfile hardening issues ({len(issues)})",
"; ".join(issues))
return TestResult("T6", "PASS", "Dockerfile hardening checks passed", "")
# ─────────────────────────────────────────────────────────────────────────────
# Runner
# ─────────────────────────────────────────────────────────────────────────────
def main() -> int:
parser = argparse.ArgumentParser(
description="LEDMatrix security proof tests",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--output", "-o", default=None,
help="Write JSON results to this file")
parser.add_argument("--verbose", "-v", action="store_true",
help="Show details for each check")
args = parser.parse_args()
print("=" * 60)
print("LEDMatrix Security Proof Tests")
print(f"Project root: {PROJECT_ROOT}")
print("=" * 60)
all_results: list[TestResult] = []
# Run all test groups
all_results.append(test_t1a_zip_slip_protection())
all_results.extend(test_t1b_dangerous_plugin_calls())
all_results.append(test_t2a_api_surface_inventory())
all_results.append(test_t3a_hardcoded_secrets())
all_results.append(test_t3b_plaintext_password_storage())
all_results.append(test_t4a_path_traversal())
all_results.append(test_t5a_auth_bypass_patterns())
all_results.append(test_t6_docker_hardening())
# Print results
print()
for r in all_results:
line = f" {r.icon} [{r.severity:<8}] {r.test_id}: {r.message}"
print(line)
if args.verbose and r.details:
print(f" {r.details}")
# Tally
critical = [r for r in all_results if r.severity == "CRITICAL"]
warnings = [r for r in all_results if r.severity == "WARNING"]
passed = [r for r in all_results if r.severity == "PASS"]
skipped = [r for r in all_results if r.severity == "SKIP"]
print()
print(f" Results: {len(passed)} PASS {len(warnings)} WARN "
f"{len(critical)} CRITICAL {len(skipped)} SKIP")
# Write JSON output
if args.output:
output_data = [r.to_dict() for r in all_results]
Path(args.output).write_text(
json.dumps(output_data, indent=2), encoding="utf-8"
)
print(f" Results written to: {args.output}")
if critical:
print(f"\n 🚨 {len(critical)} CRITICAL issue(s) found — blocking")
return 1
if warnings:
print(f"\n ⚠️ {len(warnings)} warning(s) found — non-blocking")
print("\n ✅ All checks passed (warnings are non-blocking)")
return 0
if __name__ == "__main__":
sys.exit(main())
+43 -5
View File
@@ -17,6 +17,7 @@ import os
import json
import argparse
from pathlib import Path
from typing import Any, Dict, Optional, Sequence, Union
# Add project root to path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
@@ -27,15 +28,49 @@ os.environ['EMULATOR'] = 'true'
# Import logger after path setup so src.logging_config is importable
from src.logging_config import get_logger # noqa: E402
from src.plugin_system.testing.loading import ( # noqa: E402
build_full_config, find_plugin_dir, load_manifest,
)
logger = get_logger("[Render Plugin]")
MIN_DIMENSION = 1
MAX_DIMENSION = 512
def find_plugin_dir(plugin_id: str, search_dirs: Sequence[Union[str, Path]]) -> Optional[Path]:
"""Find a plugin directory by searching multiple paths."""
from src.plugin_system.plugin_loader import PluginLoader
loader = PluginLoader()
for search_dir in search_dirs:
search_path = Path(search_dir)
if not search_path.exists():
continue
result = loader.find_plugin_directory(plugin_id, search_path)
if result:
return Path(result)
return None
def load_manifest(plugin_dir: Path) -> Dict[str, Any]:
"""Load and return manifest.json from plugin directory."""
manifest_path = plugin_dir / 'manifest.json'
if not manifest_path.exists():
raise FileNotFoundError(f"No manifest.json in {plugin_dir}")
with open(manifest_path, 'r') as f:
return json.load(f)
def load_config_defaults(plugin_dir: Path) -> Dict[str, Any]:
"""Extract default values from config_schema.json."""
schema_path = plugin_dir / 'config_schema.json'
if not schema_path.exists():
return {}
with open(schema_path, 'r') as f:
schema = json.load(f)
defaults: Dict[str, Any] = {}
for key, prop in schema.get('properties', {}).items():
if 'default' in prop:
defaults[key] = prop['default']
return defaults
def main() -> int:
"""Load a plugin, call update() + display(), and save the result as a PNG image."""
parser = argparse.ArgumentParser(description='Render a plugin display to a PNG image')
@@ -46,7 +81,7 @@ def main() -> int:
help='Plugin config as JSON string')
parser.add_argument('--mock-data', '-m', default=None,
help='Path to JSON file with mock cache data')
parser.add_argument('--output', '-o', default='/tmp/plugin_render.png', # nosec B108 - dev script default; user can override
parser.add_argument('--output', '-o', default='/tmp/plugin_render.png',
help='Output PNG path (default: /tmp/plugin_render.png)')
parser.add_argument('--width', type=int, default=128, help='Display width (default: 128)')
parser.add_argument('--height', type=int, default=32, help='Display height (default: 32)')
@@ -83,13 +118,16 @@ def main() -> int:
manifest = load_manifest(Path(plugin_dir))
# Parse config: start with schema defaults, then apply overrides
config_defaults = load_config_defaults(Path(plugin_dir))
try:
user_config = json.loads(args.config)
except json.JSONDecodeError as e:
logger.error("Invalid JSON config: %s", e)
return 1
config = build_full_config(Path(plugin_dir), cli_config=user_config)
config = {'enabled': True}
config.update(config_defaults)
config.update(user_config)
# Load mock data if provided
mock_data = {}
+5
View File
@@ -7,7 +7,9 @@ Supports both unittest and pytest.
"""
import sys
import os
import argparse
import subprocess
from pathlib import Path
from typing import Optional
@@ -196,14 +198,17 @@ def main():
if runner == 'auto':
# Try pytest first, fall back to unittest
try:
import pytest
runner = 'pytest'
except ImportError:
runner = 'unittest'
# Run tests
if runner == 'pytest':
import importlib.util
return run_pytest_tests(test_files, args.verbose, args.coverage)
else:
import importlib.util
return run_unittest_tests(test_files, args.verbose)
+1 -125
View File
@@ -209,11 +209,6 @@
onchange="onConfigChange()">
<span class="text-xs ml-2" style="color: var(--text-secondary);">px</span>
</div>
<select id="sizePreset" onchange="applySizePreset()"
class="w-full mt-2 px-2 py-1.5 rounded text-xs"
style="background: var(--bg-primary); color: var(--text-secondary); border: 1px solid var(--border-color);">
<option value="">Preset sizes…</option>
</select>
</div>
<!-- Config form -->
@@ -247,18 +242,13 @@
</div>
</details>
<!-- Render buttons -->
<!-- Render button -->
<div class="flex gap-2">
<button onclick="renderPlugin()" id="renderBtn"
class="flex-1 px-4 py-2.5 rounded-lg text-sm font-medium text-white"
style="background: var(--accent);">
Render
</button>
<button onclick="renderAllSizes()" id="renderAllBtn" title="Render at every harness test size"
class="px-4 py-2.5 rounded-lg text-sm font-medium"
style="background: var(--bg-tertiary); color: var(--text-primary); border: 1px solid var(--border-color);">
All Sizes
</button>
</div>
</div>
@@ -321,15 +311,6 @@
<div id="messagesPanel" class="panel p-3 hidden">
<div id="messagesList" class="text-xs font-mono space-y-1"></div>
</div>
<!-- Multi-size gallery -->
<div id="galleryPanel" class="panel p-4 hidden">
<div class="flex items-center justify-between mb-3">
<span class="text-xs font-medium" style="color: var(--text-secondary);">All Sizes</span>
<span class="text-xs" style="color: var(--text-secondary);" id="galleryStatus"></span>
</div>
<div id="galleryGrid" class="flex flex-wrap gap-4 items-start"></div>
</div>
</div>
</div>
@@ -359,30 +340,8 @@
opt.textContent = `${p.name} (${p.id})`;
select.appendChild(opt);
});
// Load harness size presets
try {
const sizesRes = await fetch('/api/sizes');
const sizesData = await sizesRes.json();
const preset = document.getElementById('sizePreset');
(sizesData.sizes || []).forEach(([w, h]) => {
const opt = document.createElement('option');
opt.value = `${w}x${h}`;
opt.textContent = `${w} x ${h}`;
preset.appendChild(opt);
});
} catch (e) { /* presets are a convenience; ignore */ }
});
function applySizePreset() {
const value = document.getElementById('sizePreset').value;
if (!value) return;
const [w, h] = value.split('x');
document.getElementById('displayWidth').value = w;
document.getElementById('displayHeight').value = h;
onConfigChange();
}
// ---------- Plugin selection ----------
async function onPluginChange() {
const pluginId = document.getElementById('pluginSelect').value;
@@ -526,89 +485,6 @@
}
}
// ---------- Multi-size gallery ----------
async function renderAllSizes() {
if (!currentPluginId) return;
const btn = document.getElementById('renderAllBtn');
const panel = document.getElementById('galleryPanel');
const grid = document.getElementById('galleryGrid');
const status = document.getElementById('galleryStatus');
btn.disabled = true;
btn.textContent = 'Rendering…';
panel.classList.remove('hidden');
grid.innerHTML = '';
status.textContent = 'Rendering at all harness sizes…';
const config = jsonEditor ? jsonEditor.getValue() : {};
config.enabled = true;
let mockData = {};
const mockInput = document.getElementById('mockDataInput').value.trim();
if (mockInput) {
try { mockData = JSON.parse(mockInput); }
catch (e) { showMessages([], [`Mock data JSON error: ${e.message}`]); }
}
try {
const res = await fetch('/api/render-matrix', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
plugin_id: currentPluginId,
config: config,
mock_data: mockData,
}),
});
const data = await res.json();
if (data.error) {
status.textContent = data.error;
return;
}
let failures = 0;
(data.results || []).forEach(r => {
const cell = document.createElement('div');
cell.style.cssText = 'display:flex;flex-direction:column;gap:4px;';
const failed = (r.errors || []).length > 0 || !r.image;
if (failed) failures++;
const label = document.createElement('span');
label.className = 'text-xs font-mono';
label.style.color = failed ? '#f87171' : 'var(--text-secondary)';
label.textContent = `${r.width}x${r.height} · ${r.render_time_ms}ms`;
cell.appendChild(label);
if (r.image) {
const img = document.createElement('img');
img.src = r.image;
// Small panels get 2x zoom so they stay legible in the grid
const zoom = r.height >= 128 ? 1 : 2;
img.style.cssText =
`image-rendering: pixelated; width:${r.width * zoom}px; ` +
`height:${r.height * zoom}px; ` +
`border:1px solid ${failed ? '#f87171' : 'var(--border-color)'};`;
cell.appendChild(img);
}
if (failed) {
const err = document.createElement('span');
err.className = 'text-xs font-mono';
err.style.color = '#f87171';
err.textContent = (r.errors || ['render failed']).join('; ');
cell.appendChild(err);
}
grid.appendChild(cell);
});
status.textContent = failures
? `${failures} size(s) failed`
: `${(data.results || []).length} sizes rendered`;
} catch (e) {
status.textContent = `Network error: ${e.message}`;
} finally {
btn.disabled = false;
btn.textContent = 'All Sizes';
}
}
// ---------- Zoom ----------
function updateZoom() {
const zoom = parseInt(document.getElementById('zoomSlider').value);
+2
View File
@@ -6,7 +6,9 @@ This script allows manual clearing of specific cache keys or all cache data.
import os
import sys
import json
import argparse
from pathlib import Path
# Add the src directory to the path so we can import our modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+1 -1
View File
@@ -111,7 +111,7 @@ def main():
# Ensure PYTHONPATH is set correctly if web_interface.py has relative imports to src
# The WorkingDirectory in systemd service should handle this for web_interface.py
print(f"Launching web interface v3: {sys.executable} {WEB_INTERFACE_SCRIPT}")
os.execvp(sys.executable, [sys.executable, WEB_INTERFACE_SCRIPT]) # nosec B606 - both args are fixed constants
os.execvp(sys.executable, [sys.executable, WEB_INTERFACE_SCRIPT])
except Exception as e:
print(f"Failed to exec web interface: {e}")
sys.exit(1) # Failed to start
+16 -12
View File
@@ -78,17 +78,21 @@ class WiFiMonitorDaemon:
while self.running:
try:
# One combined check that also returns the state it observed —
# the previous flow fetched status before AND after the check
# on top of the check's own internal fetch, each one several
# nmcli subprocess forks, every 30s, forever.
(state_changed, updated_status, updated_ethernet,
ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
# Get current status before checking
status = self.wifi_manager.get_wifi_status()
ethernet_connected = self.wifi_manager._is_ethernet_connected()
# Check WiFi status and manage AP mode
state_changed = self.wifi_manager.check_and_manage_ap_mode()
# Get updated status after check
updated_status = self.wifi_manager.get_wifi_status()
updated_ethernet = self.wifi_manager._is_ethernet_connected()
current_state = {
'connected': updated_status.connected,
'ethernet_connected': updated_ethernet,
'ap_active': ap_active,
'ap_active': updated_status.ap_mode_active,
'ssid': updated_status.ssid
}
@@ -105,7 +109,7 @@ class WiFiMonitorDaemon:
else:
logger.debug("Ethernet not connected")
if ap_active:
if updated_status.ap_mode_active:
logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
else:
logger.debug("AP mode inactive")
@@ -119,16 +123,16 @@ class WiFiMonitorDaemon:
# Log periodic status (less verbose)
if updated_status.connected:
logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), "
f"Ethernet={updated_ethernet}, AP={ap_active}")
f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
else:
logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}")
logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
# Escalating recovery: if nmcli reports connected but actual internet
# is unreachable for several consecutive checks, restart NetworkManager.
# This is done HERE (not inside check_and_manage_ap_mode) to keep the
# AP-enable trigger clean and avoid false-positive AP enables from
# transient packet loss on otherwise working WiFi.
if updated_status.connected and not ap_active:
if updated_status.connected and not updated_status.ap_mode_active:
if not self.wifi_manager.check_internet_connectivity():
self._consecutive_internet_failures += 1
logger.warning(
+1 -1
View File
@@ -4,5 +4,5 @@ LEDMatrix Display System
Core source package for the LED Matrix Display project.
"""
__version__ = "3.1.0"
__version__ = "1.0.0"
-174
View File
@@ -1,174 +0,0 @@
"""
Adaptive image fitting for plugins the image counterpart to
src/adaptive_layout.py's text fitting.
Promotes the proven in-field image patterns into one shared helper so
plugins stop hand-copying resize/cache code:
- "crop transparent padding, then fill the row height" (football/hockey
logo pattern) -> ``crop_to_ink=True, mode="fill_height"``
- "crop-to-fill with a top anchor for faces" (masters-tournament headshot
pattern) -> ``mode="cover", anchor="top"``
- "letterbox to fit, centered on a background" (static-image pattern)
-> ``mode="contain"``
- NEAREST for pixel art/flags vs LANCZOS for photos (masters flag pattern)
-> ``resample=RESAMPLE_NEAREST``
Unlike PIL's ``thumbnail()`` (downscale-only — the reason plugin imagery
stays tiny on big panels), ``fit_image`` upscales by default so content
genuinely adapts to larger displays; pass ``upscale=False`` for the old
behavior.
Use via ``LayoutContext.fit_image(...)`` (cached per panel size) or
``BasePlugin.draw_image(...)``; the module-level functions are the
uncached primitives.
"""
from dataclasses import dataclass
from typing import Any, Optional, Tuple
from PIL import Image
# The one Pillow >= 9.1 compat shim (replaces the per-plugin copies).
try:
RESAMPLE_LANCZOS = Image.Resampling.LANCZOS
RESAMPLE_NEAREST = Image.Resampling.NEAREST
except AttributeError: # Pillow < 9.1
RESAMPLE_LANCZOS = Image.LANCZOS
RESAMPLE_NEAREST = Image.NEAREST
FIT_MODES = ("contain", "cover", "fill_height", "stretch")
@dataclass(frozen=True)
class ImageFitResult:
"""A processed RGBA copy of a source image, sized for a target box."""
image: Image.Image
width: int
height: int
scale: float # scale applied vs the (possibly ink-cropped) source
mode: str
source_size: Tuple[int, int]
@property
def is_empty(self) -> bool:
return self.width <= 0 or self.height <= 0
_EMPTY_IMAGE = Image.new("RGBA", (1, 1), (0, 0, 0, 0))
def _empty_result(mode: str, source_size: Tuple[int, int]) -> ImageFitResult:
return ImageFitResult(_EMPTY_IMAGE, 0, 0, 0.0, mode, source_size)
def _box_dims(box: Any) -> Tuple[int, int]:
"""Accept a Region (duck-typed .w/.h) or a (w, h) tuple."""
if hasattr(box, "w") and hasattr(box, "h"):
return (int(box.w), int(box.h))
w, h = box
return (int(w), int(h))
def fit_image(img: Image.Image, box: Any, *, mode: str = "contain",
crop_to_ink: bool = False, anchor: str = "center",
resample: Any = None, upscale: bool = True) -> ImageFitResult:
"""Fit an image into a box, preserving crispness policy per content type.
Args:
img: Source PIL image (any mode; output is always RGBA).
box: Region or (w, h) target box.
mode: "contain" (letterbox), "cover" (crop-to-fill),
"fill_height" (height == box height, contain-capped by width),
"stretch" (exact resize).
crop_to_ink: Trim fully-transparent padding (getbbox) before fitting
logos shipped with generous padding otherwise render small.
anchor: For "cover" crops: "center" or "top" (keeps faces/tops).
resample: PIL resampling filter; defaults to RESAMPLE_LANCZOS.
Use RESAMPLE_NEAREST for pixel art, flags, and sprite icons.
upscale: Allow scaling above source size (default True the adaptive
point). False mimics the legacy thumbnail() behavior.
"""
if mode not in FIT_MODES:
raise ValueError(f"Unknown fit mode '{mode}' (expected one of {FIT_MODES})")
box_w, box_h = _box_dims(box)
if box_w <= 0 or box_h <= 0 or img.width <= 0 or img.height <= 0:
return _empty_result(mode, img.size)
resample = RESAMPLE_LANCZOS if resample is None else resample
work = img if img.mode == "RGBA" else img.convert("RGBA")
if crop_to_ink:
bbox = work.getbbox()
if bbox is None: # fully transparent
return _empty_result(mode, img.size)
work = work.crop(bbox)
src_w, src_h = work.size
if mode == "stretch":
out = work.resize((box_w, box_h), resample)
return ImageFitResult(out, box_w, box_h, box_w / src_w, mode, (src_w, src_h))
if mode == "cover":
scale = max(box_w / src_w, box_h / src_h)
if not upscale:
scale = min(scale, 1.0)
scaled_w = max(1, round(src_w * scale))
scaled_h = max(1, round(src_h * scale))
out = work.resize((scaled_w, scaled_h), resample)
# Crop the overhang down to the box (only when the scaled image is
# larger; with upscale=False it may be smaller and is left as-is).
crop_w, crop_h = min(box_w, scaled_w), min(box_h, scaled_h)
left = (scaled_w - crop_w) // 2
top = 0 if anchor == "top" else (scaled_h - crop_h) // 2
out = out.crop((left, top, left + crop_w, top + crop_h))
return ImageFitResult(out, out.width, out.height, scale, mode, (src_w, src_h))
# contain / fill_height share the "preserve aspect, no crop" path
if mode == "fill_height":
scale = box_h / src_h
# contain-cap: never exceed the box width (football's logo_slot rule)
scale = min(scale, box_w / src_w)
else: # contain
scale = min(box_w / src_w, box_h / src_h)
if not upscale:
scale = min(scale, 1.0)
out_w = max(1, round(src_w * scale))
out_h = max(1, round(src_h * scale))
if (out_w, out_h) == (src_w, src_h):
# No resize needed — but `work` may still BE the caller's original
# image (RGBA source, no ink crop). The result must always be an
# independent copy: LayoutContext caches ImageFitResults, and an
# aliased image would let later mutations of the source corrupt
# cached fits (or vice versa).
out = work.copy() if work is img else work
else:
out = work.resize((out_w, out_h), resample)
return ImageFitResult(out, out_w, out_h, scale, mode, (src_w, src_h))
def draw_fitted_image(display_manager: Any, ifit: ImageFitResult, box: Any, *,
align: str = "center", valign: str = "center",
offset: Tuple[int, int] = (0, 0)) -> Optional[Tuple[int, int]]:
"""Paste a fitted image aligned within a Region onto the display canvas.
Pastes with the image's own alpha mask. Returns the (x, y) actually used
so callers can position adjacent decorations, or None when nothing was
drawn (empty fit / no canvas).
"""
if ifit is None or ifit.is_empty:
return None
image = getattr(display_manager, "image", None)
if image is None:
return None
if hasattr(box, "align_xy"):
x, y = box.align_xy(ifit.width, ifit.height, align, valign)
else:
box_w, box_h = _box_dims(box)
x = (box_w - ifit.width) // 2
y = (box_h - ifit.height) // 2
x += int(offset[0])
y += int(offset[1])
image.paste(ifit.image, (x, y), ifit.image)
return (x, y)
-746
View File
@@ -1,746 +0,0 @@
"""
Adaptive layout and font scaling helpers for plugins.
Generalizes the three size-adaptation patterns proven in the plugin
ecosystem into small composable core helpers, so plugins render legibly on
any panel size (64x32, 128x32, 96x48, 128x64, 256x64, ...) without
hand-tuned per-display layouts:
- Region: integer rect algebra (bands, columns, weighted splits, centering).
Regions partition space, so text bands can't overlap by construction —
replacing the magic ``y = 1`` / ``y = height - 7`` offsets tuned for 128x32.
- Font ladders: ordered (family, size) steps known to render crisply.
Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes,
so fonts are never scaled continuously fitting walks a ladder from the
largest rung down until the measured text fits the target box. This is
baseball-scoreboard's fallback-ladder pattern promoted to core.
- LayoutContext: per-(width, height) facts breakpoint tiers
(masters-tournament's pattern), a geometry scale factor vs. a declared
design size (f1-scoreboard's pattern), and cached fit-text queries.
Everything is opt-in: plugins get a context via ``self.layout`` on
BasePlugin (or construct one directly) and existing plugins are unaffected.
Fonts are resolved through FontManager's catalog (family names are
lowercased file stems from assets/fonts, e.g. "9x15", "tom-thumb", plus
aliases like "press_start"). FitResult.font is a plain PIL font or
freetype.Face, so it drops straight into DisplayManager.draw_text().
"""
import logging
from collections import OrderedDict
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import freetype
logger = logging.getLogger(__name__)
# Height-based breakpoint tiers, smallest to largest. A 32px-tall panel is
# the ecosystem baseline ("sm"); 96x48 lands in "md"; 128x64 in "lg".
_HEIGHT_TIERS: Tuple[Tuple[str, int], ...] = (
("xs", 16), ("sm", 32), ("md", 48), ("lg", 64), ("xl", 10 ** 9),
)
TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _HEIGHT_TIERS)
_WIDTH_TIERS: Tuple[Tuple[str, int], ...] = (
("narrow", 64), ("normal", 128), ("wide", 256), ("ultrawide", 10 ** 9),
)
WIDTH_TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _WIDTH_TIERS)
# The panel size most existing plugins were authored against.
DEFAULT_DESIGN_SIZE: Tuple[int, int] = (128, 32)
@dataclass(frozen=True)
class Region:
"""An integer rectangle. Carving methods return sub-Regions clamped to
non-negative dimensions, so degenerate panels never produce negative
boxes a band request larger than the region simply consumes it all."""
x: int
y: int
w: int
h: int
def __post_init__(self):
object.__setattr__(self, "w", max(0, int(self.w)))
object.__setattr__(self, "h", max(0, int(self.h)))
object.__setattr__(self, "x", int(self.x))
object.__setattr__(self, "y", int(self.y))
@property
def right(self) -> int:
return self.x + self.w
@property
def bottom(self) -> int:
return self.y + self.h
@property
def center(self) -> Tuple[int, int]:
return (self.x + self.w // 2, self.y + self.h // 2)
# ---- carving -----------------------------------------------------
def inset(self, dx: int, dy: Optional[int] = None) -> "Region":
"""Shrink by dx horizontally and dy (default dx) vertically, each side."""
if dy is None:
dy = dx
return Region(self.x + dx, self.y + dy, self.w - 2 * dx, self.h - 2 * dy)
def offset(self, dx: int, dy: int) -> "Region":
"""Translate without resizing — the hook for user x/y-offset
customization: compute regions first, then apply the user's
configured offsets as a final translation."""
return Region(self.x + dx, self.y + dy, self.w, self.h)
def top_band(self, h: int) -> "Region":
return Region(self.x, self.y, self.w, min(h, self.h))
def bottom_band(self, h: int) -> "Region":
h = min(h, self.h)
return Region(self.x, self.bottom - h, self.w, h)
def middle(self, top_h: int = 0, bottom_h: int = 0) -> "Region":
"""What remains between a top band and a bottom band."""
return Region(self.x, self.y + top_h, self.w, self.h - top_h - bottom_h)
def left_col(self, w: int) -> "Region":
return Region(self.x, self.y, min(w, self.w), self.h)
def right_col(self, w: int) -> "Region":
w = min(w, self.w)
return Region(self.right - w, self.y, w, self.h)
def split_h(self, *weights: float, gap: int = 0) -> List["Region"]:
"""Side-by-side columns sized by weight; gaps between them."""
sizes = _weighted_sizes(self.w, weights, gap)
cols, cursor = [], self.x
for size in sizes:
cols.append(Region(cursor, self.y, size, self.h))
cursor += size + gap
return cols
def split_v(self, *weights: float, gap: int = 0) -> List["Region"]:
"""Stacked rows sized by weight; gaps between them."""
sizes = _weighted_sizes(self.h, weights, gap)
rows, cursor = [], self.y
for size in sizes:
rows.append(Region(self.x, cursor, self.w, size))
cursor += size + gap
return rows
# ---- placement ---------------------------------------------------
def align_xy(self, w: int, h: int, align: str = "center",
valign: str = "center") -> Tuple[int, int]:
"""Top-left position for a w x h box aligned within this region.
align: left|center|right; valign: top|center|bottom."""
if align == "left":
x = self.x
elif align == "right":
x = self.right - w
else:
x = self.x + (self.w - w) // 2
if valign == "top":
y = self.y
elif valign == "bottom":
y = self.bottom - h
else:
y = self.y + (self.h - h) // 2
return (x, y)
def center_xy(self, w: int, h: int) -> Tuple[int, int]:
return self.align_xy(w, h)
def contains(self, w: int, h: int) -> bool:
return w <= self.w and h <= self.h
def _weighted_sizes(total: int, weights: Sequence[float], gap: int) -> List[int]:
"""Integer sizes proportional to weights, remainder spread left-to-right."""
if not weights:
return []
usable = max(0, total - gap * (len(weights) - 1))
weight_sum = sum(weights) or 1
sizes = [int(usable * w / weight_sum) for w in weights]
remainder = usable - sum(sizes)
for i in range(remainder):
sizes[i % len(sizes)] += 1
return sizes
# ---------------------------------------------------------------------------
# Font ladders
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class FontStep:
"""One rung: a FontManager catalog family at a size it renders crisply."""
family: str
size_px: int
FontLadder = Tuple[FontStep, ...]
# X11 BDF bitmap fonts at their native pixel sizes, largest to smallest —
# baseball-scoreboard's fallback ladder extended upward. Same-height rungs
# are ordered widest first so width-constrained text steps to a narrower
# face before dropping a size.
LADDER_GRID: FontLadder = (
FontStep("10x20", 20),
FontStep("9x18", 18),
FontStep("9x15", 15),
FontStep("8x13", 13),
FontStep("7x13", 13),
FontStep("6x13", 13),
FontStep("6x12", 12),
FontStep("6x10", 10),
FontStep("6x9", 9),
FontStep("5x8", 8),
FontStep("5x7", 7),
FontStep("4x6", 6),
FontStep("tom-thumb", 6),
)
# PressStart2P at integer multiples of its 8px pixel grid only — fractional
# sizes blur a pixel font. For headline text (clocks, scores).
LADDER_ARCADE: FontLadder = (
FontStep("press_start", 32),
FontStep("press_start", 24),
FontStep("press_start", 16),
FontStep("press_start", 8),
)
LADDER_DEFAULT: FontLadder = LADDER_GRID
ELLIPSIS = ""
@dataclass(frozen=True)
class FitResult:
"""A fitted font plus the ink metrics of the (possibly ellipsized) text.
``y_offset`` is the gap between the y passed to draw_text() and where
ink actually starts; subtract it from the desired ink-top position when
drawing (draw_fitted_text does this for you).
"""
font: Any
family: str
size_px: int
text: str
width: int
height: int
baseline: int
y_offset: int
fits: bool
line_height: int = 0
def measure_ink(text: str, font: Any) -> Tuple[int, int, int, int]:
"""Measure the ink box of text: (width, height, baseline, y_offset).
y_offset is the distance from the y coordinate DisplayManager.draw_text()
is given to the top of the actual ink PIL draws TTF from the em-box
top and _draw_bdf_text derives the baseline from y + ascender, so both
leave a font-dependent gap that matters when centering in short bands.
"""
if isinstance(font, freetype.Face):
width = 0
ascender = font.size.ascender >> 6
ink_top, ink_bottom = None, None
for char in text:
font.load_char(char)
width += font.glyph.advance.x >> 6
rows = font.glyph.bitmap.rows
if rows:
top = ascender - font.glyph.bitmap_top
ink_top = top if ink_top is None else min(ink_top, top)
ink_bottom = top + rows if ink_bottom is None else max(ink_bottom, top + rows)
if ink_top is None:
ink_top, ink_bottom = 0, 0
return (width, ink_bottom - ink_top, ascender, ink_top)
bbox = font.getbbox(text)
return (bbox[2] - bbox[0], bbox[3] - bbox[1], -bbox[1], bbox[1])
def font_line_height(font: Any) -> int:
"""Recommended line spacing for a font (matches DisplayManager.get_font_height)."""
if isinstance(font, freetype.Face):
return font.size.height >> 6
ascent, descent = font.getmetrics()
return ascent + descent
def measure_font_crispness(font: Any, sample_text: str = "Ay0",
canvas_size: Tuple[int, int] = (250, 60)) -> float:
"""Fraction of the rendered sample's ink-bbox pixels that are neither
pure black nor pure white i.e. antialiased.
BDF (freetype.Face) glyphs are true bitmaps and always render at 0.0.
"Pixel-style" TTFs (PressStart2P, and similar fonts bundled for
plugins that draw through ImageDraw.text() and so can't take a BDF
face) are NOT automatically crisp at arbitrary sizes PIL antialiases
TTF outlines by default, and a pixel-grid font only lands on whole
pixels at specific sizes (for PressStart2P: exact multiples of 8).
Requesting an unverified size silently produces soft/blurry glyphs on
an LED panel, which reads as fuzzy compared to a true BDF rung.
Use this to vet any custom FontLadder rung that mixes TTF fonts before
shipping it see test_adaptive_layout.py::test_ladder_is_crisp for the
pattern. A rung should score 0.0 (or very close, to allow for the odd
diagonal stroke) before it belongs in a "crisp" ladder.
"""
if isinstance(font, freetype.Face):
return 0.0
from PIL import Image, ImageDraw
img = Image.new("L", canvas_size, 0)
ImageDraw.Draw(img).text((2, 2), sample_text, font=font, fill=255)
bbox = img.getbbox()
if bbox is None:
return 0.0
pixels = img.crop(bbox).tobytes()
pure = sum(1 for p in pixels if p == 0 or p == 255)
return (len(pixels) - pure) / len(pixels)
class LayoutContext:
"""Per-render-size layout facts and fit-text queries for one panel size.
Construct once per (width, height); BasePlugin.layout does this and
rebuilds automatically when the logical display size changes.
"""
def __init__(self, width: int, height: int, font_manager: Any,
design_size: Tuple[int, int] = DEFAULT_DESIGN_SIZE):
self.width = int(width)
self.height = int(height)
self.font_manager = font_manager
self.design_size = design_size
self.bounds = Region(0, 0, self.width, self.height)
self.aspect = self.width / max(1, self.height)
self.tier = _pick_tier(_HEIGHT_TIERS, self.height)
self.width_tier = _pick_tier(_WIDTH_TIERS, self.width)
self.is_wide_short = self.aspect >= 2.5 and self.height <= 32
design_w, design_h = design_size
# Geometry scale only (gaps, icon/logo sizes) — never applied to
# fonts, which step between crisp ladder rungs instead.
self.scale = min(self.width / max(1, design_w),
self.height / max(1, design_h))
# LRU-bounded: entries are small, but keys embed the fitted TEXT —
# a plugin fitting changing text (a live game clock, a ticker) on a
# 24/7 service would otherwise grow this without bound.
self._fit_cache: "OrderedDict[Any, FitResult]" = OrderedDict()
# LRU-bounded (images are big). Entries hold a strong reference to
# the source image when keyed by id() so the id can't be recycled
# out from under the cache.
self._image_cache: "OrderedDict[Any, Tuple[Any, Any]]" = OrderedDict()
_IMAGE_CACHE_MAX = 64
_FIT_CACHE_MAX = 512
def _fit_cache_get(self, key: Any) -> Optional["FitResult"]:
cached = self._fit_cache.get(key)
if cached is not None:
self._fit_cache.move_to_end(key)
return cached
def _fit_cache_put(self, key: Any, result: "FitResult") -> None:
self._fit_cache[key] = result
while len(self._fit_cache) > self._FIT_CACHE_MAX:
self._fit_cache.popitem(last=False)
# ---- the three adaptation patterns --------------------------------
def px(self, base: int, minimum: int = 1, maximum: Optional[int] = None) -> int:
"""Scale a design-size pixel measurement (f1's pattern): gaps,
icon sizes, logo slots. Clamped to [minimum, maximum]."""
value = max(minimum, round(base * self.scale))
if maximum is not None:
value = min(value, maximum)
return value
def by_tier(self, mapping: Dict[str, Any], default: Any = None) -> Any:
"""Pick the value for the nearest defined tier at-or-below the
panel's height tier (masters' pattern). Falls forward to the
smallest defined tier above, then to default.
by_tier({"sm": 10, "lg": 18}) -> 10 on 128x32, 18 on 128x64.
Keys may also use width tiers ("narrow", "wide", ...)."""
order = TIER_ORDER if any(k in TIER_ORDER for k in mapping) else WIDTH_TIER_ORDER
current = self.tier if order is TIER_ORDER else self.width_tier
idx = order.index(current)
for name in reversed(order[: idx + 1]):
if name in mapping:
return mapping[name]
for name in order[idx + 1:]:
if name in mapping:
return mapping[name]
return default
def fit_text(self, text: str, box: Union[Region, Tuple[int, int]],
ladder: FontLadder = LADDER_DEFAULT,
ellipsis: bool = True) -> FitResult:
"""Largest ladder rung whose rendered text fits the box (baseball's
pattern). If even the smallest rung is too wide, the text is
ellipsized to fit (unless ellipsis=False); fits=False only when no
acceptable rendering exists."""
box_w, box_h = _box_dims(box)
key = ("text", text, box_w, box_h, ladder, ellipsis)
cached = self._fit_cache_get(key)
if cached is not None:
return cached
result = self._walk_ladder(text, ladder, box_w, box_h, ellipsis)
self._fit_cache_put(key, result)
return result
def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
base_size_px: int, ladder: FontLadder = LADDER_DEFAULT,
ellipsis: bool = True,
scale: Optional[float] = None) -> FitResult:
"""Ladder rung closest to (but not exceeding) ``base_size_px * scale``
that still fits the box proportional sizing instead of ``fit_text``'s
"always maximize" behavior.
Use this when several independently-fitted elements need to stay
visually harmonious as the panel grows (e.g. a scoreboard's score,
status, and detail text) ``fit_text`` maximizes each one within
its own region, which can make one element balloon out of
proportion to its neighbors (a huge score overlapping logos it fit
fine at the design size) even though every individual pick is
independently "correct". ``base_size_px`` is the size that element
renders at on the design size (``design_size``, typically 128x32)
commonly a plugin's existing classic/fixed font size for that
element.
``scale`` defaults to ``self.scale`` (the same conservative
min(width_ratio, height_ratio) factor ``px()`` uses safe for
content whose aspect ratio matters). Pass an explicit axis-specific
value when the surrounding composition already scales that way
e.g. a scoreboard whose logos scale with height alone
(``logo_slot = min(height, width // 2)``) should size its score
text by ``height / design_height`` too, or its text will look
under-scaled next to bigger logos on a panel that only grew taller.
Falls back to the smallest rung when even that exceeds the target
(a tiny scale factor), and to fit_text's ordinary smaller-rung
fallback when the closest-to-target rung doesn't actually fit the
box.
"""
box_w, box_h = _box_dims(box)
effective_scale = self.scale if scale is None else scale
key = ("text_prop", text, box_w, box_h, ladder, base_size_px, ellipsis, effective_scale)
cached = self._fit_cache_get(key)
if cached is not None:
return cached
target = base_size_px * effective_scale
eligible = [step for step in ladder if step.size_px <= target]
candidates = eligible if eligible else (min(ladder, key=lambda s: s.size_px),)
result = self._walk_ladder(text, candidates, box_w, box_h, ellipsis)
self._fit_cache_put(key, result)
return result
def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
box_w: int, box_h: int, ellipsis: bool) -> FitResult:
"""Shared by fit_text/fit_text_proportional: first ladder entry (in
the order given) whose rendered text fits, ellipsizing the last one
tried if none do."""
result = None
for step in ladder:
font = self.font_manager.get_font(step.family, step.size_px)
width, height, baseline, y_offset = measure_ink(text, font)
result = FitResult(font, step.family, step.size_px, text,
width, height, baseline, y_offset,
fits=(width <= box_w and height <= box_h),
line_height=font_line_height(font))
if result.fits:
break
if result is not None and not result.fits and ellipsis:
short = self.ellipsize(text, result.font, box_w)
width, height, baseline, y_offset = measure_ink(short, result.font)
result = FitResult(result.font, result.family, result.size_px,
short, width, height, baseline, y_offset,
fits=(width <= box_w and height <= box_h),
line_height=result.line_height)
return result
def fit_lines(self, lines: Sequence[str], box: Union[Region, Tuple[int, int]],
ladder: FontLadder = LADDER_DEFAULT,
spacing: int = 1) -> FitResult:
"""Largest rung where every line fits the box width and the stacked
lines (line_height + spacing apart) fit the box height. Measures the
actual strings, so a long line pushes the ladder down a rung a short
one wouldn't (baseball's multiline pattern). Text is the widest line."""
box_w, box_h = _box_dims(box)
key = ("lines", tuple(lines), box_w, box_h, ladder, spacing)
cached = self._fit_cache_get(key)
if cached is not None:
return cached
rows = max(1, len(lines))
result = None
for step in ladder:
font = self.font_manager.get_font(step.family, step.size_px)
line_h = font_line_height(font)
widest, metrics = "", (0, 0, 0, 0)
for line in lines:
m = measure_ink(line, font)
if m[0] >= metrics[0]:
widest, metrics = line, m
total_h = rows * line_h + (rows - 1) * spacing
result = FitResult(font, step.family, step.size_px, widest,
metrics[0], metrics[1], metrics[2], metrics[3],
fits=(metrics[0] <= box_w and total_h <= box_h),
line_height=line_h)
if result.fits:
break
self._fit_cache_put(key, result)
return result
def font_for_rows(self, rows: int, box_h: int,
ladder: FontLadder = LADDER_GRID) -> FitResult:
"""Largest rung whose line height lets `rows` rows fit in box_h
(baseball's traditional-scoreboard pattern). Measures a digit/cap
sample rather than specific strings."""
key = ("rows", rows, box_h, ladder)
cached = self._fit_cache_get(key)
if cached is not None:
return cached
sample = "0Ay"
result = None
for step in ladder:
font = self.font_manager.get_font(step.family, step.size_px)
line_h = font_line_height(font)
width, height, baseline, y_offset = measure_ink(sample, font)
result = FitResult(font, step.family, step.size_px, sample,
width, height, baseline, y_offset,
fits=(max(1, rows) * line_h <= box_h),
line_height=line_h)
if result.fits:
break
self._fit_cache_put(key, result)
return result
# ---- images ---------------------------------------------------------
def fit_image(self, img: Any, box: Union[Region, Tuple[int, int]], *,
mode: str = "contain", crop_to_ink: bool = False,
anchor: str = "center", resample: Any = None,
upscale: bool = True, cache_key: Any = None) -> Any:
"""Fit an image into a box (see src/adaptive_images.py for modes),
cached per (image, box size, options) for this panel size.
Prefer a stable ``cache_key`` (e.g. "logo:KC") for images that get
reloaded the default id()-based key is safe (the entry pins the
source image) but misses across reloads of the same content.
"""
from src.adaptive_images import fit_image as _fit_image
box_w, box_h = _box_dims(box)
resample_name = getattr(resample, "name", repr(resample)) if resample is not None else "default"
identity = cache_key if cache_key is not None else ("id", id(img))
key = ("image", identity, img.size, box_w, box_h, mode,
crop_to_ink, anchor, resample_name, upscale)
cached = self._image_cache.get(key)
if cached is not None:
self._image_cache.move_to_end(key)
return cached[0]
result = _fit_image(img, (box_w, box_h), mode=mode,
crop_to_ink=crop_to_ink, anchor=anchor,
resample=resample, upscale=upscale)
# Pin the source only for id()-keyed entries (see docstring).
self._image_cache[key] = (result, img if cache_key is None else None)
while len(self._image_cache) > self._IMAGE_CACHE_MAX:
self._image_cache.popitem(last=False)
return result
# ---- text utilities ------------------------------------------------
def ellipsize(self, text: str, font: Any, max_w: int) -> str:
"""Trim text to fit max_w, appending an ellipsis. Returns '' when
not even the ellipsis fits."""
if measure_ink(text, font)[0] <= max_w:
return text
for end in range(len(text) - 1, 0, -1):
candidate = text[:end].rstrip() + ELLIPSIS
if measure_ink(candidate, font)[0] <= max_w:
return candidate
return ELLIPSIS if measure_ink(ELLIPSIS, font)[0] <= max_w else ""
def measure(self, text: str, font: Any) -> Tuple[int, int, int]:
"""Ink (width, height, baseline) of text — see measure_ink."""
width, height, baseline, _ = measure_ink(text, font)
return (width, height, baseline)
def clear_cache(self) -> None:
"""Drop cached fit results (call after fonts are reloaded)."""
self._fit_cache.clear()
self._image_cache.clear()
def _pick_tier(tiers: Tuple[Tuple[str, int], ...], value: int) -> str:
for name, limit in tiers:
if value <= limit:
return name
return tiers[-1][0]
def _box_dims(box: Union[Region, Tuple[int, int]]) -> Tuple[int, int]:
if isinstance(box, Region):
return (box.w, box.h)
w, h = box
return (int(w), int(h))
def draw_fitted_text(display_manager: Any, fit: FitResult,
box: Union[Region, Tuple[int, int]],
color: Tuple[int, int, int] = (255, 255, 255),
align: str = "center", valign: str = "center") -> None:
"""Draw a FitResult's text aligned within a Region via
DisplayManager.draw_text(), compensating for the font's ink offset so
the ink (not the em box) is what gets aligned."""
region = box if isinstance(box, Region) else Region(0, 0, box[0], box[1])
x, y = region.align_xy(fit.width, fit.height, align, valign)
display_manager.draw_text(fit.text, x=x, y=y - fit.y_offset,
color=color, font=fit.font)
# ---------------------------------------------------------------------------
# Composite layouts — the region arrangements repeated across plugins,
# expressed as Region math so migrated plugins stop hand-copying coordinate
# formulas. Deliberately tiny: these return Regions, they don't draw.
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ScoreboardRegions:
"""The two-logos-plus-center-score card shared by the sports plugins."""
bounds: Region
logo_slot: int # width of each logo slot: min(H, W // 2), center-reserved
away_slot: Region # left logo slot
home_slot: Region # right logo slot
center_col: Region # column between the slots (>= min_center_fraction of width)
status_band: Region # top band (replaces the magic y = 1)
score_area: Region # center_col's true width, between the bands (replaces y = H//2 - 3)
detail_band: Region # bottom band (replaces the magic y = H - 7)
bottom_left: Region # bottom corner: away records / timeouts
bottom_right: Region # bottom corner: home records / timeouts
def scoreboard_regions(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
status_h: Optional[int] = None,
detail_h: Optional[int] = None,
min_center_fraction: float = 0.15,
min_center_design_px: int = 40,
score_bleed_fraction: float = 0.5) -> ScoreboardRegions:
"""Carve a game-card Region into the standard scoreboard arrangement.
Encodes the invariant duplicated across the sports plugins:
``logo_slot = min(height, width // 2)`` (capped at half the card so the
home slot never collapses), away logo centered in the left slot, home in
the right.
That formula alone has a blind spot: at exactly 2:1 aspect ratio
(width == 2 * height a very common shape, e.g. two, four, or more
square modules stacked into a taller panel) ``width // 2`` and
``height`` are equal, so the two logo slots claim the *entire* width
and leave zero pixels for a center column, no matter how large the
panel gets. It isn't a "small panel" problem: 96x48, 128x64, and
256x128 (all exactly 2:1) hit it identically, while wide panels like
the 128x32 design baseline or a 192x48/256x32 panel never do, because
height is already the tighter constraint there.
Two knobs fix it, both defaulted to values verified against the full
harness size spread (see test_adaptive_layout.py::TestScoreboardRegions):
- ``min_center_fraction`` / ``min_center_design_px`` reserve at least
``max(width * min_center_fraction, min_center_design_px * ctx.scale)``
for the center column, capping ``logo_slot`` further when needed. The
design-px term (scaled by the context's geometry factor, so it grows
on bigger panels like everything else in ``px()``) matters most on
small panels where a flat fraction alone reserves too little absolute
space for even a short score string. On wide panels the height
constraint already leaves more room than either reserves, so both are
a no-op there 128x32/192x48-style layouts are unaffected.
- ``score_bleed_fraction`` extends the score's own *fit box* (not the
logo slots themselves) an extra ``logo_slot * score_bleed_fraction``
into each side controlled, intentional overlap with the logo art,
the same way real broadcast scoreboards let a big score number's
edges cross into the team marks flanking it. Without this, on a
square-ish panel the center reserve alone can be too narrow for even
a modest score to render without truncating (`"17-21"` -> `"17-2…"`),
which is worse than a little overlap.
status_band and detail_band span the FULL card width and overlay the
logo slots matching the classic layouts, where short outlined status/
date text is drawn over the logos without issue; only score_area (the
one element whose size actively grows with the panel) uses the
narrower, bleed-adjusted box. Band heights default to the classic
128x32 values, scaled by the context's geometry factor when one is
provided. Works on a full panel or on a scroll-mode card Region.
"""
if status_h is None:
status_h = ctx.px(9, minimum=7) if ctx else 9
if detail_h is None:
detail_h = ctx.px(8, minimum=7) if ctx else 8
logo_slot = min(bounds.h, bounds.w // 2)
design_reserve = int(min_center_design_px * (ctx.scale if ctx else 1.0))
min_center_w = max(1, int(bounds.w * min_center_fraction), design_reserve)
max_logo_slot_by_center = max(1, (bounds.w - min_center_w) // 2)
logo_slot = min(logo_slot, max_logo_slot_by_center)
away_slot = bounds.left_col(logo_slot)
home_slot = bounds.right_col(logo_slot)
center_col = Region(bounds.x + logo_slot, bounds.y,
bounds.w - 2 * logo_slot, bounds.h)
status_band = bounds.top_band(status_h)
detail_band = bounds.bottom_band(detail_h)
middle = bounds.middle(status_band.h, detail_band.h)
# score_area is the true center gap's width plus a controlled bleed
# into each logo slot (see score_bleed_fraction above) -- narrower than
# the full card width status/detail get, since it's the one element
# whose size actively grows with the panel and needs its *fit box* to
# reflect real available space, but generous enough that a short score
# string never has to truncate on a square-ish panel.
bleed = int(logo_slot * score_bleed_fraction)
score_area = Region(center_col.x - bleed, middle.y,
center_col.w + 2 * bleed, middle.h)
bottom = bounds.bottom_band(detail_h)
return ScoreboardRegions(
bounds=bounds, logo_slot=logo_slot,
away_slot=away_slot, home_slot=home_slot, center_col=center_col,
status_band=status_band, score_area=score_area, detail_band=detail_band,
bottom_left=bottom.left_col(logo_slot),
bottom_right=bottom.right_col(logo_slot),
)
@dataclass(frozen=True)
class MediaRow:
"""Art/icon on the left, text column on the right (music's idiom)."""
art: Region
body: Region
def media_row(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
square: bool = True, gap: Optional[int] = None) -> MediaRow:
"""Split a Region into an art slot and a body column.
With ``square=True`` the art slot is bounds.h wide (album-art style);
otherwise it takes the left half. The gap defaults to 2px scaled by the
context's geometry factor.
"""
if gap is None:
gap = ctx.px(2, minimum=1) if ctx else 2
art_w = bounds.h if square else bounds.w // 2
art_w = min(art_w, bounds.w)
art = bounds.left_col(art_w)
body = Region(bounds.x + art_w + gap, bounds.y,
bounds.w - art_w - gap, bounds.h)
return MediaRow(art=art, body=body)
+137
View File
@@ -0,0 +1,137 @@
"""
Background Cache Mixin for Sports Managers
This mixin provides common caching functionality to eliminate code duplication
across all sports managers. It implements the background service cache pattern
where Recent/Upcoming managers consume data from the background service cache.
"""
import time
import logging
from typing import Dict, Optional, Any, Callable
from datetime import datetime
import pytz
class BackgroundCacheMixin:
"""
Mixin class that provides background service cache functionality to sports managers.
This mixin eliminates code duplication by providing a common implementation
for the background service cache pattern used across all sports managers.
Note: For non-sports managers (weather, stocks, news, etc.), use
GenericCacheMixin instead. See src/generic_cache_mixin.py for details.
"""
def _fetch_data_with_background_cache(self,
sport_key: str,
api_fetch_method: Callable,
live_manager_class: type = None) -> Optional[Dict]:
"""
Common logic for fetching data with background service cache support.
This method implements the background service cache pattern:
1. Live managers always fetch fresh data
2. Recent/Upcoming managers try background cache first
3. Fallback to direct API call if background data unavailable
Args:
sport_key: Sport identifier (e.g., 'nba', 'nfl', 'ncaa_fb')
api_fetch_method: Method to call for direct API fetch
live_manager_class: Class to check if this is a live manager
Returns:
Cached or fresh data from API
"""
start_time = time.time()
cache_hit = False
cache_source = None
try:
# For Live managers, always fetch fresh data
if live_manager_class and isinstance(self, live_manager_class):
self.logger.info(f"[{sport_key.upper()}] Live manager - fetching fresh data")
result = api_fetch_method(use_cache=False)
cache_source = "live_fresh"
else:
# For Recent/Upcoming managers, try background service cache first
cache_key = self.cache_manager.generate_sport_cache_key(sport_key)
# Check if background service has fresh data
if self.cache_manager.is_background_data_available(cache_key, sport_key):
cached_data = self.cache_manager.get_background_cached_data(cache_key, sport_key)
if cached_data:
self.logger.info(f"[{sport_key.upper()}] Using background service cache for {cache_key}")
result = cached_data
cache_hit = True
cache_source = "background_cache"
else:
self.logger.warning(f"[{sport_key.upper()}] Background cache check passed but no data returned for {cache_key}")
result = None
cache_source = "background_miss"
else:
self.logger.info(f"[{sport_key.upper()}] Background data not available for {cache_key}")
result = None
cache_source = "background_unavailable"
# Fallback to direct API call if background data not available
if result is None:
self.logger.info(f"[{sport_key.upper()}] Fetching directly from API for {cache_key}")
result = api_fetch_method(use_cache=True)
cache_source = "api_fallback"
# Record performance metrics
duration = time.time() - start_time
self.cache_manager.record_fetch_time(duration)
# Log performance metrics
self._log_fetch_performance(sport_key, duration, cache_hit, cache_source)
return result
except Exception as e:
duration = time.time() - start_time
self.logger.error(f"[{sport_key.upper()}] Error in background cache fetch after {duration:.2f}s: {e}")
self.cache_manager.record_fetch_time(duration)
raise
def _log_fetch_performance(self, sport_key: str, duration: float, cache_hit: bool, cache_source: str):
"""
Log detailed performance metrics for fetch operations.
Args:
sport_key: Sport identifier
duration: Fetch operation duration in seconds
cache_hit: Whether this was a cache hit
cache_source: Source of the data (background_cache, api_fallback, etc.)
"""
# Log basic performance info
self.logger.info(f"[{sport_key.upper()}] Fetch completed in {duration:.2f}s "
f"(cache_hit={cache_hit}, source={cache_source})")
# Log detailed metrics every 10 operations
if hasattr(self, '_fetch_count'):
self._fetch_count += 1
else:
self._fetch_count = 1
if self._fetch_count % 10 == 0:
metrics = self.cache_manager.get_cache_metrics()
self.logger.info(f"[{sport_key.upper()}] Cache Performance Summary - "
f"Hit Rate: {metrics['cache_hit_rate']:.2%}, "
f"Background Hit Rate: {metrics['background_hit_rate']:.2%}, "
f"API Calls Saved: {metrics['api_calls_saved']}")
def get_cache_performance_summary(self) -> Dict[str, Any]:
"""
Get cache performance summary for this manager.
Returns:
Dictionary containing cache performance metrics
"""
return self.cache_manager.get_cache_metrics()
def log_cache_performance(self):
"""Log current cache performance metrics."""
self.cache_manager.log_cache_metrics()
+22 -7
View File
@@ -14,15 +14,19 @@ Key Features:
- Memory-efficient data storage
"""
import os
import time
import logging
import threading
import requests
from typing import Dict, Any, Optional, Callable
from typing import Dict, Any, Optional, List, Callable, Union
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from enum import Enum
import json
import queue
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor, Future
import weakref
from src.cache_manager import CacheManager
# Configure logging
logger = logging.getLogger(__name__)
@@ -223,7 +227,7 @@ class BackgroundDataService:
self.stats['cache_misses'] += 1
# Submit to executor
self.executor.submit(self._fetch_data_worker, request)
future = self.executor.submit(self._fetch_data_worker, request)
logger.info(f"Submitted background fetch request {request_id} for {sport} {year}")
return request_id
@@ -549,12 +553,13 @@ class BackgroundDataService:
if to_remove:
logger.info(f"Cleared {len(to_remove)} old completed requests")
def shutdown(self, wait: bool = True):
def shutdown(self, wait: bool = True, timeout: int = 30):
"""
Shutdown the background data service.
Args:
wait: Whether to wait for active requests to complete
timeout: Maximum time to wait for shutdown
"""
logger.info("Shutting down BackgroundDataService...")
@@ -565,14 +570,24 @@ class BackgroundDataService:
for request_id in list(self.active_requests.keys()):
self.cancel_request(request_id)
self.executor.shutdown(wait=wait)
# Shutdown executor with compatibility for older Python versions
try:
# Try with timeout parameter (Python 3.9+)
self.executor.shutdown(wait=wait, timeout=timeout)
except TypeError:
# Fallback for older Python versions that don't support timeout
if wait and timeout:
# For older versions, we can't specify timeout, so just wait
self.executor.shutdown(wait=True)
else:
self.executor.shutdown(wait=wait)
logger.info("BackgroundDataService shutdown complete")
def __del__(self):
"""Cleanup when service is destroyed."""
if not self._shutdown:
self.shutdown(wait=False)
self.shutdown(wait=False, timeout=None)
# Global service instance
_background_service: Optional[BackgroundDataService] = None

Some files were not shown because too many files have changed in this diff Show More