diff --git a/requirements.txt b/requirements.txt index ffacb9d1..f796095a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,4 +14,5 @@ icalevents python-socketio python-engineio websockets -websocket-client \ No newline at end of file +websocket-client +pydantic>=2.11.0 \ No newline at end of file diff --git a/src/base_classes/football.py b/src/base_classes/football.py index cf67ca6c..d73d7861 100644 --- a/src/base_classes/football.py +++ b/src/base_classes/football.py @@ -6,6 +6,7 @@ from PIL import Image, ImageDraw, ImageFont import time import pytz +from src.config.config_models import RootConfig from src.base_classes.sports import SportsCore from src.base_classes.api_extractors import ESPNFootballExtractor from src.base_classes.data_sources import ESPNDataSource @@ -30,7 +31,7 @@ class Football(SportsCore): 'api_base_url': 'https://site.api.espn.com/apis/site/v2/sports/football' } - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): super().__init__(config, display_manager, cache_manager, logger, sport_key) # Initialize football-specific architecture components @@ -167,15 +168,15 @@ def _get_weeks_data(self, league: str) -> Optional[Dict]: return super()._get_weeks_data("football", league) class FootballLive(Football): - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): super().__init__(config, display_manager, cache_manager, logger, sport_key) - self.update_interval = self.mode_config.get("live_update_interval", 15) + self.update_interval = self.mode_config.live_update_interval self.no_data_interval = 300 self.last_update = 0 self.live_games = [] self.current_game_index = 0 self.last_game_switch = 0 - self.game_display_duration = self.mode_config.get("live_game_duration", 20) + self.game_display_duration = self.mode_config.live_game_duration self.last_display_update = 0 self.last_log_time = 0 self.log_interval = 300 @@ -255,7 +256,7 @@ def update(self): if details and (details["is_live"] or details["is_halftime"]): # If show_favorite_teams_only is true, only add if it's a favorite. # Otherwise, add all games. - if self.mode_config.get("show_favorite_teams_only", False): + if self.mode_config.show_favorite_teams_only: if details["home_abbr"] in self.favorite_teams or details["away_abbr"] in self.favorite_teams: if self.show_odds: self._fetch_game_odds(details) @@ -276,12 +277,12 @@ def update(self): if should_log: if new_live_games: - filter_text = "favorite teams" if self.mode_config.get("show_favorite_teams_only", False) else "all teams" + filter_text = "favorite teams" if self.mode_config.show_favorite_teams_only else "all teams" self.logger.info(f"Found {len(new_live_games)} live/halftime games for {filter_text}.") for game_info in new_live_games: # Renamed game to game_info self.logger.info(f" - {game_info['away_abbr']}@{game_info['home_abbr']} ({game_info.get('status_text', 'N/A')})") else: - filter_text = "favorite teams" if self.mode_config.get("show_favorite_teams_only", False) else "criteria" + filter_text = "favorite teams" if self.mode_config.show_favorite_teams_only else "criteria" self.logger.info(f"No live/halftime games found for {filter_text}.") self.last_log_time = current_time_for_log diff --git a/src/base_classes/sports.py b/src/base_classes/sports.py index 16bf8aad..357103c5 100644 --- a/src/base_classes/sports.py +++ b/src/base_classes/sports.py @@ -16,12 +16,13 @@ from pathlib import Path # Import new architecture components (individual classes will import what they need) -from .api_extractors import ESPNFootballExtractor, ESPNBaseballExtractor, ESPNHockeyExtractor -from .data_sources import ESPNDataSource, MLBAPIDataSource +from src.base_classes.api_extractors import ESPNFootballExtractor, ESPNBaseballExtractor, ESPNHockeyExtractor +from src.base_classes.data_sources import ESPNDataSource, MLBAPIDataSource from src.dynamic_team_resolver import DynamicTeamResolver +from src.config.config_models import RootConfig, ScoreboardBaseConfig class SportsCore: - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): self.logger = logger self.config = config self.cache_manager = cache_manager @@ -38,20 +39,17 @@ def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cach self.sport_config = None self.api_extractor = None self.data_source = None - self.mode_config = config.get(f"{sport_key}_scoreboard", {}) # Changed config key - self.is_enabled = self.mode_config.get("enabled", False) - self.show_odds = self.mode_config.get("show_odds", False) - self.test_mode = self.mode_config.get("test_mode", False) - self.logo_dir = Path(self.mode_config.get("logo_dir", "assets/sports/ncaa_logos")) # Changed logo dir - self.update_interval = self.mode_config.get( - "update_interval_seconds", 60) - self.show_records = self.mode_config.get('show_records', False) - self.show_ranking = self.mode_config.get('show_ranking', False) + self.mode_config: ScoreboardBaseConfig = config.__getattribute__(f"{sport_key}_scoreboard") # Changed config key + self.is_enabled = self.mode_config.enabled + self.show_odds = self.mode_config.show_odds + self.test_mode = self.mode_config.test_mode + self.logo_dir = Path(self.mode_config.logo_dir) # Changed logo dir + self.update_interval = self.mode_config.update_interval_seconds + self.show_records = self.mode_config.show_records + self.show_ranking = self.mode_config.show_ranking # Number of games to show (instead of time-based windows) - self.recent_games_to_show = self.mode_config.get( - "recent_games_to_show", 5) # Show last 5 games - self.upcoming_games_to_show = self.mode_config.get( - "upcoming_games_to_show", 10) # Show next 10 games + self.recent_games_to_show = self.mode_config.recent_games_to_show + self.upcoming_games_to_show = self.mode_config.upcoming_games_to_show self.session = requests.Session() retry_strategy = Retry( @@ -81,7 +79,7 @@ def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cach # Initialize dynamic team resolver and resolve favorite teams self.dynamic_resolver = DynamicTeamResolver() - raw_favorite_teams = self.mode_config.get("favorite_teams", []) + raw_favorite_teams = self.mode_config.favorite_teams self.favorite_teams = self.dynamic_resolver.resolve_teams(raw_favorite_teams, sport_key) # Log dynamic team resolution @@ -98,9 +96,9 @@ def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cach self._rankings_cache_duration = 3600 # Cache rankings for 1 hour # Initialize background data service - background_config = self.mode_config.get("background_service", {}) - if background_config.get("enabled", True): # Default to enabled - max_workers = background_config.get("max_workers", 3) + background_config = self.mode_config.background_service + if background_config and background_config.enabled: # Default to enabled + max_workers = background_config.max_workers self.background_service = get_background_service(self.cache_manager, max_workers) self.background_fetch_requests = {} # Track background fetch requests self.background_enabled = True @@ -412,7 +410,7 @@ def _fetch_game_odds(self, game: Dict) -> None: return # Check if we should only fetch for favorite teams - is_favorites_only = self.mode_config.get("show_favorite_teams_only", False) + is_favorites_only = self.mode_config.show_favorite_teams_only if is_favorites_only: home_abbr = game.get('home_abbr') away_abbr = game.get('away_abbr') @@ -449,7 +447,7 @@ def _fetch_odds(self, game: Dict, sport: str, league: str) -> None: return # Check if we should only fetch for favorite teams - is_favorites_only = self.mode_config.get("show_favorite_teams_only", False) + is_favorites_only = self.mode_config.gshow_favorite_teams_only if is_favorites_only: home_abbr = game.get('home_abbr') away_abbr = game.get('away_abbr') @@ -463,8 +461,8 @@ def _fetch_odds(self, game: Dict, sport: str, league: str) -> None: try: # Determine update interval based on game state is_live = game.get('status', '').lower() == 'in' - update_interval = self.mode_config.get("live_odds_update_interval", 60) if is_live \ - else self.mode_config.get("odds_update_interval", 3600) + update_interval = self.mode_config.live_odds_update_interval if is_live \ + else self.mode_config.odds_update_interval odds_data = self.odds_manager.get_odds( sport=sport, @@ -485,7 +483,7 @@ def _fetch_odds(self, game: Dict, sport: str, league: str) -> None: def _get_timezone(self): try: - timezone_str = self.config.get('timezone', 'UTC') + timezone_str = self.config.timezone return pytz.timezone(timezone_str) except pytz.UnknownTimeZoneError: return pytz.utc @@ -559,7 +557,7 @@ def _extract_game_details_common(self, game_event: Dict) -> tuple[Dict | None, D game_time = local_time.strftime("%I:%M%p").lstrip('0') # Check date format from config - use_short_date_format = self.config.get('display', {}).get('use_short_date_format', False) + use_short_date_format = self.config.display.use_short_date_format if use_short_date_format: game_date = local_time.strftime("%-m/%-d") else: @@ -663,13 +661,13 @@ def _get_weeks_data(self, sport: str, league: str) -> Optional[Dict]: return None class SportsUpcoming(SportsCore): - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): super().__init__(config, display_manager, cache_manager, logger, sport_key) self.upcoming_games = [] # Store all fetched upcoming games initially self.games_list = [] # Filtered list for display (favorite teams) self.current_game_index = 0 self.last_update = 0 - self.update_interval = self.mode_config.get("upcoming_update_interval", 3600) # Check for recent games every hour + self.update_interval = self.mode_config.upcoming_update_interval # Check for recent games every hour self.last_log_time = 0 self.log_interval = 300 self.last_warning_time = 0 @@ -713,7 +711,7 @@ def update(self): # Filter criteria: must be upcoming ('pre' state) if game and game['is_upcoming']: # Only fetch odds for games that will be displayed - if self.mode_config.get("show_favorite_teams_only", False): + if self.mode_config.show_favorite_teams_only: if not self.favorite_teams: continue if game['home_abbr'] not in self.favorite_teams and game['away_abbr'] not in self.favorite_teams: @@ -771,7 +769,7 @@ def update(self): self.logger.info(f"Found {favorite_games_found} favorite team upcoming games") # Filter for favorite teams only if the config is set - if self.mode_config.get("show_favorite_teams_only", False): + if self.mode_config.show_favorite_teams_only: # Get all games involving favorite teams favorite_team_games = [game for game in processed_games if game['home_abbr'] in self.favorite_teams or @@ -1035,13 +1033,13 @@ def display(self, force_clear=False): class SportsRecent(SportsCore): - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): super().__init__(config, display_manager, cache_manager, logger, sport_key) self.recent_games = [] # Store all fetched recent games initially self.games_list = [] # Filtered list for display (favorite teams) self.current_game_index = 0 self.last_update = 0 - self.update_interval = self.mode_config.get("recent_update_interval", 3600) # Check for recent games every hour + self.update_interval = self.mode_config.recent_update_interval # Check for recent games every hour self.last_game_switch = 0 self.game_display_duration = 15 # Display each recent game for 15 seconds diff --git a/src/clock.py b/src/clock.py index de358e27..78fa7990 100644 --- a/src/clock.py +++ b/src/clock.py @@ -5,25 +5,19 @@ from typing import Dict, Any from src.config_manager import ConfigManager from src.display_manager import DisplayManager +from src.config.config_models import RootConfig # Get logger without configuring logger = logging.getLogger(__name__) class Clock: - def __init__(self, display_manager: DisplayManager = None, config: Dict[str, Any] = None): - if config is not None: - # Use provided config - self.config = config - self.config_manager = None # Not needed when config is provided - else: - # Fallback: create ConfigManager and load config (for standalone usage) - self.config_manager = ConfigManager() - self.config = self.config_manager.load_config() + def __init__(self, display_manager: DisplayManager, config: RootConfig): # Use the provided display_manager or create a new one if none provided - self.display_manager = display_manager or DisplayManager(self.config.get('display', {})) + self.config = config + self.display_manager = display_manager logger.info("Clock initialized with display_manager: %s", id(self.display_manager)) - self.location = self.config.get('location', {}) - self.clock_config = self.config.get('clock', {}) + self.location = self.config.location + self.clock_config = self.config.clock # Use configured timezone if available, otherwise try to determine it self.timezone = self._get_timezone() self.last_time = None @@ -37,7 +31,7 @@ def __init__(self, display_manager: DisplayManager = None, config: Dict[str, Any def _get_timezone(self) -> pytz.timezone: """Get timezone from the config file.""" - config_timezone = self.config.get('timezone', 'UTC') + config_timezone = self.config.timezone try: return pytz.timezone(config_timezone) except pytz.exceptions.UnknownTimeZoneError: @@ -137,7 +131,7 @@ def display_time(self, force_clear: bool = False) -> None: try: while True: clock.display_time() - time.sleep(clock.clock_config.get('update_interval', 1)) + time.sleep(clock.clock_config.update_interval) except KeyboardInterrupt: print("\nClock stopped by user") finally: diff --git a/src/config/config_models.py b/src/config/config_models.py new file mode 100644 index 00000000..a2964043 --- /dev/null +++ b/src/config/config_models.py @@ -0,0 +1,981 @@ +import re +from typing import Dict, List, Optional + +from pydantic import BaseModel, Field, field_validator + + +# ------------------------- +# Small utility models +# ------------------------- +class Schedule(BaseModel): + enabled: bool = Field(default=True, description="Whether the schedule is active") + start_time: str = Field( + default="07:00", description="Daily start time in HH:MM format" + ) + end_time: str = Field(default="23:00", description="Daily end time in HH:MM format") + + @field_validator("start_time", "end_time") + def _validate_time_format(cls, v: str) -> str: + if not isinstance(v, str) or not re.match(r"^\d{2}:\d{2}$", v): + raise ValueError("Time must be a string in HH:MM format") + hours, minutes = v.split(":") + h = int(hours) + m = int(minutes) + if not (0 <= h <= 23 and 0 <= m <= 59): + raise ValueError("Hour must be 0-23 and minute 0-59") + return v + + +class Location(BaseModel): + city: str = Field(default="Dallas", description="City of the display location") + state: str = Field(default="Texas", description="State of the display location") + country: str = Field(default="US", description="Country code (ISO 3166-1 alpha-2)") + + +# ------------------------- +# Display models +# ------------------------- +class HardwareConfig(BaseModel): + rows: int = Field(default=32, description="Number of LED matrix rows", ge=1) + cols: int = Field(default=64, description="Number of LED matrix columns", ge=1) + chain_length: int = Field( + default=2, description="Number of daisy-chained panels", ge=1 + ) + parallel: int = Field(default=1, description="Parallel chains supported", ge=1) + brightness: int = Field( + default=95, description="Panel brightness (0–100)", ge=0, le=100 + ) + hardware_mapping: str = Field( + default="adafruit-hat-pwm", description="Hardware mapping mode" + ) + scan_mode: int = Field( + default=0, description="Scan mode (0 = progressive, 1 = interlaced)", ge=0, le=1 + ) + pwm_bits: int = Field( + default=9, description="PWM bits for brightness control", ge=1 + ) + pwm_dither_bits: int = Field(default=1, description="Dithering bits for PWM", ge=0) + pwm_lsb_nanoseconds: int = Field( + default=130, description="Nanoseconds for LSB timing", ge=1 + ) + disable_hardware_pulsing: bool = Field( + default=False, description="Disable hardware pulsing if True" + ) + inverse_colors: bool = Field( + default=False, description="Invert display colors if True" + ) + show_refresh_rate: bool = Field( + default=False, description="Show panel refresh rate overlay" + ) + limit_refresh_rate_hz: int = Field( + default=120, description="Maximum refresh rate in Hz", ge=1 + ) + + +class RuntimeConfig(BaseModel): + gpio_slowdown: int = Field( + default=3, description="GPIO slowdown factor for Raspberry Pi", ge=0 + ) + + +class DisplayDurations(BaseModel): + clock: int = Field(default=15, description="Duration (sec) to show clock", ge=0) + weather: int = Field(default=30, description="Duration (sec) to show weather", ge=0) + stocks: int = Field(default=30, description="Duration (sec) to show stocks", ge=0) + hourly_forecast: int = Field( + default=30, description="Duration (sec) to show hourly forecast", ge=0 + ) + daily_forecast: int = Field( + default=30, description="Duration (sec) to show daily forecast", ge=0 + ) + stock_news: int = Field( + default=20, description="Duration (sec) to show stock news", ge=0 + ) + odds_ticker: int = Field( + default=60, description="Duration (sec) to show odds ticker", ge=0 + ) + leaderboard: int = Field( + default=300, description="Duration (sec) to show leaderboard", ge=0 + ) + nhl_live: int = Field(default=30, description="Duration (sec) for nhl_live", ge=0) + nhl_recent: int = Field( + default=30, description="Duration (sec) for nhl_recent", ge=0 + ) + nhl_upcoming: int = Field( + default=30, description="Duration (sec) for nhl_upcoming", ge=0 + ) + nba_live: int = Field(default=30, description="Duration (sec) for nba_live", ge=0) + nba_recent: int = Field( + default=30, description="Duration (sec) for nba_recent", ge=0 + ) + nba_upcoming: int = Field( + default=30, description="Duration (sec) for nba_upcoming", ge=0 + ) + nfl_live: int = Field(default=30, description="Duration (sec) for nfl_live", ge=0) + nfl_recent: int = Field( + default=30, description="Duration (sec) for nfl_recent", ge=0 + ) + nfl_upcoming: int = Field( + default=30, description="Duration (sec) for nfl_upcoming", ge=0 + ) + ncaa_fb_live: int = Field( + default=30, description="Duration (sec) for ncaa_fb_live", ge=0 + ) + ncaa_fb_recent: int = Field( + default=30, description="Duration (sec) for ncaa_fb_recent", ge=0 + ) + ncaa_fb_upcoming: int = Field( + default=30, description="Duration (sec) for ncaa_fb_upcoming", ge=0 + ) + ncaa_baseball_live: int = Field( + default=30, description="Duration (sec) for ncaa_baseball_live", ge=0 + ) + ncaa_baseball_recent: int = Field( + default=30, description="Duration (sec) for ncaa_baseball_recent", ge=0 + ) + ncaa_baseball_upcoming: int = Field( + default=30, description="Duration (sec) for ncaa_baseball_upcoming", ge=0 + ) + calendar: int = Field( + default=30, description="Duration (sec) to show calendar", ge=0 + ) + youtube: int = Field( + default=30, description="Duration (sec) for youtube items", ge=0 + ) + mlb_live: int = Field(default=30, description="Duration (sec) for mlb_live", ge=0) + mlb_recent: int = Field( + default=30, description="Duration (sec) for mlb_recent", ge=0 + ) + mlb_upcoming: int = Field( + default=30, description="Duration (sec) for mlb_upcoming", ge=0 + ) + milb_live: int = Field(default=30, description="Duration (sec) for milb_live", ge=0) + milb_recent: int = Field( + default=30, description="Duration (sec) for milb_recent", ge=0 + ) + milb_upcoming: int = Field( + default=30, description="Duration (sec) for milb_upcoming", ge=0 + ) + text_display: int = Field( + default=10, description="Duration (sec) for text display", ge=0 + ) + soccer_live: int = Field( + default=30, description="Duration (sec) for soccer_live", ge=0 + ) + soccer_recent: int = Field( + default=30, description="Duration (sec) for soccer_recent", ge=0 + ) + soccer_upcoming: int = Field( + default=30, description="Duration (sec) for soccer_upcoming", ge=0 + ) + ncaam_basketball_live: int = Field( + default=30, description="Duration (sec) for ncaam_basketball_live", ge=0 + ) + ncaam_basketball_recent: int = Field( + default=30, description="Duration (sec) for ncaam_basketball_recent", ge=0 + ) + ncaam_basketball_upcoming: int = Field( + default=30, description="Duration (sec) for ncaam_basketball_upcoming", ge=0 + ) + music: int = Field(default=30, description="Duration (sec) for music items", ge=0) + of_the_day: int = Field( + default=40, description="Duration (sec) for of the day", ge=0 + ) + news_manager: int = Field( + default=60, description="Duration (sec) for news manager", ge=0 + ) + + @field_validator("*", mode="after") + def _ensure_non_negative(cls, v): + if isinstance(v, int) and v < 0: + raise ValueError("Durations must be non-negative integers") + return v + + +class DisplayConfig(BaseModel): + hardware: HardwareConfig = Field( + default_factory=HardwareConfig, description="Low-level hardware configuration" + ) + runtime: RuntimeConfig = Field( + default_factory=RuntimeConfig, description="Runtime tweaks for the runtime" + ) + display_durations: DisplayDurations = Field( + default_factory=DisplayDurations, description="Per-module display durations" + ) + use_short_date_format: bool = Field( + default=True, description="Whether to use short date format on the display" + ) + + +# ------------------------- +# Clock +# ------------------------- +class ClockConfig(BaseModel): + enabled: bool = Field(default=True, description="Whether the clock is enabled") + format: str = Field( + default="%I:%M %p", description="Format string for clock display (strftime)" + ) + update_interval: int = Field( + default=1, description="Clock update interval in seconds", ge=1 + ) + + +# ------------------------- +# Weather +# ------------------------- +class WeatherConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable weather display") + update_interval: int = Field( + default=1800, description="Weather update interval in seconds", ge=30 + ) + units: str = Field( + default="imperial", description="Units for weather (imperial/metric)" + ) + display_format: str = Field( + default="{temp}°F\n{condition}", description="Display format for weather" + ) + + +# ------------------------- +# Stocks & Crypto & Stock News +# ------------------------- +class StocksConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable stock display") + update_interval: int = Field( + default=600, description="Update interval in seconds", ge=10 + ) + scroll_speed: int = Field(default=1, description="Scroll speed", ge=0) + scroll_delay: float = Field( + default=0.01, description="Scroll delay (seconds)", ge=0 + ) + toggle_chart: bool = Field(default=True, description="Toggle chart on ticker") + dynamic_duration: bool = Field( + default=True, description="Enable dynamic duration calculation" + ) + min_duration: int = Field( + default=30, description="Minimum display duration (sec)", ge=0 + ) + max_duration: int = Field( + default=300, description="Maximum display duration (sec)", ge=0 + ) + duration_buffer: float = Field( + default=0.1, description="Duration buffer multiplier", ge=0.0 + ) + symbols: List[str] = Field( + default_factory=lambda: ["ASTS", "SCHD", "INTC", "NVDA", "T", "VOO", "SMCI"], + description="List of stock symbols", + ) + display_format: str = Field( + default="{symbol}: ${price} ({change}%)", description="Stock display format" + ) + + +class CryptoConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable crypto display") + update_interval: int = Field( + default=600, description="Update interval in seconds", ge=10 + ) + symbols: List[str] = Field( + default_factory=lambda: ["BTC-USD", "ETH-USD"], + description="List of crypto symbols", + ) + display_format: str = Field( + default="{symbol}: ${price} ({change}%)", description="Crypto display format" + ) + + +class StockNewsConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable stock news display") + update_interval: int = Field( + default=3600, description="Update interval in seconds", ge=10 + ) + scroll_speed: int = Field(default=1, description="Scroll speed", ge=0) + scroll_delay: float = Field( + default=0.01, description="Scroll delay (seconds)", ge=0 + ) + max_headlines_per_symbol: int = Field( + default=1, description="Maximum headlines per symbol", ge=0 + ) + headlines_per_rotation: int = Field( + default=2, description="Headlines per rotation", ge=0 + ) + dynamic_duration: bool = Field(default=True, description="Enable dynamic duration") + min_duration: int = Field( + default=30, description="Minimum display duration (sec)", ge=0 + ) + max_duration: int = Field( + default=300, description="Maximum display duration (sec)", ge=0 + ) + duration_buffer: float = Field( + default=0.1, description="Duration buffer multiplier", ge=0.0 + ) + + +# ------------------------- +# Background service (shared) +# ------------------------- +class BackgroundServiceConfig(BaseModel): + enabled: bool = Field(default=True, description="Background service enabled") + max_workers: int = Field(default=3, description="Max number of workers", ge=1) + request_timeout: int = Field( + default=30, description="Request timeout (seconds)", ge=1 + ) + max_retries: int = Field(default=3, description="Maximum retries", ge=0) + priority: int = Field(default=2, description="Priority for background tasks", ge=0) + + +# ------------------------- +# Odds Ticker +# ------------------------- +class OddsTickerConfig(BaseModel): + enabled: bool = Field(default=True, description="Enable odds ticker") + show_favorite_teams_only: bool = Field( + default=True, description="Show only favorite teams" + ) + games_per_favorite_team: int = Field( + default=1, description="Games per favorite team", ge=0 + ) + max_games_per_league: int = Field( + default=5, description="Max games per league", ge=0 + ) + show_odds_only: bool = Field(default=False, description="Show only odds (no teams)") + sort_order: str = Field(default="soonest", description="Sort order for events") + enabled_leagues: List[str] = Field( + default_factory=lambda: ["nfl", "mlb", "ncaa_fb", "milb"], + description="Enabled leagues", + ) + update_interval: int = Field( + default=3600, description="Update interval in seconds", ge=1 + ) + scroll_speed: int = Field(default=1, description="Scroll speed", ge=0) + scroll_delay: float = Field( + default=0.01, description="Scroll delay (seconds)", ge=0 + ) + loop: bool = Field(default=True, description="Whether to loop the ticker") + future_fetch_days: int = Field( + default=50, description="How many days into the future to fetch", ge=0 + ) + show_channel_logos: bool = Field(default=True, description="Show channel logos") + dynamic_duration: bool = Field(default=True, description="Enable dynamic duration") + min_duration: int = Field(default=30, description="Minimum duration (sec)", ge=0) + max_duration: int = Field(default=300, description="Maximum duration (sec)", ge=0) + duration_buffer: float = Field( + default=0.1, description="Duration buffer multiplier", ge=0.0 + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service config for odds ticker", + ) + + +# ------------------------- +# Leaderboard +# ------------------------- +class LeaderboardSportsConfig(BaseModel): + enabled: bool = Field( + default=True, description="Whether this sport leaderboard is enabled" + ) + top_teams: int = Field(default=10, description="Number of top teams to show", ge=0) + show_ranking: Optional[bool] = Field( + default=None, description="Whether to display ranking (optional)" + ) + + +class LeaderboardConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable leaderboard") + enabled_sports: Dict[str, LeaderboardSportsConfig] = Field( + default_factory=lambda: { + "nfl": LeaderboardSportsConfig(enabled=True, top_teams=10), + "nba": LeaderboardSportsConfig(enabled=False, top_teams=10), + "mlb": LeaderboardSportsConfig(enabled=False, top_teams=10), + "ncaa_fb": LeaderboardSportsConfig( + enabled=True, top_teams=25, show_ranking=True + ), + "nhl": LeaderboardSportsConfig(enabled=False, top_teams=10), + "ncaam_basketball": LeaderboardSportsConfig(enabled=False, top_teams=25), + "ncaam_hockey": LeaderboardSportsConfig( + enabled=True, top_teams=10, show_ranking=True + ), + }, + description="Per-sport leaderboard configuration", + ) + update_interval: int = Field( + default=3600, description="Update interval in seconds", ge=1 + ) + scroll_speed: int = Field(default=1, description="Scroll speed", ge=0) + scroll_delay: float = Field( + default=0.01, description="Scroll delay (seconds)", ge=0 + ) + loop: bool = Field(default=False, description="Whether to loop the leaderboard") + request_timeout: int = Field( + default=30, description="Request timeout seconds", ge=1 + ) + dynamic_duration: bool = Field(default=True, description="Dynamic duration enabled") + min_duration: int = Field(default=30, description="Minimum duration (sec)", ge=0) + max_display_time: int = Field( + default=600, description="Maximum display time (sec)", ge=0 + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for leaderboard", + ) + + +# ------------------------- +# Calendar +# ------------------------- +class CalendarConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable calendar display") + credentials_file: str = Field( + default="credentials.json", description="Path to credentials file" + ) + token_file: str = Field(default="token.pickle", description="Path to token file") + update_interval: int = Field( + default=3600, description="Update interval in seconds", ge=1 + ) + max_events: int = Field( + default=3, description="Maximum number of events to show", ge=0 + ) + calendars: List[str] = Field( + default_factory=lambda: ["birthdays"], description="List of calendar IDs/names" + ) + + +# ------------------------- +# Generic scoreboard base and specific scoreboard models +# ------------------------- +class ScoreboardBaseConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable this scoreboard") + live_priority: bool = Field(default=True, description="Give live games priority") + live_game_duration: int = Field( + default=20, description="How long to show a live game (sec)", ge=0 + ) + show_odds: bool = Field( + default=True, description="Display betting odds if available" + ) + test_mode: bool = Field(default=False, description="If true, run in test mode") + update_interval_seconds: int = Field( + default=3600, description="General update interval (sec)", ge=1 + ) + live_update_interval: int = Field( + default=30, description="Live update frequency (sec)", ge=1 + ) + live_odds_update_interval: Optional[int] = Field( + default=None, description="Live odds update interval (sec)" + ) + odds_update_interval: Optional[int] = Field( + default=None, description="Odds update interval (sec)" + ) + recent_update_interval: Optional[int] = Field( + default=None, description="Recent games update interval (sec)" + ) + upcoming_update_interval: Optional[int] = Field( + default=None, description="Upcoming games update interval (sec)" + ) + recent_games_to_show: int = Field( + default=1, description="How many recent games to show", ge=0 + ) + upcoming_games_to_show: int = Field( + default=1, description="How many upcoming games to show", ge=0 + ) + show_favorite_teams_only: bool = Field( + default=True, description="Only show favorite teams" + ) + favorite_teams: List[str] = Field( + default_factory=list, description="List of favorite team codes" + ) + logo_dir: str = Field(default="", description="Directory for team logos") + show_records: bool = Field(default=True, description="Show team records") + show_ranking: Optional[bool] = Field( + default=None, description="Show ranking where applicable" + ) + upcoming_fetch_days: Optional[int] = Field( + default=None, description="How many days ahead to fetch upcoming games" + ) + background_service: Optional[BackgroundServiceConfig] = Field( + default=None, description="Background service config" + ) + display_modes: Dict[str, bool] = Field( + default_factory=dict, description="Which display modes are enabled" + ) + + +class NHLScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable NHL scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["TB"], description="Favorite NHL teams" + ) + logo_dir: str = Field( + default="assets/sports/nhl_logos", description="NHL logos directory" + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for NHL", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "nhl_live": True, + "nhl_recent": True, + "nhl_upcoming": True, + } + ) + + +class NBAScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable NBA scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["DAL"], description="Favorite NBA teams" + ) + logo_dir: str = Field( + default="assets/sports/nba_logos", description="NBA logos directory" + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for NBA", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "nba_live": True, + "nba_recent": True, + "nba_upcoming": True, + } + ) + + +class NFLScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable NFL scoreboard") + live_game_duration: int = Field( + default=30, description="Live game duration for NFL (sec)", ge=0 + ) + favorite_teams: List[str] = Field( + default_factory=lambda: ["TB", "DAL"], description="Favorite NFL teams" + ) + logo_dir: str = Field( + default="assets/sports/nfl_logos", description="NFL logos directory" + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for NFL", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "nfl_live": True, + "nfl_recent": True, + "nfl_upcoming": True, + } + ) + + +class NCAAFBScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable NCAA Football scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["UGA", "AUB", "AP_TOP_25"], + description="Favorite NCAA football teams", + ) + logo_dir: str = Field( + default="assets/sports/ncaa_logos", description="NCAA logos directory" + ) + show_ranking: bool = Field( + default=True, description="Show ranking for NCAA football" + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for NCAA FB", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "ncaa_fb_live": True, + "ncaa_fb_recent": True, + "ncaa_fb_upcoming": True, + } + ) + + +class NCAABaseballScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable NCAA baseball scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["UGA", "AUB"], + description="Favorite NCAA baseball teams", + ) + logo_dir: str = Field( + default="assets/sports/ncaa_logos", description="NCAA logos directory" + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "ncaa_baseball_live": True, + "ncaa_baseball_recent": True, + "ncaa_baseball_upcoming": True, + } + ) + + +class NCAAMBasketballScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field( + default=False, description="Enable NCAAM basketball scoreboard" + ) + favorite_teams: List[str] = Field( + default_factory=lambda: ["UGA", "AUB"], + description="Favorite NCAAM basketball teams", + ) + logo_dir: str = Field( + default="assets/sports/ncaa_logos", description="NCAA logos directory" + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "ncaam_basketball_live": True, + "ncaam_basketball_recent": True, + "ncaam_basketball_upcoming": True, + } + ) + + +class NCAAMHockeyScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=True, description="Enable NCAAM hockey scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["RIT"], description="Favorite NCAAM hockey teams" + ) + logo_dir: str = Field( + default="assets/sports/ncaa_logos", description="NCAA logos directory" + ) + show_ranking: bool = Field( + default=True, description="Show ranking for NCAAM hockey" + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "ncaam_hockey_live": True, + "ncaam_hockey_recent": True, + "ncaam_hockey_upcoming": True, + } + ) + + +class MLBScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable MLB scoreboard") + live_priority: bool = Field( + default=False, description="Whether live priority is used for MLB" + ) + favorite_teams: List[str] = Field( + default_factory=lambda: ["TB", "TEX"], description="Favorite MLB teams" + ) + logo_dir: str = Field( + default="assets/sports/mlb_logos", description="MLB logos directory" + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "mlb_live": True, + "mlb_recent": True, + "mlb_upcoming": True, + } + ) + + +class MILBScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable MiLB scoreboard") + live_priority: bool = Field( + default=False, description="Whether live priority is used for MiLB" + ) + favorite_teams: List[str] = Field( + default_factory=lambda: ["TAM"], description="Favorite MiLB teams" + ) + logo_dir: str = Field( + default="assets/sports/milb_logos", description="MiLB logos directory" + ) + upcoming_fetch_days: int = Field( + default=7, description="Days to look ahead for upcoming MiLB games", ge=0 + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for MiLB", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "milb_live": True, + "milb_recent": True, + "milb_upcoming": True, + } + ) + + +class SoccerScoreboardConfig(ScoreboardBaseConfig): + enabled: bool = Field(default=False, description="Enable soccer scoreboard") + favorite_teams: List[str] = Field( + default_factory=lambda: ["DAL"], description="Favorite soccer teams" + ) + leagues: List[str] = Field( + default_factory=lambda: ["usa.1"], description="Soccer leagues to include" + ) + logo_dir: str = Field( + default="assets/sports/soccer_logos", description="Soccer logos directory" + ) + background_service: BackgroundServiceConfig = Field( + default_factory=BackgroundServiceConfig, + description="Background service for soccer", + ) + display_modes: Dict[str, bool] = Field( + default_factory=lambda: { + "soccer_live": True, + "soccer_recent": True, + "soccer_upcoming": True, + } + ) + + +# ------------------------- +# YouTube +# ------------------------- +class YouTubeConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable YouTube module") + update_interval: int = Field( + default=3600, description="YouTube update interval in seconds", ge=1 + ) + + +# ------------------------- +# Text display +# ------------------------- +class TextDisplayConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable custom text display") + text: str = Field(default="Subscribe to ChuckBuilds", description="Text to display") + font_path: str = Field( + default="assets/fonts/press-start-2p.ttf", description="Path to font file" + ) + font_size: int = Field(default=8, description="Font size in pixels", ge=1) + scroll: bool = Field(default=True, description="Whether the text scrolls") + scroll_speed: int = Field(default=40, description="Scroll speed", ge=0) + text_color: List[int] = Field( + default_factory=lambda: [255, 0, 0], description="RGB color for text (3 ints)" + ) + background_color: List[int] = Field( + default_factory=lambda: [0, 0, 0], description="RGB for background (3 ints)" + ) + scroll_gap_width: int = Field( + default=32, description="Gap width between repeated scrolls", ge=0 + ) + + @field_validator("text_color", "background_color") + def _validate_color_array(cls, v): + if not isinstance(v, list) or len(v) != 3: + raise ValueError("Color must be a list of 3 integers [R, G, B]") + for c in v: + if not isinstance(c, int) or not (0 <= c <= 255): + raise ValueError("Color components must be integers 0-255") + return v + + +# ------------------------- +# Music +# ------------------------- +class MusicConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable music module") + preferred_source: str = Field(default="ytm", description="Preferred music source") + YTM_COMPANION_URL: str = Field( + default="http://192.168.86.12:9863", description="YTM companion URL" + ) + POLLING_INTERVAL_SECONDS: int = Field( + default=1, description="Polling interval for music (sec)", ge=1 + ) + + +# ------------------------- +# Of the day +# ------------------------- +class OfTheDayCategory(BaseModel): + enabled: bool = Field(default=True, description="Enable this category") + data_file: str = Field(default="", description="Path to data file") + display_name: str = Field(default="", description="Display name for category") + + +class OfTheDayConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable of-the-day feature") + display_rotate_interval: int = Field( + default=20, description="Rotate interval in seconds", ge=0 + ) + update_interval: int = Field( + default=3600, description="Update interval in seconds", ge=1 + ) + subtitle_rotate_interval: int = Field( + default=10, description="Subtitle rotate interval in seconds", ge=0 + ) + category_order: List[str] = Field( + default_factory=lambda: ["word_of_the_day", "slovenian_word_of_the_day"], + description="Order of categories to rotate", + ) + categories: Dict[str, OfTheDayCategory] = Field( + default_factory=lambda: { + "word_of_the_day": OfTheDayCategory( + enabled=True, + data_file="of_the_day/word_of_the_day.json", + display_name="Word of the Day", + ), + "slovenian_word_of_the_day": OfTheDayCategory( + enabled=True, + data_file="of_the_day/slovenian_word_of_the_day.json", + display_name="Slovenian Word of the Day", + ), + }, + description="Category specific configs", + ) + + +# ------------------------- +# News manager +# ------------------------- +class NewsManagerConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable news manager") + update_interval: int = Field( + default=300, description="Update interval in seconds", ge=1 + ) + scroll_speed: int = Field(default=1, description="Scroll speed", ge=0) + scroll_delay: float = Field( + default=0.01, description="Scroll delay (seconds)", ge=0 + ) + headlines_per_feed: int = Field( + default=2, description="Headlines per feed to show", ge=0 + ) + enabled_feeds: List[str] = Field( + default_factory=lambda: ["NFL", "NCAA FB", "F1", "BBC F1"], + description="Pre-enabled feed names", + ) + custom_feeds: Dict[str, str] = Field( + default_factory=lambda: { + "F1": "https://www.espn.com/espn/rss/rpm/news", + "BBC F1": "http://feeds.bbci.co.uk/sport/formula1/rss.xml", + }, + description="User-specified custom RSS feeds", + ) + rotation_enabled: bool = Field( + default=True, description="Enable rotation between feeds" + ) + rotation_threshold: int = Field( + default=3, description="Rotation threshold value", ge=0 + ) + dynamic_duration: bool = Field(default=True, description="Enable dynamic duration") + min_duration: int = Field( + default=30, description="Minimum duration in seconds", ge=0 + ) + max_duration: int = Field( + default=300, description="Maximum duration in seconds", ge=0 + ) + duration_buffer: float = Field( + default=0.1, description="Duration buffer multiplier", ge=0.0 + ) + font_size: int = Field(default=8, description="Font size for news text", ge=1) + font_path: str = Field( + default="assets/fonts/PressStart2P-Regular.ttf", description="Path to font" + ) + text_color: List[int] = Field( + default_factory=lambda: [255, 255, 255], description="RGB text color (3 ints)" + ) + separator_color: List[int] = Field( + default_factory=lambda: [255, 0, 0], description="RGB separator color (3 ints)" + ) + + @field_validator("text_color", "separator_color") + def _validate_colors(cls, v): + if not isinstance(v, list) or len(v) != 3: + raise ValueError("Color must be a list of 3 integers [R, G, B]") + for c in v: + if not isinstance(c, int) or not (0 <= c <= 255): + raise ValueError("Color components must be integers 0-255") + return v + + +# ------------------------- +# Root configuration (everything together) +# ------------------------- +class RootConfig(BaseModel): + web_display_autostart: bool = Field( + default=True, description="Autostart the web display service" + ) + schedule: Schedule = Field( + default_factory=Schedule, description="Schedule configuration" + ) + timezone: str = Field(default="America/Chicago", description="System timezone") + location: Location = Field( + default_factory=Location, description="Geographic location of display" + ) + display: DisplayConfig = Field( + default_factory=DisplayConfig, description="Display configuration" + ) + clock: ClockConfig = Field( + default_factory=ClockConfig, description="Clock configuration" + ) + weather: WeatherConfig = Field( + default_factory=WeatherConfig, description="Weather configuration" + ) + stocks: StocksConfig = Field( + default_factory=StocksConfig, description="Stock ticker configuration" + ) + crypto: CryptoConfig = Field( + default_factory=CryptoConfig, description="Crypto ticker configuration" + ) + stock_news: StockNewsConfig = Field( + default_factory=StockNewsConfig, description="Stock news configuration" + ) + odds_ticker: OddsTickerConfig = Field( + default_factory=OddsTickerConfig, description="Odds ticker configuration" + ) + leaderboard: LeaderboardConfig = Field( + default_factory=LeaderboardConfig, description="Leaderboard configuration" + ) + calendar: CalendarConfig = Field( + default_factory=CalendarConfig, description="Calendar configuration" + ) + nhl_scoreboard: NHLScoreboardConfig = Field( + default_factory=NHLScoreboardConfig, description="NHL scoreboard configuration" + ) + nba_scoreboard: NBAScoreboardConfig = Field( + default_factory=NBAScoreboardConfig, description="NBA scoreboard configuration" + ) + nfl_scoreboard: NFLScoreboardConfig = Field( + default_factory=NFLScoreboardConfig, description="NFL scoreboard configuration" + ) + ncaa_fb_scoreboard: NCAAFBScoreboardConfig = Field( + default_factory=NCAAFBScoreboardConfig, + description="NCAA Football scoreboard configuration", + ) + ncaa_baseball_scoreboard: NCAABaseballScoreboardConfig = Field( + default_factory=NCAABaseballScoreboardConfig, + description="NCAA Baseball scoreboard configuration", + ) + ncaam_basketball_scoreboard: NCAAMBasketballScoreboardConfig = Field( + default_factory=NCAAMBasketballScoreboardConfig, + description="NCAAM Basketball scoreboard configuration", + ) + ncaam_hockey_scoreboard: NCAAMHockeyScoreboardConfig = Field( + default_factory=NCAAMHockeyScoreboardConfig, + description="NCAAM Hockey scoreboard configuration", + ) + youtube: YouTubeConfig = Field( + default_factory=YouTubeConfig, description="YouTube configuration" + ) + mlb_scoreboard: MLBScoreboardConfig = Field( + default_factory=MLBScoreboardConfig, description="MLB scoreboard configuration" + ) + milb_scoreboard: MILBScoreboardConfig = Field( + default_factory=MILBScoreboardConfig, + description="MiLB scoreboard configuration", + ) + text_display: TextDisplayConfig = Field( + default_factory=TextDisplayConfig, description="Text display configuration" + ) + soccer_scoreboard: SoccerScoreboardConfig = Field( + default_factory=SoccerScoreboardConfig, + description="Soccer scoreboard configuration", + ) + music: MusicConfig = Field( + default_factory=MusicConfig, description="Music configuration" + ) + of_the_day: OfTheDayConfig = Field( + default_factory=OfTheDayConfig, description="Of-the-day configuration" + ) + news_manager: NewsManagerConfig = Field( + default_factory=NewsManagerConfig, description="News manager configuration" + ) + + @field_validator("timezone") + def _cross_validate(cls, tz): + # example: ensure timezone non-empty + if not tz or not isinstance(tz, str): + raise ValueError("timezone must be a non-empty string") + return tz diff --git a/src/config/secrets_models.py b/src/config/secrets_models.py new file mode 100644 index 00000000..953994af --- /dev/null +++ b/src/config/secrets_models.py @@ -0,0 +1,48 @@ +from pydantic import BaseModel, Field + + +class WeatherSecrets(BaseModel): + api_key: str = Field( + default="YOUR_OPENWEATHERMAP_API_KEY", + description="API key for accessing OpenWeatherMap services.", + ) + + +class YoutubeSecrets(BaseModel): + api_key: str = Field( + default="YOUR_YOUTUBE_API_KEY", + description="API key for accessing YouTube Data API.", + ) + channel_id: str = Field( + default="YOUR_YOUTUBE_CHANNEL_ID", + description="Channel ID of the YouTube channel to fetch data from.", + ) + + +class MusicSecrets(BaseModel): + SPOTIFY_CLIENT_ID: str = Field( + default="YOUR_SPOTIFY_CLIENT_ID_HERE", + description="Spotify application Client ID.", + ) + SPOTIFY_CLIENT_SECRET: str = Field( + default="YOUR_SPOTIFY_CLIENT_SECRET_HERE", + description="Spotify application Client Secret.", + ) + SPOTIFY_REDIRECT_URI: str = Field( + default="http://127.0.0.1:8888/callback", + description="Redirect URI for Spotify OAuth authentication.", + ) + + +class SecretsConfig(BaseModel): + weather: WeatherSecrets = Field( + default=WeatherSecrets(), + description="Weather API authentication configuration.", + ) + youtube: YoutubeSecrets = Field( + default=YoutubeSecrets(), + description="YouTube API authentication configuration.", + ) + music: MusicSecrets = Field( + default=MusicSecrets(), description="Spotify authentication configuration." + ) diff --git a/src/config_manager.py b/src/config_manager.py index 6d2a2991..733df38e 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -1,14 +1,17 @@ import json import os from typing import Dict, Any, Optional - +from src.config.config_models import RootConfig, DisplayConfig, ClockConfig +from src.config.secrets_models import SecretsConfig class ConfigManager: - def __init__(self, config_path: str = None, secrets_path: str = None): + def __init__(self, config_path: str | None = None, secrets_path: str | None = None): # Use current working directory as base self.config_path = config_path or "config/config.json" self.secrets_path = secrets_path or "config/config_secrets.json" self.template_path = "config/config.template.json" self.config: Dict[str, Any] = {} + self.new_config: RootConfig = RootConfig() + self.new_secrets_config: SecretsConfig = SecretsConfig() def get_config_path(self) -> str: return self.config_path @@ -16,6 +19,43 @@ def get_config_path(self) -> str: def get_secrets_path(self) -> str: return self.secrets_path + def load_new_config(self) -> tuple[RootConfig, SecretsConfig]: + """Load configuration from JSON files.""" + try: + # Check if config file exists, if not create from template + if not os.path.exists(self.config_path): + self._create_config_from_template() + + # Load main config + print(f"Attempting to load config from: {os.path.abspath(self.config_path)}") + with open(self.config_path, 'r') as f: + self.new_config = RootConfig(**json.load(f)) + + # Load and merge secrets if they exist (be permissive on errors) + if os.path.exists(self.secrets_path): + try: + with open(self.secrets_path, 'r') as f: + self.new_secrets_config = SecretsConfig(**json.load(f)) + except PermissionError as e: + print(f"Secrets file not readable ({self.secrets_path}): {e}. Continuing without secrets.") + except (json.JSONDecodeError, OSError) as e: + print(f"Error reading secrets file ({self.secrets_path}): {e}. Continuing without secrets.") + + return self.new_config, self.new_secrets_config + + except FileNotFoundError as e: + if str(e).find('config_secrets.json') == -1: # Only raise if main config is missing + print(f"Configuration file not found at {os.path.abspath(self.config_path)}") + raise + except json.JSONDecodeError: + print("Error parsing configuration file") + raise + except Exception as e: + print(f"Error loading configuration: {str(e)}") + raise + return RootConfig(), SecretsConfig() + + def load_config(self) -> Dict[str, Any]: """Load configuration from JSON files.""" try: @@ -57,48 +97,15 @@ def load_config(self) -> Dict[str, Any]: print(f"Error loading configuration: {str(e)}") raise - def _strip_secrets_recursive(self, data_to_filter: Dict[str, Any], secrets: Dict[str, Any]) -> Dict[str, Any]: - """Recursively remove secret keys from a dictionary.""" - result = {} - for key, value in data_to_filter.items(): - if key in secrets: - if isinstance(value, dict) and isinstance(secrets[key], dict): - # This key is a shared group, recurse - stripped_sub_dict = self._strip_secrets_recursive(value, secrets[key]) - if stripped_sub_dict: # Only add if there's non-secret data left - result[key] = stripped_sub_dict - # Else, it's a secret key at this level, so we skip it - else: - # This key is not in secrets, so we keep it - result[key] = value - return result - def save_config(self, new_config_data: Dict[str, Any]) -> None: """Save configuration to the main JSON file, stripping out secrets.""" - secrets_content = {} - if os.path.exists(self.secrets_path): - try: - with open(self.secrets_path, 'r') as f_secrets: - secrets_content = json.load(f_secrets) - except Exception as e: - print(f"Warning: Could not load secrets file {self.secrets_path} during save: {e}") - # Continue without stripping if secrets can't be loaded, or handle as critical error - # For now, we'll proceed cautiously and save the full new_config_data if secrets are unreadable - # to prevent accidental data loss if the secrets file is temporarily corrupt. - # A more robust approach might be to fail the save or use a cached version of secrets. - - config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content) - try: with open(self.config_path, 'w') as f: - json.dump(config_to_write, f, indent=4) + json.dump(new_config_data, f, indent=4) # Update the in-memory config to the new state (which includes secrets for runtime) - self.config = new_config_data + self.config = new_config_data print(f"Configuration successfully saved to {os.path.abspath(self.config_path)}") - if secrets_content: - print("Secret values were preserved in memory and not written to the main config file.") - except IOError as e: print(f"Error writing configuration to file {os.path.abspath(self.config_path)}: {e}") raise @@ -136,12 +143,8 @@ def _create_config_from_template(self) -> None: # Ensure config directory exists os.makedirs(os.path.dirname(self.config_path), exist_ok=True) - # Copy template to config - with open(self.template_path, 'r') as template_file: - template_data = json.load(template_file) - with open(self.config_path, 'w') as config_file: - json.dump(template_data, config_file, indent=4) + json.dump(RootConfig().model_dump(), config_file, indent=4) print(f"Created config.json from template at {os.path.abspath(self.config_path)}") @@ -207,42 +210,15 @@ def _merge_template_defaults(self, current: Dict[str, Any], template: Dict[str, def get_timezone(self) -> str: """Get the configured timezone.""" - return self.config.get('timezone', 'UTC') + return self.new_config.timezone - def get_display_config(self) -> Dict[str, Any]: + def get_display_config(self) -> DisplayConfig: """Get display configuration.""" - return self.config.get('display', {}) + return self.new_config.display - def get_clock_config(self) -> Dict[str, Any]: + def get_clock_config(self) -> ClockConfig: """Get clock configuration.""" - return self.config.get('clock', {}) - - def get_raw_file_content(self, file_type: str) -> Dict[str, Any]: - """Load raw content of 'main' config or 'secrets' config file.""" - path_to_load = "" - if file_type == "main": - path_to_load = self.config_path - elif file_type == "secrets": - path_to_load = self.secrets_path - else: - raise ValueError("Invalid file_type specified. Must be 'main' or 'secrets'.") - - if not os.path.exists(path_to_load): - # If a secrets file doesn't exist, it's not an error, just return empty - if file_type == "secrets": - return {} - print(f"{file_type.capitalize()} configuration file not found at {os.path.abspath(path_to_load)}") - raise FileNotFoundError(f"{file_type.capitalize()} configuration file not found at {os.path.abspath(path_to_load)}") - - try: - with open(path_to_load, 'r') as f: - return json.load(f) - except json.JSONDecodeError: - print(f"Error parsing {file_type} configuration file: {path_to_load}") - raise - except Exception as e: - print(f"Error loading {file_type} configuration file {path_to_load}: {str(e)}") - raise + return self.new_config.clock def save_raw_file_content(self, file_type: str, data: Dict[str, Any]) -> None: """Save data directly to 'main' config or 'secrets' config file.""" diff --git a/src/display_controller.py b/src/display_controller.py index 9a55df8e..3510b9c5 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -48,16 +48,17 @@ def __init__(self): self.config_manager = ConfigManager() self.config = self.config_manager.load_config() + self.new_config, self.secrets_config = self.config_manager.load_new_config() self.cache_manager = CacheManager() logger.info("Config loaded in %.3f seconds", time.time() - start_time) config_time = time.time() - self.display_manager = DisplayManager(self.config) + self.display_manager = DisplayManager(self.new_config) logger.info("DisplayManager initialized in %.3f seconds", time.time() - config_time) # Initialize display modes init_time = time.time() - self.clock = Clock(self.display_manager, self.config) if self.config.get('clock', {}).get('enabled', True) else None + self.clock = Clock(self.display_manager, self.new_config) if self.new_config.clock.enabled else None self.weather = WeatherManager(self.config, self.display_manager) if self.config.get('weather', {}).get('enabled', False) else None self.stocks = StockManager(self.config, self.display_manager) if self.config.get('stocks', {}).get('enabled', False) else None self.news = StockNewsManager(self.config, self.display_manager) if self.config.get('stock_news', {}).get('enabled', False) else None @@ -184,9 +185,9 @@ def __init__(self): nfl_display_modes = self.config.get('nfl_scoreboard', {}).get('display_modes', {}) if nfl_enabled: - self.nfl_live = NFLLiveManager(self.config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_live', True) else None - self.nfl_recent = NFLRecentManager(self.config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_recent', True) else None - self.nfl_upcoming = NFLUpcomingManager(self.config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_upcoming', True) else None + self.nfl_live = NFLLiveManager(self.new_config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_live', True) else None + self.nfl_recent = NFLRecentManager(self.new_config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_recent', True) else None + self.nfl_upcoming = NFLUpcomingManager(self.new_config, self.display_manager, self.cache_manager) if nfl_display_modes.get('nfl_upcoming', True) else None else: self.nfl_live = None self.nfl_recent = None diff --git a/src/display_manager.py b/src/display_manager.py index 90ea31b3..592610ea 100644 --- a/src/display_manager.py +++ b/src/display_manager.py @@ -11,6 +11,7 @@ from .weather_icons import WeatherIcons import os import freetype +from src.config.config_models import RootConfig # Get logger without configuring logger = logging.getLogger(__name__) @@ -25,7 +26,7 @@ def __new__(cls, *args, **kwargs): cls._instance = super(DisplayManager, cls).__new__(cls) return cls._instance - def __init__(self, config: Dict[str, Any] = None, force_fallback: bool = False, suppress_test_pattern: bool = False): + def __init__(self, config: RootConfig, force_fallback: bool = False, suppress_test_pattern: bool = False): start_time = time.time() self.config = config or {} self._force_fallback = force_fallback @@ -66,36 +67,36 @@ def _setup_matrix(self): options = RGBMatrixOptions() # Hardware configuration - hardware_config = self.config.get('display', {}).get('hardware', {}) - runtime_config = self.config.get('display', {}).get('runtime', {}) + hardware_config = self.config.display.hardware + runtime_config = self.config.display.runtime # Basic hardware settings - options.rows = hardware_config.get('rows', 32) - options.cols = hardware_config.get('cols', 64) - options.chain_length = hardware_config.get('chain_length', 2) - options.parallel = hardware_config.get('parallel', 1) - options.hardware_mapping = hardware_config.get('hardware_mapping', 'adafruit-hat-pwm') + options.rows = hardware_config.rows + options.cols = hardware_config.cols + options.chain_length = hardware_config.chain_length + options.parallel = hardware_config.parallel + options.hardware_mapping = hardware_config.hardware_mapping # Performance and stability settings - options.brightness = hardware_config.get('brightness', 90) - options.pwm_bits = hardware_config.get('pwm_bits', 10) - options.pwm_lsb_nanoseconds = hardware_config.get('pwm_lsb_nanoseconds', 150) - options.led_rgb_sequence = hardware_config.get('led_rgb_sequence', 'RGB') - options.pixel_mapper_config = hardware_config.get('pixel_mapper_config', '') - options.row_address_type = hardware_config.get('row_address_type', 0) - options.multiplexing = hardware_config.get('multiplexing', 0) - options.disable_hardware_pulsing = hardware_config.get('disable_hardware_pulsing', False) - options.show_refresh_rate = hardware_config.get('show_refresh_rate', False) - options.limit_refresh_rate_hz = hardware_config.get('limit_refresh_rate_hz', 90) - options.gpio_slowdown = runtime_config.get('gpio_slowdown', 2) + options.brightness = hardware_config.brightness + options.pwm_bits = hardware_config.pwm_bits + options.pwm_lsb_nanoseconds = hardware_config.pwm_lsb_nanoseconds + # options.led_rgb_sequence = hardware_config.led_rgb_sequence + # options.pixel_mapper_config = hardware_config.pixel_mapper_config + # options.row_address_type = hardware_config.row_address_type + # options.multiplexing = hardware_config.multiplexing + options.disable_hardware_pulsing = hardware_config.disable_hardware_pulsing + options.show_refresh_rate = hardware_config.show_refresh_rate + # options.limit_refresh_rate_hz = hardware_config.glimit_refresh_rate_hz + options.gpio_slowdown = runtime_config.gpio_slowdown # Additional settings from config - if 'scan_mode' in hardware_config: - options.scan_mode = hardware_config.get('scan_mode') - if 'pwm_dither_bits' in hardware_config: - options.pwm_dither_bits = hardware_config.get('pwm_dither_bits') - if 'inverse_colors' in hardware_config: - options.inverse_colors = hardware_config.get('inverse_colors') + if hardware_config.scan_mode: + options.scan_mode = hardware_config.scan_mode + if hardware_config.pwm_dither_bits: + options.pwm_dither_bits = hardware_config.pwm_dither_bits + if hardware_config.inverse_colors: + options.inverse_colors = hardware_config.inverse_colors logger.info(f"Initializing RGB Matrix with settings: rows={options.rows}, cols={options.cols}, chain_length={options.chain_length}, parallel={options.parallel}, hardware_mapping={options.hardware_mapping}") @@ -130,10 +131,10 @@ def _setup_matrix(self): # Create a fallback image for web preview using configured dimensions when available self.matrix = None try: - hardware_config = self.config.get('display', {}).get('hardware', {}) if self.config else {} - rows = int(hardware_config.get('rows', 32)) - cols = int(hardware_config.get('cols', 64)) - chain_length = int(hardware_config.get('chain_length', 2)) + hardware_config = self.config.display.hardware + rows = hardware_config.rows + cols = hardware_config.cols + chain_length = hardware_config.chain_length fallback_width = max(1, cols * chain_length) fallback_height = max(1, rows) except Exception: diff --git a/src/nfl_managers.py b/src/nfl_managers.py index 917cf42f..1f546f93 100644 --- a/src/nfl_managers.py +++ b/src/nfl_managers.py @@ -7,6 +7,7 @@ from src.display_manager import DisplayManager from src.cache_manager import CacheManager import pytz +from src.config.config_models import RootConfig from src.base_classes.sports import SportsRecent, SportsUpcoming from src.base_classes.football import Football, FootballLive @@ -22,7 +23,7 @@ class BaseNFLManager(Football): # Renamed class _shared_data = None _last_shared_update = 0 - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager): self.logger = logging.getLogger('NFL') # Changed logger name super().__init__(config=config, display_manager=display_manager, cache_manager=cache_manager, logger=self.logger, sport_key="nfl") @@ -30,7 +31,7 @@ def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cach # self.logo_dir and self.update_interval are already configured # Check display modes to determine what data to fetch - display_modes = self.mode_config.get("display_modes", {}) + display_modes = self.mode_config.display_modes self.recent_enabled = display_modes.get("nfl_recent", False) self.upcoming_enabled = display_modes.get("nfl_upcoming", False) self.live_enabled = display_modes.get("nfl_live", False) @@ -91,27 +92,28 @@ def fetch_callback(result): del self.background_fetch_requests[season_year] # Get background service configuration - background_config = self.mode_config.get("background_service", {}) - timeout = background_config.get("request_timeout", 30) - max_retries = background_config.get("max_retries", 3) - priority = background_config.get("priority", 2) - - # Submit background fetch request - request_id = self.background_service.submit_fetch_request( - sport="nfl", - year=season_year, - url=ESPN_NFL_SCOREBOARD_URL, - cache_key=cache_key, - params={"dates": datestring, "limit": 1000}, - headers=self.headers, - timeout=timeout, - max_retries=max_retries, - priority=priority, - callback=fetch_callback - ) - - # Track the request - self.background_fetch_requests[season_year] = request_id + background_config = self.mode_config.background_service + if background_config: + timeout = background_config.request_timeout + max_retries = background_config.max_retries + priority = background_config.priority + + # Submit background fetch request + request_id = self.background_service.submit_fetch_request( + sport="nfl", + year=season_year, + url=ESPN_NFL_SCOREBOARD_URL, + cache_key=cache_key, + params={"dates": datestring, "limit": 1000}, + headers=self.headers, + timeout=timeout, + max_retries=max_retries, + priority=priority, + callback=fetch_callback + ) + + # Track the request + self.background_fetch_requests[season_year] = request_id # For immediate response, try to get partial data partial_data = self._get_weeks_data("nfl") @@ -155,7 +157,7 @@ def _fetch_data(self) -> Optional[Dict]: class NFLLiveManager(BaseNFLManager, FootballLive): # Renamed class """Manager for live NFL games.""" - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager): super().__init__(config, display_manager, cache_manager) self.logger = logging.getLogger('NFLLiveManager') # Changed logger name @@ -183,14 +185,14 @@ def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cach class NFLRecentManager(BaseNFLManager, SportsRecent): # Renamed class """Manager for recently completed NFL games.""" - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager): super().__init__(config, display_manager, cache_manager) self.logger = logging.getLogger('NFLRecentManager') # Changed logger name self.logger.info(f"Initialized NFLRecentManager with {len(self.favorite_teams)} favorite teams") class NFLUpcomingManager(BaseNFLManager, SportsUpcoming): # Renamed class """Manager for upcoming NFL games.""" - def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager): + def __init__(self, config: RootConfig, display_manager: DisplayManager, cache_manager: CacheManager): super().__init__(config, display_manager, cache_manager) self.logger = logging.getLogger('NFLUpcomingManager') # Changed logger name self.logger.info(f"Initialized NFLUpcomingManager with {len(self.favorite_teams)} favorite teams")