Compare commits

..
Author SHA1 Message Date
ChuckBuildsandClaude Sonnet 5 cd7e16e58e fix(security): validate plugin_id before path construction in /api/install
CodeQL flagged 16 high-severity "path depends on user-provided value"
alerts. Investigated each:

- install_locally() (/api/install) built a filesystem path from
  metadata.id without validating it at that point -- it was only
  implicitly safe because _generate_plugin_files() validates the same
  field (re-extracted independently) earlier in the same request. That's
  a real gap: reorder or change that earlier call and it's an exploitable
  path traversal / arbitrary file write. Fixed by validating plugin_id
  directly against _PLUGIN_ID_RE at the point the path is built, matching
  the pattern already used correctly in validate_id() and load_plugin().
- The other 10 flagged locations (serve_font's allowlist check,
  validate_id, load_plugin and its downstream reads) were already
  guarded by an explicit check earlier in the same function -- false
  positives from CodeQL not modeling those as sanitizers.

Also fixed 2 of the 5 "stack trace exposed" warnings that were genuine:
install_locally() and load_plugin() returned raw OSError/Exception text
to the client in a 500 response; now logged server-side with a generic
client-facing message. The other 3 (generate_zip/install_locally/
preview_code returning str(ValueError) from _generate_plugin_files) are
deliberate, human-authored validation messages, not exception internals
-- left as-is.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
2026-07-14 17:23:25 -04:00
ChuckBuildsandClaude Sonnet 5 47e3021fc3 fix: address Codacy findings in the Composer blueprint
- Dropped a pointless f-string prefix (no placeholders) on the default
  plugin description.
- Replaced two bare except:pass/continue blocks (manifest.json listing,
  config_schema.json parsing) with a logged warning before falling
  through to the same skip-this-entry behavior -- same control flow,
  now visible in logs instead of silent.

Skipped as false positives (verified against actual usage, not fixed):
- Jinja2 Environment(autoescape=False) -- this env renders manager.py.j2,
  a Python source-code generator, never HTML; autoescaping would corrupt
  generated code. Flagged by a generic XSS rule that assumes all Jinja2
  environments render HTML.
- "Flask route directly returning a formatted string" on _as_rgb_filter
  -- that's a Jinja *filter* function, not a Flask route.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
2026-07-14 16:31:43 -04:00
ChuckBuildsandClaude Sonnet 5 e319540c6e feat(web): add Plugin Composer -- visual drag-and-drop plugin builder
Web UI (/composer/) for building a working LEDMatrix plugin without
writing Python: drop elements (text, time, date, countdown, scrolling
text, bar/waveform, groups, custom config variables) onto a canvas
matching the real panel's pixel grid, configure them with live preview,
then generate a real plugin (manager.py + manifest.json + config_schema.json)
from manager.py.j2 -- downloadable as a ZIP or installed directly.

NOTE: composer_bp is not yet registered in web_interface/app.py, so this
blueprint is currently inert. Split out of the original chore/dead-code-
removal commit, which had accidentally bundled this in alongside unrelated
dead-code deletions; app.py registration was not part of that commit
either and still needs to be added before this is reachable.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
2026-07-14 16:27:50 -04:00
15 changed files with 6670 additions and 1 deletions
+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
+7
View File
@@ -8,11 +8,16 @@ 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
# 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
@@ -24,8 +29,10 @@ spotipy>=2.25.2,<3.0.0
Flask>=3.1.3,<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
+1 -1
View File
@@ -55,7 +55,7 @@ def main():
failures += not check("draw.textbbox",
lambda: draw.textbbox((0, 0), "Test", font=font))
print("\nResampling (used in logo_helper, sports base):")
print("\nResampling (used in logo_helper, image_utils, sports base):")
logo = Image.new('RGBA', (200, 200), (255, 128, 0, 200))
failures += not check("Image.Resampling.LANCZOS exists",
lambda: str(Image.Resampling.LANCZOS))
+134
View File
@@ -0,0 +1,134 @@
"""
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
from typing import Dict, Optional, Any, Callable
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()
+328
View File
@@ -0,0 +1,328 @@
"""
Example: Basketball Plugin using LEDMatrix Common Helpers
This example shows how to refactor the basketball plugin to use the
ledmatrix-common package for cleaner, more maintainable code.
"""
from pathlib import Path
from typing import Any, Dict, List, Optional
# Import common helpers
from src.common import (
LogoHelper, TextHelper, APIHelper, DisplayHelper,
GameHelper, ConfigHelper
)
from src.plugin_system.base_plugin import BasePlugin
class BasketballPluginManager(BasePlugin):
"""
Basketball scoreboard plugin using LEDMatrix Common helpers.
This version is much cleaner and more maintainable than the original
because it delegates common functionality to the shared helpers.
"""
def __init__(
self,
plugin_id: str,
config: Dict[str, Any],
display_manager,
cache_manager,
plugin_manager
):
"""Initialize the basketball plugin with common helpers."""
super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
# Get display dimensions
self.display_width = display_manager.matrix.width
self.display_height = display_manager.matrix.height
# Initialize common helpers
self._init_helpers()
# Load configuration
self._load_config()
# State tracking
self.current_games = []
self.current_game = None
# Log initialization
enabled_leagues = [k for k, v in self.league_configs.items() if v['enabled']]
self.logger.info(f"Basketball plugin initialized with leagues: {enabled_leagues}")
def _init_helpers(self):
"""Initialize all common helpers."""
# Logo helper for team logos
self.logo_helper = LogoHelper(
display_width=self.display_width,
display_height=self.display_height,
logger=self.logger
)
# Text helper for rendering
self.text_helper = TextHelper(logger=self.logger)
self.fonts = self.text_helper.load_fonts()
# API helper for ESPN data
self.api_helper = APIHelper(
cache_manager=self.cache_manager,
logger=self.logger
)
# Display helper for layouts
self.display_helper = DisplayHelper(
display_width=self.display_width,
display_height=self.display_height,
logger=self.logger
)
# Game helper for data processing
self.game_helper = GameHelper(
timezone_str=self.config.get('timezone', 'UTC'),
logger=self.logger
)
# Config helper for configuration management
self.config_helper = ConfigHelper(logger=self.logger)
def _load_config(self):
"""Load and validate configuration."""
# Get basketball-specific config
basketball_config = self.config_helper.get_sports_config(self.config, 'basketball')
# Build league configurations
self.league_configs = {
'nba': {
'enabled': basketball_config.get('nba_enabled', True),
'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/nba/scoreboard',
'logo_dir': Path('assets/sports/nba_logos'),
'favorite_teams': basketball_config.get('nba_favorite_teams', []),
'display_modes': {
'nba_live': basketball_config.get('nba_display_modes_live', True),
'nba_recent': basketball_config.get('nba_display_modes_recent', True),
'nba_upcoming': basketball_config.get('nba_display_modes_upcoming', True),
},
},
'wnba': {
'enabled': basketball_config.get('wnba_enabled', False),
'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/wnba/scoreboard',
'logo_dir': Path('assets/sports/wnba_logos'),
'favorite_teams': basketball_config.get('wnba_favorite_teams', []),
'display_modes': {
'wnba_live': basketball_config.get('wnba_display_modes_live', True),
'wnba_recent': basketball_config.get('wnba_display_modes_recent', True),
'wnba_upcoming': basketball_config.get('wnba_display_modes_upcoming', True),
},
},
'ncaam': {
'enabled': basketball_config.get('ncaam_basketball_enabled', False),
'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/mens-college-basketball/scoreboard',
'logo_dir': Path('assets/sports/ncaa_logos'),
'favorite_teams': basketball_config.get('ncaam_basketball_favorite_teams', []),
'display_modes': {
'ncaam_basketball_live': basketball_config.get('ncaam_basketball_display_modes_live', True),
'ncaam_basketball_recent': basketball_config.get('ncaam_basketball_display_modes_recent', True),
'ncaam_basketball_upcoming': basketball_config.get('ncaam_basketball_display_modes_upcoming', True),
},
},
'ncaaw': {
'enabled': basketball_config.get('ncaaw_basketball_enabled', False),
'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/womens-college-basketball/scoreboard',
'logo_dir': Path('assets/sports/ncaa_logos'),
'favorite_teams': basketball_config.get('ncaaw_basketball_favorite_teams', []),
'display_modes': {
'ncaaw_basketball_live': basketball_config.get('ncaaw_basketball_display_modes_live', True),
'ncaaw_basketball_recent': basketball_config.get('ncaaw_basketball_display_modes_recent', True),
'ncaaw_basketball_upcoming': basketball_config.get('ncaaw_basketball_display_modes_upcoming', True),
},
},
}
def update(self) -> None:
"""Update game data for all enabled leagues."""
try:
all_games = []
for league_key, league_config in self.league_configs.items():
if not league_config['enabled']:
continue
games = self._fetch_league_games(league_key, league_config)
for game in games:
game['league_key'] = league_key
game['league_config'] = league_config
all_games.extend(games)
self.current_games = all_games
self.logger.debug(f"Updated basketball data: {len(all_games)} total games")
except Exception as e:
self.logger.error(f"Error updating basketball data: {e}", exc_info=True)
def _fetch_league_games(self, league_key: str, league_config: Dict) -> List[Dict]:
"""Fetch games for a specific league using API helper."""
try:
# Use API helper to fetch ESPN data with caching
data = self.api_helper.fetch_espn_scoreboard(
sport='basketball',
league=league_key,
cache_key=f"basketball_{league_key}",
cache_ttl=300 # 5 minutes cache
)
if not data:
return []
# Use game helper to process events
events = data.get('events', [])
games = self.game_helper.process_games(events, sport='basketball')
# Add logo paths to games
for game in games:
logo_dir = league_config['logo_dir']
game['home_logo_path'] = logo_dir / f"{game['home_abbr']}.png"
game['away_logo_path'] = logo_dir / f"{game['away_abbr']}.png"
return games
except Exception as e:
self.logger.error(f"Error fetching {league_key} games: {e}", exc_info=True)
return []
def display(self, force_clear: bool = False, display_mode: str = None) -> None:
"""Display basketball games using display helper."""
try:
mode = display_mode or self._determine_display_mode()
if not mode:
self._display_no_games()
return
# Filter games for mode
filtered_games = self._filter_games_for_mode(mode)
if not filtered_games:
self._display_no_games()
return
# Display first game
self.current_game = filtered_games[0]
self._draw_scorebug_layout(self.current_game, force_clear)
except Exception as e:
self.logger.error(f"Error displaying game: {e}", exc_info=True)
def _determine_display_mode(self) -> Optional[str]:
"""Determine display mode based on available games."""
# Priority: live > recent > upcoming
for game in self.current_games:
if game.get('is_live'):
return f"{game['league_key']}_live"
for game in self.current_games:
if game.get('is_final'):
return f"{game['league_key']}_recent"
for game in self.current_games:
if game.get('is_upcoming'):
return f"{game['league_key']}_upcoming"
return None
def _filter_games_for_mode(self, mode: str) -> List[Dict]:
"""Filter games based on display mode."""
filtered = []
for game in self.current_games:
league_config = game.get('league_config', {})
display_modes = league_config.get('display_modes', {})
if mode in display_modes and display_modes[mode]:
if 'live' in mode and game.get('is_live'):
filtered.append(game)
elif 'recent' in mode and game.get('is_final'):
filtered.append(game)
elif 'upcoming' in mode and game.get('is_upcoming'):
filtered.append(game)
return filtered[:5]
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
"""Draw the basketball scorebug layout using display helper."""
try:
# Load logos using logo helper
home_logo = self.logo_helper.load_logo(
game['home_abbr'],
game['home_logo_path']
)
away_logo = self.logo_helper.load_logo(
game['away_abbr'],
game['away_logo_path']
)
if not home_logo or not away_logo:
self.logger.error("Failed to load logos")
self._display_error("Logo Error")
return
# Use display helper to create scorebug layout
final_img = self.display_helper.draw_scorebug_layout(
game_data=game,
fonts=self.fonts,
home_logo=home_logo,
away_logo=away_logo
)
# Display the image
self.display_manager.image.paste(final_img, (0, 0))
self.display_manager.update_display()
except Exception as e:
self.logger.error(f"Error drawing scorebug: {e}", exc_info=True)
def _display_no_games(self) -> None:
"""Display 'no games' message using display helper."""
try:
img = self.display_helper.draw_no_data_message("No Games")
self.display_manager.image = img.copy()
self.display_manager.update_display()
except Exception as e:
self.logger.error(f"Error displaying no games: {e}", exc_info=True)
def _display_error(self, message: str) -> None:
"""Display error message using display helper."""
try:
img = self.display_helper.draw_error_message(message)
self.display_manager.image = img.copy()
self.display_manager.update_display()
except Exception as e:
self.logger.error(f"Error displaying error message: {e}", exc_info=True)
def get_display_duration(self) -> float:
"""Get display duration."""
return self.config.get('display_duration', 15)
def cleanup(self) -> None:
"""Cleanup resources."""
self.current_games = []
self.logger.info("Basketball plugin cleaned up")
# Example usage and benefits:
"""
Benefits of using LEDMatrix Common helpers:
1. **Cleaner Code**: The plugin is much shorter and more readable
2. **Reusable Components**: Common functionality is shared across plugins
3. **Better Testing**: Each helper can be tested independently
4. **Easier Maintenance**: Bug fixes in helpers benefit all plugins
5. **Consistent Behavior**: All plugins use the same underlying logic
6. **Reduced Dependencies**: Plugins don't need to import LEDMatrix core
7. **Better Error Handling**: Centralized error handling in helpers
8. **Configuration Management**: Consistent config handling across plugins
The original basketball plugin was 326 lines. This version is much cleaner
and delegates most functionality to the common helpers, making it easier to
maintain and extend.
"""
+135
View File
@@ -0,0 +1,135 @@
import os
import freetype
from PIL import ImageDraw, ImageFont
import logging
from typing import Dict, Any
from src.display_manager import DisplayManager
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FontTestManager:
"""Manager for testing fonts with easy BDF/TTF switching."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
self.display_manager = display_manager
self.config = config
self.logger = logging.getLogger('FontTest')
# FONT CONFIGURATION - EASY SWITCHING
# Set to 'bdf' or 'ttf' to switch font types
self.font_type = 'bdf' # Change this to 'ttf' to use TTF font
# Font configurations
self.font_configs = {
'bdf': {
'path': "assets/fonts/cozette.bdf",
'display_name': "Cozette BTF",
'description': "BTF font Test"
},
'ttf': {
'path': "assets/fonts/5by7.regular.ttf",
'display_name': "5by7 TTF",
'description': "TTF font test"
}
}
# Get current font configuration
self.current_config = self.font_configs[self.font_type]
self.font_path = self.current_config['path']
# Verify font exists
if not os.path.exists(self.font_path):
self.logger.error(f"Font file not found: {self.font_path}")
raise FileNotFoundError(f"Font file not found: {self.font_path}")
# Load the font based on type
if self.font_type == 'bdf':
self._load_bdf_font()
else:
self._load_ttf_font()
self.logger.info(f"Initialized FontTestManager with {self.current_config['description']}")
def _load_bdf_font(self):
"""Load BDF font using freetype."""
try:
self.face = freetype.Face(self.font_path)
self.logger.info(f"Successfully loaded BDF font from {self.font_path}")
except Exception as e:
self.logger.error(f"Failed to load BDF font: {e}")
raise
def _load_ttf_font(self):
"""Load TTF font using PIL."""
try:
self.font = ImageFont.truetype(self.font_path, 8) # Size 8 for 5x7 font
self.logger.info(f"Successfully loaded TTF font from {self.font_path}")
except Exception as e:
self.logger.error(f"Failed to load TTF font: {e}")
raise
def update(self):
"""No update needed for static display."""
def display(self, force_clear: bool = False):
"""Display the font with sample text."""
try:
# Clear the display
self.display_manager.clear()
# Draw font name at the top
self.display_manager.draw_text(self.current_config['display_name'], y=2, color=(255, 255, 255))
# Draw sample text
draw = ImageDraw.Draw(self.display_manager.image)
sample_text = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
# Calculate starting position
x = 10 # Start 10 pixels from the left
y = 10 # Start 10 pixels from the top
# Draw text based on font type
if self.font_type == 'bdf':
self._draw_bdf_text(draw, sample_text, x, y)
else:
self._draw_ttf_text(draw, sample_text, x, y)
# Update the display once
self.display_manager.update_display()
# Log that display is complete
self.logger.info("Font test display complete.")
except Exception as e:
self.logger.error(f"Error displaying font test: {e}", exc_info=True)
def _draw_bdf_text(self, draw, text, x, y):
"""Draw text using BDF font."""
for char in text:
# Load the glyph
self.face.load_char(char)
bitmap = self.face.glyph.bitmap
# Draw the glyph
for i in range(bitmap.rows):
for j in range(bitmap.width):
try:
# Get the byte containing the pixel
byte_index = i * bitmap.pitch + (j // 8)
if byte_index < len(bitmap.buffer):
byte = bitmap.buffer[byte_index]
# Check if the specific bit is set
if byte & (1 << (7 - (j % 8))):
draw.point((x + j, y + i), fill=(255, 255, 255))
except IndexError:
self.logger.warning(f"Index out of range for char '{char}' at position ({i}, {j})")
continue
# Move to next character position
x += self.face.glyph.advance.x >> 6
def _draw_ttf_text(self, draw, text, x, y):
"""Draw text using TTF font."""
draw.text((x, y), text, font=self.font, fill=(255, 255, 255))
+150
View File
@@ -0,0 +1,150 @@
"""
Generic Cache Mixin for Any Manager
This mixin provides caching functionality that can be used by any manager
that needs to cache data, not just sports managers. It's a more general
version of BackgroundCacheMixin that works for weather, stocks, news, etc.
"""
import time
from typing import Dict, Optional, Any, Callable
class GenericCacheMixin:
"""
Generic mixin class that provides caching functionality to any manager.
This mixin can be used by weather, stock, news, or any other manager
that needs to cache data with performance monitoring.
Note: For sports managers that need background service cache integration,
use BackgroundCacheMixin instead. See src/background_cache_mixin.py for details.
"""
def _fetch_data_with_cache(self,
cache_key: str,
api_fetch_method: Callable,
cache_ttl: int = 300,
force_refresh: bool = False) -> Optional[Dict]:
"""
Generic caching pattern for any manager.
Args:
cache_key: Unique cache key for this data
api_fetch_method: Method to call for fresh data
cache_ttl: Time-to-live in seconds (default: 5 minutes)
force_refresh: Skip cache and fetch fresh data
Returns:
Cached or fresh data from API
"""
start_time = time.time()
cache_hit = False
cache_source = None
try:
# Check cache first (unless forcing refresh)
if not force_refresh:
cached_data = self.cache_manager.get_cached_data(cache_key, cache_ttl)
if cached_data:
self.logger.info(f"Using cached data for {cache_key}")
cache_hit = True
cache_source = "cache"
self.cache_manager.record_cache_hit('regular')
# Record performance metrics
duration = time.time() - start_time
self.cache_manager.record_fetch_time(duration)
self._log_fetch_performance(cache_key, duration, cache_hit, cache_source)
return cached_data
# Fetch fresh data
self.logger.info(f"Fetching fresh data for {cache_key}")
result = api_fetch_method()
cache_source = "api_fresh"
# Store in cache if we got data
if result:
self.cache_manager.save_cache(cache_key, result)
self.cache_manager.record_cache_miss('regular')
else:
self.logger.warning(f"No data returned for {cache_key}")
# Record performance metrics
duration = time.time() - start_time
self.cache_manager.record_fetch_time(duration)
# Log performance
self._log_fetch_performance(cache_key, duration, cache_hit, cache_source)
return result
except Exception as e:
duration = time.time() - start_time
self.logger.error(f"Error fetching data for {cache_key} after {duration:.2f}s: {e}")
self.cache_manager.record_fetch_time(duration)
raise
def _log_fetch_performance(self, cache_key: str, duration: float, cache_hit: bool, cache_source: str):
"""
Log detailed performance metrics for fetch operations.
Args:
cache_key: Cache key that was accessed
duration: Fetch operation duration in seconds
cache_hit: Whether this was a cache hit
cache_source: Source of the data (cache, api_fresh, etc.)
"""
# Log basic performance info
self.logger.info(f"Fetch completed for {cache_key} 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"Cache Performance Summary - "
f"Hit Rate: {metrics['cache_hit_rate']:.2%}, "
f"API Calls Saved: {metrics['api_calls_saved']}, "
f"Avg Fetch Time: {metrics['average_fetch_time']:.2f}s")
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()
def clear_cache_for_key(self, cache_key: str):
"""Clear cache for a specific key."""
self.cache_manager.clear_cache(cache_key)
self.logger.info(f"Cleared cache for {cache_key}")
def get_cache_info(self, cache_key: str) -> Dict[str, Any]:
"""
Get information about a cached item.
Args:
cache_key: Cache key to check
Returns:
Dictionary with cache information
"""
# This would need to be implemented in CacheManager
# For now, just return basic info
return {
'key': cache_key,
'exists': self.cache_manager.get_cached_data(cache_key, 0) is not None,
'ttl': 'unknown' # Would need to be implemented
}
+22
View File
@@ -0,0 +1,22 @@
"""Deprecated: use src/adaptive_images.py (fit_image) instead.
This module predates the adaptive image system and has no known callers.
It is kept only so any out-of-tree code importing it keeps working.
"""
import logging
from PIL import Image
logger = logging.getLogger(__name__)
def scale_to_max_dimensions(img, max_width, max_height):
h_to_w_ratio = img.height / img.width
w_to_h_ratio = img.width / img.height
if img.height > max_height:
img = img.resize((int(max_height * w_to_h_ratio), max_height), Image.Resampling.LANCZOS)
if img.width > max_width:
img = img.resize((max_width, int(max_width * h_to_w_ratio)), Image.Resampling.LANCZOS)
return img
+409
View File
@@ -0,0 +1,409 @@
"""
Layout Manager for LED Matrix Display
Handles custom layouts, element positioning, and display composition.
"""
import json
import os
import logging
from typing import Dict, List, Any
from datetime import datetime
logger = logging.getLogger(__name__)
class LayoutManager:
def __init__(self, display_manager=None, config_path="config/custom_layouts.json"):
self.display_manager = display_manager
self.config_path = config_path
self.layouts = self.load_layouts()
self.current_layout = None
def load_layouts(self) -> Dict[str, Any]:
"""Load saved layouts from file."""
try:
if os.path.exists(self.config_path):
with open(self.config_path, 'r') as f:
return json.load(f)
return {}
except Exception as e:
logger.error(f"Error loading layouts: {e}")
return {}
def save_layouts(self) -> bool:
"""Save layouts to file."""
try:
from pathlib import Path
from src.common.permission_utils import (
ensure_directory_permissions,
get_config_dir_mode
)
config_path_obj = Path(self.config_path)
ensure_directory_permissions(config_path_obj.parent, get_config_dir_mode())
with open(self.config_path, 'w') as f:
json.dump(self.layouts, f, indent=2)
return True
except Exception as e:
logger.error(f"Error saving layouts: {e}")
return False
def create_layout(self, name: str, elements: List[Dict], description: str = "") -> bool:
"""Create a new layout."""
try:
self.layouts[name] = {
'elements': elements,
'description': description,
'created': datetime.now().isoformat(),
'modified': datetime.now().isoformat()
}
return self.save_layouts()
except Exception as e:
logger.error(f"Error creating layout '{name}': {e}")
return False
def update_layout(self, name: str, elements: List[Dict], description: str = None) -> bool:
"""Update an existing layout."""
try:
if name not in self.layouts:
return False
self.layouts[name]['elements'] = elements
self.layouts[name]['modified'] = datetime.now().isoformat()
if description is not None:
self.layouts[name]['description'] = description
return self.save_layouts()
except Exception as e:
logger.error(f"Error updating layout '{name}': {e}")
return False
def delete_layout(self, name: str) -> bool:
"""Delete a layout."""
try:
if name in self.layouts:
del self.layouts[name]
return self.save_layouts()
return False
except Exception as e:
logger.error(f"Error deleting layout '{name}': {e}")
return False
def get_layout(self, name: str) -> Dict[str, Any]:
"""Get a specific layout."""
return self.layouts.get(name, {})
def list_layouts(self) -> List[str]:
"""Get list of all layout names."""
return list(self.layouts.keys())
def set_current_layout(self, name: str) -> bool:
"""Set the current active layout."""
if name in self.layouts:
self.current_layout = name
return True
return False
def render_layout(self, layout_name: str = None, data_context: Dict = None) -> bool:
"""Render a layout to the display."""
if not self.display_manager:
logger.error("No display manager available")
return False
layout_name = layout_name or self.current_layout
if not layout_name or layout_name not in self.layouts:
logger.error(f"Layout '{layout_name}' not found")
return False
try:
# Clear the display
self.display_manager.clear()
# Get layout elements
elements = self.layouts[layout_name]['elements']
# Render each element
for element in elements:
self.render_element(element, data_context or {})
# Update the display
self.display_manager.update_display()
return True
except Exception as e:
logger.error(f"Error rendering layout '{layout_name}': {e}")
return False
def render_element(self, element: Dict, data_context: Dict) -> None:
"""Render a single element."""
element_type = element.get('type')
x = element.get('x', 0)
y = element.get('y', 0)
properties = element.get('properties', {})
try:
if element_type == 'text':
self._render_text_element(x, y, properties, data_context)
elif element_type == 'weather_icon':
self._render_weather_icon_element(x, y, properties, data_context)
elif element_type == 'rectangle':
self._render_rectangle_element(x, y, properties)
elif element_type == 'line':
self._render_line_element(x, y, properties)
elif element_type == 'clock':
self._render_clock_element(x, y, properties)
elif element_type == 'data_text':
self._render_data_text_element(x, y, properties, data_context)
else:
logger.warning(f"Unknown element type: {element_type}")
except Exception as e:
logger.error(f"Error rendering element {element_type}: {e}")
def _render_text_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
"""Render a text element."""
text = properties.get('text', 'Sample Text')
color = tuple(properties.get('color', [255, 255, 255]))
font_size = properties.get('font_size', 'normal')
# Support template variables in text
text = self._process_template_text(text, data_context)
# Select font
if font_size == 'small':
font = self.display_manager.small_font
elif font_size == 'large':
font = self.display_manager.regular_font
else:
font = self.display_manager.regular_font
self.display_manager.draw_text(text, x, y, color, font=font)
def _render_weather_icon_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
"""Render a weather icon element."""
condition = properties.get('condition', 'sunny')
size = properties.get('size', 16)
# Use weather data from context if available
if 'weather' in data_context and 'condition' in data_context['weather']:
condition = data_context['weather']['condition'].lower()
self.display_manager.draw_weather_icon(condition, x, y, size)
def _render_rectangle_element(self, x: int, y: int, properties: Dict) -> None:
"""Render a rectangle element."""
width = properties.get('width', 10)
height = properties.get('height', 10)
color = tuple(properties.get('color', [255, 255, 255]))
filled = properties.get('filled', False)
if filled:
self.display_manager.draw.rectangle(
[x, y, x + width, y + height],
fill=color
)
else:
self.display_manager.draw.rectangle(
[x, y, x + width, y + height],
outline=color
)
def _render_line_element(self, x: int, y: int, properties: Dict) -> None:
"""Render a line element."""
x2 = properties.get('x2', x + 10)
y2 = properties.get('y2', y)
color = tuple(properties.get('color', [255, 255, 255]))
width = properties.get('width', 1)
self.display_manager.draw.line([x, y, x2, y2], fill=color, width=width)
def _render_clock_element(self, x: int, y: int, properties: Dict) -> None:
"""Render a clock element."""
format_str = properties.get('format', '%H:%M')
color = tuple(properties.get('color', [255, 255, 255]))
current_time = datetime.now().strftime(format_str)
self.display_manager.draw_text(current_time, x, y, color)
def _render_data_text_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
"""Render a data-driven text element."""
data_key = properties.get('data_key', '')
format_str = properties.get('format', '{value}')
color = tuple(properties.get('color', [255, 255, 255]))
default_value = properties.get('default', 'N/A')
# Extract data from context
value = self._get_nested_value(data_context, data_key, default_value)
# Format the text
try:
text = format_str.format(value=value)
except (ValueError, TypeError, KeyError, IndexError):
text = str(value)
self.display_manager.draw_text(text, x, y, color)
def _process_template_text(self, text: str, data_context: Dict) -> str:
"""Process template variables in text."""
try:
# Simple template processing - replace {key} with values from context
for key, value in data_context.items():
placeholder = f"{{{key}}}"
if placeholder in text:
text = text.replace(placeholder, str(value))
return text
except Exception as e:
logger.error(f"Error processing template text: {e}")
return text
def _get_nested_value(self, data: Dict, key: str, default=None):
"""Get a nested value from a dictionary using dot notation."""
try:
keys = key.split('.')
value = data
for k in keys:
value = value[k]
return value
except (KeyError, TypeError):
return default
def create_preset_layouts(self) -> None:
"""Create some preset layouts for common use cases."""
# Basic clock layout
clock_layout = [
{
'type': 'clock',
'x': 10,
'y': 10,
'properties': {
'format': '%H:%M',
'color': [255, 255, 255]
}
},
{
'type': 'clock',
'x': 10,
'y': 20,
'properties': {
'format': '%m/%d',
'color': [100, 100, 255]
}
}
]
self.create_layout('basic_clock', clock_layout, 'Simple clock with date')
# Weather layout
weather_layout = [
{
'type': 'weather_icon',
'x': 5,
'y': 5,
'properties': {
'condition': 'sunny',
'size': 20
}
},
{
'type': 'data_text',
'x': 30,
'y': 8,
'properties': {
'data_key': 'weather.temperature',
'format': '{value}°',
'color': [255, 200, 0],
'default': '--°'
}
},
{
'type': 'data_text',
'x': 30,
'y': 18,
'properties': {
'data_key': 'weather.condition',
'format': '{value}',
'color': [200, 200, 200],
'default': 'Unknown'
}
}
]
self.create_layout('weather_display', weather_layout, 'Weather icon with temperature and condition')
# Mixed dashboard layout
dashboard_layout = [
{
'type': 'clock',
'x': 2,
'y': 2,
'properties': {
'format': '%H:%M',
'color': [255, 255, 255]
}
},
{
'type': 'weather_icon',
'x': 50,
'y': 2,
'properties': {
'size': 16
}
},
{
'type': 'data_text',
'x': 70,
'y': 5,
'properties': {
'data_key': 'weather.temperature',
'format': '{value}°',
'color': [255, 200, 0],
'default': '--°'
}
},
{
'type': 'line',
'x': 0,
'y': 15,
'properties': {
'x2': 128,
'y2': 15,
'color': [100, 100, 100]
}
},
{
'type': 'data_text',
'x': 2,
'y': 18,
'properties': {
'data_key': 'stocks.AAPL.price',
'format': 'AAPL: ${value}',
'color': [0, 255, 0],
'default': 'AAPL: N/A'
}
}
]
self.create_layout('dashboard', dashboard_layout, 'Mixed dashboard with clock, weather, and stocks')
logger.info("Created preset layouts")
def get_layout_preview(self, layout_name: str) -> Dict[str, Any]:
"""Get a preview representation of a layout."""
if layout_name not in self.layouts:
return {}
layout = self.layouts[layout_name]
elements = layout['elements']
# Create a simple preview representation
preview = {
'name': layout_name,
'description': layout.get('description', ''),
'element_count': len(elements),
'elements': []
}
for element in elements:
preview['elements'].append({
'type': element.get('type'),
'position': f"({element.get('x', 0)}, {element.get('y', 0)})",
'properties': list(element.get('properties', {}).keys())
})
return preview
+392
View File
@@ -0,0 +1,392 @@
"""
Tests for LayoutManager.
Tests layout creation, management, rendering, and element positioning.
"""
import pytest
import json
from unittest.mock import MagicMock
from src.layout_manager import LayoutManager
class TestLayoutManager:
"""Test LayoutManager functionality."""
@pytest.fixture
def tmp_layout_file(self, tmp_path):
"""Create a temporary layout file."""
layout_file = tmp_path / "custom_layouts.json"
return str(layout_file)
@pytest.fixture
def mock_display_manager(self):
"""Create a mock display manager."""
dm = MagicMock()
dm.clear = MagicMock()
dm.update_display = MagicMock()
dm.draw_text = MagicMock()
dm.draw_weather_icon = MagicMock()
dm.small_font = MagicMock()
dm.regular_font = MagicMock()
return dm
@pytest.fixture
def layout_manager(self, tmp_layout_file, mock_display_manager):
"""Create a LayoutManager instance."""
return LayoutManager(
display_manager=mock_display_manager,
config_path=tmp_layout_file
)
def test_init(self, tmp_layout_file, mock_display_manager):
"""Test LayoutManager initialization."""
lm = LayoutManager(
display_manager=mock_display_manager,
config_path=tmp_layout_file
)
assert lm.display_manager == mock_display_manager
assert lm.config_path == tmp_layout_file
assert lm.layouts == {}
assert lm.current_layout is None
def test_load_layouts_file_exists(self, tmp_path, mock_display_manager):
"""Test loading layouts from existing file."""
layout_file = tmp_path / "custom_layouts.json"
layout_data = {
"test_layout": {
"elements": [{"type": "text", "x": 0, "y": 0}],
"description": "Test layout"
}
}
with open(layout_file, 'w') as f:
json.dump(layout_data, f)
lm = LayoutManager(
display_manager=mock_display_manager,
config_path=str(layout_file)
)
assert "test_layout" in lm.layouts
assert lm.layouts["test_layout"]["description"] == "Test layout"
def test_load_layouts_file_not_exists(self, tmp_layout_file, mock_display_manager):
"""Test loading layouts when file doesn't exist."""
lm = LayoutManager(
display_manager=mock_display_manager,
config_path=tmp_layout_file
)
assert lm.layouts == {}
def test_create_layout(self, layout_manager):
"""Test creating a new layout."""
elements = [{"type": "text", "x": 10, "y": 20, "properties": {"text": "Hello"}}]
result = layout_manager.create_layout("test_layout", elements, "Test description")
assert result is True
assert "test_layout" in layout_manager.layouts
assert layout_manager.layouts["test_layout"]["elements"] == elements
assert layout_manager.layouts["test_layout"]["description"] == "Test description"
assert "created" in layout_manager.layouts["test_layout"]
assert "modified" in layout_manager.layouts["test_layout"]
def test_update_layout(self, layout_manager):
"""Test updating an existing layout."""
# Create a layout first
elements1 = [{"type": "text", "x": 0, "y": 0}]
layout_manager.create_layout("test_layout", elements1, "Original")
# Update it
elements2 = [{"type": "text", "x": 10, "y": 20}]
result = layout_manager.update_layout("test_layout", elements2, "Updated")
assert result is True
assert layout_manager.layouts["test_layout"]["elements"] == elements2
assert layout_manager.layouts["test_layout"]["description"] == "Updated"
assert "modified" in layout_manager.layouts["test_layout"]
def test_update_layout_not_exists(self, layout_manager):
"""Test updating a non-existent layout."""
elements = [{"type": "text", "x": 0, "y": 0}]
result = layout_manager.update_layout("nonexistent", elements)
assert result is False
def test_delete_layout(self, layout_manager):
"""Test deleting a layout."""
elements = [{"type": "text", "x": 0, "y": 0}]
layout_manager.create_layout("test_layout", elements)
result = layout_manager.delete_layout("test_layout")
assert result is True
assert "test_layout" not in layout_manager.layouts
def test_delete_layout_not_exists(self, layout_manager):
"""Test deleting a non-existent layout."""
result = layout_manager.delete_layout("nonexistent")
assert result is False
def test_get_layout(self, layout_manager):
"""Test getting a specific layout."""
elements = [{"type": "text", "x": 0, "y": 0}]
layout_manager.create_layout("test_layout", elements)
layout = layout_manager.get_layout("test_layout")
assert layout is not None
assert layout["elements"] == elements
def test_get_layout_not_exists(self, layout_manager):
"""Test getting a non-existent layout."""
layout = layout_manager.get_layout("nonexistent")
assert layout == {}
def test_list_layouts(self, layout_manager):
"""Test listing all layouts."""
layout_manager.create_layout("layout1", [])
layout_manager.create_layout("layout2", [])
layout_manager.create_layout("layout3", [])
layouts = layout_manager.list_layouts()
assert len(layouts) == 3
assert "layout1" in layouts
assert "layout2" in layouts
assert "layout3" in layouts
def test_set_current_layout(self, layout_manager):
"""Test setting the current layout."""
layout_manager.create_layout("test_layout", [])
result = layout_manager.set_current_layout("test_layout")
assert result is True
assert layout_manager.current_layout == "test_layout"
def test_set_current_layout_not_exists(self, layout_manager):
"""Test setting a non-existent layout as current."""
result = layout_manager.set_current_layout("nonexistent")
assert result is False
assert layout_manager.current_layout is None
def test_render_layout(self, layout_manager, mock_display_manager):
"""Test rendering a layout."""
elements = [
{"type": "text", "x": 0, "y": 0, "properties": {"text": "Hello"}},
{"type": "text", "x": 10, "y": 10, "properties": {"text": "World"}}
]
layout_manager.create_layout("test_layout", elements)
result = layout_manager.render_layout("test_layout")
assert result is True
mock_display_manager.clear.assert_called_once()
mock_display_manager.update_display.assert_called_once()
assert mock_display_manager.draw_text.call_count == 2
def test_render_layout_no_display_manager(self, tmp_layout_file):
"""Test rendering without display manager."""
lm = LayoutManager(display_manager=None, config_path=tmp_layout_file)
lm.create_layout("test_layout", [])
result = lm.render_layout("test_layout")
assert result is False
def test_render_layout_not_exists(self, layout_manager):
"""Test rendering a non-existent layout."""
result = layout_manager.render_layout("nonexistent")
assert result is False
def test_render_element_text(self, layout_manager, mock_display_manager):
"""Test rendering a text element."""
element = {
"type": "text",
"x": 10,
"y": 20,
"properties": {
"text": "Hello",
"color": [255, 0, 0],
"font_size": "small"
}
}
layout_manager.render_element(element, {})
mock_display_manager.draw_text.assert_called_once()
call_args = mock_display_manager.draw_text.call_args
assert call_args[0][0] == "Hello" # text
assert call_args[0][1] == 10 # x
assert call_args[0][2] == 20 # y
def test_render_element_weather_icon(self, layout_manager, mock_display_manager):
"""Test rendering a weather icon element."""
element = {
"type": "weather_icon",
"x": 10,
"y": 20,
"properties": {
"condition": "sunny",
"size": 16
}
}
layout_manager.render_element(element, {})
mock_display_manager.draw_weather_icon.assert_called_once_with("sunny", 10, 20, 16)
def test_render_element_weather_icon_from_context(self, layout_manager, mock_display_manager):
"""Test rendering weather icon with data from context."""
element = {
"type": "weather_icon",
"x": 10,
"y": 20,
"properties": {"size": 16}
}
data_context = {
"weather": {
"condition": "cloudy"
}
}
layout_manager.render_element(element, data_context)
mock_display_manager.draw_weather_icon.assert_called_once_with("cloudy", 10, 20, 16)
def test_render_element_rectangle(self, layout_manager, mock_display_manager):
"""Test rendering a rectangle element."""
element = {
"type": "rectangle",
"x": 10,
"y": 20,
"properties": {
"width": 50,
"height": 30,
"color": [255, 0, 0],
"filled": True
}
}
# Mock the draw object and rectangle method
mock_draw = MagicMock()
mock_display_manager.draw = mock_draw
layout_manager.render_element(element, {})
# Verify rectangle was drawn
mock_draw.rectangle.assert_called_once()
def test_render_element_unknown_type(self, layout_manager):
"""Test rendering an unknown element type."""
element = {
"type": "unknown_type",
"x": 0,
"y": 0,
"properties": {}
}
# Should not raise an exception
layout_manager.render_element(element, {})
def test_process_template_text(self, layout_manager):
"""Test template text processing."""
text = "Hello {name}, temperature is {temp}°F"
data_context = {
"name": "World",
"temp": 72
}
result = layout_manager._process_template_text(text, data_context)
assert result == "Hello World, temperature is 72°F"
def test_process_template_text_no_context(self, layout_manager):
"""Test template text with missing context."""
text = "Hello {name}"
data_context = {}
result = layout_manager._process_template_text(text, data_context)
# Should leave template as-is or handle gracefully
assert "{name}" in result or result == "Hello "
def test_save_layouts_error_handling(self, layout_manager):
"""Test error handling when saving layouts."""
# Create a layout
layout_manager.create_layout("test", [])
# Make save fail by using invalid path
layout_manager.config_path = "/nonexistent/directory/layouts.json"
result = layout_manager.save_layouts()
# Should handle error gracefully
assert result is False
def test_render_element_line(self, layout_manager, mock_display_manager):
"""Test rendering a line element."""
element = {
"type": "line",
"x": 10,
"y": 20,
"properties": {
"x2": 50,
"y2": 30,
"color": [255, 0, 0],
"width": 2
}
}
mock_draw = MagicMock()
mock_display_manager.draw = mock_draw
layout_manager.render_element(element, {})
mock_draw.line.assert_called_once()
def test_render_element_clock(self, layout_manager, mock_display_manager):
"""Test rendering a clock element."""
element = {
"type": "clock",
"x": 10,
"y": 20,
"properties": {
"format": "%H:%M",
"color": [255, 255, 255]
}
}
layout_manager.render_element(element, {})
mock_display_manager.draw_text.assert_called_once()
def test_render_element_data_text(self, layout_manager, mock_display_manager):
"""Test rendering a data text element."""
element = {
"type": "data_text",
"x": 10,
"y": 20,
"properties": {
"data_key": "weather.temperature",
"format": "Temp: {value}°F",
"color": [255, 255, 255],
"default": "N/A"
}
}
data_context = {
"weather": {
"temperature": 72
}
}
layout_manager.render_element(element, data_context)
mock_display_manager.draw_text.assert_called_once()
+847
View File
@@ -0,0 +1,847 @@
"""
Plugin Composer blueprint drag-and-drop plugin builder for LEDMatrix.
Routes:
GET /composer/ Composer page
POST /composer/api/generate Generate and return plugin ZIP
POST /composer/api/install Write plugin directly to plugins_dir
GET /composer/api/fonts/<name> Serve TTF font files for canvas rendering
GET /composer/api/validate-id/<id> Check if a plugin ID is already taken
"""
import ast
import io
import json
import logging
import re
import zipfile
from datetime import datetime
from pathlib import Path
import jinja2
import jsonschema
from flask import Blueprint, jsonify, render_template, request, send_file
logger = logging.getLogger(__name__)
composer_bp = Blueprint('composer', __name__)
# Module-level attributes injected by app.py at registration time
composer_bp.config_manager = None
composer_bp.plugin_manager = None
composer_bp.plugins_dir = None
composer_bp.project_root = None
# Fonts safe to serve to the browser for canvas rendering
_ALLOWED_FONTS = frozenset({'PressStart2P-Regular.ttf', '4x6-font.ttf', '5by7.regular.ttf'})
# Map composer font keys → DisplayManager attribute names
_FONT_ATTR_MAP = {
'press_start': 'regular_font',
'four_by_six': 'extra_small_font',
'five_by_seven': 'bdf_5x7_font',
}
# Font sizes in LED pixels (used to compute second-line Y offsets)
_FONT_SIZE_MAP = {
'press_start': 8,
'four_by_six': 6,
'five_by_seven': 7,
}
_PLUGIN_ID_RE = re.compile(r'^[a-z][a-z0-9-]{0,62}$')
_PYTHON_IDENT_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
# ── Jinja2 environment (separate from Flask's; autoescape=False for code gen) ──
_jinja_env: jinja2.Environment | None = None
def _get_jinja_env() -> jinja2.Environment:
global _jinja_env
if _jinja_env is None:
template_dir = Path(__file__).parent.parent / 'templates' / 'v3' / 'composer'
_jinja_env = jinja2.Environment(
loader=jinja2.FileSystemLoader(str(template_dir)),
autoescape=False,
trim_blocks=True,
lstrip_blocks=True,
)
_jinja_env.filters['as_rgb'] = _as_rgb_filter
_jinja_env.filters['as_fill'] = _as_fill_filter
return _jinja_env
def _as_rgb_filter(val) -> str:
"""[r, g, b] → '(r, g, b)'"""
if val is None:
return 'None'
return f'({int(val[0])}, {int(val[1])}, {int(val[2])})'
def _as_fill_filter(val) -> str:
"""[r, g, b] or None → '(r, g, b)' or 'None'"""
if val is None:
return 'None'
return _as_rgb_filter(val)
# ── Helper functions ──────────────────────────────────────────────────────────
def _to_class_name(name: str) -> str:
"""'My Clock''MyClockPlugin' (avoids double-suffix if name already ends with Plugin)"""
words = re.sub(r'[^a-zA-Z0-9]', ' ', name).split()
base = ''.join(w.capitalize() for w in words)
return base if base.endswith('Plugin') else base + 'Plugin'
def _compute_pos_expr(val: int, anchor: str | None, dim_var: str) -> str:
"""Produce a Python expression string for an anchored or fixed position.
anchor=None/'left'/'top' fixed pixel value
anchor='center' dim_var // 2 ± offset
anchor='right'/'bottom' dim_var - offset
"""
if not anchor or anchor in ('left', 'top'):
return str(val)
if anchor in ('center', 'middle'):
if val == 0:
return f"{dim_var} // 2"
return f"{dim_var} // 2 + {val}" if val > 0 else f"{dim_var} // 2 - {abs(val)}"
if anchor in ('right', 'bottom'):
return dim_var if val == 0 else f"{dim_var} - {val}"
return str(val)
# Character widths in LED pixels per font (for text-alignment x offset math)
_FONT_CHAR_W = {
'press_start': 8,
'four_by_six': 4,
'five_by_seven': 5,
}
def _aligned_x_expr(x_base_expr: str, text_align: str, char_count: int, char_w: int) -> str:
"""Return Python x expression for text alignment.
left x_base_expr (no change)
center x_base_expr - half_text_width
right x_base_expr - text_width
"""
if text_align == 'left' or not text_align:
return x_base_expr
text_px = char_count * char_w
if text_align == 'center':
offset = text_px // 2
return f"({x_base_expr}) - {offset}" if offset else x_base_expr
if text_align == 'right':
return f"({x_base_expr}) - {text_px}" if text_px else x_base_expr
return x_base_expr
def _preprocess_elements(elements: list) -> list:
"""Expand raw element dicts into template-ready dicts with anchor expressions.
Invisible elements (visible=False) are excluded from generated code entirely.
"""
result = []
for el in elements:
# Skip hidden elements — they exist only in the preview
if el.get('visible') is False:
continue
p = dict(el)
t = el.get('type', '')
# Section elements are layer-list annotations only — no canvas output
if t == 'section':
continue
x_anchor = el.get('xAnchor') or None
y_anchor = el.get('yAnchor') or None
p['min_width'] = int(el.get('minWidth', 0) or 0)
if t in ('text', 'clock'):
font_key = el.get('font', 'press_start')
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
text_align = el.get('textAlign', 'left')
raw_x = el.get('x', 0)
x_base_expr = _compute_pos_expr(raw_x, x_anchor, 'width')
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
font_size = _FONT_SIZE_MAP.get(font_key, 8)
char_w = _FONT_CHAR_W.get(font_key, 8)
line_spacing = int(el.get('lineSpacing', 2))
y_expr = p['y_expr']
p['y2_expr'] = f"({y_expr}) + {font_size + line_spacing}"
if t == 'text':
t1 = el.get('text', '') or ''
t2 = el.get('text2', '') or ''
p['text2'] = t2
# Detect {variable} tokens — generate format_map() call instead of literal
_var_re = re.compile(r'\{([a-zA-Z_]\w*)\}')
p['text_is_template'] = bool(_var_re.search(t1) or _var_re.search(t2))
ref_len = max(len(t1), len(t2)) if t2 else len(t1)
p['x_expr'] = _aligned_x_expr(x_base_expr, text_align, ref_len, char_w)
p['x2_expr'] = p['x_expr'] # second line uses same x
else: # clock
fmt1 = el.get('format', '%H:%M') or '%H:%M'
fmt2 = el.get('format2', '') or ''
p['format2'] = fmt2
ref_len = max(len(fmt1), len(fmt2)) if fmt2 else len(fmt1)
p['x_expr'] = _aligned_x_expr(x_base_expr, text_align, ref_len, char_w)
p['x2_expr'] = p['x_expr']
p['blink'] = bool(el.get('blink', False))
elif t == 'dynamic_text':
binding = el.get('binding', {})
p['binding_source'] = binding.get('source', 'config')
p['binding_key'] = binding.get('key', '')
p['binding_format'] = binding.get('format')
font_key = el.get('font', 'press_start')
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 100)})"
x_base_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
p['x_expr'] = x_base_expr # dynamic text: runtime content determines width; use raw pos
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['blink'] = bool(el.get('blink', False))
elif t == 'rectangle':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
w = el.get('width', 10)
h = el.get('height', 8)
p['x_expr'] = x_expr
p['y_expr'] = y_expr
# x2/y2 as runtime expressions to support anchored positions
p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}"
fill = (
[el.get('fillR', 0), el.get('fillG', 0), el.get('fillB', 128)]
if el.get('hasFill', True) else None
)
outline = (
[el.get('outR', 255), el.get('outG', 255), el.get('outB', 255)]
if el.get('hasOutline', True) else None
)
p['fill_tuple'] = _as_fill_filter(fill)
p['outline_tuple'] = _as_fill_filter(outline)
p['blink'] = bool(el.get('blink', False))
elif t in ('line', 'divider'):
if t == 'divider':
orient = el.get('orientation', 'horizontal')
if orient == 'horizontal':
y_val = el.get('y', 16)
y_expr = _compute_pos_expr(y_val, y_anchor, 'height')
p.update(x0_expr='0', y0_expr=y_expr, x1_expr='width - 1', y1_expr=y_expr)
else:
x_val = el.get('x', 64)
x_expr = _compute_pos_expr(x_val, x_anchor, 'width')
p.update(x0_expr=x_expr, y0_expr='0', x1_expr=x_expr, y1_expr='height - 1')
else:
p['x0_expr'] = _compute_pos_expr(el.get('x0', 0), x_anchor, 'width')
p['y0_expr'] = _compute_pos_expr(el.get('y0', 0), y_anchor, 'height')
p['x1_expr'] = str(el.get('x1', 127))
p['y1_expr'] = str(el.get('y1', 0))
p['rgb_tuple'] = f"({el.get('r', 180)}, {el.get('g', 180)}, {el.get('b', 180)})"
p['line_width'] = el.get('lineWidth', 1)
p['blink'] = bool(el.get('blink', False))
elif t == 'progress_bar':
p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['bar_width'] = int(el.get('barWidth', 40))
p['bar_height'] = int(el.get('barHeight', 6))
binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '')
p['fill_tuple'] = f"({el.get('r', 100)}, {el.get('g', 200)}, {el.get('b', 100)})"
bg = (
[el.get('bgR', 30), el.get('bgG', 30), el.get('bgB', 30)]
if el.get('hasBg', True) else None
)
outline = (
[el.get('outR', 100), el.get('outG', 100), el.get('outB', 100)]
if el.get('hasOutline', True) else None
)
p['bg_tuple'] = _as_fill_filter(bg)
p['outline_tuple'] = _as_fill_filter(outline)
p['blink'] = bool(el.get('blink', False))
elif t == 'arc':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
w = el.get('width', 24)
h = el.get('height', 24)
p['x_expr'] = x_expr
p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}"
p['start_angle'] = int(el.get('startAngle', 0))
p['end_angle'] = int(el.get('endAngle', 270))
p['line_width'] = max(1, int(el.get('lineWidth', 2)))
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 0)})"
p['blink'] = bool(el.get('blink', False))
elif t == 'ellipse':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
w = el.get('width', 24)
h = el.get('height', 12)
p['x_expr'] = x_expr
p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}"
fill = (
[el.get('fillR', 0), el.get('fillG', 100), el.get('fillB', 200)]
if el.get('hasFill', True) else None
)
outline = (
[el.get('outR', 100), el.get('outG', 180), el.get('outB', 255)]
if el.get('hasOutline', True) else None
)
p['fill_tuple'] = _as_fill_filter(fill)
p['outline_tuple'] = _as_fill_filter(outline)
p['blink'] = bool(el.get('blink', False))
elif t == 'pixel':
p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
p['blink'] = bool(el.get('blink', False))
elif t == 'rounded_rectangle':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
w = el.get('width', 24)
h = el.get('height', 10)
p['x_expr'] = x_expr
p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}"
p['border_radius'] = int(el.get('borderRadius', 3))
fill = (
[el.get('fillR', 0), el.get('fillG', 80), el.get('fillB', 180)]
if el.get('hasFill', True) else None
)
outline = (
[el.get('outR', 120), el.get('outG', 180), el.get('outB', 255)]
if el.get('hasOutline', True) else None
)
p['fill_tuple'] = _as_fill_filter(fill)
p['outline_tuple'] = _as_fill_filter(outline)
p['blink'] = bool(el.get('blink', False))
elif t == 'countdown':
font_key = el.get('font', 'four_by_six')
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'extra_small_font')
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 180)}, {el.get('b', 0)})"
binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '')
p['countdown_format'] = el.get('countdownFormat', 'dh')
x_base_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
p['x_expr'] = x_base_expr
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['blink'] = bool(el.get('blink', False))
elif t == 'pips':
p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['pip_count'] = max(1, int(el.get('count', 5)))
p['pip_size'] = max(1, int(el.get('pipSize', 4)))
p['pip_spacing'] = max(0, int(el.get('pipSpacing', 2)))
p['show_empty'] = bool(el.get('showEmpty', True))
binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '')
p['fill_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 0)})"
p['empty_tuple'] = f"({el.get('emptyR', 50)}, {el.get('emptyG', 50)}, {el.get('emptyB', 50)})"
p['blink'] = bool(el.get('blink', False))
elif t == 'sparkline':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['x_expr'] = x_expr
p['y_expr'] = y_expr
p['bar_width_px'] = int(el.get('width', 40))
p['bar_height_px'] = int(el.get('height', 12))
p['bar_count'] = max(1, int(el.get('barCount', 8)))
p['bar_spacing'] = max(0, int(el.get('barSpacing', 1)))
binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '')
p['fill_tuple'] = f"({el.get('r', 80)}, {el.get('g', 200)}, {el.get('b', 120)})"
bg = [el.get('bgR', 30), el.get('bgG', 30), el.get('bgB', 30)] if el.get('hasBg', False) else None
p['bg_tuple'] = _as_fill_filter(bg)
p['blink'] = bool(el.get('blink', False))
elif t == 'gauge':
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
w = el.get('width', 32)
h = el.get('height', 32)
p['x_expr'] = x_expr
p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}"
p['start_angle'] = int(el.get('startAngle', 135))
p['end_angle'] = int(el.get('endAngle', 45))
p['line_width'] = max(1, int(el.get('lineWidth', 3)))
p['rgb_tuple'] = f"({el.get('r', 80)}, {el.get('g', 220)}, {el.get('b', 80)})"
track = (
[el.get('trackR', 40), el.get('trackG', 40), el.get('trackB', 40)]
if el.get('hasTrack', True) else None
)
p['track_tuple'] = _as_fill_filter(track)
binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '')
font_key = el.get('font', 'four_by_six')
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'extra_small_font')
p['show_label'] = bool(el.get('showLabel', True))
p['label_tuple'] = f"({el.get('labelR', 200)}, {el.get('labelG', 200)}, {el.get('labelB', 200)})"
p['blink'] = bool(el.get('blink', False))
elif t == 'marquee':
font_key = el.get('font', 'press_start')
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['text'] = el.get('text', 'Scrolling text')
p['char_w'] = _FONT_CHAR_W.get(font_key, 8)
p['gap'] = int(el.get('gap', 16))
p['scroll_speed'] = max(1, int(el.get('scrollSpeed', 1)))
p['direction'] = el.get('direction', 'left')
# Data key stored in self._data for stateful scrolling across display() calls
raw_id = str(el.get('id', 0)).replace('-', '_')
p['data_key'] = f"mq_{raw_id}"
p['blink'] = bool(el.get('blink', False))
result.append(p)
return result
def _generate_plugin_files(data: dict) -> dict:
"""
Generate all plugin file contents as strings.
Returns dict: {'manager.py', 'manifest.json', 'config_schema.json', 'requirements.txt'}
Raises ValueError with a human-readable message on any validation failure.
"""
metadata = data.get('metadata', {})
elements = data.get('elements', [])
data_model = data.get('dataModel', {})
config_vars = data_model.get('configVars', [])
plugin_id = metadata.get('id', '').strip()
if not _PLUGIN_ID_RE.match(plugin_id):
raise ValueError(
'Plugin ID must start with a lowercase letter and contain only '
'lowercase letters, numbers, and hyphens (max 63 chars).'
)
plugin_name = metadata.get('name', '').strip()
if not plugin_name:
raise ValueError('Plugin name is required.')
author = metadata.get('author', '').strip()
if not author:
raise ValueError('Author is required.')
version = metadata.get('version', '1.0.0').strip()
# Validate config var keys are valid Python identifiers
for cv in config_vars:
key = cv.get('key', '')
if not _PYTHON_IDENT_RE.match(key):
raise ValueError(f'Config variable key "{key}" is not a valid Python identifier.')
class_name = _to_class_name(plugin_name)
# Only consider visible elements for code generation flags
visible_elements = [e for e in elements if e.get('visible') is not False]
processed = _preprocess_elements(elements)
has_clock = any(e.get('type') == 'clock' for e in visible_elements)
has_blink = any(e.get('blink') for e in visible_elements)
has_countdown = any(e.get('type') == 'countdown' for e in visible_elements)
_var_re = re.compile(r'\{[a-zA-Z_]\w*\}')
has_text_template = any(
e.get('type') == 'text' and (
_var_re.search(e.get('text', '') or '') or
_var_re.search(e.get('text2', '') or '')
)
for e in visible_elements
)
# Background fill color (None → don't render, use LED panel's native black)
bg_color: str | None = None
bg_raw = metadata.get('bgColor')
if isinstance(bg_raw, dict):
r, g, b = int(bg_raw.get('r', 0)), int(bg_raw.get('g', 0)), int(bg_raw.get('b', 0))
if r or g or b:
bg_color = f'({r}, {g}, {b})'
# Render manager.py
env = _get_jinja_env()
try:
tmpl = env.get_template('manager.py.j2')
except jinja2.TemplateNotFound:
raise ValueError('Code generation template not found. This is a server configuration issue.')
manager_py = tmpl.render(
plugin_name=plugin_name,
class_name=class_name,
plugin_id=plugin_id,
generated_date=datetime.now().strftime('%Y-%m-%d'),
config_vars=config_vars,
elements=processed,
has_clock=has_clock,
has_blink=has_blink,
has_countdown=has_countdown,
has_text_template=has_text_template,
bg_color=bg_color,
)
# Syntax-check the generated Python
try:
ast.parse(manager_py)
except SyntaxError as exc:
raise ValueError(f'Generated code has a syntax error: {exc}') from exc
# Build manifest
manifest = {
'id': plugin_id,
'name': plugin_name,
'version': version,
'author': author,
'description': metadata.get('description', 'Custom plugin created with LEDMatrix Plugin Composer'),
'category': metadata.get('category', 'custom'),
'tags': ['composer', 'custom'],
'entry_point': 'manager.py',
'class_name': class_name,
'display_modes': [plugin_id],
'compatible_versions': ['>=2.0.0'],
'last_updated': datetime.now().strftime('%Y-%m-%d'),
'update_interval': int(metadata.get('update_interval', 60)),
'default_duration': float(metadata.get('display_duration', 15)),
'versions': [
{'released': datetime.now().strftime('%Y-%m-%d'), 'version': version}
],
}
# Validate manifest against the project's schema
if composer_bp.project_root:
schema_path = Path(composer_bp.project_root) / 'schema' / 'manifest_schema.json'
if schema_path.exists():
schema = json.loads(schema_path.read_text())
validator = jsonschema.Draft7Validator(schema)
errors = list(validator.iter_errors(manifest))
if errors:
msgs = '; '.join(e.message for e in errors[:3])
raise ValueError(f'Manifest validation failed: {msgs}')
# Build config_schema
type_map = {
'string': {'type': 'string'},
'number': {'type': 'number', 'minimum': 0},
'boolean': {'type': 'boolean'},
'color': {
'type': 'array',
'items': {'type': 'integer', 'minimum': 0, 'maximum': 255},
'minItems': 3,
'maxItems': 3,
},
}
config_properties = {
'enabled': {'type': 'boolean', 'default': True},
'display_duration': {'type': 'number', 'minimum': 1, 'default': float(metadata.get('display_duration', 15))},
}
for cv in config_vars:
cv_type = cv.get('type', 'string')
prop = dict(type_map.get(cv_type, {'type': 'string'}))
if cv.get('description'):
prop['description'] = cv['description']
if cv.get('label'):
prop['title'] = cv['label']
default = cv.get('default', '')
if cv_type == 'number':
try:
prop['default'] = float(default) if default != '' else 0
except (TypeError, ValueError):
prop['default'] = 0
elif cv_type == 'boolean':
prop['default'] = bool(default)
else:
prop['default'] = default
config_properties[cv['key']] = prop
config_schema = {
'$schema': 'http://json-schema.org/draft-07/schema#',
'type': 'object',
'properties': config_properties,
}
return {
'manager.py': manager_py,
'manifest.json': json.dumps(manifest, indent=2),
'config_schema.json': json.dumps(config_schema, indent=2),
'requirements.txt': '',
}
def _save_composer_state(target_dir: Path, payload: dict) -> None:
"""Persist the raw composer payload alongside the generated plugin files."""
(target_dir / '_composer_state.json').write_text(
json.dumps(payload, indent=2, ensure_ascii=False), encoding='utf-8'
)
def _pack_zip(files: dict, plugin_id: str) -> io.BytesIO:
"""Pack generated plugin files into an in-memory ZIP."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', compression=zipfile.ZIP_DEFLATED) as zf:
for filename, content in files.items():
info = zipfile.ZipInfo(f'{plugin_id}/{filename}')
info.compress_type = zipfile.ZIP_DEFLATED
zf.writestr(info, content.encode('utf-8') if isinstance(content, str) else content)
buf.seek(0)
return buf
# ── Routes ────────────────────────────────────────────────────────────────────
@composer_bp.route('/')
def index():
return render_template('v3/composer.html')
@composer_bp.route('/api/generate', methods=['POST'])
def generate_zip():
data = request.get_json(force=True, silent=True)
if not data:
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
try:
files = _generate_plugin_files(data)
except ValueError as exc:
return jsonify({'status': 'error', 'message': str(exc)}), 422
plugin_id = data.get('metadata', {}).get('id', 'plugin')
files['_composer_state.json'] = json.dumps(data, indent=2, ensure_ascii=False)
zip_buf = _pack_zip(files, plugin_id)
return send_file(
zip_buf,
mimetype='application/zip',
as_attachment=True,
download_name=f'{plugin_id}.zip',
)
@composer_bp.route('/api/install', methods=['POST'])
def install_locally():
if not composer_bp.plugins_dir:
return jsonify({'status': 'error', 'message': 'Plugin directory not configured'}), 503
data = request.get_json(force=True, silent=True)
if not data:
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
try:
files = _generate_plugin_files(data)
except ValueError as exc:
return jsonify({'status': 'error', 'message': str(exc)}), 422
plugin_id = data.get('metadata', {}).get('id', '')
# _generate_plugin_files() above already validates metadata.id via this
# same regex before it will return, but that guarantee lives in a
# different function -- re-check here, at the point the path is actually
# built, so this route stays safe on its own if that call is ever
# reordered or changed.
if not _PLUGIN_ID_RE.match(plugin_id):
return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400
target = Path(composer_bp.plugins_dir) / plugin_id
force = bool(data.get('_force', False))
if target.exists() and not force:
return jsonify({
'status': 'conflict',
'message': f'Plugin "{plugin_id}" is already installed.',
}), 409
try:
if target.exists() and force:
import shutil as _shutil
_shutil.rmtree(target)
target.mkdir(parents=True, exist_ok=False)
for filename, content in files.items():
(target / filename).write_text(content, encoding='utf-8')
_save_composer_state(target, data)
except OSError as exc:
logger.error('Failed to write plugin files for %s: %s', plugin_id, exc)
return jsonify({'status': 'error', 'message': 'Failed to write plugin files'}), 500
# Trigger plugin discovery so it shows up in the Plugin Manager immediately
if composer_bp.plugin_manager:
try:
composer_bp.plugin_manager.discover_plugins()
except Exception as exc:
logger.warning('discover_plugins() failed after composer install: %s', exc)
return jsonify({
'status': 'success',
'message': f'Plugin "{plugin_id}" installed successfully.',
'plugin_id': plugin_id,
})
@composer_bp.route('/api/fonts/<font_name>')
def serve_font(font_name):
"""Serve an allowlisted font file for canvas FontFace loading."""
if font_name not in _ALLOWED_FONTS:
return '', 404
if not composer_bp.project_root:
return '', 503
font_path = Path(composer_bp.project_root) / 'assets' / 'fonts' / font_name
if not font_path.exists():
return '', 404
return send_file(str(font_path), mimetype='font/ttf')
@composer_bp.route('/api/validate-id/<plugin_id>')
def validate_id(plugin_id):
"""Check whether a plugin ID is valid and available."""
if not _PLUGIN_ID_RE.match(plugin_id):
return jsonify({'valid': False, 'available': False, 'reason': 'Invalid format'})
if composer_bp.plugins_dir:
taken = (Path(composer_bp.plugins_dir) / plugin_id).exists()
if taken:
return jsonify({'valid': True, 'available': False, 'reason': 'Already installed'})
return jsonify({'valid': True, 'available': True})
@composer_bp.route('/api/plugins')
def list_plugins():
"""List installed plugins, flagging which ones have a saved composer state."""
if not composer_bp.plugins_dir:
return jsonify([])
plugins_dir = Path(composer_bp.plugins_dir)
results = []
for entry in sorted(plugins_dir.iterdir()):
if not entry.is_dir():
continue
manifest_path = entry / 'manifest.json'
if not manifest_path.exists():
continue
try:
manifest = json.loads(manifest_path.read_text())
except Exception as e:
logger.warning("Skipping %s: unreadable manifest.json (%s)", entry.name, e)
continue
has_state = (entry / '_composer_state.json').exists()
results.append({
'id': manifest.get('id', entry.name),
'name': manifest.get('name', entry.name),
'version': manifest.get('version', ''),
'author': manifest.get('author', ''),
'has_composer_state': has_state,
})
return jsonify(results)
@composer_bp.route('/api/preview', methods=['POST'])
def preview_code():
"""Generate plugin files and return them as JSON for the code preview modal."""
data = request.get_json(force=True, silent=True)
if not data:
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
try:
files = _generate_plugin_files(data)
except ValueError as exc:
return jsonify({'status': 'error', 'message': str(exc)}), 422
return jsonify({
'status': 'ok',
'files': {
'manager.py': files['manager.py'],
'manifest.json': files['manifest.json'],
'config_schema.json': files['config_schema.json'],
},
})
@composer_bp.route('/api/load/<plugin_id>')
def load_plugin(plugin_id):
"""Load a plugin's composer state for editing.
If a _composer_state.json exists, return it verbatim.
Otherwise, extract config vars from config_schema.json for a partial import.
"""
if not composer_bp.plugins_dir:
return jsonify({'status': 'error', 'message': 'Plugin directory not configured'}), 503
if not _PLUGIN_ID_RE.match(plugin_id):
return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400
plugin_dir = Path(composer_bp.plugins_dir) / plugin_id
if not plugin_dir.exists():
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
# Full composer state
state_path = plugin_dir / '_composer_state.json'
if state_path.exists():
try:
state = json.loads(state_path.read_text())
return jsonify({'status': 'ok', 'source': 'composer', 'state': state})
except Exception as exc:
logger.error('Failed to read composer state for %s: %s', plugin_id, exc)
return jsonify({'status': 'error', 'message': 'Failed to read state'}), 500
# Partial import from config_schema.json
schema_path = plugin_dir / 'config_schema.json'
manifest_path = plugin_dir / 'manifest.json'
config_vars = []
if schema_path.exists():
try:
schema = json.loads(schema_path.read_text())
props = schema.get('properties', {})
skip = {'enabled', 'display_duration', 'update_interval'}
type_map = {'boolean': 'boolean', 'number': 'number', 'integer': 'number', 'string': 'string'}
for key, prop in props.items():
if key in skip:
continue
prop_type = prop.get('type', 'string')
if isinstance(prop_type, list):
prop_type = next((t for t in prop_type if t != 'null'), 'string')
# Detect color arrays
if prop_type == 'array' and prop.get('maxItems') == 3:
cv_type = 'color'
else:
cv_type = type_map.get(prop_type, 'string')
config_vars.append({
'key': key,
'label': prop.get('title', key.replace('_', ' ').title()),
'type': cv_type,
'default': prop.get('default', ''),
'description': prop.get('description', ''),
})
except Exception as e:
logger.warning("Failed to parse config_schema.json for %s: %s", plugin_id, e)
manifest = {}
if manifest_path.exists():
try:
manifest = json.loads(manifest_path.read_text())
except Exception:
pass
partial_state = {
'composer_version': '1.0',
'metadata': {
'id': manifest.get('id', plugin_id),
'name': manifest.get('name', plugin_id),
'author': manifest.get('author', ''),
'version': manifest.get('version', '1.0.0'),
'description': manifest.get('description', ''),
'category': manifest.get('category', 'custom'),
'display_duration': manifest.get('default_duration', 15),
'update_interval': manifest.get('update_interval', 60),
'api_requirements': manifest.get('api_requirements', []),
},
'elements': [],
'dataModel': {'configVars': config_vars, 'dataSources': [], 'computedVars': []},
}
return jsonify({'status': 'ok', 'source': 'schema_import', 'state': partial_state})
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,757 @@
/**
* ComposerCanvas stateless LED matrix canvas renderer.
*
* Coordinate system: LED pixels (integers). All drawing multiplies by SCALE.
* PIL draw.text(x,y) is top-left; canvas fillText(x,y) is baseline.
* Canvas text cy = (actualY + fontSizePx) * SCALE
*
* Anchors: element x/y are offsets from their anchor point:
* xAnchor=null/'left' x is fixed offset from left
* xAnchor='center' x is offset from width/2
* xAnchor='right' x is offset inward from right edge
* yAnchor follows the same pattern with 'top'/'middle'/'bottom'
*
* Breakpoints: elements with minWidth > currentMatrixW are rendered at 25% opacity.
*
* Resize handles: drawn on selected rectangles; 8 handles (corners + edge mids).
*/
window.ComposerCanvas = (() => {
'use strict';
let _canvas = null;
let _ctx = null;
let _showGrid = true;
const DISPLAY_PRESETS = [
{ label: '64×32', w: 64, h: 32 },
{ label: '128×32', w: 128, h: 32 },
{ label: '128×64', w: 128, h: 64 },
{ label: '256×32', w: 256, h: 32 },
{ label: '256×64', w: 256, h: 64 },
];
const FONT_MAP = {
press_start: { family: "'PressStart2P', monospace", sizePx: 8, charW: 8 },
four_by_six: { family: 'monospace', sizePx: 6, charW: 4 },
five_by_seven: { family: 'monospace', sizePx: 7, charW: 5 },
};
const ELEMENT_DEFAULTS = {
text: {
text: 'Hello', font: 'press_start',
r: 255, g: 255, b: 255,
text2: '', lineSpacing: 2, textAlign: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
dynamic_text: {
binding: { source: 'config', key: '', format: null },
font: 'press_start', textAlign: 'left',
r: 255, g: 200, b: 100,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
clock: {
format: '%H:%M', font: 'press_start',
r: 100, g: 255, b: 100,
format2: '', lineSpacing: 2, textAlign: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
rectangle: {
width: 20, height: 8,
fillR: 0, fillG: 0, fillB: 128, hasFill: true,
outR: 255, outG: 255, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
ellipse: {
width: 24, height: 12,
fillR: 0, fillG: 100, fillB: 200, hasFill: true,
outR: 100, outG: 180, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
arc: {
width: 24, height: 24,
startAngle: 0, endAngle: 270, lineWidth: 2,
r: 255, g: 200, b: 0,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
pixel: {
r: 255, g: 255, b: 255,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
rounded_rectangle: {
width: 24, height: 10, borderRadius: 3,
fillR: 0, fillG: 80, fillB: 180, hasFill: true,
outR: 120, outG: 180, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
line: {
x0: 0, y0: 16, x1: 63, y1: 16,
r: 180, g: 180, b: 180, lineWidth: 1,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
divider: {
orientation: 'horizontal', y: 16, x: 64,
r: 100, g: 100, b: 100,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
progress_bar: {
barWidth: 60, barHeight: 6,
binding: { source: 'config', key: '', format: null },
r: 80, g: 200, b: 80,
bgR: 30, bgG: 30, bgB: 30, hasBg: true,
outR: 100, outG: 100, outB: 100, hasOutline: true,
previewPct: 65,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
countdown: {
binding: { source: 'config', key: '', format: null },
countdownFormat: 'dh',
font: 'four_by_six', textAlign: 'left',
r: 255, g: 180, b: 0,
previewText: '42d 3h',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
marquee: {
text: 'Scrolling text', font: 'press_start',
r: 255, g: 255, b: 255,
scrollSpeed: 1, gap: 16, direction: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
section: {
label: 'Section',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
pips: {
count: 5, filled: 3, pipSize: 4, pipSpacing: 2,
r: 255, g: 200, b: 0,
emptyR: 50, emptyG: 50, emptyB: 50, showEmpty: true,
binding: { source: 'config', key: '', format: null },
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
sparkline: {
width: 40, height: 12,
barCount: 8, barSpacing: 1,
r: 80, g: 200, b: 120,
bgR: 30, bgG: 30, bgB: 30, hasBg: false,
binding: { source: 'config', key: '', format: null },
previewData: '0.3,0.6,0.4,0.8,0.5,0.9,0.7,0.85',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
gauge: {
width: 32, height: 32,
startAngle: 135, endAngle: 45, lineWidth: 3,
binding: { source: 'config', key: '', format: null },
r: 80, g: 220, b: 80,
trackR: 40, trackG: 40, trackB: 40, hasTrack: true,
showLabel: true, font: 'four_by_six', labelR: 200, labelG: 200, labelB: 200,
previewPct: 65,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
};
// ── Anchor resolution ────────────────────────────────────────────────
function resolveAnchor(val, anchor, dim) {
if (!anchor || anchor === 'left' || anchor === 'top') return val;
if (anchor === 'center' || anchor === 'middle') return Math.floor(dim / 2) + val;
if (anchor === 'right' || anchor === 'bottom') return dim - val;
return val;
}
function computeActualPos(el, matrixW, matrixH) {
const ax = resolveAnchor(el.x ?? el.x0 ?? 0, el.xAnchor, matrixW);
const ay = resolveAnchor(el.y ?? el.y0 ?? 0, el.yAnchor, matrixH);
return { x: ax, y: ay };
}
// ── Bounding box (LED pixel space) ──────────────────────────────────
function getBoundingBox(el, matrixW, matrixH) {
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
switch (el.type) {
case 'text': {
const t1 = el.text || '', t2 = el.text2 || '';
const w = Math.max(t1.length, t2.length) * finfo.charW;
const h = t2 ? finfo.sizePx * 2 + (el.lineSpacing ?? 2) : finfo.sizePx;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h };
}
case 'dynamic_text': {
const key = el.binding?.key || '?';
const w = (`{${key}}`).length * finfo.charW;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h: finfo.sizePx };
}
case 'clock': {
const t1 = el.format || '%H:%M', t2 = el.format2 || '';
const w = Math.max(t1.length, t2.length) * finfo.charW;
const h = t2 ? finfo.sizePx * 2 + (el.lineSpacing ?? 2) : finfo.sizePx;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h };
}
case 'countdown': {
const pt = el.previewText || '--d --h';
const w = pt.length * finfo.charW;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h: finfo.sizePx };
}
case 'rectangle':
case 'rounded_rectangle':
case 'ellipse':
case 'arc':
return { x: ax, y: ay, w: el.width, h: el.height };
case 'pixel':
return { x: ax, y: ay, w: 1, h: 1 };
case 'line':
return {
x: Math.min(el.x0, el.x1), y: Math.min(el.y0, el.y1),
w: Math.max(1, Math.abs(el.x1 - el.x0)),
h: Math.max(1, Math.abs(el.y1 - el.y0)),
};
case 'divider':
return el.orientation === 'horizontal'
? { x: 0, y: ay, w: matrixW, h: 1 }
: { x: ax, y: 0, w: 1, h: matrixH };
case 'progress_bar':
return { x: ax, y: ay, w: el.barWidth ?? 60, h: el.barHeight ?? 6 };
case 'marquee': {
const mfinfo = FONT_MAP[el.font] || FONT_MAP.press_start;
return { x: 0, y: ay, w: matrixW, h: mfinfo.sizePx };
}
case 'gauge':
return { x: ax, y: ay, w: el.width ?? 32, h: el.height ?? 32 };
case 'sparkline':
return { x: ax, y: ay, w: el.width ?? 40, h: el.height ?? 12 };
case 'pips': {
const pc = el.count ?? 5, ps = el.pipSize ?? 4, pg = el.pipSpacing ?? 2;
return { x: ax, y: ay, w: pc * ps + (pc - 1) * pg, h: ps };
}
case 'section':
return { x: ax, y: ay, w: 0, h: 0 };
default:
return { x: ax, y: ay, w: 4, h: 4 };
}
}
// ── Resize handle support ─────────────────────────────────────────────
// Returns 8 handle points for a rectangle in LED pixel space
function _getRectHandles(el, matrixW, matrixH) {
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const w = el.width, h = el.height;
const cx = ax + w / 2, cy = ay + h / 2;
return {
nw: { x: ax, y: ay },
n: { x: cx, y: ay },
ne: { x: ax + w, y: ay },
w: { x: ax, y: cy },
e: { x: ax + w, y: cy },
sw: { x: ax, y: ay + h },
s: { x: cx, y: ay + h },
se: { x: ax + w, y: ay + h },
};
}
// Returns the handle direction under LED-space point (lx, ly), or null
function getResizeHandle(el, lx, ly, matrixW, matrixH) {
if (!['rectangle', 'rounded_rectangle', 'ellipse', 'arc', 'gauge', 'sparkline'].includes(el.type)) return null;
const handles = _getRectHandles(el, matrixW, matrixH);
const PAD = 4;
for (const [dir, pt] of Object.entries(handles)) {
if (Math.abs(lx - pt.x) <= PAD && Math.abs(ly - pt.y) <= PAD) return dir;
}
return null;
}
const _HANDLE_CURSORS = {
nw: 'nw-resize', n: 'n-resize', ne: 'ne-resize',
w: 'w-resize', e: 'e-resize',
sw: 'sw-resize', s: 's-resize', se: 'se-resize',
};
function getCursorForHandle(handle) {
return _HANDLE_CURSORS[handle] || 'crosshair';
}
// ── Hit test ─────────────────────────────────────────────────────────
function hitTest(el, lx, ly, matrixW, matrixH) {
const PAD = 3;
const bb = getBoundingBox(el, matrixW, matrixH);
return (
lx >= bb.x - PAD && lx <= bb.x + bb.w + PAD &&
ly >= bb.y - PAD && ly <= bb.y + bb.h + PAD
);
}
// ── Draw a single element ─────────────────────────────────────────────
function _drawElement(ctx, el, SCALE, matrixW, matrixH, opts = {}) {
const s = SCALE;
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const belowBreakpoint = el.minWidth > 0 && matrixW < el.minWidth;
const hidden = el.visible === false;
ctx.save();
if (hidden) ctx.globalAlpha = 0.12;
else if (belowBreakpoint) ctx.globalAlpha = 0.25;
// Blink animation: when blinkOff, fully hide blinking elements
if (el.blink) {
if (opts.blinkOff) { ctx.restore(); return; }
ctx.globalAlpha *= 0.55;
}
// Helper: compute draw X for text alignment
const _textX = (text, finfo) => {
const tw = text.length * finfo.charW * s;
if (el.textAlign === 'center') return ax * s - tw / 2;
if (el.textAlign === 'right') return ax * s - tw;
return ax * s;
};
try {
switch (el.type) {
case 'text':
case 'dynamic_text':
case 'clock': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const key = el.binding?.key || '?';
const pv = opts.previewValues?.[key];
// Substitute {variable} tokens in text using previewValues
const _subVars = str => (str || '').replace(/\{(\w+)\}/g, (_, k) => {
const v = opts.previewValues?.[k];
return v !== undefined && v !== '' ? String(v) : `{${k}}`;
});
const displayText =
el.type === 'text' ? _subVars(el.text || '')
: el.type === 'clock' ? (el.format || '%H:%M')
: (pv !== undefined && pv !== '' ? String(pv) : `{${key}}`);
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillText(displayText, _textX(displayText, finfo), (ay + finfo.sizePx) * s);
// Second line (text and clock)
if (el.type === 'text' && el.text2) {
const t2 = _subVars(el.text2);
const y2 = ay + finfo.sizePx + (el.lineSpacing ?? 2);
ctx.fillText(t2, _textX(t2, finfo), (y2 + finfo.sizePx) * s);
}
if (el.type === 'clock' && el.format2) {
const y2 = ay + finfo.sizePx + (el.lineSpacing ?? 2);
ctx.fillText(el.format2, _textX(el.format2, finfo), (y2 + finfo.sizePx) * s);
}
break;
}
case 'countdown': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const t = el.previewText || '--d --h';
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillText(t, _textX(t, finfo), (ay + finfo.sizePx) * s);
break;
}
case 'rectangle': {
const rx = ax * s, ry = ay * s;
const rw = el.width * s, rh = el.height * s;
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fillRect(rx, ry, rw, rh);
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
ctx.lineWidth = 1;
ctx.strokeRect(rx, ry, rw, rh);
}
break;
}
case 'ellipse': {
const cx = (ax + el.width / 2) * s;
const cy = (ay + el.height / 2) * s;
const rx = (el.width / 2) * s;
const ry = (el.height / 2) * s;
ctx.beginPath();
ctx.ellipse(cx, cy, rx, ry, 0, 0, Math.PI * 2);
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fill();
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
ctx.lineWidth = 1;
ctx.stroke();
}
break;
}
case 'arc': {
const cx = (ax + el.width / 2) * s;
const cy = (ay + el.height / 2) * s;
const rx = (el.width / 2) * s;
const ry = (el.height / 2) * s;
// PIL: 0°=right, clockwise. Canvas: same with anticlockwise=false
const startRad = (el.startAngle ?? 0) * Math.PI / 180;
const endRad = (el.endAngle ?? 270) * Math.PI / 180;
ctx.beginPath();
ctx.ellipse(cx, cy, rx, ry, 0, startRad, endRad, false);
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = Math.max(1, el.lineWidth || 2);
ctx.stroke();
break;
}
case 'pixel': {
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillRect(ax * s, ay * s, s, s);
break;
}
case 'rounded_rectangle': {
const rx = ax * s, ry = ay * s;
const rw = el.width * s, rh = el.height * s;
const rad = Math.min((el.borderRadius ?? 3) * s, rw / 2, rh / 2);
ctx.beginPath();
ctx.roundRect(rx, ry, rw, rh, rad);
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fill();
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
ctx.lineWidth = 1;
ctx.stroke();
}
break;
}
case 'line': {
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = Math.max(1, el.lineWidth || 1);
ctx.beginPath();
ctx.moveTo(el.x0 * s, el.y0 * s);
ctx.lineTo(el.x1 * s, el.y1 * s);
ctx.stroke();
break;
}
case 'divider': {
const isH = (el.orientation || 'horizontal') === 'horizontal';
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = 1;
ctx.beginPath();
if (isH) {
ctx.moveTo(0, ay * s + 0.5);
ctx.lineTo(_canvas.width, ay * s + 0.5);
} else {
ctx.moveTo(ax * s + 0.5, 0);
ctx.lineTo(ax * s + 0.5, _canvas.height);
}
ctx.stroke();
break;
}
case 'pips': {
const pipCount = Math.max(1, el.count ?? 5);
const pvPips = opts.previewValues?.[el.binding?.key];
const filledN = pvPips !== undefined
? Math.max(0, Math.min(pipCount, Math.round(parseFloat(pvPips) || 0)))
: Math.max(0, Math.min(pipCount, el.filled ?? 3));
const ps = Math.max(1, el.pipSize ?? 4);
const pg = Math.max(0, el.pipSpacing ?? 2);
for (let i = 0; i < pipCount; i++) {
const isFilled = i < filledN;
if (!isFilled && !el.showEmpty) continue;
ctx.fillStyle = isFilled
? `rgb(${el.r},${el.g},${el.b})`
: `rgb(${el.emptyR ?? 50},${el.emptyG ?? 50},${el.emptyB ?? 50})`;
ctx.fillRect((ax + i * (ps + pg)) * s, ay * s, ps * s, ps * s);
}
break;
}
case 'sparkline': {
const slW = el.width ?? 40, slH = el.height ?? 12;
const count = Math.max(1, el.barCount ?? 8);
const spacing = el.barSpacing ?? 1;
const barW = Math.max(1, Math.floor((slW - spacing * (count - 1)) / count));
const rawVals = (el.previewData || '').split(',')
.map(v => parseFloat(v.trim())).filter(n => !isNaN(n));
while (rawVals.length < count) rawVals.push(0);
const maxV = Math.max(...rawVals.slice(0, count), 0.001);
const rx = ax * s, ry = ay * s;
if (el.hasBg) {
ctx.fillStyle = `rgb(${el.bgR ?? 30},${el.bgG ?? 30},${el.bgB ?? 30})`;
ctx.fillRect(rx, ry, slW * s, slH * s);
}
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
for (let i = 0; i < count; i++) {
const norm = Math.max(0, Math.min(1, rawVals[i] / maxV));
const barH = Math.max(1, Math.round(slH * norm));
const bx = rx + (barW + spacing) * i * s;
const by = ry + (slH - barH) * s;
ctx.fillRect(bx, by, barW * s, barH * s);
}
break;
}
case 'gauge': {
const gw = (el.width ?? 32), gh = (el.height ?? 32);
const cx = (ax + gw / 2) * s, cy = (ay + gh / 2) * s;
const rx = (gw / 2) * s, ry = (gh / 2) * s;
const lw = Math.max(1, (el.lineWidth ?? 3));
const startDeg = el.startAngle ?? 135;
const endDeg = el.endAngle ?? 45;
// Arc sweep: from startDeg clockwise to endDeg (PIL convention)
const totalSweep = ((endDeg - startDeg) + 360) % 360 || 360;
const pvGauge = opts.previewValues?.[el.binding?.key];
const pct = pvGauge !== undefined
? Math.max(0, Math.min(100, parseFloat(pvGauge) || 0)) / 100
: Math.max(0, Math.min(100, el.previewPct ?? 65)) / 100;
const fillSweep = totalSweep * pct;
const toRad = deg => (deg - 90) * Math.PI / 180; // canvas 0=top, PIL 0=right → offset -90
// Track arc
if (el.hasTrack !== false) {
ctx.beginPath();
ctx.ellipse(cx, cy, rx - lw / 2, ry - lw / 2, 0, toRad(startDeg), toRad(startDeg + totalSweep), false);
ctx.strokeStyle = `rgb(${el.trackR ?? 40},${el.trackG ?? 40},${el.trackB ?? 40})`;
ctx.lineWidth = lw * s;
ctx.stroke();
}
// Fill arc
if (pct > 0) {
ctx.beginPath();
ctx.ellipse(cx, cy, rx - lw / 2, ry - lw / 2, 0, toRad(startDeg), toRad(startDeg + fillSweep), false);
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = lw * s;
ctx.stroke();
}
// Centre label
if (el.showLabel) {
const gfinfo = FONT_MAP[el.font || 'four_by_six'] || FONT_MAP.four_by_six;
const labelText = Math.round(pct * 100) + '%';
ctx.font = `${gfinfo.sizePx * s}px ${gfinfo.family}`;
ctx.fillStyle = `rgb(${el.labelR ?? 200},${el.labelG ?? 200},${el.labelB ?? 200})`;
const ltw = ctx.measureText(labelText).width;
ctx.fillText(labelText, cx - ltw / 2, cy + (gfinfo.sizePx * s) / 2);
}
break;
}
case 'marquee': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const text = el.text || 'Scrolling text';
const tw = text.length * finfo.charW * s;
const gap = (el.gap ?? 16) * s;
const totalW = tw + gap;
const tick = opts.animTick ?? 0;
const speed = (el.scrollSpeed ?? 1) * 2;
const scrolled = (tick * speed) % totalW;
// left: text enters from right; right: text enters from left
const startX = el.direction === 'right'
? scrolled - tw
: matrixW * s - scrolled;
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
// Clip to canvas width so text doesn't bleed outside
ctx.save();
ctx.beginPath();
ctx.rect(0, ay * s - 1, matrixW * s, (finfo.sizePx + 2) * s);
ctx.clip();
for (let i = -1; i <= 2; i++) {
ctx.fillText(text, startX + i * totalW, (ay + finfo.sizePx) * s);
}
ctx.restore();
break;
}
case 'progress_bar': {
const bw = el.barWidth ?? 60, bh = el.barHeight ?? 6;
const pvPb = opts.previewValues?.[el.binding?.key];
const pct = pvPb !== undefined
? Math.max(0, Math.min(100, parseFloat(pvPb) || 0)) / 100
: Math.max(0, Math.min(100, el.previewPct ?? 65)) / 100;
const rx = ax * s, ry = ay * s;
if (el.hasBg) {
ctx.fillStyle = `rgb(${el.bgR ?? 30},${el.bgG ?? 30},${el.bgB ?? 30})`;
ctx.fillRect(rx, ry, bw * s, bh * s);
}
const fillW = Math.max(0, Math.round(bw * pct));
if (fillW > 0) {
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillRect(rx, ry, fillW * s, bh * s);
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR ?? 100},${el.outG ?? 100},${el.outB ?? 100})`;
ctx.lineWidth = 1;
ctx.strokeRect(rx, ry, bw * s, bh * s);
}
break;
}
}
if (belowBreakpoint) {
ctx.globalAlpha = 0.6;
const bb = getBoundingBox(el, matrixW, matrixH);
ctx.font = `${Math.max(8, s * 2)}px monospace`;
ctx.fillStyle = '#facc15';
ctx.fillText(`${el.minWidth}px`, bb.x * s, (bb.y + 4) * s);
}
} finally {
ctx.restore();
}
}
// ── Selection indicator ──────────────────────────────────────────────
function _drawSelection(ctx, el, SCALE, matrixW, matrixH) {
const bb = getBoundingBox(el, matrixW, matrixH);
const PAD = 2, s = SCALE;
const rx = bb.x * s - PAD, ry = bb.y * s - PAD;
const rw = bb.w * s + PAD * 2, rh = bb.h * s + PAD * 2;
ctx.save();
ctx.strokeStyle = '#3b82f6';
ctx.lineWidth = 1;
ctx.setLineDash([3, 2]);
ctx.strokeRect(rx, ry, rw, rh);
ctx.setLineDash([]);
if (el.xAnchor || el.yAnchor) {
ctx.font = `${Math.max(7, s)}px sans-serif`;
ctx.fillStyle = '#a78bfa';
const anchorText = [
el.xAnchor ? `x:${el.xAnchor[0]}` : '',
el.yAnchor ? `y:${el.yAnchor[0]}` : '',
].filter(Boolean).join(' ');
if (anchorText) ctx.fillText(anchorText, rx + 1, ry - 2);
}
// Resize handles: on rect, rounded rect, ellipse
if (['rectangle', 'rounded_rectangle', 'ellipse', 'arc', 'gauge', 'sparkline'].includes(el.type)) {
const handles = _getRectHandles(el, matrixW, matrixH);
const HS = 5;
ctx.fillStyle = 'white';
ctx.strokeStyle = '#2563eb';
ctx.lineWidth = 1;
for (const pt of Object.values(handles)) {
const hx = pt.x * s - HS / 2;
const hy = pt.y * s - HS / 2;
ctx.fillRect(hx, hy, HS, HS);
ctx.strokeRect(hx, hy, HS, HS);
}
} else {
// Corner dots for non-rectangle elements
ctx.fillStyle = '#3b82f6';
const HS = 4;
for (const [hx, hy] of [
[rx - HS / 2, ry - HS / 2], [rx + rw - HS / 2, ry - HS / 2],
[rx - HS / 2, ry + rh - HS / 2], [rx + rw - HS / 2, ry + rh - HS / 2],
]) ctx.fillRect(hx, hy, HS, HS);
}
ctx.restore();
}
// ── Dimension tooltip while dragging ─────────────────────────────────
function drawDragTooltip(ctx, el, SCALE, matrixW, matrixH) {
const bb = getBoundingBox(el, matrixW, matrixH);
const label = el.type === 'rectangle'
? `${el.width}×${el.height}`
: `${bb.x},${bb.y}`;
const s = SCALE;
ctx.save();
ctx.font = `${Math.max(9, s * 1.5)}px monospace`;
const tw = ctx.measureText(label).width;
const tx = bb.x * s, ty = (bb.y - 2) * s;
ctx.fillStyle = 'rgba(0,0,0,0.7)';
ctx.fillRect(tx - 2, ty - 10, tw + 4, 12);
ctx.fillStyle = 'white';
ctx.fillText(label, tx, ty);
ctx.restore();
}
// ── Public API ───────────────────────────────────────────────────────
function init(canvasEl) {
_canvas = canvasEl;
_ctx = canvasEl.getContext('2d');
}
function setGrid(show) { _showGrid = show; }
function updateCanvasSize(matrixW, matrixH, SCALE) {
if (!_canvas) return;
_canvas.width = matrixW * SCALE;
_canvas.height = matrixH * SCALE;
}
function render(elements, selectedId, matrixW, matrixH, SCALE, opts = {}) {
if (!_ctx) return;
const cW = matrixW * SCALE, cH = matrixH * SCALE;
const bg = opts.bgColor;
_ctx.fillStyle = bg ? `rgb(${bg.r},${bg.g},${bg.b})` : '#000';
_ctx.fillRect(0, 0, cW, cH);
if (_showGrid) {
_ctx.strokeStyle = 'rgba(255,255,255,0.07)';
_ctx.lineWidth = 0.5;
for (let x = SCALE; x < cW; x += SCALE) {
_ctx.beginPath(); _ctx.moveTo(x, 0); _ctx.lineTo(x, cH); _ctx.stroke();
}
for (let y = SCALE; y < cH; y += SCALE) {
_ctx.beginPath(); _ctx.moveTo(0, y); _ctx.lineTo(cW, y); _ctx.stroke();
}
}
for (const el of elements) _drawElement(_ctx, el, SCALE, matrixW, matrixH, opts);
if (opts.showRuler) {
_ctx.save();
_ctx.fillStyle = 'rgba(255,255,255,0.08)';
_ctx.fillRect(0, 0, cW, SCALE); // top strip
_ctx.fillRect(0, 0, SCALE, cH); // left strip
_ctx.strokeStyle = 'rgba(255,255,255,0.5)';
_ctx.fillStyle = 'rgba(255,255,255,0.6)';
_ctx.font = `${Math.max(5, SCALE - 1)}px monospace`;
const step = SCALE >= 4 ? 8 : 16;
for (let px = 0; px <= matrixW; px += step) {
const cx = px * SCALE;
const major = px % 32 === 0;
_ctx.lineWidth = 0.5;
_ctx.beginPath(); _ctx.moveTo(cx, 0); _ctx.lineTo(cx, major ? SCALE : SCALE * 0.5); _ctx.stroke();
if (major && px > 0 && px < matrixW - 4) _ctx.fillText(String(px), cx + 1, SCALE - 1);
}
for (let py = 0; py <= matrixH; py += step) {
const cy = py * SCALE;
const major = py % 32 === 0;
_ctx.beginPath(); _ctx.moveTo(0, cy); _ctx.lineTo(major ? SCALE : SCALE * 0.5, cy); _ctx.stroke();
if (major && py > 0 && py < matrixH - 4) _ctx.fillText(String(py), 1, cy + SCALE - 1);
}
_ctx.restore();
}
if (opts.showGuides) {
_ctx.save();
_ctx.strokeStyle = 'rgba(255,60,60,0.45)';
_ctx.lineWidth = 1;
_ctx.setLineDash([4, 3]);
const mx = Math.floor(cW / 2) + 0.5;
const my = Math.floor(cH / 2) + 0.5;
_ctx.beginPath(); _ctx.moveTo(mx, 0); _ctx.lineTo(mx, cH); _ctx.stroke();
_ctx.beginPath(); _ctx.moveTo(0, my); _ctx.lineTo(cW, my); _ctx.stroke();
_ctx.setLineDash([]);
_ctx.restore();
}
const sel = selectedId != null ? elements.find(e => e.id === selectedId) : null;
if (sel) {
_drawSelection(_ctx, sel, SCALE, matrixW, matrixH);
if (opts.showTooltip) drawDragTooltip(_ctx, sel, SCALE, matrixW, matrixH);
}
}
return {
init, render, setGrid, updateCanvasSize,
hitTest, getBoundingBox, computeActualPos, resolveAnchor,
getResizeHandle, getCursorForHandle,
ELEMENT_DEFAULTS, FONT_MAP, DISPLAY_PRESETS,
};
})();
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,298 @@
"""
{{ plugin_name }} — LEDMatrix Plugin
Generated by LEDMatrix Plugin Composer on {{ generated_date }}
Extension points:
update() → add HTTP/MQTT data-fetching logic here
_get_display_values() → map fetched data to display strings
display() → add new elements or adapt layout per display size
"""
from src.plugin_system.base_plugin import BasePlugin
{% if has_clock or has_countdown %}
from datetime import datetime
{% endif %}
{% if has_blink %}
import time
{% endif %}
{% if has_text_template %}
from collections import defaultdict
{% endif %}
class {{ class_name }}(BasePlugin):
def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):
super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
{% for var in config_vars %}
self.{{ var.key }} = config.get({{ var.key | tojson }}, {{ var.default | tojson }})
{% endfor %}
# Live data cache — populated by update(); always {} in static layouts
self._data = {}
def update(self):
"""Fetch and refresh display data.
For dynamic plugins: fetch from APIs/MQTT here and store in self._data.
_get_display_values() will read self._data to produce display strings.
"""
# --- Data sources (add fetch logic here for dynamic plugins) ---
pass
def _get_display_values(self):
"""Map config variables and live data to display-ready strings.
This is the single extension point for v2 data sources:
add self._data lookups here once update() populates them.
"""
return {
{% for var in config_vars %}
{{ var.key | tojson }}: str(self.{{ var.key }}),
{% endfor %}
}
def display(self, force_clear=False):
try:
{% if has_text_template %}
values = defaultdict(str, self._get_display_values())
{% else %}
values = self._get_display_values()
{% endif %}
if force_clear:
self.display_manager.clear()
width = self.display_manager.width
height = self.display_manager.height
{% if bg_color %}
self.display_manager.draw.rectangle([0, 0, width, height], fill={{ bg_color }})
{% endif %}
# ── Elements (rendered bottom to top) ──────────────────────────
{% for el in elements %}
{% set p = " " if el.min_width > 0 else " " %}
{% set pi = (p + " ") if el.blink else p %}
{% if el.min_width > 0 %}
if width >= {{ el.min_width }}: # breakpoint: {{ el.min_width }}px+ displays only
{% endif %}
{% if el.blink %}
{{ p }}if int(time.time() * 2) % 2:
{% endif %}
{% if el.type == 'text' %}
{{ pi }}self.display_manager.draw_text(
{% if el.text_is_template %}
{{ pi }} {{ el.text | tojson }}.format_map(values),
{% else %}
{{ pi }} {{ el.text | tojson }},
{% endif %}
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% if el.text2 %}
{{ pi }}self.display_manager.draw_text(
{% if el.text_is_template %}
{{ pi }} {{ el.text2 | tojson }}.format_map(values),
{% else %}
{{ pi }} {{ el.text2 | tojson }},
{% endif %}
{{ pi }} x={{ el.x2_expr }}, y={{ el.y2_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'dynamic_text' %}
{% if el.binding_source == 'config' %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} values.get({{ el.binding_key | tojson }}, ''),
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'clock' %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} datetime.now().strftime({{ el.format | tojson }}),
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% if el.format2 %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} datetime.now().strftime({{ el.format2 | tojson }}),
{{ pi }} x={{ el.x2_expr }}, y={{ el.y2_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'countdown' %}
{{ pi }}_cd_target = float(values.get({{ el.binding_key | tojson }}, 0) or 0)
{{ pi }}_cd_secs = max(0.0, _cd_target - datetime.now().timestamp())
{% if el.countdown_format == 'dhms' %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h, _cd_rem = divmod(_cd_rem, 3600)
{{ pi }}_cd_m, _cd_s = divmod(_cd_rem, 60)
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h:02d}:{_cd_m:02d}:{_cd_s:02d}'
{% elif el.countdown_format == 'hms' %}
{{ pi }}_cd_h, _cd_rem = divmod(int(_cd_secs), 3600)
{{ pi }}_cd_m, _cd_s = divmod(_cd_rem, 60)
{{ pi }}_cd_str = f'{_cd_h}h {_cd_m:02d}:{_cd_s:02d}'
{% elif el.countdown_format == 'dhm' %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h, _cd_m = divmod(_cd_rem // 60, 60)
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h:02d}h {_cd_m:02d}m'
{% else %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h = _cd_rem // 3600
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h}h'
{% endif %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} _cd_str,
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% elif el.type == 'rectangle' %}
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type == 'arc' %}
{{ pi }}self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.end_angle }},
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% elif el.type == 'ellipse' %}
{{ pi }}self.display_manager.draw.ellipse(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type == 'pixel' %}
{{ pi }}self.display_manager.draw.point(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}],
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }})
{% elif el.type == 'rounded_rectangle' %}
{{ pi }}self.display_manager.draw.rounded_rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} radius={{ el.border_radius }},
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type in ('line', 'divider') %}
{{ pi }}self.display_manager.draw.line(
{{ pi }} [{{ el.x0_expr }}, {{ el.y0_expr }}, {{ el.x1_expr }}, {{ el.y1_expr }}],
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% elif el.type == 'pips' %}
{{ pi }}_pip_filled = max(0, min({{ el.pip_count }}, int(float(values.get({{ el.binding_key | tojson }}, 0) or 0))))
{{ pi }}for _pip_i in range({{ el.pip_count }}):
{{ pi }} _pip_x = ({{ el.x_expr }}) + _pip_i * ({{ el.pip_size }} + {{ el.pip_spacing }})
{{ pi }} _pip_color = {{ el.fill_tuple }} if _pip_i < _pip_filled else {{ el.empty_tuple }}
{% if not el.show_empty %}
{{ pi }} if _pip_i >= _pip_filled:
{{ pi }} continue
{% endif %}
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_pip_x, {{ el.y_expr }}, _pip_x + {{ el.pip_size }} - 1, ({{ el.y_expr }}) + {{ el.pip_size }} - 1],
{{ pi }} fill=_pip_color,
{{ pi }} )
{% elif el.type == 'sparkline' %}
{{ pi }}_sl_raw = str(values.get({{ el.binding_key | tojson }}, '') or '')
{{ pi }}_sl_vals = [float(v.strip()) for v in _sl_raw.split(',') if v.strip()][:{{ el.bar_count }}]
{{ pi }}_sl_vals += [0.0] * max(0, {{ el.bar_count }} - len(_sl_vals))
{{ pi }}_sl_max = max(_sl_vals) if any(_sl_vals) else 1.0
{{ pi }}_sl_bw = max(1, ({{ el.bar_width_px }} - {{ el.bar_spacing }} * ({{ el.bar_count }} - 1)) // {{ el.bar_count }})
{% if el.bg_tuple != 'None' %}
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, ({{ el.x_expr }}) + {{ el.bar_width_px }}, ({{ el.y_expr }}) + {{ el.bar_height_px }}],
{{ pi }} fill={{ el.bg_tuple }},
{{ pi }})
{% endif %}
{{ pi }}for _sl_i, _sl_v in enumerate(_sl_vals):
{{ pi }} _sl_norm = max(0.0, min(1.0, _sl_v / (_sl_max or 1)))
{{ pi }} _sl_bh = max(1, round({{ el.bar_height_px }} * _sl_norm))
{{ pi }} _sl_bx = ({{ el.x_expr }}) + (_sl_bw + {{ el.bar_spacing }}) * _sl_i
{{ pi }} _sl_by = ({{ el.y_expr }}) + {{ el.bar_height_px }} - _sl_bh
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_sl_bx, _sl_by, _sl_bx + _sl_bw - 1, _sl_by + _sl_bh - 1],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} )
{% elif el.type == 'gauge' %}
{{ pi }}_gv = max(0.0, min(100.0, float(values.get({{ el.binding_key | tojson }}, 0) or 0)))
{{ pi }}_g_total = (({{ el.end_angle }} - {{ el.start_angle }}) % 360) or 360
{{ pi }}_g_sweep = _g_total * _gv / 100.0
{% if el.track_tuple != 'None' %}
{{ pi }}self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.start_angle }} + _g_total,
{{ pi }} fill={{ el.track_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% endif %}
{{ pi }}if _g_sweep > 0:
{{ pi }} self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.start_angle }} + _g_sweep,
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }} )
{% if el.show_label %}
{{ pi }}_g_cx = ({{ el.x_expr }}) + ({{ el.x2_expr }} - ({{ el.x_expr }})) // 2
{{ pi }}_g_cy = ({{ el.y_expr }}) + ({{ el.y2_expr }} - ({{ el.y_expr }})) // 2
{{ pi }}self.display_manager.draw_text(
{{ pi }} f'{int(_gv)}%',
{{ pi }} x=_g_cx, y=_g_cy,
{{ pi }} color={{ el.label_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'marquee' %}
{{ pi }}_{{ el.data_key }}_text = {{ el.text | tojson }}
{{ pi }}_{{ el.data_key }}_tw = len(_{{ el.data_key }}_text) * {{ el.char_w }}
{{ pi }}_{{ el.data_key }}_x = int(self._data.get({{ el.data_key | tojson }}, width))
{% if el.direction == 'right' %}
{{ pi }}_{{ el.data_key }}_x += {{ el.scroll_speed }}
{{ pi }}if _{{ el.data_key }}_x > width:
{{ pi }} _{{ el.data_key }}_x = -(_{{ el.data_key }}_tw + {{ el.gap }})
{% else %}
{{ pi }}_{{ el.data_key }}_x -= {{ el.scroll_speed }}
{{ pi }}if _{{ el.data_key }}_x < -(_{{ el.data_key }}_tw + {{ el.gap }}):
{{ pi }} _{{ el.data_key }}_x = width
{% endif %}
{{ pi }}self._data[{{ el.data_key | tojson }}] = _{{ el.data_key }}_x
{{ pi }}self.display_manager.draw_text(
{{ pi }} _{{ el.data_key }}_text,
{{ pi }} x=_{{ el.data_key }}_x, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% elif el.type == 'progress_bar' %}
{{ pi }}_pb_x = {{ el.x_expr }}
{{ pi }}_pb_y = {{ el.y_expr }}
{{ pi }}_pb_pct = max(0.0, min(100.0, float(values.get({{ el.binding_key | tojson }}, 0) or 0))) / 100.0
{{ pi }}_pb_fill_w = int({{ el.bar_width }} * _pb_pct)
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [_pb_x, _pb_y, _pb_x + {{ el.bar_width }}, _pb_y + {{ el.bar_height }}],
{{ pi }} fill={{ el.bg_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{{ pi }}if _pb_fill_w > 0:
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_pb_x, _pb_y, _pb_x + _pb_fill_w, _pb_y + {{ el.bar_height }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} )
{% endif %}
{% endfor %}
# ── End elements ───────────────────────────────────────────────
self.display_manager.update_display()
except Exception as e:
self.logger.error('Display error: %s', e, exc_info=True)