diff --git a/plugins.json b/plugins.json index ea541df9..fba95296 100644 --- a/plugins.json +++ b/plugins.json @@ -76,7 +76,7 @@ "last_updated": "2026-07-31", "verified": true, "screenshot": "", - "latest_version": "1.21.1" + "latest_version": "1.22.0" }, { "id": "basketball-scoreboard", diff --git a/plugins/baseball-scoreboard/CHANGELOG.md b/plugins/baseball-scoreboard/CHANGELOG.md index 55035952..b875e37b 100644 --- a/plugins/baseball-scoreboard/CHANGELOG.md +++ b/plugins/baseball-scoreboard/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## [1.22.0] - 2026-08-03 + +### Changed +- **Scroll display now runs on the core's shared implementation.** The orchestration half of `scroll_display.py` — scroll-helper configuration, frame pumping, completion, settings resolution, and native `global_config['target_fps']` support — moves to the core's `src.common.sports_scroll` (LEDMatrix 3.2.0). Only the baseball-specific content half stays here: game cards and league separator icons. A fix to the shared behaviour now lands once in the core instead of being replicated across nine scoreboards. +- **Nothing changes on an older core.** The import is guarded: a core without `src.common.sports_scroll` falls back to `scroll_display_legacy.py`, the previous self-contained implementation, and the plugin behaves exactly as it did. This is why the minimum core version is unchanged at 2.0.0 — the plugin does not *require* 3.2.0, it merely prefers it. The fallback goes away in a later release, and the floor rises then. +- Verified byte-for-byte: all 24 safety-harness renders (8 panel sizes × 3 screens) are identical to 1.21.1, before and after. + ## [1.21.1] - 2026-08-03 ### Fixed diff --git a/plugins/baseball-scoreboard/manifest.json b/plugins/baseball-scoreboard/manifest.json index 960359d8..3359a5c5 100644 --- a/plugins/baseball-scoreboard/manifest.json +++ b/plugins/baseball-scoreboard/manifest.json @@ -1,7 +1,7 @@ { "id": "baseball-scoreboard", "name": "Baseball Scoreboard", - "version": "1.21.1", + "version": "1.22.0", "author": "ChuckBuilds", "description": "Live, recent, and upcoming baseball games across MLB, MiLB, and NCAA Baseball with real-time scores and schedules", "category": "sports", @@ -30,6 +30,12 @@ "branch": "main", "plugin_path": "plugins/baseball-scoreboard", "versions": [ + { + "released": "2026-08-03", + "version": "1.22.0", + "notes": "Scroll display now uses the core's shared sports-scroll orchestration (LEDMatrix 3.2.0) when available, falling back to the bundled implementation on older cores. No behaviour change: all 24 harness renders are byte-for-byte identical.", + "ledmatrix_min_version": "2.0.0" + }, { "released": "2026-08-03", "version": "1.21.1", diff --git a/plugins/baseball-scoreboard/scroll_display.py b/plugins/baseball-scoreboard/scroll_display.py index 24eac27d..5f19e52d 100644 --- a/plugins/baseball-scoreboard/scroll_display.py +++ b/plugins/baseball-scoreboard/scroll_display.py @@ -1,27 +1,29 @@ """ -Scroll Display Handler for Baseball Scoreboard Plugin - -Implements high-FPS horizontal scrolling of all matching games with league separator icons. -Uses ScrollHelper for efficient numpy-based scrolling and dynamic duration calculation. - -Features: -- Pre-rendered game cards for smooth scrolling -- League separator icons (MLB logo, MiLB logo, NCAA baseball logos) between different leagues -- Dynamic duration based on total content width -- FPS logging and performance monitoring -- Live priority support for scroll mode +Scroll Display Handler for Baseball Scoreboard Plugin. + +Orchestration (scroll-helper configuration, frame pumping, completion, +settings resolution, `global_config['target_fps']`) comes from the core's +`src.common.sports_scroll`, shipped in LEDMatrix 3.2.0. Only the *content* +half lives here: building this sport's game cards and separator icons. + +On a core that predates that module we fall back to `scroll_display_legacy`, +the previous self-contained implementation, so the plugin keeps working +unchanged. That fallback is why this plugin is safe to adopt core code ahead +of the B6 sunset -- the version floor alone does not protect users whose core +misreports its version (the v3.1.0 release reports "1.0.0"), and they would +otherwise get a plugin that fails to load. + +The content methods below are duplicated in the legacy module by design: it is +frozen, and deleting it at B6 leaves this file as the only copy. Fix bugs +here, not there. """ import logging -import time import os +import time from typing import Dict, Any, List, Optional -from PIL import Image -try: - from src.common.scroll_helper import ScrollHelper -except ImportError: - ScrollHelper = None +from PIL import Image try: from game_renderer import GameRenderer @@ -30,724 +32,345 @@ logger = logging.getLogger(__name__) -# Pillow compatibility: Image.Resampling.LANCZOS is available in Pillow >= 9.1 -# Fall back to Image.LANCZOS for older versions try: RESAMPLE_FILTER = Image.Resampling.LANCZOS except AttributeError: RESAMPLE_FILTER = Image.LANCZOS +_USING_CORE_SCROLL = False +try: + from src.common.sports_scroll import ( + SportsScrollDisplay as _ScrollDisplayBase, + SportsScrollDisplayManager as _ScrollDisplayManagerBase, + ) + _USING_CORE_SCROLL = True +except ModuleNotFoundError as exc: + # Fall back only when the CORE module is absent. A bare `except + # ImportError` would also swallow a failure raised *inside* a core module + # that is present, silently loading the legacy copy and hiding a broken + # core install. + if exc.name not in {"src", "src.common", "src.common.sports_scroll"}: + raise + _ScrollDisplayBase = None + _ScrollDisplayManagerBase = None + + +if not _USING_CORE_SCROLL: + # Pre-3.2.0 core: use the previous implementation wholesale. + from scroll_display_legacy import ( # noqa: F401 + LegacyScrollDisplay as ScrollDisplay, + LegacyScrollDisplayManager as ScrollDisplayManager, + ) + logger.info( + "baseball-scoreboard: core src.common.sports_scroll not available; " + "using the bundled legacy scroll display" + ) +else: + + class ScrollDisplay(_ScrollDisplayBase): + """Baseball game cards and separator icons on the core scroll engine.""" + + # The ladder the legacy _get_scroll_settings walked, same order. + SCROLL_LEAGUE_KEYS = ("mlb", "milb", "ncaa_baseball") + + # Paths to league separator icons + MLB_SEPARATOR_ICON = "assets/sports/mlb_logos/MLB.png" + MILB_SEPARATOR_ICON = "assets/sports/milb_logos/MiLB.png" + NCAA_BASEBALL_SEPARATOR_ICON = "assets/sports/ncaa_logos/ncaa_baseball.png" + + def __init__(self, *args, **kwargs): + # Set before super(): the base calls _load_separator_icons() from + # its __init__, and this plugin's renderer cache must already exist. + self._game_renderer = None + super().__init__(*args, **kwargs) + + def _get_game_renderer(self, game_card_width: int = 128) -> Optional[GameRenderer]: + """Get or create the cached GameRenderer instance. + + Args: + game_card_width: Width for each game card. Cached renderer is recreated + if this differs from the current renderer's width. + """ + if GameRenderer is None: + self.logger.error("GameRenderer not available") + return None + + # Recreate renderer if card width changed (e.g. config update) + if self._game_renderer is None or getattr(self._game_renderer, "display_width", None) != game_card_width: + self._game_renderer = GameRenderer( + game_card_width, + self.display_height, + self.config, + logo_cache=self._logo_cache, + custom_logger=self.logger + ) + return self._game_renderer -class ScrollDisplay: - """ - Handles scroll display mode for the baseball scoreboard plugin. - - This class: - - Collects all games matching criteria (respecting live priority) - - Pre-renders each game using GameRenderer - - Adds league separator icons between different leagues - - Composes a single wide image using ScrollHelper - - Implements dynamic duration based on total content width - - Logs FPS and game count during scrolling - """ - - # Paths to league separator icons - MLB_SEPARATOR_ICON = "assets/sports/mlb_logos/MLB.png" - MILB_SEPARATOR_ICON = "assets/sports/milb_logos/MiLB.png" - NCAA_BASEBALL_SEPARATOR_ICON = "assets/sports/ncaa_logos/ncaa_baseball.png" - - def __init__( - self, - display_manager, - config: Dict[str, Any], - custom_logger: Optional[logging.Logger] = None, - global_config: Optional[Dict[str, Any]] = None - ): - """ - Initialize the ScrollDisplay handler. - - Args: - display_manager: Display manager instance - config: Plugin configuration dictionary - custom_logger: Optional custom logger instance - global_config: Optional global LEDMatrix config (for target_fps) - """ - self.display_manager = display_manager - self.config = config - self.logger = custom_logger or logger - self.global_config = global_config or {} - - # Get display dimensions - if hasattr(display_manager, 'matrix') and display_manager.matrix is not None: - self.display_width = display_manager.matrix.width - self.display_height = display_manager.matrix.height - else: - self.display_width = getattr(display_manager, "width", 128) - self.display_height = getattr(display_manager, "height", 32) - - # Initialize ScrollHelper - if ScrollHelper: - self.scroll_helper = ScrollHelper( - self.display_width, - self.display_height, - self.logger - ) - # Configure scroll settings - self._configure_scroll_helper() - else: - self.scroll_helper = None - self.logger.error("ScrollHelper not available - scroll mode will not work") - - # Shared logo cache for game renderer - self._logo_cache: Dict[str, Image.Image] = {} - - # Cached GameRenderer instance (created lazily) - self._game_renderer: Optional[GameRenderer] = None - - # League separator icons cache - self._separator_icons: Dict[str, Image.Image] = {} - self._load_separator_icons() - - # Tracking state - self._current_games: List[Dict] = [] - self._current_game_type: str = "" - self._current_leagues: List[str] = [] - self._vegas_content_items: List[Image.Image] = [] - self._is_scrolling = False - self._scroll_start_time: Optional[float] = None - self._last_log_time: float = 0 - self._log_interval: float = 5.0 # Log every 5 seconds - - # Performance tracking - self._frame_count: int = 0 - self._fps_sample_start: float = time.time() - - def _configure_scroll_helper(self) -> None: - """Configure scroll helper with settings from config.""" - if not self.scroll_helper: - return - - # Get global scroll settings, then per-league overrides - scroll_settings = self._get_scroll_settings() - - # Set scroll speed (pixels per second in time-based mode) - scroll_speed = scroll_settings.get("scroll_speed", 50.0) - self.scroll_helper.set_scroll_speed(scroll_speed) - - # Set scroll delay - scroll_delay = scroll_settings.get("scroll_delay", 0.01) - self.scroll_helper.set_scroll_delay(scroll_delay) - - # Honor the global smooth-scrolling FPS target (older cores lack the setter) - target_fps = self.global_config.get('target_fps') or self.global_config.get('scroll_target_fps') - try: - # Coerce before comparing: a malformed global config value - # must degrade to today's scroll_delay pacing, not raise. - target_fps = float(target_fps) if target_fps is not None else None - except (TypeError, ValueError): - target_fps = None - if target_fps: - if hasattr(self.scroll_helper, 'set_target_fps'): - self.scroll_helper.set_target_fps(target_fps) - else: - self.scroll_helper.target_fps = max(30.0, min(200.0, target_fps)) - self.scroll_helper.frame_time_target = 1.0 / self.scroll_helper.target_fps - - # Enable dynamic duration - dynamic_duration = scroll_settings.get("dynamic_duration", True) - self.scroll_helper.set_dynamic_duration_settings( - enabled=dynamic_duration, - min_duration=30, - max_duration=600, # 10 minutes max - buffer=0.2 # 20% buffer to ensure scroll completes fully off screen - ) - - # Use frame-based scrolling for better FPS control - self.scroll_helper.set_frame_based_scrolling(True) - - # Convert scroll_speed to pixels/frame, handling both interpretations: - # - scroll_speed as pixels/second: multiply by scroll_delay to get pixels/frame - # - scroll_speed as pixels/frame: use directly - # Pick the interpretation that yields a reasonable value (0.1-5.0 range) - valid_range = (0.1, 5.0) - candidate_pps = scroll_speed * scroll_delay # pixels/sec interpretation - - if valid_range[0] <= candidate_pps <= valid_range[1]: - # pixels/second interpretation yields valid pixels/frame - pixels_per_frame = candidate_pps - elif valid_range[0] <= scroll_speed <= valid_range[1]: - # scroll_speed is already a valid pixels/frame value - pixels_per_frame = scroll_speed - else: - # Neither interpretation is valid, use pixels/sec and clamp - pixels_per_frame = candidate_pps - - # Clamp to reasonable range (0.1 to 5 pixels per frame for smooth scrolling) - pixels_per_frame = max(valid_range[0], min(valid_range[1], pixels_per_frame)) - self.scroll_helper.set_scroll_speed(pixels_per_frame) - - # Calculate effective pixels per second for logging - effective_pps = pixels_per_frame / scroll_delay if scroll_delay > 0 else pixels_per_frame * 100 - - self.logger.info( - f"ScrollHelper configured: {pixels_per_frame:.2f} px/frame, delay={scroll_delay}s " - f"(effective {effective_pps:.1f} px/s), dynamic_duration={dynamic_duration}" - ) - - def _get_scroll_settings(self, league: Optional[str] = None) -> Dict[str, Any]: - """Get scroll settings, optionally for a specific league.""" - # Default scroll settings - defaults = { - "scroll_speed": 50.0, - "scroll_delay": 0.01, - "gap_between_games": 48, - "show_league_separators": True, - "dynamic_duration": True, - "game_card_width": self.display_width, - } - - # Try to get league-specific settings first - if league: - league_config = self.config.get(league, {}) - league_scroll = league_config.get("scroll_settings", {}) - if league_scroll: - return {**defaults, **league_scroll} - - # Fall back to MLB settings (usually first enabled) - mlb_config = self.config.get("mlb", {}) - mlb_scroll = mlb_config.get("scroll_settings", {}) - if mlb_scroll: - return {**defaults, **mlb_scroll} - - # Fall back to MiLB settings - milb_config = self.config.get("milb", {}) - milb_scroll = milb_config.get("scroll_settings", {}) - if milb_scroll: - return {**defaults, **milb_scroll} - - # Fall back to NCAA Baseball settings - ncaa_config = self.config.get("ncaa_baseball", {}) - ncaa_scroll = ncaa_config.get("scroll_settings", {}) - if ncaa_scroll: - return {**defaults, **ncaa_scroll} - - return defaults - - def _get_game_renderer(self, game_card_width: int = 128) -> Optional[GameRenderer]: - """Get or create the cached GameRenderer instance. - - Args: - game_card_width: Width for each game card. Cached renderer is recreated - if this differs from the current renderer's width. - """ - if GameRenderer is None: - self.logger.error("GameRenderer not available") - return None - - # Recreate renderer if card width changed (e.g. config update) - if self._game_renderer is None or getattr(self._game_renderer, "display_width", None) != game_card_width: - self._game_renderer = GameRenderer( - game_card_width, - self.display_height, - self.config, - logo_cache=self._logo_cache, - custom_logger=self.logger - ) - return self._game_renderer - - def _load_separator_icon(self, icon_path: str, league_key: str, target_height: int) -> None: - """ - Load and resize a single league separator icon. - - Args: - icon_path: Path to the icon file - league_key: Key to store the icon under in _separator_icons - target_height: Target height for the resized icon - """ - if not os.path.exists(icon_path): - self.logger.warning(f"{league_key.upper()} separator icon not found at {icon_path}") - return - - try: - with Image.open(icon_path) as icon: - if icon.mode != "RGBA": - icon = icon.convert("RGBA") - # Resize to fit height while maintaining aspect ratio - aspect = icon.width / icon.height - new_width = int(target_height * aspect) - icon = icon.resize((new_width, target_height), resample=RESAMPLE_FILTER) - self._separator_icons[league_key] = icon.copy() - self.logger.debug(f"Loaded {league_key.upper()} separator icon: {new_width}x{target_height}") - except OSError: - self.logger.exception(f"Error loading {league_key.upper()} separator icon") - - def _load_separator_icons(self) -> None: - """Load and resize league separator icons.""" - separator_height = self.display_height - 4 # Leave some padding - - # Load all league separator icons - icons_to_load = [ - (self.MLB_SEPARATOR_ICON, "mlb"), - (self.MILB_SEPARATOR_ICON, "milb"), - (self.NCAA_BASEBALL_SEPARATOR_ICON, "ncaa_baseball"), - ] - for icon_path, league_key in icons_to_load: - self._load_separator_icon(icon_path, league_key, separator_height) - - def _determine_game_type(self, game: Dict) -> str: - """ - Determine the game type from the game's status. - - Args: - game: Game dictionary - - Returns: - Game type: 'live', 'recent', or 'upcoming' - """ - if game.get('is_live'): - return 'live' - elif game.get('is_final'): - return 'recent' - elif game.get('is_upcoming'): - return 'upcoming' - else: - # Default to upcoming if state is unknown - return 'upcoming' - - def prepare_scroll_content( - self, - games: List[Dict], - game_type: str, - leagues: List[str], - rankings_cache: Optional[Dict[str, int]] = None - ) -> bool: - """ - Prepare scrolling content from a list of games. - - Args: - games: List of game dictionaries with league info - game_type: Type hint ('live', 'recent', 'upcoming', or 'mixed' for mixed types) - leagues: List of leagues in order (e.g., ['mlb', 'milb', 'ncaa_baseball']) - rankings_cache: Optional team rankings cache for displaying team rankings - - Returns: - True if content was prepared successfully, False otherwise - """ - if not self.scroll_helper: - self.logger.error("ScrollHelper not available") - return False + def _load_separator_icon(self, icon_path: str, league_key: str, target_height: int) -> None: + """ + Load and resize a single league separator icon. - if not games: - self.logger.debug("No games to prepare for scrolling") - self.scroll_helper.clear_cache() - self._current_games = [] - self._vegas_content_items = [] - self._is_scrolling = False - return False + Args: + icon_path: Path to the icon file + league_key: Key to store the icon under in _separator_icons + target_height: Target height for the resized icon + """ + if not os.path.exists(icon_path): + self.logger.warning(f"{league_key.upper()} separator icon not found at {icon_path}") + return - self._current_games = games - self._current_game_type = game_type - self._current_leagues = leagues - - # Get scroll settings - scroll_settings = self._get_scroll_settings() - gap_between_games = scroll_settings.get("gap_between_games", 24) - show_separators = scroll_settings.get("show_league_separators", True) - game_card_width = scroll_settings.get("game_card_width", self.display_width) - - # Get or create cached game renderer; default card width is the full display width - # so each game card fills the viewport and logos sit at the display edges - renderer = self._get_game_renderer(game_card_width) - - # Pass rankings cache to renderer if available - if renderer and rankings_cache: - renderer.set_rankings_cache(rankings_cache) - - # Pre-render all game cards - content_items: List[Image.Image] = [] - current_league = None - game_count = 0 - league_counts: Dict[str, int] = {} - - for game in games: - game_league = game.get("league", "mlb") # Default to MLB if not specified - - # Add league separator when entering a new league (first or switching) - if show_separators and game_league != current_league: - separator = self._separator_icons.get(game_league) - if separator: - # Create a separator image with proper background - sep_img = Image.new('RGB', (separator.width + 8, self.display_height), (0, 0, 0)) - # Center the separator vertically - y_offset = (self.display_height - separator.height) // 2 - sep_img.paste(separator, (4, y_offset), separator) - content_items.append(sep_img) - context = "at start" if current_league is None else "" - self.logger.debug(f"Added {game_league} separator icon {context}".strip()) - - current_league = game_league - - # Render game card - determine type from game state try: - individual_game_type = self._determine_game_type(game) - game_img = renderer.render_game_card(game, individual_game_type) - - # Only pad when card is narrower than the viewport; full-width cards - # need no padding or the card becomes wider than the display. - padding = 0 if game_img.width >= self.display_width else 12 - padded_width = game_img.width + (padding * 2) - padded_img = Image.new('RGB', (padded_width, game_img.height), (0, 0, 0)) - padded_img.paste(game_img, (padding, 0)) - - content_items.append(padded_img) - game_count += 1 - league_counts[game_league] = league_counts.get(game_league, 0) + 1 - except Exception: - self.logger.exception("Error rendering game card") - continue - - if not content_items: - self.logger.warning("No game cards rendered") - return False - - # Store individual items for Vegas mode (avoids scroll_helper padding) - self._vegas_content_items = list(content_items) - - # Create scrolling image using ScrollHelper - self.scroll_helper.create_scrolling_image( - content_items, - item_gap=gap_between_games, - element_gap=0 # No element gap - each item is a complete game card - ) - - # Log what we loaded - league_summary = ", ".join([f"{league.upper()}({count})" for league, count in league_counts.items()]) - self.logger.info( - f"[Baseball Scroll] Prepared {game_count} games for scrolling: {league_summary}" - ) - self.logger.info( - f"[Baseball Scroll] Total scroll width: {self.scroll_helper.total_scroll_width}px, " - f"Dynamic duration: {self.scroll_helper.calculated_duration}s" - ) - - # Reset tracking state - self._is_scrolling = True - self._scroll_start_time = time.time() - self._frame_count = 0 - self._fps_sample_start = time.time() - - return True - - def display_scroll_frame(self) -> bool: - """ - Display the next frame of the scrolling content. - - Returns: - True if a frame was displayed, False if scroll is complete or no content - """ - if not self.scroll_helper or not self.scroll_helper.cached_image: - return False - - # Update scroll position - self.scroll_helper.update_scroll_position() - - # Get visible portion - visible = self.scroll_helper.get_visible_portion() - if not visible: - return False - - # Display the visible portion - try: - self.display_manager.image = visible - self.display_manager.update_display() + with Image.open(icon_path) as icon: + if icon.mode != "RGBA": + icon = icon.convert("RGBA") + # Resize to fit height while maintaining aspect ratio + aspect = icon.width / icon.height + new_width = int(target_height * aspect) + icon = icon.resize((new_width, target_height), resample=RESAMPLE_FILTER) + self._separator_icons[league_key] = icon.copy() + self.logger.debug(f"Loaded {league_key.upper()} separator icon: {new_width}x{target_height}") + except OSError: + self.logger.exception(f"Error loading {league_key.upper()} separator icon") + + def _load_separator_icons(self) -> None: + """Load and resize league separator icons.""" + separator_height = self.display_height - 4 # Leave some padding + + # Load all league separator icons + icons_to_load = [ + (self.MLB_SEPARATOR_ICON, "mlb"), + (self.MILB_SEPARATOR_ICON, "milb"), + (self.NCAA_BASEBALL_SEPARATOR_ICON, "ncaa_baseball"), + ] + for icon_path, league_key in icons_to_load: + self._load_separator_icon(icon_path, league_key, separator_height) + + def _determine_game_type(self, game: Dict) -> str: + """ + Determine the game type from the game's status. + + Args: + game: Game dictionary + + Returns: + Game type: 'live', 'recent', or 'upcoming' + """ + if game.get('is_live'): + return 'live' + elif game.get('is_final'): + return 'recent' + elif game.get('is_upcoming'): + return 'upcoming' + else: + # Default to upcoming if state is unknown + return 'upcoming' + + def prepare_scroll_content( + self, + games: List[Dict], + game_type: str, + leagues: List[str], + rankings_cache: Optional[Dict[str, int]] = None + ) -> bool: + """ + Prepare scrolling content from a list of games. + + Args: + games: List of game dictionaries with league info + game_type: Type hint ('live', 'recent', 'upcoming', or 'mixed' for mixed types) + leagues: List of leagues in order (e.g., ['mlb', 'milb', 'ncaa_baseball']) + rankings_cache: Optional team rankings cache for displaying team rankings + + Returns: + True if content was prepared successfully, False otherwise + """ + if not self.scroll_helper: + self.logger.error("ScrollHelper not available") + return False + + if not games: + self.logger.debug("No games to prepare for scrolling") + self.scroll_helper.clear_cache() + self._current_games = [] + self._vegas_content_items = [] + self._is_scrolling = False + return False + + self._current_games = games + self._current_game_type = game_type + self._current_leagues = leagues + + # Get scroll settings + scroll_settings = self._get_scroll_settings() + gap_between_games = scroll_settings.get("gap_between_games", 24) + show_separators = scroll_settings.get("show_league_separators", True) + game_card_width = scroll_settings.get("game_card_width", self.display_width) + + # Get or create cached game renderer; default card width is the full display width + # so each game card fills the viewport and logos sit at the display edges + renderer = self._get_game_renderer(game_card_width) + + # Pass rankings cache to renderer if available + if renderer and rankings_cache: + renderer.set_rankings_cache(rankings_cache) + + # Pre-render all game cards + content_items: List[Image.Image] = [] + current_league = None + game_count = 0 + league_counts: Dict[str, int] = {} + + for game in games: + game_league = game.get("league", "mlb") # Default to MLB if not specified + + # Add league separator when entering a new league (first or switching) + if show_separators and game_league != current_league: + separator = self._separator_icons.get(game_league) + if separator: + # Create a separator image with proper background + sep_img = Image.new('RGB', (separator.width + 8, self.display_height), (0, 0, 0)) + # Center the separator vertically + y_offset = (self.display_height - separator.height) // 2 + sep_img.paste(separator, (4, y_offset), separator) + content_items.append(sep_img) + context = "at start" if current_league is None else "" + self.logger.debug(f"Added {game_league} separator icon {context}".strip()) + + current_league = game_league + + # Render game card - determine type from game state + try: + individual_game_type = self._determine_game_type(game) + game_img = renderer.render_game_card(game, individual_game_type) + + # Only pad when card is narrower than the viewport; full-width cards + # need no padding or the card becomes wider than the display. + padding = 0 if game_img.width >= self.display_width else 12 + padded_width = game_img.width + (padding * 2) + padded_img = Image.new('RGB', (padded_width, game_img.height), (0, 0, 0)) + padded_img.paste(game_img, (padding, 0)) + + content_items.append(padded_img) + game_count += 1 + league_counts[game_league] = league_counts.get(game_league, 0) + 1 + except Exception: + self.logger.exception("Error rendering game card") + continue + + if not content_items: + self.logger.warning("No game cards rendered") + return False + + # Store individual items for Vegas mode (avoids scroll_helper padding) + self._vegas_content_items = list(content_items) + + # Create scrolling image using ScrollHelper + self.scroll_helper.create_scrolling_image( + content_items, + item_gap=gap_between_games, + element_gap=0 # No element gap - each item is a complete game card + ) - # Track frame rate - self._frame_count += 1 - self.scroll_helper.log_frame_rate() + # Log what we loaded + league_summary = ", ".join([f"{league.upper()}({count})" for league, count in league_counts.items()]) + self.logger.info( + f"[Baseball Scroll] Prepared {game_count} games for scrolling: {league_summary}" + ) + self.logger.info( + f"[Baseball Scroll] Total scroll width: {self.scroll_helper.total_scroll_width}px, " + f"Dynamic duration: {self.scroll_helper.calculated_duration}s" + ) - # Periodic logging - self._log_scroll_progress() + # Reset tracking state + self._is_scrolling = True + self._scroll_start_time = time.time() + self._frame_count = 0 + self._fps_sample_start = time.time() return True - except Exception: - self.logger.exception("Error displaying scroll frame") - return False - def _log_scroll_progress(self) -> None: - """Log scroll progress and FPS periodically.""" - current_time = time.time() - if current_time - self._last_log_time >= self._log_interval: - # Calculate FPS - elapsed = current_time - self._fps_sample_start - if elapsed > 0: - fps = self._frame_count / elapsed + class ScrollDisplayManager(_ScrollDisplayManagerBase): + """Baseball scroll manager -- everything but the extras below is core.""" + + display_class = ScrollDisplay + + def prepare_content( + self, + games: List[Dict], + game_type: str, + leagues: List[str], + rankings_cache: Dict[str, int] = None + ) -> bool: + """ + Render content for one scroll display without making it the active one. + + Vegas mode builds its own combined slate in the background while the + standalone rotation may be mid-scroll on a different game type. Going + through prepare_and_display() for that would repoint + ``_current_game_type``, so the next display_frame() would render the + Vegas slate instead of the mode the rotation is actually showing. + + Args: + games: List of game dictionaries + game_type: Scroll display key to render into + leagues: List of leagues + rankings_cache: Optional team rankings cache + + Returns: + True if content was prepared successfully + """ + scroll_display = self.get_scroll_display(game_type) + return scroll_display.prepare_scroll_content( + games, game_type, leagues, rankings_cache + ) - # Get scroll info - scroll_info = self.scroll_helper.get_scroll_info() + def get_dynamic_duration(self, game_type: Optional[str] = None) -> int: + """Get the dynamic duration for the current scroll.""" + if game_type is None: + game_type = self._current_game_type - self.logger.info( - f"[Baseball Scroll] FPS: {fps:.1f}, " - f"Position: {scroll_info['scroll_position']:.0f}/{scroll_info['total_width']}px, " - f"Elapsed: {scroll_info.get('elapsed_time', 0):.1f}s/{scroll_info['dynamic_duration']}s" - ) + if game_type is None: + return 60 - # Reset FPS tracking - self._frame_count = 0 - self._fps_sample_start = current_time - self._last_log_time = current_time - - def is_scroll_complete(self) -> bool: - """Check if the scroll cycle is complete.""" - if not self.scroll_helper: - return True - return self.scroll_helper.is_scroll_complete() + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return 60 - def reset_scroll(self) -> None: - """Reset the scroll position to the beginning.""" - if self.scroll_helper: - self.scroll_helper.reset_scroll() - self._frame_count = 0 - self._fps_sample_start = time.time() - self.logger.debug("Scroll position reset") - - def get_scroll_info(self) -> Dict[str, Any]: - """Get current scroll state information.""" - if not self.scroll_helper: - return {"error": "ScrollHelper not available"} - - info = self.scroll_helper.get_scroll_info() - info.update({ - "game_count": len(self._current_games), - "game_type": self._current_game_type, - "leagues": self._current_leagues, - "is_scrolling": self._is_scrolling - }) - return info - - def get_dynamic_duration(self) -> int: - """Get the calculated dynamic duration for this scroll content.""" - if self.scroll_helper: - return self.scroll_helper.get_dynamic_duration() - return 60 # Default fallback - - def clear(self) -> None: - """Clear scroll content and reset state.""" - if self.scroll_helper: - self.scroll_helper.clear_cache() - self._current_games = [] - self._current_game_type = "" - self._current_leagues = [] - self._vegas_content_items = [] - self._is_scrolling = False - self._scroll_start_time = None - self.logger.debug("Scroll display cleared") - - -class ScrollDisplayManager: - """ - Manages scroll display instances for different game types. - - This class provides a higher-level interface for the baseball plugin - to manage scroll displays for live, recent, and upcoming games. - """ - - def __init__( - self, - display_manager, - config: Dict[str, Any], - custom_logger: Optional[logging.Logger] = None, - global_config: Optional[Dict[str, Any]] = None - ): - """ - Initialize the ScrollDisplayManager. - - Args: - display_manager: Display manager instance - config: Plugin configuration dictionary - custom_logger: Optional custom logger instance - global_config: Optional global LEDMatrix config (for target_fps) - """ - self.display_manager = display_manager - self.config = config - self.logger = custom_logger or logger - self.global_config = global_config or {} - - # Create scroll displays for each game type - self._scroll_displays: Dict[str, ScrollDisplay] = {} - self._current_game_type: Optional[str] = None - - def get_scroll_display(self, game_type: str) -> ScrollDisplay: - """ - Get or create a scroll display for a game type. - - Args: - game_type: Type of games ('live', 'recent', 'upcoming') - - Returns: - ScrollDisplay instance for the game type - """ - if game_type not in self._scroll_displays: - self._scroll_displays[game_type] = ScrollDisplay( - self.display_manager, - self.config, - self.logger, - global_config=self.global_config - ) - return self._scroll_displays[game_type] - - def prepare_and_display( - self, - games: List[Dict], - game_type: str, - leagues: List[str], - rankings_cache: Dict[str, int] = None - ) -> bool: - """ - Prepare content and start displaying scroll. - - Args: - games: List of game dictionaries - game_type: Type of games - leagues: List of leagues - rankings_cache: Optional team rankings cache - - Returns: - True if scroll was started successfully - """ - success = self.prepare_content(games, game_type, leagues, rankings_cache) - - if success: - self._current_game_type = game_type + return scroll_display.get_dynamic_duration() - return success - - def prepare_content( - self, - games: List[Dict], - game_type: str, - leagues: List[str], - rankings_cache: Dict[str, int] = None - ) -> bool: - """ - Render content for one scroll display without making it the active one. - - Vegas mode builds its own combined slate in the background while the - standalone rotation may be mid-scroll on a different game type. Going - through prepare_and_display() for that would repoint - ``_current_game_type``, so the next display_frame() would render the - Vegas slate instead of the mode the rotation is actually showing. - - Args: - games: List of game dictionaries - game_type: Scroll display key to render into - leagues: List of leagues - rankings_cache: Optional team rankings cache - - Returns: - True if content was prepared successfully - """ - scroll_display = self.get_scroll_display(game_type) - return scroll_display.prepare_scroll_content( - games, game_type, leagues, rankings_cache - ) - - def display_frame(self, game_type: Optional[str] = None) -> bool: - """ - Display the next frame of the current scroll. - - Args: - game_type: Optional game type (uses current if not specified) - - Returns: - True if a frame was displayed - """ - if game_type is None: - game_type = self._current_game_type - - if game_type is None: - return False + def has_cached_content(self) -> bool: + """ + Check if any scroll display has cached content. - scroll_display = self._scroll_displays.get(game_type) - if scroll_display is None: + Returns: + True if any scroll display has a cached image, False otherwise + """ + for scroll_display in self._scroll_displays.values(): + if hasattr(scroll_display, 'scroll_helper') and scroll_display.scroll_helper: + if scroll_display.scroll_helper.cached_image is not None: + return True return False - return scroll_display.display_scroll_frame() + def get_vegas_content_items_for(self, game_type: str) -> list: + """ + Return the Vegas item list for a single scroll display. - def is_complete(self, game_type: Optional[str] = None) -> bool: - """Check if the current scroll is complete.""" - if game_type is None: - game_type = self._current_game_type + Vegas mode needs the items from one specific display (the combined + live/recent/upcoming set), not the union across all of them. + get_all_vegas_content_items() returns whatever the standalone display + modes happen to have rendered, which both under-reports (only the last + rendered mode's games) and can double-count a game that appears in two + displays. - if game_type is None: - return True + Args: + game_type: Scroll display key, e.g. 'mixed' - scroll_display = self._scroll_displays.get(game_type) - if scroll_display is None: - return True + Returns: + Copy of that display's Vegas items, or an empty list if absent. + """ + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return [] + return list(getattr(scroll_display, '_vegas_content_items', None) or []) - return scroll_display.is_scroll_complete() - - def get_dynamic_duration(self, game_type: Optional[str] = None) -> int: - """Get the dynamic duration for the current scroll.""" - if game_type is None: - game_type = self._current_game_type - - if game_type is None: - return 60 - - scroll_display = self._scroll_displays.get(game_type) - if scroll_display is None: - return 60 - - return scroll_display.get_dynamic_duration() - - def clear_all(self) -> None: - """Clear all scroll displays.""" - for scroll_display in self._scroll_displays.values(): - scroll_display.clear() - self._current_game_type = None - - def has_cached_content(self) -> bool: - """ - Check if any scroll display has cached content. - - Returns: - True if any scroll display has a cached image, False otherwise - """ - for scroll_display in self._scroll_displays.values(): - if hasattr(scroll_display, 'scroll_helper') and scroll_display.scroll_helper: - if scroll_display.scroll_helper.cached_image is not None: - return True - return False - - def get_all_vegas_content_items(self) -> list: - """Collect _vegas_content_items from all scroll displays.""" - items = [] - for sd in self._scroll_displays.values(): - vegas_items = getattr(sd, '_vegas_content_items', None) - if vegas_items: - items.extend(vegas_items) - return items - - def get_vegas_content_items_for(self, game_type: str) -> list: - """ - Return the Vegas item list for a single scroll display. - - Vegas mode needs the items from one specific display (the combined - live/recent/upcoming set), not the union across all of them. - get_all_vegas_content_items() returns whatever the standalone display - modes happen to have rendered, which both under-reports (only the last - rendered mode's games) and can double-count a game that appears in two - displays. - - Args: - game_type: Scroll display key, e.g. 'mixed' - - Returns: - Copy of that display's Vegas items, or an empty list if absent. - """ - scroll_display = self._scroll_displays.get(game_type) - if scroll_display is None: - return [] - return list(getattr(scroll_display, '_vegas_content_items', None) or []) diff --git a/plugins/baseball-scoreboard/scroll_display_legacy.py b/plugins/baseball-scoreboard/scroll_display_legacy.py new file mode 100644 index 00000000..a5ef29a3 --- /dev/null +++ b/plugins/baseball-scoreboard/scroll_display_legacy.py @@ -0,0 +1,758 @@ +""" +Scroll Display Handler for Baseball Scoreboard Plugin -- LEGACY FALLBACK. + +Used only on a core that predates `src.common.sports_scroll` (LEDMatrix +3.2.0). `scroll_display.py` prefers the core implementation and falls back +here, so this file is frozen: fix bugs in the core module, not here. It goes +away at the B6 sunset, once cores without that module are gone. + +Implements high-FPS horizontal scrolling of all matching games with league separator icons. +Uses ScrollHelper for efficient numpy-based scrolling and dynamic duration calculation. + +Features: +- Pre-rendered game cards for smooth scrolling +- League separator icons (MLB logo, MiLB logo, NCAA baseball logos) between different leagues +- Dynamic duration based on total content width +- FPS logging and performance monitoring +- Live priority support for scroll mode +""" + +import logging +import time +import os +from typing import Dict, Any, List, Optional +from PIL import Image + +try: + from src.common.scroll_helper import ScrollHelper +except ImportError: + ScrollHelper = None + +try: + from game_renderer import GameRenderer +except ImportError: + GameRenderer = None + +logger = logging.getLogger(__name__) + +# Pillow compatibility: Image.Resampling.LANCZOS is available in Pillow >= 9.1 +# Fall back to Image.LANCZOS for older versions +try: + RESAMPLE_FILTER = Image.Resampling.LANCZOS +except AttributeError: + RESAMPLE_FILTER = Image.LANCZOS + + +class LegacyScrollDisplay: + """ + Handles scroll display mode for the baseball scoreboard plugin. + + This class: + - Collects all games matching criteria (respecting live priority) + - Pre-renders each game using GameRenderer + - Adds league separator icons between different leagues + - Composes a single wide image using ScrollHelper + - Implements dynamic duration based on total content width + - Logs FPS and game count during scrolling + """ + + # Paths to league separator icons + MLB_SEPARATOR_ICON = "assets/sports/mlb_logos/MLB.png" + MILB_SEPARATOR_ICON = "assets/sports/milb_logos/MiLB.png" + NCAA_BASEBALL_SEPARATOR_ICON = "assets/sports/ncaa_logos/ncaa_baseball.png" + + def __init__( + self, + display_manager, + config: Dict[str, Any], + custom_logger: Optional[logging.Logger] = None, + global_config: Optional[Dict[str, Any]] = None + ): + """ + Initialize the ScrollDisplay handler. + + Args: + display_manager: Display manager instance + config: Plugin configuration dictionary + custom_logger: Optional custom logger instance + global_config: Optional global LEDMatrix config (for target_fps) + """ + self.display_manager = display_manager + self.config = config + self.logger = custom_logger or logger + self.global_config = global_config or {} + + # Get display dimensions + if hasattr(display_manager, 'matrix') and display_manager.matrix is not None: + self.display_width = display_manager.matrix.width + self.display_height = display_manager.matrix.height + else: + self.display_width = getattr(display_manager, "width", 128) + self.display_height = getattr(display_manager, "height", 32) + + # Initialize ScrollHelper + if ScrollHelper: + self.scroll_helper = ScrollHelper( + self.display_width, + self.display_height, + self.logger + ) + # Configure scroll settings + self._configure_scroll_helper() + else: + self.scroll_helper = None + self.logger.error("ScrollHelper not available - scroll mode will not work") + + # Shared logo cache for game renderer + self._logo_cache: Dict[str, Image.Image] = {} + + # Cached GameRenderer instance (created lazily) + self._game_renderer: Optional[GameRenderer] = None + + # League separator icons cache + self._separator_icons: Dict[str, Image.Image] = {} + self._load_separator_icons() + + # Tracking state + self._current_games: List[Dict] = [] + self._current_game_type: str = "" + self._current_leagues: List[str] = [] + self._vegas_content_items: List[Image.Image] = [] + self._is_scrolling = False + self._scroll_start_time: Optional[float] = None + self._last_log_time: float = 0 + self._log_interval: float = 5.0 # Log every 5 seconds + + # Performance tracking + self._frame_count: int = 0 + self._fps_sample_start: float = time.time() + + def _configure_scroll_helper(self) -> None: + """Configure scroll helper with settings from config.""" + if not self.scroll_helper: + return + + # Get global scroll settings, then per-league overrides + scroll_settings = self._get_scroll_settings() + + # Set scroll speed (pixels per second in time-based mode) + scroll_speed = scroll_settings.get("scroll_speed", 50.0) + self.scroll_helper.set_scroll_speed(scroll_speed) + + # Set scroll delay + scroll_delay = scroll_settings.get("scroll_delay", 0.01) + self.scroll_helper.set_scroll_delay(scroll_delay) + + # Honor the global smooth-scrolling FPS target (older cores lack the setter) + target_fps = self.global_config.get('target_fps') or self.global_config.get('scroll_target_fps') + try: + # Coerce before comparing: a malformed global config value + # must degrade to today's scroll_delay pacing, not raise. + target_fps = float(target_fps) if target_fps is not None else None + except (TypeError, ValueError): + target_fps = None + if target_fps: + if hasattr(self.scroll_helper, 'set_target_fps'): + self.scroll_helper.set_target_fps(target_fps) + else: + self.scroll_helper.target_fps = max(30.0, min(200.0, target_fps)) + self.scroll_helper.frame_time_target = 1.0 / self.scroll_helper.target_fps + + # Enable dynamic duration + dynamic_duration = scroll_settings.get("dynamic_duration", True) + self.scroll_helper.set_dynamic_duration_settings( + enabled=dynamic_duration, + min_duration=30, + max_duration=600, # 10 minutes max + buffer=0.2 # 20% buffer to ensure scroll completes fully off screen + ) + + # Use frame-based scrolling for better FPS control + self.scroll_helper.set_frame_based_scrolling(True) + + # Convert scroll_speed to pixels/frame, handling both interpretations: + # - scroll_speed as pixels/second: multiply by scroll_delay to get pixels/frame + # - scroll_speed as pixels/frame: use directly + # Pick the interpretation that yields a reasonable value (0.1-5.0 range) + valid_range = (0.1, 5.0) + candidate_pps = scroll_speed * scroll_delay # pixels/sec interpretation + + if valid_range[0] <= candidate_pps <= valid_range[1]: + # pixels/second interpretation yields valid pixels/frame + pixels_per_frame = candidate_pps + elif valid_range[0] <= scroll_speed <= valid_range[1]: + # scroll_speed is already a valid pixels/frame value + pixels_per_frame = scroll_speed + else: + # Neither interpretation is valid, use pixels/sec and clamp + pixels_per_frame = candidate_pps + + # Clamp to reasonable range (0.1 to 5 pixels per frame for smooth scrolling) + pixels_per_frame = max(valid_range[0], min(valid_range[1], pixels_per_frame)) + self.scroll_helper.set_scroll_speed(pixels_per_frame) + + # Calculate effective pixels per second for logging + effective_pps = pixels_per_frame / scroll_delay if scroll_delay > 0 else pixels_per_frame * 100 + + self.logger.info( + f"ScrollHelper configured: {pixels_per_frame:.2f} px/frame, delay={scroll_delay}s " + f"(effective {effective_pps:.1f} px/s), dynamic_duration={dynamic_duration}" + ) + + def _get_scroll_settings(self, league: Optional[str] = None) -> Dict[str, Any]: + """Get scroll settings, optionally for a specific league.""" + # Default scroll settings + defaults = { + "scroll_speed": 50.0, + "scroll_delay": 0.01, + "gap_between_games": 48, + "show_league_separators": True, + "dynamic_duration": True, + "game_card_width": self.display_width, + } + + # Try to get league-specific settings first + if league: + league_config = self.config.get(league, {}) + league_scroll = league_config.get("scroll_settings", {}) + if league_scroll: + return {**defaults, **league_scroll} + + # Fall back to MLB settings (usually first enabled) + mlb_config = self.config.get("mlb", {}) + mlb_scroll = mlb_config.get("scroll_settings", {}) + if mlb_scroll: + return {**defaults, **mlb_scroll} + + # Fall back to MiLB settings + milb_config = self.config.get("milb", {}) + milb_scroll = milb_config.get("scroll_settings", {}) + if milb_scroll: + return {**defaults, **milb_scroll} + + # Fall back to NCAA Baseball settings + ncaa_config = self.config.get("ncaa_baseball", {}) + ncaa_scroll = ncaa_config.get("scroll_settings", {}) + if ncaa_scroll: + return {**defaults, **ncaa_scroll} + + return defaults + + def _get_game_renderer(self, game_card_width: int = 128) -> Optional[GameRenderer]: + """Get or create the cached GameRenderer instance. + + Args: + game_card_width: Width for each game card. Cached renderer is recreated + if this differs from the current renderer's width. + """ + if GameRenderer is None: + self.logger.error("GameRenderer not available") + return None + + # Recreate renderer if card width changed (e.g. config update) + if self._game_renderer is None or getattr(self._game_renderer, "display_width", None) != game_card_width: + self._game_renderer = GameRenderer( + game_card_width, + self.display_height, + self.config, + logo_cache=self._logo_cache, + custom_logger=self.logger + ) + return self._game_renderer + + def _load_separator_icon(self, icon_path: str, league_key: str, target_height: int) -> None: + """ + Load and resize a single league separator icon. + + Args: + icon_path: Path to the icon file + league_key: Key to store the icon under in _separator_icons + target_height: Target height for the resized icon + """ + if not os.path.exists(icon_path): + self.logger.warning(f"{league_key.upper()} separator icon not found at {icon_path}") + return + + try: + with Image.open(icon_path) as icon: + if icon.mode != "RGBA": + icon = icon.convert("RGBA") + # Resize to fit height while maintaining aspect ratio + aspect = icon.width / icon.height + new_width = int(target_height * aspect) + icon = icon.resize((new_width, target_height), resample=RESAMPLE_FILTER) + self._separator_icons[league_key] = icon.copy() + self.logger.debug(f"Loaded {league_key.upper()} separator icon: {new_width}x{target_height}") + except OSError: + self.logger.exception(f"Error loading {league_key.upper()} separator icon") + + def _load_separator_icons(self) -> None: + """Load and resize league separator icons.""" + separator_height = self.display_height - 4 # Leave some padding + + # Load all league separator icons + icons_to_load = [ + (self.MLB_SEPARATOR_ICON, "mlb"), + (self.MILB_SEPARATOR_ICON, "milb"), + (self.NCAA_BASEBALL_SEPARATOR_ICON, "ncaa_baseball"), + ] + for icon_path, league_key in icons_to_load: + self._load_separator_icon(icon_path, league_key, separator_height) + + def _determine_game_type(self, game: Dict) -> str: + """ + Determine the game type from the game's status. + + Args: + game: Game dictionary + + Returns: + Game type: 'live', 'recent', or 'upcoming' + """ + if game.get('is_live'): + return 'live' + elif game.get('is_final'): + return 'recent' + elif game.get('is_upcoming'): + return 'upcoming' + else: + # Default to upcoming if state is unknown + return 'upcoming' + + def prepare_scroll_content( + self, + games: List[Dict], + game_type: str, + leagues: List[str], + rankings_cache: Optional[Dict[str, int]] = None + ) -> bool: + """ + Prepare scrolling content from a list of games. + + Args: + games: List of game dictionaries with league info + game_type: Type hint ('live', 'recent', 'upcoming', or 'mixed' for mixed types) + leagues: List of leagues in order (e.g., ['mlb', 'milb', 'ncaa_baseball']) + rankings_cache: Optional team rankings cache for displaying team rankings + + Returns: + True if content was prepared successfully, False otherwise + """ + if not self.scroll_helper: + self.logger.error("ScrollHelper not available") + return False + + if not games: + self.logger.debug("No games to prepare for scrolling") + self.scroll_helper.clear_cache() + self._current_games = [] + self._vegas_content_items = [] + self._is_scrolling = False + return False + + self._current_games = games + self._current_game_type = game_type + self._current_leagues = leagues + + # Get scroll settings + scroll_settings = self._get_scroll_settings() + gap_between_games = scroll_settings.get("gap_between_games", 24) + show_separators = scroll_settings.get("show_league_separators", True) + game_card_width = scroll_settings.get("game_card_width", self.display_width) + + # Get or create cached game renderer; default card width is the full display width + # so each game card fills the viewport and logos sit at the display edges + renderer = self._get_game_renderer(game_card_width) + + # Pass rankings cache to renderer if available + if renderer and rankings_cache: + renderer.set_rankings_cache(rankings_cache) + + # Pre-render all game cards + content_items: List[Image.Image] = [] + current_league = None + game_count = 0 + league_counts: Dict[str, int] = {} + + for game in games: + game_league = game.get("league", "mlb") # Default to MLB if not specified + + # Add league separator when entering a new league (first or switching) + if show_separators and game_league != current_league: + separator = self._separator_icons.get(game_league) + if separator: + # Create a separator image with proper background + sep_img = Image.new('RGB', (separator.width + 8, self.display_height), (0, 0, 0)) + # Center the separator vertically + y_offset = (self.display_height - separator.height) // 2 + sep_img.paste(separator, (4, y_offset), separator) + content_items.append(sep_img) + context = "at start" if current_league is None else "" + self.logger.debug(f"Added {game_league} separator icon {context}".strip()) + + current_league = game_league + + # Render game card - determine type from game state + try: + individual_game_type = self._determine_game_type(game) + game_img = renderer.render_game_card(game, individual_game_type) + + # Only pad when card is narrower than the viewport; full-width cards + # need no padding or the card becomes wider than the display. + padding = 0 if game_img.width >= self.display_width else 12 + padded_width = game_img.width + (padding * 2) + padded_img = Image.new('RGB', (padded_width, game_img.height), (0, 0, 0)) + padded_img.paste(game_img, (padding, 0)) + + content_items.append(padded_img) + game_count += 1 + league_counts[game_league] = league_counts.get(game_league, 0) + 1 + except Exception: + self.logger.exception("Error rendering game card") + continue + + if not content_items: + self.logger.warning("No game cards rendered") + return False + + # Store individual items for Vegas mode (avoids scroll_helper padding) + self._vegas_content_items = list(content_items) + + # Create scrolling image using ScrollHelper + self.scroll_helper.create_scrolling_image( + content_items, + item_gap=gap_between_games, + element_gap=0 # No element gap - each item is a complete game card + ) + + # Log what we loaded + league_summary = ", ".join([f"{league.upper()}({count})" for league, count in league_counts.items()]) + self.logger.info( + f"[Baseball Scroll] Prepared {game_count} games for scrolling: {league_summary}" + ) + self.logger.info( + f"[Baseball Scroll] Total scroll width: {self.scroll_helper.total_scroll_width}px, " + f"Dynamic duration: {self.scroll_helper.calculated_duration}s" + ) + + # Reset tracking state + self._is_scrolling = True + self._scroll_start_time = time.time() + self._frame_count = 0 + self._fps_sample_start = time.time() + + return True + + def display_scroll_frame(self) -> bool: + """ + Display the next frame of the scrolling content. + + Returns: + True if a frame was displayed, False if scroll is complete or no content + """ + if not self.scroll_helper or not self.scroll_helper.cached_image: + return False + + # Update scroll position + self.scroll_helper.update_scroll_position() + + # Get visible portion + visible = self.scroll_helper.get_visible_portion() + if not visible: + return False + + # Display the visible portion + try: + self.display_manager.image = visible + self.display_manager.update_display() + + # Track frame rate + self._frame_count += 1 + self.scroll_helper.log_frame_rate() + + # Periodic logging + self._log_scroll_progress() + + return True + except Exception: + self.logger.exception("Error displaying scroll frame") + return False + + def _log_scroll_progress(self) -> None: + """Log scroll progress and FPS periodically.""" + current_time = time.time() + + if current_time - self._last_log_time >= self._log_interval: + # Calculate FPS + elapsed = current_time - self._fps_sample_start + if elapsed > 0: + fps = self._frame_count / elapsed + + # Get scroll info + scroll_info = self.scroll_helper.get_scroll_info() + + self.logger.info( + f"[Baseball Scroll] FPS: {fps:.1f}, " + f"Position: {scroll_info['scroll_position']:.0f}/{scroll_info['total_width']}px, " + f"Elapsed: {scroll_info.get('elapsed_time', 0):.1f}s/{scroll_info['dynamic_duration']}s" + ) + + # Reset FPS tracking + self._frame_count = 0 + self._fps_sample_start = current_time + self._last_log_time = current_time + + def is_scroll_complete(self) -> bool: + """Check if the scroll cycle is complete.""" + if not self.scroll_helper: + return True + return self.scroll_helper.is_scroll_complete() + + def reset_scroll(self) -> None: + """Reset the scroll position to the beginning.""" + if self.scroll_helper: + self.scroll_helper.reset_scroll() + self._frame_count = 0 + self._fps_sample_start = time.time() + self.logger.debug("Scroll position reset") + + def get_scroll_info(self) -> Dict[str, Any]: + """Get current scroll state information.""" + if not self.scroll_helper: + return {"error": "ScrollHelper not available"} + + info = self.scroll_helper.get_scroll_info() + info.update({ + "game_count": len(self._current_games), + "game_type": self._current_game_type, + "leagues": self._current_leagues, + "is_scrolling": self._is_scrolling + }) + return info + + def get_dynamic_duration(self) -> int: + """Get the calculated dynamic duration for this scroll content.""" + if self.scroll_helper: + return self.scroll_helper.get_dynamic_duration() + return 60 # Default fallback + + def clear(self) -> None: + """Clear scroll content and reset state.""" + if self.scroll_helper: + self.scroll_helper.clear_cache() + self._current_games = [] + self._current_game_type = "" + self._current_leagues = [] + self._vegas_content_items = [] + self._is_scrolling = False + self._scroll_start_time = None + self.logger.debug("Scroll display cleared") + + +class LegacyScrollDisplayManager: + """ + Manages scroll display instances for different game types. + + This class provides a higher-level interface for the baseball plugin + to manage scroll displays for live, recent, and upcoming games. + """ + + def __init__( + self, + display_manager, + config: Dict[str, Any], + custom_logger: Optional[logging.Logger] = None, + global_config: Optional[Dict[str, Any]] = None + ): + """ + Initialize the ScrollDisplayManager. + + Args: + display_manager: Display manager instance + config: Plugin configuration dictionary + custom_logger: Optional custom logger instance + global_config: Optional global LEDMatrix config (for target_fps) + """ + self.display_manager = display_manager + self.config = config + self.logger = custom_logger or logger + self.global_config = global_config or {} + + # Create scroll displays for each game type + self._scroll_displays: Dict[str, 'LegacyScrollDisplay'] = {} + self._current_game_type: Optional[str] = None + + def get_scroll_display(self, game_type: str) -> 'LegacyScrollDisplay': + """ + Get or create a scroll display for a game type. + + Args: + game_type: Type of games ('live', 'recent', 'upcoming') + + Returns: + ScrollDisplay instance for the game type + """ + if game_type not in self._scroll_displays: + self._scroll_displays[game_type] = LegacyScrollDisplay( + self.display_manager, + self.config, + self.logger, + global_config=self.global_config + ) + return self._scroll_displays[game_type] + + def prepare_and_display( + self, + games: List[Dict], + game_type: str, + leagues: List[str], + rankings_cache: Dict[str, int] = None + ) -> bool: + """ + Prepare content and start displaying scroll. + + Args: + games: List of game dictionaries + game_type: Type of games + leagues: List of leagues + rankings_cache: Optional team rankings cache + + Returns: + True if scroll was started successfully + """ + success = self.prepare_content(games, game_type, leagues, rankings_cache) + + if success: + self._current_game_type = game_type + + return success + + def prepare_content( + self, + games: List[Dict], + game_type: str, + leagues: List[str], + rankings_cache: Dict[str, int] = None + ) -> bool: + """ + Render content for one scroll display without making it the active one. + + Vegas mode builds its own combined slate in the background while the + standalone rotation may be mid-scroll on a different game type. Going + through prepare_and_display() for that would repoint + ``_current_game_type``, so the next display_frame() would render the + Vegas slate instead of the mode the rotation is actually showing. + + Args: + games: List of game dictionaries + game_type: Scroll display key to render into + leagues: List of leagues + rankings_cache: Optional team rankings cache + + Returns: + True if content was prepared successfully + """ + scroll_display = self.get_scroll_display(game_type) + return scroll_display.prepare_scroll_content( + games, game_type, leagues, rankings_cache + ) + + def display_frame(self, game_type: Optional[str] = None) -> bool: + """ + Display the next frame of the current scroll. + + Args: + game_type: Optional game type (uses current if not specified) + + Returns: + True if a frame was displayed + """ + if game_type is None: + game_type = self._current_game_type + + if game_type is None: + return False + + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return False + + return scroll_display.display_scroll_frame() + + def is_complete(self, game_type: Optional[str] = None) -> bool: + """Check if the current scroll is complete.""" + if game_type is None: + game_type = self._current_game_type + + if game_type is None: + return True + + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return True + + return scroll_display.is_scroll_complete() + + def get_dynamic_duration(self, game_type: Optional[str] = None) -> int: + """Get the dynamic duration for the current scroll.""" + if game_type is None: + game_type = self._current_game_type + + if game_type is None: + return 60 + + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return 60 + + return scroll_display.get_dynamic_duration() + + def clear_all(self) -> None: + """Clear all scroll displays.""" + for scroll_display in self._scroll_displays.values(): + scroll_display.clear() + self._current_game_type = None + + def has_cached_content(self) -> bool: + """ + Check if any scroll display has cached content. + + Returns: + True if any scroll display has a cached image, False otherwise + """ + for scroll_display in self._scroll_displays.values(): + if hasattr(scroll_display, 'scroll_helper') and scroll_display.scroll_helper: + if scroll_display.scroll_helper.cached_image is not None: + return True + return False + + def get_all_vegas_content_items(self) -> list: + """Collect _vegas_content_items from all scroll displays.""" + items = [] + for sd in self._scroll_displays.values(): + vegas_items = getattr(sd, '_vegas_content_items', None) + if vegas_items: + items.extend(vegas_items) + return items + + def get_vegas_content_items_for(self, game_type: str) -> list: + """ + Return the Vegas item list for a single scroll display. + + Vegas mode needs the items from one specific display (the combined + live/recent/upcoming set), not the union across all of them. + get_all_vegas_content_items() returns whatever the standalone display + modes happen to have rendered, which both under-reports (only the last + rendered mode's games) and can double-count a game that appears in two + displays. + + Args: + game_type: Scroll display key, e.g. 'mixed' + + Returns: + Copy of that display's Vegas items, or an empty list if absent. + """ + scroll_display = self._scroll_displays.get(game_type) + if scroll_display is None: + return [] + return list(getattr(scroll_display, '_vegas_content_items', None) or []) diff --git a/plugins/baseball-scoreboard/test_core_fallback.py b/plugins/baseball-scoreboard/test_core_fallback.py new file mode 100644 index 00000000..ce369154 --- /dev/null +++ b/plugins/baseball-scoreboard/test_core_fallback.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +"""The guarded core import must behave on every core this plugin can meet. + +Adopting core code (B5) is only safe because `scroll_display.py` falls back to +`scroll_display_legacy.py` when the core lacks `src.common.sports_scroll`. +Nothing verified that claim — the safety harness renders against the *current* +core, so it exercises exactly one of the cases below. + +That gap matters because the version floor does not close it. A user who +installed from the v3.1.0 release reports `__version__ = "1.0.0"`, which the +install gate treats as untrustworthy and lets through, and that core has no +`src.common.sports_scroll`. The fallback is the only thing standing between +them and a scoreboard that fails to load. + + core module bundled copy expected + ----------------- ---------------- ------------------------------------ + present present core implementation (today) + absent present legacy implementation, fully working + absent absent the B6 state: fails, and names the + exact missing module + +The third row is not a supported configuration — it is what the sunset looks +like on a core that is too old. It is asserted so that when the fallback is +finally deleted, the failure is specific and attributable rather than a vague +crash, and so nobody deletes the fallback without meeting it. + +Run: /bin/python plugins/baseball-scoreboard/test_core_fallback.py +""" + +import importlib +import os +import sys + +PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__)) +if PLUGIN_DIR not in sys.path: + sys.path.insert(0, PLUGIN_DIR) + +CORE_MODULE = "src.common.sports_scroll" + + +class _BlockModules: + """Make named modules un-importable, simulating an older core. + + A meta-path finder is used rather than deleting files: it is reversible, + leaves the checkout untouched, and reproduces exactly what Python does when + the module genuinely is not there — `ModuleNotFoundError` with `.name` set. + """ + + def __init__(self, *names): + self.names = set(names) + self._saved = {} + + def find_module(self, fullname, path=None): # pragma: no cover - legacy API + return self if fullname in self.names else None + + def find_spec(self, fullname, path=None, target=None): + if fullname in self.names: + raise ModuleNotFoundError(f"No module named {fullname!r}", name=fullname) + return None + + def __enter__(self): + for name in list(sys.modules): + if name in self.names or name.startswith("scroll_display"): + self._saved[name] = sys.modules.pop(name) + sys.meta_path.insert(0, self) + return self + + def __exit__(self, *exc): + sys.meta_path.remove(self) + for name in list(sys.modules): + if name.startswith("scroll_display"): + del sys.modules[name] + sys.modules.update(self._saved) + return False + + +def _fresh_scroll_display(): + for name in list(sys.modules): + if name.startswith("scroll_display"): + del sys.modules[name] + return importlib.import_module("scroll_display") + + +def test_core_present_uses_core(): + mod = _fresh_scroll_display() + assert mod._USING_CORE_SCROLL is True, ( + "core ships src.common.sports_scroll, so the plugin should be using it" + ) + bases = [c.__name__ for c in mod.ScrollDisplay.__mro__] + assert "SportsScrollDisplay" in bases, bases + assert mod.ScrollDisplayManager.display_class is mod.ScrollDisplay + + +def test_core_absent_falls_back_and_still_works(): + """The B5 guarantee. This is the case the harness cannot reach.""" + with _BlockModules(CORE_MODULE): + mod = _fresh_scroll_display() + + assert mod._USING_CORE_SCROLL is False, ( + "with the core module gone the plugin must fall back, not import it" + ) + assert mod.ScrollDisplay.__name__ == "LegacyScrollDisplay", ( + f"expected the bundled implementation, got {mod.ScrollDisplay.__name__}" + ) + # A fallback that loads but cannot draw is no fallback at all. + for method in ("prepare_scroll_content", "display_scroll_frame", + "is_scroll_complete", "get_dynamic_duration", + "_load_separator_icons"): + assert hasattr(mod.ScrollDisplay, method), f"fallback lost {method}()" + assert hasattr(mod.ScrollDisplayManager, "prepare_and_display") + + +def test_sunset_state_fails_specifically(): + """The B6 state: no core module and no bundled copy. + + Not supported — asserted so the failure names the missing module rather + than surfacing as something unattributable. + """ + with _BlockModules(CORE_MODULE, "scroll_display_legacy"): + for name in list(sys.modules): + if name.startswith("scroll_display"): + del sys.modules[name] + try: + importlib.import_module("scroll_display") + except ModuleNotFoundError as exc: + assert exc.name in {CORE_MODULE, "scroll_display_legacy"}, ( + f"failure should name the missing module, got {exc.name!r}" + ) + else: + raise AssertionError( + "expected ModuleNotFoundError with neither implementation present" + ) + + + +def _unresolvable_globals(cls, module): + """Globals a class's own methods read that nothing can resolve. + + Walks each method's AST for `Name` loads rather than its bytecode: the + bytecode's co_names mixes in attribute names, so `Image.Resampling.LANCZOS` + looked like a missing global. Locals, arguments and comprehension targets + are excluded, leaving only names Python would resolve globally. + + Reading the source rather than calling the method is deliberate -- building + a real display needs a display manager, fonts and assets, but an + unresolvable global is a load-time fact and needs none of that. `hasattr` + could not see this at all: the method exists; what it reaches for does not. + """ + import ast + import builtins + import inspect + import textwrap + import types + + # Resolve against the module the CLASS lives in, not the one we imported. + # On the fallback path ScrollDisplay is LegacyScrollDisplay, whose globals + # are scroll_display_legacy's -- checking scroll_display's namespace made + # every fallback look broken. + import sys as _sys + module = _sys.modules.get(cls.__module__, module) + + try: + tree = ast.parse(textwrap.dedent(inspect.getsource(cls))) + except (OSError, TypeError): # pragma: no cover - source always available here + return [] + + missing = set() + for node in ast.walk(tree): + if not isinstance(node, ast.FunctionDef): + continue + bound = {a.arg for a in node.args.args + node.args.kwonlyargs} + if node.args.vararg: + bound.add(node.args.vararg.arg) + if node.args.kwarg: + bound.add(node.args.kwarg.arg) + for sub in ast.walk(node): + if isinstance(sub, ast.Name) and isinstance(sub.ctx, (ast.Store,)): + bound.add(sub.id) + elif isinstance(sub, (ast.Import, ast.ImportFrom)): + for alias in sub.names: + bound.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(sub, ast.ExceptHandler) and sub.name: + bound.add(sub.name) + for sub in ast.walk(node): + if isinstance(sub, ast.Name) and isinstance(sub.ctx, ast.Load): + name = sub.id + if (name in bound or hasattr(module, name) or hasattr(cls, name) + or hasattr(builtins, name)): + continue + missing.add(name) + return sorted(missing) + + +def test_content_methods_can_resolve_what_they_use(): + """The core path must be able to draw, not merely import. + + `_load_separator_icons` and `prepare_scroll_content` were lifted out of the + legacy module; their dependencies were not. Nothing caught it: the safety + harness renders the scoreboard screens rather than scroll mode, and the + earlier version of this file only checked that method names existed. + """ + mod = _fresh_scroll_display() + for cls in (mod.ScrollDisplay, mod.ScrollDisplayManager): + missing = _unresolvable_globals(cls, mod) + assert not missing, ( + f"{cls.__name__} methods reference {missing}, which their module " + f"cannot resolve — they raise NameError on the core path" + ) + + +def test_fallback_content_methods_can_resolve_what_they_use(): + """Same check on the bundled implementation.""" + with _BlockModules(CORE_MODULE): + mod = _fresh_scroll_display() + for cls in (mod.ScrollDisplay, mod.ScrollDisplayManager): + missing = _unresolvable_globals(cls, mod) + assert not missing, ( + f"the fallback's {cls.__name__} references {missing}, which " + f"its module cannot resolve" + ) + +if __name__ == "__main__": + # Pre-flight, deliberately BEFORE any test runs. Deciding "skip" from an + # exception raised *during* a test is what this suite is guarding against: + # when the fallback was removed, the escaping ModuleNotFoundError named the + # core module and an in-test skip handler swallowed it as "no core on + # PYTHONPATH" -- hiding the exact regression this file exists to catch. + # Once we know the core is importable, any ModuleNotFoundError from here on + # is a real failure. + try: + importlib.import_module(CORE_MODULE) + except ModuleNotFoundError as exc: + print(f"SKIP: needs a LEDMatrix core with {CORE_MODULE} on PYTHONPATH " + f"({exc})") + sys.exit(2) + + print("guarded core-import fallback tests") + print("=" * 55) + failures = [] + for t in (test_core_present_uses_core, + test_core_absent_falls_back_and_still_works, + test_content_methods_can_resolve_what_they_use, + test_fallback_content_methods_can_resolve_what_they_use, + test_sunset_state_fails_specifically): + try: + t() + print(f"PASS {t.__name__}") + except (AssertionError, ModuleNotFoundError) as e: + failures.append(t.__name__) + print(f"FAIL {t.__name__}: {e}") + print("=" * 55) + if failures: + print(f"{len(failures)} test(s) failed: {failures}") + sys.exit(1) + print("All tests passed.")