mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-10 21:03:01 +00:00
1041 lines
52 KiB
Python
1041 lines
52 KiB
Python
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 # Keep CacheManager import
|
|
|
|
# Constants
|
|
ESPN_NFL_SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/football/nfl/scoreboard" # Changed URL
|
|
|
|
# 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'
|
|
)
|
|
|
|
# Re-add CacheManager definition temporarily until it's confirmed where it lives
|
|
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)
|
|
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 (usually the date string)
|
|
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 BaseNFLManager: # Renamed class
|
|
"""Base class for NFL 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 = None
|
|
_last_shared_update = 0
|
|
cache_manager = CacheManager()
|
|
logger = logging.getLogger('NFL') # Changed logger name
|
|
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
|
self.display_manager = display_manager
|
|
self.config = config
|
|
self.nfl_config = config.get("nfl_scoreboard", {}) # Changed config key
|
|
self.is_enabled = self.nfl_config.get("enabled", False)
|
|
self.test_mode = self.nfl_config.get("test_mode", False)
|
|
self.logo_dir = self.nfl_config.get("logo_dir", "assets/sports/nfl_logos") # Changed logo dir
|
|
self.update_interval = self.nfl_config.get("update_interval_seconds", 60)
|
|
self.last_update = 0
|
|
self.current_game = None
|
|
self.fonts = self._load_fonts()
|
|
self.favorite_teams = self.nfl_config.get("favorite_teams", [])
|
|
self.past_fetch_days = self.nfl_config.get("past_fetch_days", 7)
|
|
self.future_fetch_days = self.nfl_config.get("future_fetch_days", 7)
|
|
|
|
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 NFL manager with display dimensions: {self.display_width}x{self.display_height}")
|
|
self.logger.info(f"Logo directory: {self.logo_dir}")
|
|
|
|
@classmethod
|
|
def _fetch_shared_data(cls, past_days: int, future_days: int, date_str: str = None) -> Optional[Dict]:
|
|
"""Fetch and cache data for all managers to share."""
|
|
current_time = time.time()
|
|
|
|
if cls._shared_data and (current_time - cls._last_shared_update) < 300:
|
|
return cls._shared_data
|
|
|
|
try:
|
|
cache_key = date_str if date_str else 'today_nfl' # Changed cache key prefix
|
|
cached_data = cls.cache_manager.get(cache_key, max_age=300)
|
|
if cached_data:
|
|
cls.logger.info(f"[NFL] Using cached data for {cache_key}")
|
|
cls._shared_data = cached_data
|
|
cls._last_shared_update = current_time
|
|
return cached_data
|
|
|
|
url = ESPN_NFL_SCOREBOARD_URL # Use NFL URL
|
|
params = {}
|
|
if date_str:
|
|
params['dates'] = date_str
|
|
|
|
response = requests.get(url, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
cls.logger.info(f"[NFL] Successfully fetched data from ESPN API")
|
|
|
|
cls.cache_manager.set(cache_key, data)
|
|
cls._shared_data = data
|
|
cls._last_shared_update = current_time
|
|
|
|
if not date_str:
|
|
today = datetime.now(timezone.utc).date()
|
|
dates_to_fetch = []
|
|
# Generate dates from past_days ago to future_days ahead
|
|
for i in range(-past_days, future_days + 1):
|
|
fetch_dt = today + timedelta(days=i)
|
|
dates_to_fetch.append(fetch_dt.strftime('%Y%m%d'))
|
|
|
|
cls.logger.info(f"[NFL] Fetching data for dates: {dates_to_fetch}")
|
|
|
|
all_events = []
|
|
# Fetch data for each date (excluding today if already fetched)
|
|
for fetch_date in dates_to_fetch:
|
|
if fetch_date == today.strftime('%Y%m%d') and cache_key == 'today_nfl': # Skip today if already fetched initially
|
|
continue
|
|
|
|
date_cache_key = f"{fetch_date}_nfl"
|
|
cached_date_data = cls.cache_manager.get(date_cache_key, max_age=300)
|
|
if cached_date_data:
|
|
cls.logger.info(f"[NFL] Using cached data for date {fetch_date}")
|
|
if "events" in cached_date_data:
|
|
all_events.extend(cached_date_data["events"])
|
|
continue
|
|
|
|
params['dates'] = fetch_date
|
|
response = requests.get(url, params=params)
|
|
response.raise_for_status()
|
|
date_data = response.json()
|
|
if date_data and "events" in date_data:
|
|
all_events.extend(date_data["events"])
|
|
cls.logger.info(f"[NFL] Fetched {len(date_data['events'])} events for date {fetch_date}")
|
|
cls.cache_manager.set(date_cache_key, date_data)
|
|
|
|
if all_events:
|
|
if "events" not in data: data["events"] = [] # Ensure 'events' key exists
|
|
data["events"].extend(all_events)
|
|
cls.logger.info(f"[NFL] Combined {len(data['events'])} total events from all dates")
|
|
cls._shared_data = data
|
|
cls._last_shared_update = current_time
|
|
|
|
return data
|
|
except requests.exceptions.RequestException as e:
|
|
cls.logger.error(f"[NFL] Error fetching data from ESPN: {e}")
|
|
return None
|
|
|
|
def _fetch_data(self, date_str: str = None) -> Optional[Dict]:
|
|
"""Fetch data using shared data mechanism or direct fetch for live."""
|
|
# Check if the instance is NFLLiveManager
|
|
if isinstance(self, NFLLiveManager):
|
|
try:
|
|
url = ESPN_NFL_SCOREBOARD_URL # Use NFL URL
|
|
params = {}
|
|
if date_str:
|
|
params['dates'] = date_str
|
|
|
|
response = requests.get(url, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
self.logger.info(f"[NFL] Successfully fetched live game data from ESPN API")
|
|
return data
|
|
except requests.exceptions.RequestException as e:
|
|
self.logger.error(f"[NFL] Error fetching live game data from ESPN: {e}")
|
|
return None
|
|
else:
|
|
# For non-live games, use the shared cache
|
|
return self._fetch_shared_data(self.past_fetch_days, self.future_fetch_days, date_str)
|
|
|
|
def _load_fonts(self):
|
|
"""Load fonts used by the scoreboard."""
|
|
fonts = {}
|
|
try:
|
|
fonts['score'] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 10)
|
|
fonts['time'] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
|
|
fonts['team'] = ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8)
|
|
fonts['status'] = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6) # Using 4x6 for status
|
|
fonts['detail'] = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6) # Added detail font
|
|
logging.info("[NFL] Successfully loaded fonts")
|
|
except IOError:
|
|
logging.warning("[NFL] 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()
|
|
fonts['detail'] = 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
|
|
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((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:
|
|
# Create placeholder if logo doesn't exist (useful for testing)
|
|
if not os.path.exists(logo_path):
|
|
self.logger.warning(f"Logo not found for {team_abbrev} at {logo_path}. Creating placeholder.")
|
|
os.makedirs(os.path.dirname(logo_path), exist_ok=True)
|
|
logo = Image.new('RGBA', (32, 32), (200, 200, 200, 255)) # Gray placeholder
|
|
draw = ImageDraw.Draw(logo)
|
|
draw.text((2, 10), team_abbrev, fill=(0, 0, 0, 255))
|
|
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')
|
|
|
|
max_width = int(self.display_width * 1.5)
|
|
max_height = int(self.display_height * 1.5)
|
|
logo.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
|
|
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 _extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
|
"""Extract relevant game details from ESPN NFL API response."""
|
|
# --- THIS METHOD NEEDS SIGNIFICANT ADAPTATION FOR NFL API ---
|
|
if not game_event: return None
|
|
|
|
try:
|
|
competition = game_event["competitions"][0]
|
|
status = competition["status"]
|
|
competitors = competition["competitors"]
|
|
game_date_str = game_event["date"]
|
|
|
|
start_time_utc = None
|
|
try:
|
|
start_time_utc = datetime.fromisoformat(game_date_str.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
logging.warning(f"[NFL] Could not parse game date: {game_date_str}")
|
|
|
|
home_team = next((c for c in competitors if c.get("homeAway") == "home"), None)
|
|
away_team = next((c for c in competitors if c.get("homeAway") == "away"), None)
|
|
|
|
if not home_team or not away_team:
|
|
self.logger.warning(f"[NFL] Could not find home or away team in event: {game_event.get('id')}")
|
|
return None
|
|
|
|
game_time, game_date = "", ""
|
|
if start_time_utc:
|
|
local_time = start_time_utc.astimezone()
|
|
game_time = local_time.strftime("%-I:%M %p")
|
|
game_date = local_time.strftime("%-m/%-d")
|
|
|
|
# --- NFL Specific Details ---
|
|
situation = competition.get("situation")
|
|
down_distance_text = ""
|
|
if situation and status["type"]["state"] == "in":
|
|
down = situation.get("down")
|
|
distance = situation.get("distance")
|
|
if down and distance is not None:
|
|
down_str = {1: "1st", 2: "2nd", 3: "3rd", 4: "4th"}.get(down, f"{down}th")
|
|
dist_str = f"& {distance}" if distance > 0 else "& Goal"
|
|
down_distance_text = f"{down_str} {dist_str}"
|
|
elif situation.get("isRedZone"):
|
|
down_distance_text = "Red Zone" # Simplified if down/distance not present but in redzone
|
|
|
|
# Format period/quarter
|
|
period = status.get("period", 0)
|
|
period_text = ""
|
|
if status["type"]["state"] == "in":
|
|
if period == 0: period_text = "Start" # Before kickoff
|
|
elif period == 1: period_text = "Q1"
|
|
elif period == 2: period_text = "Q2"
|
|
elif period == 3: period_text = "HALF" # Halftime is usually period 3 in API
|
|
elif period == 4: period_text = "Q3"
|
|
elif period == 5: period_text = "Q4"
|
|
elif period > 5: period_text = "OT" # Assuming OT starts at period 6+
|
|
elif status["type"]["state"] == "halftime": # Check explicit halftime state
|
|
period_text = "HALF"
|
|
elif status["type"]["state"] == "post":
|
|
if period > 5 : period_text = "Final/OT"
|
|
else: period_text = "Final"
|
|
elif status["type"]["state"] == "pre":
|
|
period_text = game_time # Show time for upcoming
|
|
|
|
# Timeouts (assuming max 3 per half, not carried over well in standard API)
|
|
# API often provides 'timeouts' directly under team, but reset logic is tricky
|
|
# We might need to simplify this or just use a fixed display if API is unreliable
|
|
home_timeouts = home_team.get("timeouts", 3) # Default to 3 if not specified
|
|
away_timeouts = away_team.get("timeouts", 3) # Default to 3 if not specified
|
|
|
|
|
|
details = {
|
|
"id": game_event.get("id"),
|
|
"start_time_utc": start_time_utc,
|
|
"status_text": status["type"]["shortDetail"], # e.g., "Final", "7:30 PM", "Q1 12:34"
|
|
"period": period,
|
|
"period_text": period_text, # Formatted quarter/status
|
|
"clock": status.get("displayClock", "0:00"),
|
|
"is_live": status["type"]["state"] == "in",
|
|
"is_final": status["type"]["state"] == "post",
|
|
"is_upcoming": status["type"]["state"] == "pre",
|
|
"is_halftime": status["type"]["state"] == "halftime", # Added halftime check
|
|
"home_abbr": home_team["team"]["abbreviation"],
|
|
"home_score": home_team.get("score", "0"),
|
|
"home_logo_path": os.path.join(self.logo_dir, f"{home_team['team']['abbreviation']}.png"),
|
|
"home_timeouts": home_timeouts,
|
|
"away_abbr": away_team["team"]["abbreviation"],
|
|
"away_score": away_team.get("score", "0"),
|
|
"away_logo_path": os.path.join(self.logo_dir, f"{away_team['team']['abbreviation']}.png"),
|
|
"away_timeouts": away_timeouts,
|
|
"game_time": game_time,
|
|
"game_date": game_date,
|
|
"down_distance_text": down_distance_text, # Added Down/Distance
|
|
"possession": situation.get("possession") if situation else None, # ID of team with possession
|
|
}
|
|
|
|
# Basic validation (can be expanded)
|
|
if not details['home_abbr'] or not details['away_abbr']:
|
|
self.logger.warning(f"[NFL] Missing team abbreviation in event: {details['id']}")
|
|
return None
|
|
|
|
self.logger.debug(f"[NFL] Extracted: {details['away_abbr']}@{details['home_abbr']}, Status: {status['type']['name']}, Live: {details['is_live']}, Final: {details['is_final']}, Upcoming: {details['is_upcoming']}")
|
|
|
|
# Logo validation (optional but good practice)
|
|
for team in ["home", "away"]:
|
|
logo_path = details[f"{team}_logo_path"]
|
|
# No need to check file existence here, _load_and_resize_logo handles it
|
|
|
|
return details
|
|
except Exception as e:
|
|
# Log the problematic event structure if possible
|
|
logging.error(f"[NFL] Error extracting game details: {e} from event: {game_event.get('id')}", exc_info=True)
|
|
return None
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Placeholder draw method - subclasses should override."""
|
|
# This base method will be simple, subclasses provide specifics
|
|
try:
|
|
img = Image.new('RGB', (self.display_width, self.display_height), (0, 0, 0))
|
|
draw = ImageDraw.Draw(img)
|
|
status = game.get("status_text", "N/A")
|
|
self._draw_text_with_outline(draw, status, (2, 2), self.fonts['status'])
|
|
self.display_manager.image.paste(img, (0, 0))
|
|
# Don't call update_display here, let subclasses handle it after drawing
|
|
except Exception as e:
|
|
self.logger.error(f"Error in base _draw_scorebug_layout: {e}", exc_info=True)
|
|
|
|
|
|
def display(self, force_clear: bool = False) -> None:
|
|
"""Common display method for all NFL managers"""
|
|
if not self.is_enabled: # Check if module is enabled
|
|
return
|
|
|
|
if not self.current_game:
|
|
current_time = time.time()
|
|
if not hasattr(self, '_last_warning_time'):
|
|
self._last_warning_time = 0
|
|
if current_time - getattr(self, '_last_warning_time', 0) > 300:
|
|
self.logger.warning(f"[NFL] No game data available to display in {self.__class__.__name__}")
|
|
setattr(self, '_last_warning_time', current_time)
|
|
return
|
|
|
|
try:
|
|
self._draw_scorebug_layout(self.current_game, force_clear)
|
|
# display_manager.update_display() should be called within subclass draw methods
|
|
# or after calling display() in the main loop. Let's keep it out of the base display.
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL] Error during display call in {self.__class__.__name__}: {e}", exc_info=True)
|
|
|
|
|
|
class NFLLiveManager(BaseNFLManager): # Renamed class
|
|
"""Manager for live NFL games."""
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
|
super().__init__(config, display_manager)
|
|
self.update_interval = self.nfl_config.get("live_update_interval", 15)
|
|
self.no_data_interval = 300
|
|
self.last_update = 0
|
|
self.logger.info("Initialized NFL Live Manager")
|
|
self.live_games = []
|
|
self.current_game_index = 0
|
|
self.last_game_switch = 0
|
|
self.game_display_duration = self.nfl_config.get("live_game_duration", 20)
|
|
self.last_display_update = 0
|
|
self.last_log_time = 0
|
|
self.log_interval = 300
|
|
|
|
if self.test_mode:
|
|
# More detailed test game for NFL
|
|
self.current_game = {
|
|
"id": "test001",
|
|
"home_abbr": "TB", "away_abbr": "DAL",
|
|
"home_score": "21", "away_score": "17",
|
|
"period": 4, "period_text": "Q4", "clock": "02:35",
|
|
"down_distance_text": "1st & 10", "possession": "TB_ID", # Placeholder ID
|
|
"home_timeouts": 2, "away_timeouts": 3,
|
|
"home_logo_path": os.path.join(self.logo_dir, "TB.png"),
|
|
"away_logo_path": os.path.join(self.logo_dir, "DAL.png"),
|
|
"is_live": True, "is_final": False, "is_upcoming": False, "is_halftime": False,
|
|
"status_text": "Q4 02:35"
|
|
}
|
|
self.live_games = [self.current_game]
|
|
logging.info("[NFL] Initialized NFLLiveManager with test game: BUF vs KC")
|
|
else:
|
|
logging.info("[NFL] Initialized NFLLiveManager in live mode")
|
|
|
|
def update(self):
|
|
"""Update live game data."""
|
|
if not self.is_enabled: return
|
|
current_time = time.time()
|
|
interval = self.no_data_interval if not self.live_games and not self.test_mode else self.update_interval
|
|
|
|
if current_time - self.last_update >= interval:
|
|
self.last_update = current_time
|
|
|
|
if self.test_mode:
|
|
# Simulate clock running down in test mode
|
|
if self.current_game and self.current_game["is_live"]:
|
|
try:
|
|
minutes, seconds = map(int, self.current_game["clock"].split(':'))
|
|
seconds -= 1
|
|
if seconds < 0:
|
|
seconds = 59
|
|
minutes -= 1
|
|
if minutes < 0:
|
|
# Simulate end of quarter/game
|
|
if self.current_game["period"] < 5: # Assuming 5 is Q4 end
|
|
self.current_game["period"] += 1
|
|
# Update period_text based on new period
|
|
if self.current_game["period"] == 3: self.current_game["period_text"] = "HALF"
|
|
elif self.current_game["period"] == 5: self.current_game["period_text"] = "Q4"
|
|
# Reset clock for next quarter (e.g., 15:00)
|
|
minutes, seconds = 15, 0
|
|
else:
|
|
# Simulate game end
|
|
self.current_game["is_live"] = False
|
|
self.current_game["is_final"] = True
|
|
self.current_game["period_text"] = "Final"
|
|
minutes, seconds = 0, 0
|
|
self.current_game["clock"] = f"{minutes:02d}:{seconds:02d}"
|
|
# Simulate down change occasionally
|
|
if seconds % 15 == 0:
|
|
self.current_game["down_distance_text"] = f"{['1st','2nd','3rd','4th'][seconds % 4]} & {seconds % 10 + 1}"
|
|
self.current_game["status_text"] = f"{self.current_game['period_text']} {self.current_game['clock']}"
|
|
|
|
# Display update handled by main loop or explicit call if needed immediately
|
|
# self.display(force_clear=True) # Only if immediate update is desired here
|
|
|
|
except ValueError:
|
|
self.logger.warning("[NFL] Test mode: Could not parse clock")
|
|
# No actual display call here, let main loop handle it
|
|
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)
|
|
if details and (details["is_live"] or details["is_halftime"]): # Include halftime as 'live' display
|
|
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)
|
|
|
|
# Log changes or periodically
|
|
should_log = (
|
|
current_time - self.last_log_time >= self.log_interval or
|
|
len(new_live_games) != len(self.live_games) or
|
|
any(g1['id'] != g2.get('id') for g1, g2 in zip(self.live_games, new_live_games)) or # Check if game IDs changed
|
|
(not self.live_games and new_live_games) # Log if games appeared
|
|
)
|
|
|
|
if should_log:
|
|
if new_live_games:
|
|
self.logger.info(f"[NFL] Found {len(new_live_games)} live/halftime games for fav teams.")
|
|
for game in new_live_games:
|
|
self.logger.info(f" - {game['away_abbr']}@{game['home_abbr']} ({game.get('status_text', 'N/A')})")
|
|
else:
|
|
self.logger.info("[NFL] No live/halftime games found for favorite teams.")
|
|
self.last_log_time = current_time
|
|
|
|
|
|
# Update game list and current game
|
|
if new_live_games:
|
|
# Check if the games themselves changed, not just scores/time
|
|
new_game_ids = {g['id'] for g in new_live_games}
|
|
current_game_ids = {g['id'] for g in self.live_games}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.live_games = sorted(new_live_games, key=lambda g: g.get('start_time_utc') or datetime.now(timezone.utc)) # Sort by start 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
|
|
self.current_game = self.live_games[0] if self.live_games else None
|
|
self.last_game_switch = current_time
|
|
else:
|
|
# Find current game's new index if it still exists
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.live_games) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.live_games[self.current_game_index] # Update current_game with fresh data
|
|
except StopIteration: # Should not happen if check above passed, but safety first
|
|
self.current_game_index = 0
|
|
self.current_game = self.live_games[0]
|
|
self.last_game_switch = current_time
|
|
|
|
else:
|
|
# Just update the data for the existing games
|
|
temp_game_dict = {g['id']: g for g in new_live_games}
|
|
self.live_games = [temp_game_dict.get(g['id'], g) for g in self.live_games] # Update in place
|
|
if self.current_game:
|
|
self.current_game = temp_game_dict.get(self.current_game['id'], self.current_game)
|
|
|
|
# Display update handled by main loop based on interval
|
|
|
|
else:
|
|
# No live games found
|
|
if self.live_games: # Were there games before?
|
|
self.logger.info("[NFL] Live games previously showing have ended or are no longer live.")
|
|
self.live_games = []
|
|
self.current_game = None
|
|
self.current_game_index = 0
|
|
|
|
else:
|
|
# Error fetching data or no events
|
|
if self.live_games: # Were there games before?
|
|
self.logger.warning("[NFL] Could not fetch update; keeping existing live game data for now.")
|
|
else:
|
|
self.logger.warning("[NFL] Could not fetch data and no existing live games.")
|
|
self.current_game = None # Clear current game if fetch fails and no games were active
|
|
|
|
# Handle game switching (outside test mode check)
|
|
if not self.test_mode and 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
|
|
self.logger.info(f"[NFL] Switched live view to: {self.current_game['away_abbr']}@{self.current_game['home_abbr']}")
|
|
# Force display update via flag or direct call if needed, but usually let main loop handle
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Draw the detailed scorebug layout for a live NFL game."""
|
|
try:
|
|
main_img = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 255))
|
|
overlay = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 0))
|
|
draw_overlay = ImageDraw.Draw(overlay) # Draw text elements on overlay first
|
|
|
|
home_logo = self._load_and_resize_logo(game["home_abbr"])
|
|
away_logo = self._load_and_resize_logo(game["away_abbr"])
|
|
|
|
if not home_logo or not away_logo:
|
|
self.logger.error(f"[NFL] Failed to load logos for live game: {game.get('id')}")
|
|
# Draw placeholder text if logos fail
|
|
draw_final = ImageDraw.Draw(main_img.convert('RGB'))
|
|
self._draw_text_with_outline(draw_final, "Logo Error", (5,5), self.fonts['status'])
|
|
self.display_manager.image.paste(main_img.convert('RGB'), (0, 0))
|
|
self.display_manager.update_display()
|
|
return
|
|
|
|
center_y = self.display_height // 2
|
|
|
|
# Draw logos (shifted slightly more inward than NHL perhaps)
|
|
home_x = self.display_width - home_logo.width + 10 #adjusted from 18 # Adjust position as needed
|
|
home_y = center_y - (home_logo.height // 2)
|
|
main_img.paste(home_logo, (home_x, home_y), home_logo)
|
|
|
|
away_x = -10 #adjusted from 18 # Adjust position as needed
|
|
away_y = center_y - (away_logo.height // 2)
|
|
main_img.paste(away_logo, (away_x, away_y), away_logo)
|
|
|
|
# --- Draw Text Elements on Overlay ---
|
|
# Scores (centered, slightly above bottom)
|
|
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_overlay.textlength(score_text, font=self.fonts['score'])
|
|
score_x = (self.display_width - score_width) // 2
|
|
score_y = (self.display_height // 2) - 5 #centered #from 14 # Position score higher
|
|
self._draw_text_with_outline(draw_overlay, score_text, (score_x, score_y), self.fonts['score'])
|
|
|
|
# Period/Quarter and Clock (Top center)
|
|
period_clock_text = f"{game.get('period_text', '')} {game.get('clock', '')}".strip()
|
|
if game.get("is_halftime"): period_clock_text = "Halftime" # Override for halftime
|
|
|
|
status_width = draw_overlay.textlength(period_clock_text, font=self.fonts['time'])
|
|
status_x = (self.display_width - status_width) // 2
|
|
status_y = 1 # Position at top
|
|
self._draw_text_with_outline(draw_overlay, period_clock_text, (status_x, status_y), self.fonts['time'])
|
|
|
|
# Down & Distance (Below Period/Clock)
|
|
down_distance = game.get("down_distance_text", "")
|
|
if down_distance and game.get("is_live"): # Only show if live and available
|
|
dd_width = draw_overlay.textlength(down_distance, font=self.fonts['detail'])
|
|
dd_x = (self.display_width - dd_width) // 2
|
|
dd_y = score_y + 12 # Below the status/clock line
|
|
self._draw_text_with_outline(draw_overlay, down_distance, (dd_x, dd_y), self.fonts['detail'], fill=(200, 200, 0)) # Yellowish text
|
|
|
|
# Timeouts (Bottom corners) - 3 small bars per team
|
|
timeout_bar_width = 4
|
|
timeout_bar_height = 2
|
|
timeout_spacing = 1
|
|
timeout_y = self.display_height - timeout_bar_height - 1 # Bottom edge
|
|
|
|
# Away Timeouts (Bottom Left)
|
|
away_timeouts_remaining = game.get("away_timeouts", 0)
|
|
for i in range(3):
|
|
to_x = 2 + i * (timeout_bar_width + timeout_spacing)
|
|
color = (255, 255, 255) if i < away_timeouts_remaining else (80, 80, 80) # White if available, gray if used
|
|
draw_overlay.rectangle([to_x, timeout_y, to_x + timeout_bar_width, timeout_y + timeout_bar_height], fill=color, outline=(0,0,0))
|
|
|
|
# Home Timeouts (Bottom Right)
|
|
home_timeouts_remaining = game.get("home_timeouts", 0)
|
|
for i in range(3):
|
|
to_x = self.display_width - 2 - timeout_bar_width - (2-i) * (timeout_bar_width + timeout_spacing)
|
|
color = (255, 255, 255) if i < home_timeouts_remaining else (80, 80, 80) # White if available, gray if used
|
|
draw_overlay.rectangle([to_x, timeout_y, to_x + timeout_bar_width, timeout_y + timeout_bar_height], fill=color, outline=(0,0,0))
|
|
|
|
# Composite the text overlay onto the main image
|
|
main_img = Image.alpha_composite(main_img, overlay)
|
|
main_img = main_img.convert('RGB') # Convert for display
|
|
|
|
# Display the final image
|
|
self.display_manager.image.paste(main_img, (0, 0))
|
|
self.display_manager.update_display() # Update display here for live
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error displaying live NFL game: {e}", exc_info=True)
|
|
|
|
# Inherits display() method from BaseNFLManager, which calls the overridden _draw_scorebug_layout
|
|
|
|
|
|
class NFLRecentManager(BaseNFLManager): # Renamed class
|
|
"""Manager for recently completed NFL games."""
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
|
super().__init__(config, display_manager)
|
|
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 = 300 # Check for recent games every 5 mins
|
|
self.last_game_switch = 0
|
|
self.game_display_duration = 15 # Display each recent game for 15 seconds
|
|
self.logger.info(f"Initialized NFLRecentManager with {len(self.favorite_teams)} favorite teams")
|
|
|
|
def update(self):
|
|
"""Update recent games data."""
|
|
if not self.is_enabled: return
|
|
current_time = time.time()
|
|
if current_time - self.last_update < self.update_interval:
|
|
return
|
|
|
|
self.last_update = current_time # Update time even if fetch fails
|
|
try:
|
|
data = self._fetch_data() # Uses shared cache
|
|
if not data or 'events' not in data:
|
|
self.logger.warning("[NFL Recent] No events found in shared data.")
|
|
if not self.games_list: self.current_game = None # Clear display if no games were showing
|
|
return
|
|
|
|
events = data['events']
|
|
# self.logger.info(f"[NFL Recent] Processing {len(events)} events from shared data.")
|
|
|
|
# Process games and filter for final & within window & favorite teams
|
|
processed_games = []
|
|
for event in events:
|
|
game = self._extract_game_details(event)
|
|
# Filter criteria: must be final, within time window
|
|
if game and game['is_final'] and game['is_within_window']:
|
|
processed_games.append(game)
|
|
|
|
# Filter for favorite teams
|
|
if self.favorite_teams:
|
|
team_games = [game for game in processed_games
|
|
if game['home_abbr'] in self.favorite_teams or
|
|
game['away_abbr'] in self.favorite_teams]
|
|
else:
|
|
team_games = processed_games # Show all recent games if no favorites defined
|
|
|
|
# Sort by game time, most recent first
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
|
|
|
|
# Check if the list of games to display has changed
|
|
new_game_ids = {g['id'] for g in team_games}
|
|
current_game_ids = {g['id'] for g in self.games_list}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.logger.info(f"[NFL Recent] Found {len(team_games)} final games within window for display.")
|
|
self.games_list = team_games
|
|
# Reset index if list changed or current game removed
|
|
if not self.current_game or not self.games_list or self.current_game['id'] not in new_game_ids:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0] if self.games_list else None
|
|
self.last_game_switch = current_time # Reset switch timer
|
|
else:
|
|
# Try to maintain position if possible
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.games_list) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.games_list[self.current_game_index] # Update data just in case
|
|
except StopIteration:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0]
|
|
self.last_game_switch = current_time
|
|
|
|
elif self.games_list:
|
|
# List content is same, just update data for current game
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
|
|
|
|
if not self.games_list:
|
|
self.logger.info("[NFL Recent] No relevant recent games found to display.")
|
|
self.current_game = None # Ensure display clears if no games
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Recent] Error updating recent games: {e}", exc_info=True)
|
|
# Don't clear current game on error, keep showing last known state
|
|
# self.current_game = None # Decide if we want to clear display on error
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Draw the layout for a recently completed NFL game."""
|
|
try:
|
|
main_img = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 255))
|
|
overlay = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 0))
|
|
draw_overlay = ImageDraw.Draw(overlay)
|
|
|
|
home_logo = self._load_and_resize_logo(game["home_abbr"])
|
|
away_logo = self._load_and_resize_logo(game["away_abbr"])
|
|
|
|
if not home_logo or not away_logo:
|
|
self.logger.error(f"[NFL Recent] Failed to load logos for game: {game.get('id')}")
|
|
# Draw placeholder text if logos fail (similar to live)
|
|
draw_final = ImageDraw.Draw(main_img.convert('RGB'))
|
|
self._draw_text_with_outline(draw_final, "Logo Error", (5,5), self.fonts['status'])
|
|
self.display_manager.image.paste(main_img.convert('RGB'), (0, 0))
|
|
self.display_manager.update_display()
|
|
return
|
|
|
|
center_y = self.display_height // 2
|
|
|
|
home_x = self.display_width - home_logo.width + 18
|
|
home_y = center_y - (home_logo.height // 2)
|
|
main_img.paste(home_logo, (home_x, home_y), home_logo)
|
|
|
|
away_x = -18
|
|
away_y = center_y - (away_logo.height // 2)
|
|
main_img.paste(away_logo, (away_x, away_y), away_logo)
|
|
|
|
# Draw Text Elements on Overlay
|
|
# Final Scores (Centered, same position as live)
|
|
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_overlay.textlength(score_text, font=self.fonts['score'])
|
|
score_x = (self.display_width - score_width) // 2
|
|
score_y = self.display_height - 14
|
|
self._draw_text_with_outline(draw_overlay, score_text, (score_x, score_y), self.fonts['score'])
|
|
|
|
# "Final" text (Top center)
|
|
status_text = game.get("period_text", "Final") # Use formatted period text (e.g., "Final/OT") or default "Final"
|
|
status_width = draw_overlay.textlength(status_text, font=self.fonts['time'])
|
|
status_x = (self.display_width - status_width) // 2
|
|
status_y = 1
|
|
self._draw_text_with_outline(draw_overlay, status_text, (status_x, status_y), self.fonts['time'])
|
|
|
|
# Composite and display
|
|
main_img = Image.alpha_composite(main_img, overlay)
|
|
main_img = main_img.convert('RGB')
|
|
self.display_manager.image.paste(main_img, (0, 0))
|
|
self.display_manager.update_display() # Update display here
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Recent] Error displaying recent game: {e}", exc_info=True)
|
|
|
|
def display(self, force_clear=False):
|
|
"""Display recent games, handling switching."""
|
|
if not self.is_enabled or not self.games_list:
|
|
# If disabled or no games, ensure display might be cleared by main loop if needed
|
|
# Or potentially clear it here? For now, rely on main loop/other managers.
|
|
if not self.games_list and self.current_game:
|
|
self.current_game = None # Clear internal state if list becomes empty
|
|
return
|
|
|
|
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 redraw on switch
|
|
self.logger.debug(f"[NFL Recent] Switched to game index {self.current_game_index}")
|
|
|
|
if self.current_game:
|
|
self._draw_scorebug_layout(self.current_game, force_clear)
|
|
# update_display() is called within _draw_scorebug_layout for recent
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Recent] Error in display loop: {e}", exc_info=True)
|
|
|
|
|
|
class NFLUpcomingManager(BaseNFLManager): # Renamed class
|
|
"""Manager for upcoming NFL games."""
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
|
super().__init__(config, display_manager)
|
|
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 = 300 # Check for upcoming games every 5 mins
|
|
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 # Display each upcoming game for 15 seconds
|
|
self.logger.info(f"Initialized NFLUpcomingManager with {len(self.favorite_teams)} favorite teams")
|
|
|
|
def update(self):
|
|
"""Update upcoming games data."""
|
|
if not self.is_enabled: return
|
|
current_time = time.time()
|
|
if current_time - self.last_update < self.update_interval:
|
|
return
|
|
|
|
self.last_update = current_time
|
|
try:
|
|
data = self._fetch_data() # Uses shared cache
|
|
if not data or 'events' not in data:
|
|
self.logger.warning("[NFL Upcoming] No events found in shared data.")
|
|
if not self.games_list: self.current_game = None
|
|
return
|
|
|
|
events = data['events']
|
|
# self.logger.info(f"[NFL Upcoming] Processing {len(events)} events from shared data.")
|
|
|
|
processed_games = []
|
|
for event in events:
|
|
game = self._extract_game_details(event)
|
|
# Filter criteria: must be upcoming ('pre' state) and within time window
|
|
if game and game['is_upcoming'] and game['is_within_window']:
|
|
processed_games.append(game)
|
|
|
|
# Filter for favorite teams
|
|
if self.favorite_teams:
|
|
team_games = [game for game in processed_games
|
|
if game['home_abbr'] in self.favorite_teams or
|
|
game['away_abbr'] in self.favorite_teams]
|
|
else:
|
|
team_games = processed_games # Show all upcoming if no favorites
|
|
|
|
# Sort by game time, earliest first
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
|
|
|
|
# Log changes or periodically
|
|
should_log = (
|
|
current_time - self.last_log_time >= self.log_interval or
|
|
len(team_games) != len(self.games_list) or
|
|
any(g1['id'] != g2.get('id') for g1, g2 in zip(self.games_list, team_games)) or
|
|
(not self.games_list and team_games)
|
|
)
|
|
|
|
# Check if the list of games to display has changed
|
|
new_game_ids = {g['id'] for g in team_games}
|
|
current_game_ids = {g['id'] for g in self.games_list}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.logger.info(f"[NFL Upcoming] Found {len(team_games)} upcoming games within window for display.")
|
|
self.games_list = team_games
|
|
if not self.current_game or not self.games_list or self.current_game['id'] not in new_game_ids:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0] if self.games_list else None
|
|
self.last_game_switch = current_time
|
|
else:
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.games_list) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
except StopIteration:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0]
|
|
self.last_game_switch = current_time
|
|
|
|
elif self.games_list:
|
|
self.current_game = self.games_list[self.current_game_index] # Update data
|
|
|
|
if not self.games_list:
|
|
self.logger.info("[NFL Upcoming] No relevant upcoming games found to display.")
|
|
self.current_game = None
|
|
|
|
if should_log and not self.games_list:
|
|
# Log favorite teams only if no games are found and logging is needed
|
|
self.logger.debug(f"[NFL Upcoming] Favorite teams: {self.favorite_teams}")
|
|
self.logger.debug(f"[NFL Upcoming] Total upcoming games before filtering: {len(processed_games)}")
|
|
self.last_log_time = current_time
|
|
elif should_log:
|
|
self.last_log_time = current_time
|
|
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Upcoming] Error updating upcoming games: {e}", exc_info=True)
|
|
# self.current_game = None # Decide if clear on error
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Draw the layout for an upcoming NFL game."""
|
|
try:
|
|
main_img = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 255))
|
|
overlay = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 0))
|
|
draw_overlay = ImageDraw.Draw(overlay)
|
|
|
|
home_logo = self._load_and_resize_logo(game["home_abbr"])
|
|
away_logo = self._load_and_resize_logo(game["away_abbr"])
|
|
|
|
if not home_logo or not away_logo:
|
|
self.logger.error(f"[NFL Upcoming] Failed to load logos for game: {game.get('id')}")
|
|
draw_final = ImageDraw.Draw(main_img.convert('RGB'))
|
|
self._draw_text_with_outline(draw_final, "Logo Error", (5,5), self.fonts['status'])
|
|
self.display_manager.image.paste(main_img.convert('RGB'), (0, 0))
|
|
self.display_manager.update_display()
|
|
return
|
|
|
|
center_y = self.display_height // 2
|
|
|
|
home_x = self.display_width - home_logo.width + 18
|
|
home_y = center_y - (home_logo.height // 2)
|
|
main_img.paste(home_logo, (home_x, home_y), home_logo)
|
|
|
|
away_x = -18
|
|
away_y = center_y - (away_logo.height // 2)
|
|
main_img.paste(away_logo, (away_x, away_y), away_logo)
|
|
|
|
# Draw Text Elements on Overlay
|
|
game_date = game.get("game_date", "")
|
|
game_time = game.get("game_time", "")
|
|
|
|
# "Next Game" at the top (use smaller status font)
|
|
status_text = "Next Game"
|
|
status_width = draw_overlay.textlength(status_text, font=self.fonts['status'])
|
|
status_x = (self.display_width - status_width) // 2
|
|
status_y = 1 # Changed from 2
|
|
self._draw_text_with_outline(draw_overlay, status_text, (status_x, status_y), self.fonts['status'])
|
|
|
|
# Date text (centered, below "Next Game")
|
|
date_width = draw_overlay.textlength(game_date, font=self.fonts['time'])
|
|
date_x = (self.display_width - date_width) // 2
|
|
# Adjust Y position to stack date and time nicely
|
|
date_y = center_y - 7 # Raise date slightly
|
|
self._draw_text_with_outline(draw_overlay, game_date, (date_x, date_y), self.fonts['time'])
|
|
|
|
# Time text (centered, below Date)
|
|
time_width = draw_overlay.textlength(game_time, font=self.fonts['time'])
|
|
time_x = (self.display_width - time_width) // 2
|
|
time_y = date_y + 9 # Place time below date
|
|
self._draw_text_with_outline(draw_overlay, game_time, (time_x, time_y), self.fonts['time'])
|
|
|
|
# Composite and display
|
|
main_img = Image.alpha_composite(main_img, overlay)
|
|
main_img = main_img.convert('RGB')
|
|
self.display_manager.image.paste(main_img, (0, 0))
|
|
self.display_manager.update_display() # Update display here
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Upcoming] Error displaying upcoming game: {e}", exc_info=True)
|
|
|
|
def display(self, force_clear=False):
|
|
"""Display upcoming games, handling switching."""
|
|
if not self.is_enabled: return
|
|
|
|
if not self.games_list:
|
|
if self.current_game: self.current_game = None # Clear state if list empty
|
|
current_time = time.time()
|
|
# Log warning periodically if no games found
|
|
if current_time - self.last_warning_time > self.warning_cooldown:
|
|
self.logger.info("[NFL Upcoming] No upcoming games found for favorite teams to display.")
|
|
self.last_warning_time = current_time
|
|
return # Skip display update
|
|
|
|
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 redraw on switch
|
|
self.logger.debug(f"[NFL Upcoming] Switched to game index {self.current_game_index}")
|
|
|
|
if self.current_game:
|
|
self._draw_scorebug_layout(self.current_game, force_clear)
|
|
# update_display() is called within _draw_scorebug_layout for upcoming
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"[NFL Upcoming] Error in display loop: {e}", exc_info=True)
|
|
|