diff --git a/docs/HOW_TO_RUN_TESTS.md b/docs/HOW_TO_RUN_TESTS.md index 828f52cc..5dde96ec 100644 --- a/docs/HOW_TO_RUN_TESTS.md +++ b/docs/HOW_TO_RUN_TESTS.md @@ -248,7 +248,6 @@ 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 diff --git a/requirements.txt b/requirements.txt index 5e1bf365..9725d6c3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,16 +8,11 @@ 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 @@ -29,10 +24,8 @@ 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 diff --git a/scripts/dev/test_pillow_compat.py b/scripts/dev/test_pillow_compat.py index 590d632c..679b8c3a 100755 --- a/scripts/dev/test_pillow_compat.py +++ b/scripts/dev/test_pillow_compat.py @@ -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, image_utils, sports base):") + print("\nResampling (used in logo_helper, sports base):") logo = Image.new('RGBA', (200, 200), (255, 128, 0, 200)) failures += not check("Image.Resampling.LANCZOS exists", lambda: str(Image.Resampling.LANCZOS)) diff --git a/src/background_cache_mixin.py b/src/background_cache_mixin.py deleted file mode 100644 index 623a5313..00000000 --- a/src/background_cache_mixin.py +++ /dev/null @@ -1,134 +0,0 @@ -""" -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() diff --git a/src/common/basketball_plugin_example.py b/src/common/basketball_plugin_example.py deleted file mode 100644 index 9921a2fe..00000000 --- a/src/common/basketball_plugin_example.py +++ /dev/null @@ -1,328 +0,0 @@ -""" -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. -""" diff --git a/src/font_test_manager.py b/src/font_test_manager.py deleted file mode 100644 index 8869ebbd..00000000 --- a/src/font_test_manager.py +++ /dev/null @@ -1,135 +0,0 @@ -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)) \ No newline at end of file diff --git a/src/generic_cache_mixin.py b/src/generic_cache_mixin.py deleted file mode 100644 index 3c727eb0..00000000 --- a/src/generic_cache_mixin.py +++ /dev/null @@ -1,150 +0,0 @@ -""" -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 - } diff --git a/src/image_utils.py b/src/image_utils.py deleted file mode 100644 index 6179f9bf..00000000 --- a/src/image_utils.py +++ /dev/null @@ -1,22 +0,0 @@ -"""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 diff --git a/src/layout_manager.py b/src/layout_manager.py deleted file mode 100644 index 1d852fc2..00000000 --- a/src/layout_manager.py +++ /dev/null @@ -1,409 +0,0 @@ -""" -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 \ No newline at end of file diff --git a/test/test_layout_manager.py b/test/test_layout_manager.py deleted file mode 100644 index c29d2fe4..00000000 --- a/test/test_layout_manager.py +++ /dev/null @@ -1,392 +0,0 @@ -""" -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()