feat: Add soccer scoreboard display

This commit is contained in:
ChuckBuilds
2025-04-30 13:34:59 -05:00
parent af30cc1441
commit 890770bf2a
2 changed files with 1143 additions and 194 deletions

View File

@@ -20,6 +20,7 @@ from src.stock_news_manager import StockNewsManager
from src.nhl_managers import NHLLiveManager, NHLRecentManager, NHLUpcomingManager
from src.nba_managers import NBALiveManager, NBARecentManager, NBAUpcomingManager
from src.mlb_manager import MLBLiveManager, MLBRecentManager, MLBUpcomingManager
from src.soccer_managers import SoccerLiveManager, SoccerRecentManager, SoccerUpcomingManager
from src.youtube_display import YouTubeDisplay
from src.calendar_manager import CalendarManager
from src.text_display import TextDisplay
@@ -66,8 +67,10 @@ class DisplayController:
self.nhl_live = None
self.nhl_recent = None
self.nhl_upcoming = None
logger.info("NHL managers initialized in %.3f seconds", time.time() - nhl_time)
# Initialize NBA managers if enabled
nba_time = time.time()
nba_enabled = self.config.get('nba_scoreboard', {}).get('enabled', False)
nba_display_modes = self.config.get('nba_scoreboard', {}).get('display_modes', {})
@@ -79,6 +82,7 @@ class DisplayController:
self.nba_live = None
self.nba_recent = None
self.nba_upcoming = None
logger.info("NBA managers initialized in %.3f seconds", time.time() - nba_time)
# Initialize MLB managers if enabled
mlb_time = time.time()
@@ -93,7 +97,23 @@ class DisplayController:
self.mlb_live = None
self.mlb_recent = None
self.mlb_upcoming = None
logger.info("MLB managers initialized in %.3f seconds", time.time() - mlb_time)
# Initialize Soccer managers if enabled
soccer_time = time.time()
soccer_enabled = self.config.get('soccer_scoreboard', {}).get('enabled', False)
soccer_display_modes = self.config.get('soccer_scoreboard', {}).get('display_modes', {})
if soccer_enabled:
self.soccer_live = SoccerLiveManager(self.config, self.display_manager) if soccer_display_modes.get('soccer_live', True) else None
self.soccer_recent = SoccerRecentManager(self.config, self.display_manager) if soccer_display_modes.get('soccer_recent', True) else None
self.soccer_upcoming = SoccerUpcomingManager(self.config, self.display_manager) if soccer_display_modes.get('soccer_upcoming', True) else None
else:
self.soccer_live = None
self.soccer_recent = None
self.soccer_upcoming = None
logger.info("Soccer managers initialized in %.3f seconds", time.time() - soccer_time)
# Track MLB rotation state
self.mlb_current_team_index = 0
self.mlb_showing_recent = True
@@ -124,9 +144,15 @@ class DisplayController:
# Add MLB display modes if enabled
if mlb_enabled:
if mlb_display_modes.get('mlb_recent', True): self.available_modes.append('mlb_recent')
if mlb_display_modes.get('mlb_upcoming', True): self.available_modes.append('mlb_upcoming')
if self.mlb_recent: self.available_modes.append('mlb_recent') # Use recent if mode enabled
if self.mlb_upcoming: self.available_modes.append('mlb_upcoming') # Use upcoming if mode enabled
# mlb_live is handled separately when live games are available
# Add Soccer display modes if enabled
if soccer_enabled:
if self.soccer_recent: self.available_modes.append('soccer_recent')
if self.soccer_upcoming: self.available_modes.append('soccer_upcoming')
# soccer_live is handled separately when live games are available
# Set initial display to first available mode (clock)
self.current_mode_index = 0
@@ -135,7 +161,7 @@ class DisplayController:
self.force_clear = True
self.update_interval = 0.01 # Reduced from 0.1 to 0.01 for smoother scrolling
# Track team-based rotation states
# Track team-based rotation states (Add Soccer)
self.nhl_current_team_index = 0
self.nhl_showing_recent = True
self.nhl_favorite_teams = self.config.get('nhl_scoreboard', {}).get('favorite_teams', [])
@@ -146,8 +172,15 @@ class DisplayController:
self.nba_favorite_teams = self.config.get('nba_scoreboard', {}).get('favorite_teams', [])
self.in_nba_rotation = False
self.soccer_current_team_index = 0 # Soccer rotation state
self.soccer_showing_recent = True
self.soccer_favorite_teams = self.config.get('soccer_scoreboard', {}).get('favorite_teams', [])
self.in_soccer_rotation = False
# Update display durations to include all modes
self.display_durations = self.config['display'].get('display_durations', {
self.display_durations = self.config['display'].get('display_durations', {})
# Add defaults for soccer if missing
default_durations = {
'clock': 15,
'weather_current': 15,
'weather_hourly': 15,
@@ -164,27 +197,36 @@ class DisplayController:
'nba_upcoming': 20,
'mlb_live': 30,
'mlb_recent': 20,
'mlb_upcoming': 20
})
'mlb_upcoming': 20,
'soccer_live': 30, # Soccer durations
'soccer_recent': 20,
'soccer_upcoming': 20
}
# Merge loaded durations with defaults
for key, value in default_durations.items():
if key not in self.display_durations:
self.display_durations[key] = value
logger.info("DisplayController initialized with display_manager: %s", id(self.display_manager))
logger.info(f"Available display modes: {self.available_modes}")
logger.info(f"NHL Favorite teams: {self.nhl_favorite_teams}")
logger.info(f"NBA Favorite teams: {self.nba_favorite_teams}")
logger.info(f"MLB Favorite teams: {self.mlb_favorite_teams}")
logger.info("NHL managers initialized in %.3f seconds", time.time() - nhl_time)
logger.info("MLB managers initialized in %.3f seconds", time.time() - mlb_time)
logger.info(f"Soccer Favorite teams: {self.soccer_favorite_teams}") # Log Soccer teams
# Removed redundant NHL/MLB init time logs
def get_current_duration(self) -> int:
"""Get the duration for the current display mode."""
mode_key = self.current_display_mode
# Simplify weather key handling
if mode_key.startswith('weather_'):
duration_key = mode_key.split('_', 1)[1]
if duration_key == 'current': duration_key = 'weather'
elif duration_key == 'hourly': duration_key = 'hourly_forecast'
elif duration_key == 'daily': duration_key = 'daily_forecast'
else: duration_key = 'weather'
return self.display_durations.get(duration_key, 15)
return self.display_durations.get(mode_key, 15)
# duration_key = mode_key.split('_', 1)[1]
# if duration_key == 'current': duration_key = 'weather_current' # Keep specific keys
# elif duration_key == 'hourly': duration_key = 'weather_hourly'
# elif duration_key == 'daily': duration_key = 'weather_daily'
# else: duration_key = 'weather_current' # Default to current
# return self.display_durations.get(duration_key, 15)
return self.display_durations.get(mode_key, 15)
@@ -211,23 +253,29 @@ class DisplayController:
if self.mlb_live: self.mlb_live.update()
if self.mlb_recent: self.mlb_recent.update()
if self.mlb_upcoming: self.mlb_upcoming.update()
# Update Soccer managers
if self.soccer_live: self.soccer_live.update()
if self.soccer_recent: self.soccer_recent.update()
if self.soccer_upcoming: self.soccer_upcoming.update()
def _check_live_games(self) -> tuple[bool, str]:
"""
Check if there are any live games available.
Returns:
tuple[bool, str]: (has_live_games, sport_type)
sport_type will be 'nhl', 'nba', 'mlb' or None
sport_type will be 'nhl', 'nba', 'mlb', 'soccer' or None
"""
# Check NHL live games
# Prioritize sports (e.g., Soccer > NHL > NBA > MLB)
if self.soccer_live and self.soccer_live.live_games:
return True, 'soccer'
if self.nhl_live and self.nhl_live.live_games:
return True, 'nhl'
# Check NBA live games
if self.nba_live and self.nba_live.live_games:
return True, 'nba'
# Check MLB live games
if self.mlb_live and self.mlb_live.live_games:
return True, 'mlb'
@@ -238,114 +286,107 @@ class DisplayController:
Get games for a specific team and update the current game.
Args:
team: Team abbreviation
sport: 'nhl', 'nba', or 'mlb'
sport: 'nhl', 'nba', 'mlb', or 'soccer'
is_recent: Whether to look for recent or upcoming games
Returns:
bool: True if games were found and set
"""
manager_recent = None
manager_upcoming = None
games_list_attr = 'games_list' # Default for NHL/NBA
abbr_key_home = 'home_abbr'
abbr_key_away = 'away_abbr'
if sport == 'nhl':
if is_recent and self.nhl_recent:
# Find recent games for this team
for game in self.nhl_recent.games_list:
if game["home_abbr"] == team or game["away_abbr"] == team:
self.nhl_recent.current_game = game
return True
elif not is_recent and self.nhl_upcoming:
# Find upcoming games for this team
for game in self.nhl_upcoming.games_list:
if game["home_abbr"] == team or game["away_abbr"] == team:
self.nhl_upcoming.current_game = game
return True
manager_recent = self.nhl_recent
manager_upcoming = self.nhl_upcoming
elif sport == 'nba':
if is_recent and self.nba_recent:
# Find recent games for this team
for game in self.nba_recent.games_list:
if game["home_abbr"] == team or game["away_abbr"] == team:
self.nba_recent.current_game = game
return True
elif not is_recent and self.nba_upcoming:
# Find upcoming games for this team
for game in self.nba_upcoming.games_list:
if game["home_abbr"] == team or game["away_abbr"] == team:
self.nba_upcoming.current_game = game
return True
manager_recent = self.nba_recent
manager_upcoming = self.nba_upcoming
elif sport == 'mlb':
if is_recent and self.mlb_recent:
# Find recent games for this team
for game in self.mlb_recent.recent_games:
if game['home_team'] == team or game['away_team'] == team:
self.mlb_recent.current_game = game
return True
elif not is_recent and self.mlb_upcoming:
# Find upcoming games for this team
for game in self.mlb_upcoming.upcoming_games:
if game['home_team'] == team or game['away_team'] == team:
self.mlb_upcoming.current_game = game
return True
manager_recent = self.mlb_recent
manager_upcoming = self.mlb_upcoming
games_list_attr = 'recent_games' if is_recent else 'upcoming_games'
abbr_key_home = 'home_team' # MLB uses different keys
abbr_key_away = 'away_team'
elif sport == 'soccer':
manager_recent = self.soccer_recent
manager_upcoming = self.soccer_upcoming
games_list_attr = 'games_list' if is_recent else 'upcoming_games' # Soccer uses games_list/upcoming_games
manager = manager_recent if is_recent else manager_upcoming
if manager and hasattr(manager, games_list_attr):
game_list = getattr(manager, games_list_attr, [])
for game in game_list:
# Need to handle potential missing keys gracefully
home_team_abbr = game.get(abbr_key_home)
away_team_abbr = game.get(abbr_key_away)
if home_team_abbr == team or away_team_abbr == team:
manager.current_game = game
return True
return False
def _has_team_games(self, sport: str = 'nhl') -> bool:
"""Check if there are any games for favorite teams."""
favorite_teams = []
manager_recent = None
manager_upcoming = None
if sport == 'nhl':
return bool(self.nhl_favorite_teams and (self.nhl_recent or self.nhl_upcoming))
favorite_teams = self.nhl_favorite_teams
manager_recent = self.nhl_recent
manager_upcoming = self.nhl_upcoming
elif sport == 'nba':
return bool(self.nba_favorite_teams and (self.nba_recent or self.nba_upcoming))
favorite_teams = self.nba_favorite_teams
manager_recent = self.nba_recent
manager_upcoming = self.nba_upcoming
elif sport == 'mlb':
return bool(self.mlb_favorite_teams and (self.mlb_recent or self.mlb_upcoming))
return False
favorite_teams = self.mlb_favorite_teams
manager_recent = self.mlb_recent
manager_upcoming = self.mlb_upcoming
elif sport == 'soccer':
favorite_teams = self.soccer_favorite_teams
manager_recent = self.soccer_recent
manager_upcoming = self.soccer_upcoming
return bool(favorite_teams and (manager_recent or manager_upcoming))
def _rotate_team_games(self, sport: str = 'nhl') -> None:
"""Rotate through games for favorite teams."""
"""Rotate through games for favorite teams. (No longer used directly in loop)"""
# This logic is now mostly handled within each manager's display/update
# Keeping the structure in case direct rotation is needed later.
if not self._has_team_games(sport):
return
if sport == 'nhl':
if not self._has_team_games('nhl'): return
# Try to find games for current team
if not self.nhl_favorite_teams: return
current_team = self.nhl_favorite_teams[self.nhl_current_team_index]
found_games = self._get_team_games(current_team, 'nhl', self.nhl_showing_recent)
if not found_games:
# Try opposite type (recent/upcoming) for same team
self.nhl_showing_recent = not self.nhl_showing_recent
found_games = self._get_team_games(current_team, 'nhl', self.nhl_showing_recent)
if not found_games:
# Move to next team
self.nhl_current_team_index = (self.nhl_current_team_index + 1) % len(self.nhl_favorite_teams)
self.nhl_showing_recent = True # Reset to recent games for next team
# ... (rest of NHL rotation logic - now less relevant)
elif sport == 'nba':
if not self._has_team_games('nba'): return
# Try to find games for current team
current_team = self.nba_favorite_teams[self.nba_current_team_index]
found_games = self._get_team_games(current_team, 'nba', self.nba_showing_recent)
if not found_games:
# Try opposite type (recent/upcoming) for same team
self.nba_showing_recent = not self.nba_showing_recent
found_games = self._get_team_games(current_team, 'nba', self.nba_showing_recent)
if not found_games:
# Move to next team
self.nba_current_team_index = (self.nba_current_team_index + 1) % len(self.nba_favorite_teams)
self.nba_showing_recent = True # Reset to recent games for next team
if not self.nba_favorite_teams: return
current_team = self.nba_favorite_teams[self.nba_current_team_index]
# ... (rest of NBA rotation logic)
elif sport == 'mlb':
if not self._has_team_games('mlb'): return
# Try to find games for current team
if not self.mlb_favorite_teams: return
current_team = self.mlb_favorite_teams[self.mlb_current_team_index]
found_games = self._get_team_games(current_team, 'mlb', self.mlb_showing_recent)
# ... (rest of MLB rotation logic)
elif sport == 'soccer':
if not self.soccer_favorite_teams: return
current_team = self.soccer_favorite_teams[self.soccer_current_team_index]
# Try to find games for current team (recent first)
found_games = self._get_team_games(current_team, 'soccer', self.soccer_showing_recent)
if not found_games:
# Try opposite type (upcoming/recent)
self.soccer_showing_recent = not self.soccer_showing_recent
found_games = self._get_team_games(current_team, 'soccer', self.soccer_showing_recent)
if not found_games:
# Try opposite type (recent/upcoming) for same team
self.mlb_showing_recent = not self.mlb_showing_recent
found_games = self._get_team_games(current_team, 'mlb', self.mlb_showing_recent)
if not found_games:
# Move to next team
self.mlb_current_team_index = (self.mlb_current_team_index + 1) % len(self.mlb_favorite_teams)
self.mlb_showing_recent = True # Reset to recent games for next team
# Move to next team if no games found for current one
self.soccer_current_team_index = (self.soccer_current_team_index + 1) % len(self.soccer_favorite_teams)
self.soccer_showing_recent = True # Reset to recent for the new team
# Maybe try finding game for the *new* team immediately? Optional.
def run(self):
"""Run the display controller, switching between displays."""
@@ -361,109 +402,150 @@ class DisplayController:
# Update data for all modules
self._update_modules()
# Check for live games
has_live_games, sport_type = self._check_live_games()
# Check for live games (priority: Soccer > NHL > NBA > MLB)
has_live_games, live_sport_type = self._check_live_games()
# If we have live games, cycle through them
# --- Live Game Handling ---
if has_live_games:
# Check if it's time to switch live games
if current_time - self.last_switch > self.get_current_duration():
# Switch between NHL, NBA, and MLB live games if multiple are available
if sport_type == 'nhl' and self.nba_live and self.nba_live.live_games:
sport_type = 'nba'
self.last_switch = current_time
self.force_clear = True
elif sport_type == 'nba' and self.mlb_live and self.mlb_live.live_games:
sport_type = 'mlb'
self.last_switch = current_time
self.force_clear = True
elif sport_type == 'mlb' and self.nhl_live and self.nhl_live.live_games:
sport_type = 'nhl'
self.last_switch = current_time
self.force_clear = True
# Determine which live manager to use
live_manager = None
if live_sport_type == 'soccer' and self.soccer_live:
live_manager = self.soccer_live
current_mode_for_duration = 'soccer_live'
elif live_sport_type == 'nhl' and self.nhl_live:
live_manager = self.nhl_live
current_mode_for_duration = 'nhl_live'
elif live_sport_type == 'nba' and self.nba_live:
live_manager = self.nba_live
current_mode_for_duration = 'nba_live'
elif live_sport_type == 'mlb' and self.mlb_live:
live_manager = self.mlb_live
current_mode_for_duration = 'mlb_live'
# Display the current live game
if sport_type == 'nhl' and self.nhl_live:
self.nhl_live.update() # Force update to get latest data
self.nhl_live.display(force_clear=self.force_clear)
elif sport_type == 'nba' and self.nba_live:
self.nba_live.update() # Force update to get latest data
self.nba_live.display(force_clear=self.force_clear)
elif sport_type == 'mlb' and self.mlb_live:
self.mlb_live.update() # Force update to get latest data
self.mlb_live.display(force_clear=self.force_clear)
self.force_clear = False
continue # Skip the rest of the loop to stay on live games
# Only proceed with mode switching if no live games
if current_time - self.last_switch > self.get_current_duration():
# No live games, continue with regular rotation
# If we're currently on calendar, advance to next event before switching modes
if self.current_display_mode == 'calendar' and self.calendar:
self.calendar.advance_event()
self.current_mode_index = (self.current_mode_index + 1) % len(self.available_modes)
self.current_display_mode = self.available_modes[self.current_mode_index]
logger.info(f"Switching to: {self.current_display_mode}")
self.force_clear = True
self.last_switch = current_time
if live_manager:
# Check if switching *into* live mode or switching *between* live sports
if self.current_display_mode != current_mode_for_duration:
logger.info(f"Switching to LIVE mode: {current_mode_for_duration}")
self.current_display_mode = current_mode_for_duration
self.force_clear = True
self.last_switch = current_time # Reset timer for live duration
# Display current mode frame (only for non-live modes)
try:
if self.current_display_mode == 'clock' and self.clock:
self.clock.display_time(force_clear=self.force_clear)
elif self.current_display_mode == 'weather_current' and self.weather:
self.weather.display_weather(force_clear=self.force_clear)
elif self.current_display_mode == 'weather_hourly' and self.weather:
self.weather.display_hourly_forecast(force_clear=self.force_clear)
elif self.current_display_mode == 'weather_daily' and self.weather:
self.weather.display_daily_forecast(force_clear=self.force_clear)
elif self.current_display_mode == 'stocks' and self.stocks:
self.stocks.display_stocks(force_clear=self.force_clear)
elif self.current_display_mode == 'stock_news' and self.news:
self.news.display_news()
elif self.current_display_mode == 'calendar' and self.calendar:
self.calendar.display(force_clear=self.force_clear)
elif self.current_display_mode == 'nhl_recent' and self.nhl_recent:
self.nhl_recent.display(force_clear=self.force_clear)
elif self.current_display_mode == 'nhl_upcoming' and self.nhl_upcoming:
self.nhl_upcoming.display(force_clear=self.force_clear)
elif self.current_display_mode == 'nba_recent' and self.nba_recent:
self.nba_recent.display(force_clear=self.force_clear)
elif self.current_display_mode == 'nba_upcoming' and self.nba_upcoming:
self.nba_upcoming.display(force_clear=self.force_clear)
elif self.current_display_mode == 'mlb_recent' and self.mlb_recent:
self.mlb_recent.display(force_clear=self.force_clear)
elif self.current_display_mode == 'mlb_upcoming' and self.mlb_upcoming:
self.mlb_upcoming.display(force_clear=self.force_clear)
elif self.current_display_mode == 'youtube' and self.youtube:
self.youtube.display(force_clear=self.force_clear)
elif self.current_display_mode == 'text_display' and self.text_display:
self.text_display.display()
except Exception as e:
logger.error(f"Error updating display for mode {self.current_display_mode}: {e}", exc_info=True)
continue
# Display the current live game using the selected manager
live_manager.display(force_clear=self.force_clear)
self.force_clear = False # Clear only once when switching
else:
# Should not happen if _check_live_games is correct, but handle defensively
has_live_games = False # Fall back to regular rotation
logger.warning("Live game detected but corresponding manager is None.")
self.force_clear = False
# --- Regular Mode Rotation (only if NO live games) ---
if not has_live_games:
# Check if we were just in live mode and need to switch back
if self.current_display_mode.endswith('_live'):
logger.info(f"Switching back to regular rotation from {self.current_display_mode}")
# Find the next available mode in the regular list
self.current_mode_index = (self.current_mode_index + 1) % len(self.available_modes)
self.current_display_mode = self.available_modes[self.current_mode_index]
self.force_clear = True
self.last_switch = current_time
logger.info(f"Switching to: {self.current_display_mode}")
# Check if it's time to switch modes based on duration
elif current_time - self.last_switch > self.get_current_duration():
# Advance calendar event before switching away
if self.current_display_mode == 'calendar' and self.calendar:
self.calendar.advance_event()
self.current_mode_index = (self.current_mode_index + 1) % len(self.available_modes)
self.current_display_mode = self.available_modes[self.current_mode_index]
logger.info(f"Switching to: {self.current_display_mode}")
self.force_clear = True
self.last_switch = current_time
# Display current mode frame
try:
display_updated = False # Flag to track if display was handled
if self.current_display_mode == 'clock' and self.clock:
self.clock.display_time(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'weather_current' and self.weather:
self.weather.display_weather(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'weather_hourly' and self.weather:
self.weather.display_hourly_forecast(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'weather_daily' and self.weather:
self.weather.display_daily_forecast(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'stocks' and self.stocks:
self.stocks.display_stocks(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'stock_news' and self.news:
self.news.display_news() # Assumes news handles its own clearing/drawing
display_updated = True
elif self.current_display_mode == 'calendar' and self.calendar:
self.calendar.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'nhl_recent' and self.nhl_recent:
self.nhl_recent.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'nhl_upcoming' and self.nhl_upcoming:
self.nhl_upcoming.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'nba_recent' and self.nba_recent:
self.nba_recent.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'nba_upcoming' and self.nba_upcoming:
self.nba_upcoming.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'mlb_recent' and self.mlb_recent:
self.mlb_recent.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'mlb_upcoming' and self.mlb_upcoming:
self.mlb_upcoming.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'soccer_recent' and self.soccer_recent:
self.soccer_recent.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'soccer_upcoming' and self.soccer_upcoming:
self.soccer_upcoming.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'youtube' and self.youtube:
self.youtube.display(force_clear=self.force_clear)
display_updated = True
elif self.current_display_mode == 'text_display' and self.text_display:
self.text_display.display() # Assumes text handles its own drawing
display_updated = True
# Reset force_clear only if a display method was actually called
if display_updated:
self.force_clear = False
except Exception as e:
logger.error(f"Error updating display for mode {self.current_display_mode}: {e}", exc_info=True)
# Continue to next iteration after error
# Small sleep to prevent high CPU usage
time.sleep(self.update_interval)
except KeyboardInterrupt:
logger.info("Display controller stopped by user")
except Exception as e:
logger.error(f"Error in display controller: {e}", exc_info=True)
logger.error(f"Critical error in display controller run loop: {e}", exc_info=True)
finally:
logger.info("Cleaning up display manager...")
self.display_manager.cleanup()
logger.info("Cleanup complete.")
def main():
controller = DisplayController()

867
src/soccer_managers.py Normal file
View File

@@ -0,0 +1,867 @@
import os
import time
import logging
import requests
import json
from typing import Dict, Any, Optional, List
from PIL import Image, ImageDraw, ImageFont
from pathlib import Path
from datetime import datetime, timedelta, timezone
from src.display_manager import DisplayManager
from src.cache_manager import CacheManager
# Constants
ESPN_SOCCER_SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/soccer/scoreboards"
# Common league slugs (add more as needed)
LEAGUE_SLUGS = {
"eng.1": "Premier League",
"esp.1": "La Liga",
"ger.1": "Bundesliga",
"ita.1": "Serie A",
"fra.1": "Ligue 1",
"uefa.champions": "Champions League",
"uefa.europa": "Europa League",
"usa.1": "MLS",
# Add other leagues here if needed
}
# Configure logging to match main configuration
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s.%(msecs)03d - %(levelname)s:%(name)s:%(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
class CacheManager:
"""Manages caching of ESPN API responses."""
_instance = None
_cache = {}
_cache_timestamps = {}
def __new__(cls):
if cls._instance is None:
cls._instance = super(CacheManager, cls).__new__(cls)
return cls._instance
@classmethod
def get(cls, key: str, max_age: int = 60) -> Optional[Dict]:
"""
Get data from cache if it exists and is not stale.
Args:
key: Cache key (usually the date string or league)
max_age: Maximum age of cached data in seconds
Returns:
Cached data if valid, None if missing or stale
"""
if key not in cls._cache:
return None
timestamp = cls._cache_timestamps.get(key, 0)
if time.time() - timestamp > max_age:
# Data is stale, remove it
del cls._cache[key]
del cls._cache_timestamps[key]
return None
return cls._cache[key]
@classmethod
def set(cls, key: str, data: Dict) -> None:
"""
Store data in cache with current timestamp.
Args:
key: Cache key
data: Data to cache
"""
cls._cache[key] = data
cls._cache_timestamps[key] = time.time()
@classmethod
def clear(cls) -> None:
"""Clear all cached data."""
cls._cache.clear()
cls._cache_timestamps.clear()
class BaseSoccerManager:
"""Base class for Soccer managers with common functionality."""
# Class variables for warning tracking
_no_data_warning_logged = False
_last_warning_time = 0
_warning_cooldown = 60 # Only log warnings once per minute
_shared_data = {} # Dictionary to hold shared data per league/date
_last_shared_update = {} # Dictionary for update times per league/date
cache_manager = CacheManager()
logger = logging.getLogger('Soccer') # Use 'Soccer' logger
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
self.display_manager = display_manager
self.config = config
self.soccer_config = config.get("soccer_scoreboard", {}) # Use 'soccer_scoreboard' config
self.is_enabled = self.soccer_config.get("enabled", False)
self.test_mode = self.soccer_config.get("test_mode", False)
self.logo_dir = self.soccer_config.get("logo_dir", "assets/sports/soccer_logos") # Soccer logos
self.update_interval = self.soccer_config.get("update_interval_seconds", 60)
self.last_update = 0
self.current_game = None
self.fonts = self._load_fonts()
self.favorite_teams = self.soccer_config.get("favorite_teams", [])
self.target_leagues = self.soccer_config.get("leagues", list(LEAGUE_SLUGS.keys())) # Get target leagues from config
self.recent_hours = self.soccer_config.get("recent_game_hours", 48)
self.logger.setLevel(logging.DEBUG)
display_config = config.get("display", {})
hardware_config = display_config.get("hardware", {})
cols = hardware_config.get("cols", 64)
chain = hardware_config.get("chain_length", 1)
self.display_width = int(cols * chain)
self.display_height = hardware_config.get("rows", 32)
self._logo_cache = {}
self.logger.info(f"Initialized Soccer manager with display dimensions: {self.display_width}x{self.display_height}")
self.logger.info(f"Logo directory: {self.logo_dir}")
self.logger.info(f"Target leagues: {self.target_leagues}")
@classmethod
def _fetch_shared_data(cls, date_str: str = None) -> Optional[Dict]:
"""Fetch and cache data for all managers to share."""
current_time = time.time()
all_data = {"events": []} # Combine data from multiple dates/leagues
# Determine dates to fetch
today = datetime.now(timezone.utc).date()
dates_to_fetch = [
(today - timedelta(days=1)).strftime('%Y%m%d'), # Yesterday
today.strftime('%Y%m%d'), # Today
(today + timedelta(days=1)).strftime('%Y%m%d') # Tomorrow (for upcoming)
]
if date_str and date_str not in dates_to_fetch:
dates_to_fetch.append(date_str) # Ensure specific requested date is included
for fetch_date in dates_to_fetch:
cache_key = f"soccer_{fetch_date}"
# Check cache first
cached_data = cls.cache_manager.get(cache_key, max_age=300) # 5 minutes cache
if cached_data:
cls.logger.info(f"[Soccer] Using cached data for {fetch_date}")
if "events" in cached_data:
all_data["events"].extend(cached_data["events"])
continue # Skip fetching if cached
# If not in cache or stale, fetch from API
try:
url = ESPN_SOCCER_SCOREBOARD_URL
params = {'dates': fetch_date, 'limit': 500} # Fetch more events if needed
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
cls.logger.info(f"[Soccer] Successfully fetched data from ESPN API for date {fetch_date}")
# Cache the response
cls.cache_manager.set(cache_key, data)
if "events" in data:
all_data["events"].extend(data["events"])
except requests.exceptions.RequestException as e:
cls.logger.error(f"[Soccer] Error fetching data from ESPN for date {fetch_date}: {e}")
# Continue to try other dates even if one fails
# Update shared data and timestamp (using a generic key for simplicity)
cls._shared_data = all_data
cls._last_shared_update = current_time
return cls._shared_data
def _fetch_data(self, date_str: str = None) -> Optional[Dict]:
"""Fetch data using shared data mechanism, ensuring fresh data for live games."""
if isinstance(self, SoccerLiveManager) and not self.test_mode:
# Live manager bypasses shared cache for most recent data
try:
url = ESPN_SOCCER_SCOREBOARD_URL
# Fetch only today's data for live games
today_date_str = datetime.now(timezone.utc).strftime('%Y%m%d')
params = {'dates': today_date_str, 'limit': 500}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
self.logger.info(f"[Soccer] Successfully fetched live game data from ESPN API for {today_date_str}")
# Filter by target leagues immediately
if "events" in data:
data["events"] = [
event for event in data["events"]
if event.get("league", {}).get("slug") in self.target_leagues
]
return data
except requests.exceptions.RequestException as e:
self.logger.error(f"[Soccer] Error fetching live game data from ESPN: {e}")
return None
else:
# For non-live games or test mode, use the shared data fetch
data = self._fetch_shared_data(date_str)
# Filter shared data by target leagues
if data and "events" in data:
filtered_events = [
event for event in data["events"]
if event.get("league", {}).get("slug") in self.target_leagues
]
# Return a copy to avoid modifying the shared cache
return {"events": filtered_events}
return data
def _load_fonts(self):
"""Load fonts used by the scoreboard."""
fonts = {}
try:
fonts['score'] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8) # Smaller score
fonts['time'] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
fonts['team'] = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6) # Even smaller team abbr
fonts['status'] = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6) # Small status
logging.info("[Soccer] Successfully loaded custom fonts")
except IOError:
logging.warning("[Soccer] Custom fonts not found, using default PIL font.")
fonts['score'] = ImageFont.load_default()
fonts['time'] = ImageFont.load_default()
fonts['team'] = ImageFont.load_default()
fonts['status'] = ImageFont.load_default()
return fonts
def _draw_text_with_outline(self, draw, text, position, font, fill=(255, 255, 255), outline_color=(0, 0, 0)):
"""Draw text with a black outline for better readability."""
x, y = position
# Draw outline
for dx, dy in [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 1), (1, -1), (1, 0), (1, 1)]:
draw.text((x + dx, y + dy), text, font=font, fill=outline_color)
# Draw text
draw.text((x, y), text, font=font, fill=fill)
def _load_and_resize_logo(self, team_abbrev: str) -> Optional[Image.Image]:
"""Load and resize a team logo, with caching."""
if team_abbrev in self._logo_cache:
return self._logo_cache[team_abbrev]
logo_path = os.path.join(self.logo_dir, f"{team_abbrev}.png")
self.logger.debug(f"Logo path: {logo_path}")
try:
if not os.path.exists(logo_path):
self.logger.info(f"Creating placeholder logo for {team_abbrev}")
os.makedirs(os.path.dirname(logo_path), exist_ok=True)
logo = Image.new('RGBA', (24, 24), (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255), 255)) # Smaller default logo
draw = ImageDraw.Draw(logo)
# Simple placeholder: colored square
# Optionally add text, but keep it simple
# draw.text((2, 2), team_abbrev[:3], fill=(0,0,0,255)) # Draw abbreviation if needed
logo.save(logo_path)
self.logger.info(f"Created placeholder logo at {logo_path}")
logo = Image.open(logo_path)
if logo.mode != 'RGBA':
logo = logo.convert('RGBA')
# Resize logo to fit better in soccer layout (e.g., 16x16 or 20x20)
target_size = 20 # Smaller logos for soccer
logo.thumbnail((target_size, target_size), Image.Resampling.LANCZOS)
self.logger.debug(f"Resized {team_abbrev} logo to {logo.size}")
self._logo_cache[team_abbrev] = logo
return logo
except Exception as e:
self.logger.error(f"Error loading logo for {team_abbrev}: {e}", exc_info=True)
return None
def _format_game_time(self, status: Dict) -> str:
"""Format game time display for soccer (e.g., HT, FT, 45', 90+2')."""
status_type = status["type"]["name"]
clock = status.get("displayClock", "0:00")
period = status.get("period", 0)
if status_type == "STATUS_FINAL":
return "FT"
if status_type == "STATUS_HALFTIME":
return "HT"
if status_type == "STATUS_SCHEDULED":
return "" # Handled by is_upcoming
if status_type == "STATUS_POSTPONED":
return "PPD"
if status_type == "STATUS_CANCELED":
return "CANC"
# Handle live game time
if status_type == "STATUS_IN_PROGRESS":
# Simple clock display, potentially add period info if needed
# Remove seconds for cleaner display
if ':' in clock:
clock_parts = clock.split(':')
return f"{clock_parts[0]}'" # Display as minutes'
else:
return clock # Fallback
return clock # Default fallback
def _extract_game_details(self, game_event: Dict) -> Optional[Dict]:
"""Extract relevant game details from ESPN Soccer API response."""
if not game_event:
return None
try:
competition = game_event["competitions"][0]
status = competition["status"]
competitors = competition["competitors"]
game_date_str = game_event["date"]
league_info = game_event.get("league", {})
league_slug = league_info.get("slug")
league_name = league_info.get("name", league_slug) # Use name if available
# Filter out games not in target leagues (redundant check, but safe)
if league_slug not in self.target_leagues:
self.logger.debug(f"[Soccer] Skipping game from league: {league_name} ({league_slug})")
return None
try:
start_time_utc = datetime.fromisoformat(game_date_str.replace("Z", "+00:00"))
except ValueError:
logging.warning(f"[Soccer] Could not parse game date: {game_date_str}")
start_time_utc = None
home_team = next(c for c in competitors if c.get("homeAway") == "home")
away_team = next(c for c in competitors if c.get("homeAway") == "away")
game_time = ""
game_date = ""
if start_time_utc:
local_time = start_time_utc.astimezone()
game_time = local_time.strftime("%-I:%M%p").lower() # e.g., 2:30pm
game_date = local_time.strftime("%-m/%-d")
status_type = status["type"]["name"]
is_live = status_type == "STATUS_IN_PROGRESS"
is_final = status_type == "STATUS_FINAL"
is_upcoming = status_type == "STATUS_SCHEDULED"
is_halftime = status_type == "STATUS_HALFTIME"
# Calculate if game is within recent/upcoming window
is_within_window = False
if start_time_utc:
now_utc = datetime.now(timezone.utc)
if is_upcoming:
cutoff_time = now_utc + timedelta(hours=self.recent_hours)
is_within_window = start_time_utc <= cutoff_time
else: # Recent or live
cutoff_time = now_utc - timedelta(hours=self.recent_hours)
is_within_window = start_time_utc >= cutoff_time
details = {
"id": game_event["id"],
"start_time_utc": start_time_utc,
"status_text": status["type"]["shortDetail"],
"game_clock_display": self._format_game_time(status),
"period": status.get("period", 0), # 1st half, 2nd half, ET periods?
"is_live": is_live or is_halftime, # Treat halftime as live for display purposes
"is_final": is_final,
"is_upcoming": is_upcoming,
"is_within_window": is_within_window,
"home_abbr": home_team["team"]["abbreviation"],
"home_score": home_team.get("score", "0"),
"home_logo": self._load_and_resize_logo(home_team["team"]["abbreviation"]),
"away_abbr": away_team["team"]["abbreviation"],
"away_score": away_team.get("score", "0"),
"away_logo": self._load_and_resize_logo(away_team["team"]["abbreviation"]),
"game_time": game_time, # Formatted local time (e.g., 2:30pm)
"game_date": game_date, # Formatted local date (e.g., 7/21)
"league": league_name
}
self.logger.debug(f"[Soccer] Extracted game: {details['away_abbr']} {details['away_score']} @ {details['home_abbr']} {details['home_score']} ({details['game_clock_display']}) - League: {details['league']} - Final: {details['is_final']}, Upcoming: {details['is_upcoming']}, Live: {details['is_live']}, Within Window: {details['is_within_window']}")
# Basic validation (logos handled in loading)
if not details["home_abbr"] or not details["away_abbr"]:
logging.warning(f"[Soccer] Missing team abbreviation in game data: {game_event['id']}")
return None
return details
except Exception as e:
logging.error(f"[Soccer] Error extracting game details for event {game_event.get('id', 'N/A')}: {e}", exc_info=True)
return None
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
"""Draw the soccer scorebug layout."""
try:
main_img = Image.new('RGB', (self.display_width, self.display_height), (0, 0, 0))
draw = ImageDraw.Draw(main_img)
home_logo = game.get("home_logo")
away_logo = game.get("away_logo")
# --- Layout Configuration ---
logo_y = (self.display_height - (home_logo.height if home_logo else 20)) // 2
away_logo_x = 2
home_logo_x = self.display_width - (home_logo.width if home_logo else 20) - 2
center_x = self.display_width // 2
score_y = logo_y # Align score vertically with logos
abbr_y = score_y + (self.fonts['score'].size if 'score' in self.fonts else 10) + 1 # Below score
status_y = 1 # Status/Time at the top center
# --- Draw Logos ---
if away_logo:
main_img.paste(away_logo, (away_logo_x, logo_y), away_logo)
if home_logo:
main_img.paste(home_logo, (home_logo_x, logo_y), home_logo)
# --- Draw Team Abbreviations ---
away_abbr = game.get("away_abbr", "AWAY")
home_abbr = game.get("home_abbr", "HOME")
abbr_font = self.fonts['team']
away_abbr_width = draw.textlength(away_abbr, font=abbr_font)
home_abbr_width = draw.textlength(home_abbr, font=abbr_font)
# Position abbreviations near logos
away_abbr_x = away_logo_x + (away_logo.width if away_logo else 20) + 2
home_abbr_x = home_logo_x - home_abbr_width - 2
self._draw_text_with_outline(draw, away_abbr, (away_abbr_x, abbr_y), abbr_font)
self._draw_text_with_outline(draw, home_abbr, (home_abbr_x, abbr_y), abbr_font)
# --- Draw Score / Game Time ---
score_font = self.fonts['score']
status_font = self.fonts['time'] # Use 'time' font for status line
if game.get("is_upcoming"):
# Display Date and Time for upcoming games
game_date = game.get("game_date", "")
game_time = game.get("game_time", "")
date_time_text = f"{game_date} {game_time}"
date_time_width = draw.textlength(date_time_text, font=status_font)
date_time_x = center_x - date_time_width // 2
# Position below logos/abbrs
date_time_y = abbr_y + (self.fonts['team'].size if 'team' in self.fonts else 8) + 2
self._draw_text_with_outline(draw, date_time_text, (date_time_x, date_time_y), status_font)
# Show "Upcoming" status at the top
status_text = "Upcoming"
status_width = draw.textlength(status_text, font=status_font)
status_x = center_x - status_width // 2
self._draw_text_with_outline(draw, status_text, (status_x, status_y), status_font)
else:
# Display Score for live/final games
home_score = str(game.get("home_score", "0"))
away_score = str(game.get("away_score", "0"))
score_text = f"{away_score} - {home_score}"
score_width = draw.textlength(score_text, font=score_font)
score_x = center_x - score_width // 2
self._draw_text_with_outline(draw, score_text, (score_x, score_y), score_font)
# --- Draw Game Status/Time ---
status_text = game.get("game_clock_display", "")
status_width = draw.textlength(status_text, font=status_font)
status_x = center_x - status_width // 2
self._draw_text_with_outline(draw, status_text, (status_x, status_y), status_font)
# --- Display Image ---
self.display_manager.image.paste(main_img, (0, 0))
self.display_manager.update_display()
except Exception as e:
self.logger.error(f"Error displaying soccer game: {e}", exc_info=True)
def display(self, force_clear: bool = False) -> None:
"""Common display method for all Soccer managers"""
if not self.current_game:
current_time = time.time()
if not hasattr(self, '_last_warning_time'):
self._last_warning_time = 0
if current_time - self._last_warning_time > 300:
self.logger.warning("[Soccer] No game data available to display")
self._last_warning_time = current_time
return
self._draw_scorebug_layout(self.current_game, force_clear)
# ===============================================
# Manager Implementations (Live, Recent, Upcoming)
# ===============================================
class SoccerLiveManager(BaseSoccerManager):
"""Manager for live Soccer games."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
super().__init__(config, display_manager)
self.update_interval = self.soccer_config.get("live_update_interval", 20) # Slightly longer for soccer?
self.no_data_interval = 300
self.last_update = 0
self.logger.info("Initialized Soccer Live Manager")
self.live_games = []
self.current_game_index = 0
self.last_game_switch = 0
self.game_display_duration = self.soccer_config.get("live_game_duration", 20)
self.last_display_update = 0
self.last_log_time = 0
self.log_interval = 300
if self.test_mode:
# Simple test game
self.current_game = {
"id": "test001",
"home_abbr": "FCB", "away_abbr": "RMA", "home_score": "1", "away_score": "1",
"game_clock_display": "65'", "period": 2, "is_live": True, "is_final": False, "is_upcoming": False,
"home_logo": self._load_and_resize_logo("FCB"), "away_logo": self._load_and_resize_logo("RMA"),
"league": "Test League"
}
self.live_games = [self.current_game]
logging.info("[Soccer] Initialized SoccerLiveManager with test game: FCB vs RMA")
else:
logging.info("[Soccer] Initialized SoccerLiveManager in live mode")
def update(self):
"""Update live game data."""
current_time = time.time()
interval = self.no_data_interval if not self.live_games else self.update_interval
if current_time - self.last_update >= interval:
self.last_update = current_time
if self.test_mode:
# Basic test mode clock update
if self.current_game and self.current_game["is_live"]:
try:
minutes_str = self.current_game["game_clock_display"].replace("'", "")
minutes = int(minutes_str)
minutes += 1
if minutes == 45: self.current_game["game_clock_display"] = "HT"
elif minutes == 46: self.current_game["period"] = 2 # Start 2nd half
elif minutes > 90: self.current_game["game_clock_display"] = "FT"; self.current_game["is_live"]=False; self.current_game["is_final"]=True
else: self.current_game["game_clock_display"] = f"{minutes}'"
except ValueError: # Handle HT, FT states
if self.current_game["game_clock_display"] == "HT":
self.current_game["game_clock_display"] = "46'" # Start 2nd half after HT
self.current_game["period"] = 2
pass # Do nothing if FT or other non-numeric
# Always update display in test mode
# self.display(force_clear=True) # Display handled by controller loop
else:
# Fetch live game data
data = self._fetch_data()
new_live_games = []
if data and "events" in data:
for event in data["events"]:
details = self._extract_game_details(event)
# Ensure it's live and involves a favorite team (if specified)
if details and details["is_live"]:
if not self.favorite_teams or (
details["home_abbr"] in self.favorite_teams or
details["away_abbr"] in self.favorite_teams
):
new_live_games.append(details)
# Logging
should_log = (current_time - self.last_log_time >= self.log_interval or
len(new_live_games) != len(self.live_games) or
not self.live_games)
if should_log:
if new_live_games:
self.logger.info(f"[Soccer] Found {len(new_live_games)} live games involving favorite teams / all teams.")
for game in new_live_games:
self.logger.info(f"[Soccer] Live game: {game['away_abbr']} vs {game['home_abbr']} ({game['game_clock_display']}) - {game['league']}")
else:
self.logger.info("[Soccer] No live games found matching criteria.")
self.last_log_time = current_time
# Update game list and current game
if new_live_games:
# Check if the list of games actually changed (based on ID)
new_game_ids = {game['id'] for game in new_live_games}
current_game_ids = {game['id'] for game in self.live_games}
if new_game_ids != current_game_ids:
self.live_games = sorted(new_live_games, key=lambda x: x['start_time_utc'] or datetime.now(timezone.utc)) # Sort by time
# Reset index if current game is gone or list is new
if not self.current_game or self.current_game['id'] not in new_game_ids:
self.current_game_index = 0
if self.live_games:
self.current_game = self.live_games[0]
self.last_game_switch = current_time # Reset switch timer
else:
self.current_game = None
else:
# Update the currently displayed game data if it still exists
try:
current_game_id = self.current_game['id']
self.current_game = next(g for g in new_live_games if g['id'] == current_game_id)
except StopIteration:
# Should not happen if check above works, but handle defensively
self.current_game_index = 0
self.current_game = self.live_games[0] if self.live_games else None
self.last_game_switch = current_time
else: # Games are the same, just update data
updated_live_games = []
for existing_game in self.live_games:
try:
updated_game = next(g for g in new_live_games if g['id'] == existing_game['id'])
updated_live_games.append(updated_game)
# Update current_game if it's the one being displayed
if self.current_game and self.current_game['id'] == updated_game['id']:
self.current_game = updated_game
except StopIteration:
pass # Game disappeared between checks?
self.live_games = updated_live_games
# Limit display updates
# if current_time - self.last_display_update >= 1.0:
# self.display(force_clear=True) # Display handled by controller
# self.last_display_update = current_time
else:
# No live games found
if self.live_games: # Log only if previously had games
self.logger.info("[Soccer] All live games have ended or no longer match criteria.")
self.live_games = []
self.current_game = None
# Check if it's time to switch displayed game
if len(self.live_games) > 1 and (current_time - self.last_game_switch) >= self.game_display_duration:
self.current_game_index = (self.current_game_index + 1) % len(self.live_games)
self.current_game = self.live_games[self.current_game_index]
self.last_game_switch = current_time
# Force display handled by controller on mode switch
# self.display(force_clear=True) # Display handled by controller
# self.last_display_update = current_time
def display(self, force_clear: bool = False):
"""Display live game information."""
# This method might be redundant if controller handles display calls
# but keep it for potential direct calls or consistency
if not self.current_game:
# Optionally clear screen or show 'No Live Games' message
# self.display_manager.clear_display() # Example
return
super().display(force_clear)
class SoccerRecentManager(BaseSoccerManager):
"""Manager for recently completed Soccer games."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
super().__init__(config, display_manager)
self.recent_games = [] # Holds all fetched recent games matching criteria
self.games_list = [] # Holds games filtered by favorite teams (if applicable)
self.current_game_index = 0
self.last_update = 0
self.update_interval = 300 # 5 minutes for recent games
self.last_game_switch = 0
self.game_display_duration = 15 # Short display time for recent/upcoming
self.logger.info(f"Initialized SoccerRecentManager")
def update(self):
"""Update recent games data."""
current_time = time.time()
if current_time - self.last_update < self.update_interval:
return
self.last_update = current_time
try:
data = self._fetch_data() # Fetches shared data (past/present/future)
if not data or 'events' not in data:
self.logger.warning("[Soccer] No recent events found in ESPN API response")
self.recent_games = []
self.games_list = []
self.current_game = None
return
# Process and filter games
new_recent_games = []
now_utc = datetime.now(timezone.utc)
cutoff_time = now_utc - timedelta(hours=self.recent_hours)
for event in data['events']:
game = self._extract_game_details(event)
if game and game['is_final'] and game['start_time_utc'] and game['start_time_utc'] >= cutoff_time:
# Check favorite teams if list is provided
if not self.favorite_teams or (game['home_abbr'] in self.favorite_teams or game['away_abbr'] in self.favorite_teams):
new_recent_games.append(game)
# Sort games by start time, most recent first
new_recent_games.sort(key=lambda x: x['start_time_utc'], reverse=True)
# Update only if the list content changes
new_ids = {g['id'] for g in new_recent_games}
current_ids = {g['id'] for g in self.games_list}
if new_ids != current_ids:
self.logger.info(f"[Soccer] Found {len(new_recent_games)} recent games matching criteria.")
self.recent_games = new_recent_games # Keep raw filtered list
self.games_list = new_recent_games # Use the same list for display rotation
# Reset display index if needed
if not self.current_game or self.current_game['id'] not in new_ids:
self.current_game_index = 0
if self.games_list:
self.current_game = self.games_list[0]
self.last_game_switch = current_time # Reset timer when list changes
else:
self.current_game = None
except Exception as e:
self.logger.error(f"[Soccer] Error updating recent games: {e}", exc_info=True)
self.games_list = []
self.current_game = None
def display(self, force_clear=False):
"""Display recent games, rotating through games_list."""
if not self.games_list:
# self.logger.debug("[Soccer] No recent games to display") # Too noisy
return # Skip display update entirely
try:
current_time = time.time()
# Check if it's time to switch games
if len(self.games_list) > 1 and current_time - self.last_game_switch >= self.game_display_duration:
self.current_game_index = (self.current_game_index + 1) % len(self.games_list)
self.current_game = self.games_list[self.current_game_index]
self.last_game_switch = current_time
force_clear = True # Force clear when switching games
# Ensure current_game is set (it might be None initially)
if not self.current_game and self.games_list:
self.current_game = self.games_list[self.current_game_index]
force_clear = True # Force clear on first display
if self.current_game:
self._draw_scorebug_layout(self.current_game, force_clear)
# Display update handled by controller loop
# self.display_manager.update_display()
except Exception as e:
self.logger.error(f"[Soccer] Error displaying recent game: {e}", exc_info=True)
class SoccerUpcomingManager(BaseSoccerManager):
"""Manager for upcoming Soccer games."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
super().__init__(config, display_manager)
self.upcoming_games = [] # Filtered list for display
self.current_game_index = 0
self.last_update = 0
self.update_interval = 300 # 5 minutes
self.last_log_time = 0
self.log_interval = 300
self.last_warning_time = 0
self.warning_cooldown = 300
self.last_game_switch = 0
self.game_display_duration = 15 # Short display time
self.logger.info(f"Initialized SoccerUpcomingManager")
def update(self):
"""Update upcoming games data."""
current_time = time.time()
if current_time - self.last_update < self.update_interval:
return
self.last_update = current_time
try:
data = self._fetch_data() # Fetches shared data
if not data or 'events' not in data:
self.logger.warning("[Soccer] No upcoming events found in ESPN API response")
self.upcoming_games = []
self.current_game = None
return
# Process and filter games
new_upcoming_games = []
now_utc = datetime.now(timezone.utc)
cutoff_time = now_utc + timedelta(hours=self.recent_hours) # Use recent_hours as upcoming window
for event in data['events']:
game = self._extract_game_details(event)
# Must be upcoming, have a start time, and be within the window
if game and game['is_upcoming'] and game['start_time_utc'] and \
game['start_time_utc'] >= now_utc and game['start_time_utc'] <= cutoff_time:
# Check favorite teams if list is provided
if not self.favorite_teams or (game['home_abbr'] in self.favorite_teams or game['away_abbr'] in self.favorite_teams):
new_upcoming_games.append(game)
# Sort games by start time, soonest first
new_upcoming_games.sort(key=lambda x: x['start_time_utc'])
# Update only if the list content changes
new_ids = {g['id'] for g in new_upcoming_games}
current_ids = {g['id'] for g in self.upcoming_games}
if new_ids != current_ids:
# Logging
should_log = (current_time - self.last_log_time >= self.log_interval or
len(new_upcoming_games) != len(self.upcoming_games) or
not self.upcoming_games)
if should_log:
if new_upcoming_games:
self.logger.info(f"[Soccer] Found {len(new_upcoming_games)} upcoming games matching criteria.")
# Log first few games for brevity
for game in new_upcoming_games[:3]:
self.logger.info(f"[Soccer] Upcoming game: {game['away_abbr']} vs {game['home_abbr']} ({game['game_date']} {game['game_time']}) - {game['league']}")
else:
self.logger.info("[Soccer] No upcoming games found matching criteria.")
self.last_log_time = current_time
self.upcoming_games = new_upcoming_games
# Reset display index if needed
if not self.current_game or self.current_game['id'] not in new_ids:
self.current_game_index = 0
if self.upcoming_games:
self.current_game = self.upcoming_games[0]
self.last_game_switch = current_time # Reset timer
else:
self.current_game = None
except Exception as e:
self.logger.error(f"[Soccer] Error updating upcoming games: {e}", exc_info=True)
self.upcoming_games = []
self.current_game = None
def display(self, force_clear=False):
"""Display upcoming games, rotating through upcoming_games list."""
if not self.upcoming_games:
current_time = time.time()
if current_time - self.last_warning_time > self.warning_cooldown:
# self.logger.info("[Soccer] No upcoming games to display") # Too noisy
self.last_warning_time = current_time
return # Skip display update entirely
try:
current_time = time.time()
# Check if it's time to switch games
if len(self.upcoming_games) > 1 and current_time - self.last_game_switch >= self.game_display_duration:
self.current_game_index = (self.current_game_index + 1) % len(self.upcoming_games)
self.current_game = self.upcoming_games[self.current_game_index]
self.last_game_switch = current_time
force_clear = True # Force clear when switching games
# Ensure current_game is set
if not self.current_game and self.upcoming_games:
self.current_game = self.upcoming_games[self.current_game_index]
force_clear = True
if self.current_game:
self._draw_scorebug_layout(self.current_game, force_clear)
# Update display handled by controller loop
# self.display_manager.update_display()
except Exception as e:
self.logger.error(f"[Soccer] Error displaying upcoming game: {e}", exc_info=True)