Files
LEDMatrix/src/ncaa_fb_managers.py
2025-05-22 17:51:14 -05:00

1033 lines
54 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_NCAAFB_SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/football/college-football/scoreboard" # Changed URL for NCAA FB
# 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 BaseNCAAFBManager: # Renamed class
"""Base class for NCAA FB managers with common functionality.""" # Updated docstring
# 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('NCAAFB') # Changed logger name
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
self.display_manager = display_manager
self.config = config
self.ncaa_fb_config = config.get("ncaa_fb_scoreboard", {}) # Changed config key
self.is_enabled = self.ncaa_fb_config.get("enabled", False)
self.test_mode = self.ncaa_fb_config.get("test_mode", False)
self.logo_dir = self.ncaa_fb_config.get("logo_dir", "assets/sports/ncaa_fbs_logos") # Changed logo dir
self.update_interval = self.ncaa_fb_config.get("update_interval_seconds", 60)
self.last_update = 0
self.current_game = None
self.fonts = self._load_fonts()
self.favorite_teams = self.ncaa_fb_config.get("favorite_teams", [])
self.past_fetch_days = self.ncaa_fb_config.get("past_fetch_days", 7)
self.future_fetch_days = self.ncaa_fb_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 NCAAFB 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_ncaafb' # Changed cache key prefix
cached_data = cls.cache_manager.get(cache_key, max_age=300)
if cached_data:
cls.logger.info(f"[NCAAFB] Using cached data for {cache_key}")
cls._shared_data = cached_data
cls._last_shared_update = current_time
return cached_data
url = ESPN_NCAAFB_SCOREBOARD_URL # Use NCAA FB 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"[NCAAFB] 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"[NCAAFB] 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_ncaafb': # Skip today if already fetched initially
continue
date_cache_key = f"{fetch_date}_ncaafb" # Changed cache key suffix
cached_date_data = cls.cache_manager.get(date_cache_key, max_age=300)
if cached_date_data:
cls.logger.info(f"[NCAAFB] 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"[NCAAFB] 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"[NCAAFB] 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"[NCAAFB] 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 NCAAFBLiveManager
if isinstance(self, NCAAFBLiveManager): # Changed class name
try:
url = ESPN_NCAAFB_SCOREBOARD_URL # Use NCAA FB 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"[NCAAFB] Successfully fetched live game data from ESPN API")
return data
except requests.exceptions.RequestException as e:
self.logger.error(f"[NCAAFB] 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("[NCAAFB] Successfully loaded fonts") # Changed log prefix
except IOError:
logging.warning("[NCAAFB] Fonts not found, using default PIL font.") # Changed log prefix
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 NCAA FB API response."""
# --- THIS METHOD MAY NEED ADJUSTMENTS FOR NCAA FB API DIFFERENCES ---
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"[NCAAFB] 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"[NCAAFB] 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")
# --- Football Specific Details (Likely same for NFL/NCAAFB) ---
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"[NCAAFB] Missing team abbreviation in event: {details['id']}")
return None
self.logger.debug(f"[NCAAFB] 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"[NCAAFB] 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 NCAA FB managers""" # Updated docstring
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"[NCAAFB] 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"[NCAAFB] Error during display call in {self.__class__.__name__}: {e}", exc_info=True)
class NCAAFBLiveManager(BaseNCAAFBManager): # Renamed class
"""Manager for live NCAA FB games.""" # Updated docstring
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
super().__init__(config, display_manager)
self.update_interval = self.ncaa_fb_config.get("live_update_interval", 15)
self.no_data_interval = 300
self.last_update = 0
self.logger.info("Initialized NCAAFB Live Manager")
self.live_games = []
self.current_game_index = 0
self.last_game_switch = 0
self.game_display_duration = self.ncaa_fb_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 NCAA FB
self.current_game = {
"id": "testNCAAFB001",
"home_abbr": "UGA", "away_abbr": "AUB", # NCAA Examples
"home_score": "28", "away_score": "21",
"period": 4, "period_text": "Q4", "clock": "01:15",
"down_distance_text": "2nd & 5", "possession": "UGA_ID", # Placeholder ID
"home_timeouts": 1, "away_timeouts": 2,
"home_logo_path": os.path.join(self.logo_dir, "UGA.png"),
"away_logo_path": os.path.join(self.logo_dir, "AUB.png"),
"is_live": True, "is_final": False, "is_upcoming": False, "is_halftime": False,
"status_text": "Q4 01:15"
}
self.live_games = [self.current_game]
logging.info("[NCAAFB] Initialized NCAAFBLiveManager with test game: AUB vs UGA") # Updated log message
else:
logging.info("[NCAAFB] Initialized NCAAFBLiveManager in live mode") # Updated log message
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("[NCAAFB] Test mode: Could not parse clock") # Changed log prefix
# 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"[NCAAFB] Found {len(new_live_games)} live/halftime games for fav teams.") # Changed log prefix
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("[NCAAFB] No live/halftime games found for favorite teams.") # Changed log prefix
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("[NCAAFB] Live games previously showing have ended or are no longer live.") # Changed log prefix
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("[NCAAFB] Could not fetch update; keeping existing live game data for now.") # Changed log prefix
else:
self.logger.warning("[NCAAFB] Could not fetch data and no existing live games.") # Changed log prefix
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"[NCAAFB] Switched live view to: {self.current_game['away_abbr']}@{self.current_game['home_abbr']}") # Changed log prefix
# 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 NCAA FB game.""" # Updated docstring
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"[NCAAFB] Failed to load logos for live game: {game.get('id')}") # Changed log prefix
# 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) - 3 #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 = (self.display_height)- 7 #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 NCAAFB game: {e}", exc_info=True) # Changed log prefix
# Inherits display() method from BaseNCAAFBManager, which calls the overridden _draw_scorebug_layout
class NCAAFBRecentManager(BaseNCAAFBManager): # Renamed class
"""Manager for recently completed NCAA FB games.""" # Updated docstring
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 NCAAFBRecentManager with {len(self.favorite_teams)} favorite teams") # Changed log prefix
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("[NCAAFB Recent] No events found in shared data.") # Changed log prefix
if not self.games_list: self.current_game = None # Clear display if no games were showing
return
events = data['events']
# self.logger.info(f"[NCAAFB Recent] Processing {len(events)} events from shared data.") # Changed log prefix
# 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.get('is_within_window', True): # Assume within window if key missing, check logic
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"[NCAAFB Recent] Found {len(team_games)} final games within window for display.") # Changed log prefix
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("[NCAAFB Recent] No relevant recent games found to display.") # Changed log prefix
self.current_game = None # Ensure display clears if no games
except Exception as e:
self.logger.error(f"[NCAAFB Recent] Error updating recent games: {e}", exc_info=True) # Changed log prefix
# 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 NCAA FB game.""" # Updated docstring
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"[NCAAFB Recent] Failed to load logos for game: {game.get('id')}") # Changed log prefix
# 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"[NCAAFB Recent] Error displaying recent game: {e}", exc_info=True) # Changed log prefix
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"[NCAAFB Recent] Switched to game index {self.current_game_index}") # Changed log prefix
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"[NCAAFB Recent] Error in display loop: {e}", exc_info=True) # Changed log prefix
class NCAAFBUpcomingManager(BaseNCAAFBManager): # Renamed class
"""Manager for upcoming NCAA FB games.""" # Updated docstring
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 NCAAFBUpcomingManager with {len(self.favorite_teams)} favorite teams") # Changed log prefix
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("[NCAAFB Upcoming] No events found in shared data.") # Changed log prefix
if not self.games_list: self.current_game = None
return
events = data['events']
# self.logger.info(f"[NCAAFB Upcoming] Processing {len(events)} events from shared data.") # Changed log prefix
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.get('is_within_window', True): # Assume within window if key missing, check logic
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"[NCAAFB Upcoming] Found {len(team_games)} upcoming games within window for display.") # Changed log prefix
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("[NCAAFB Upcoming] No relevant upcoming games found to display.") # Changed log prefix
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"[NCAAFB Upcoming] Favorite teams: {self.favorite_teams}") # Changed log prefix
self.logger.debug(f"[NCAAFB Upcoming] Total upcoming games before filtering: {len(processed_games)}") # Changed log prefix
self.last_log_time = current_time
elif should_log:
self.last_log_time = current_time
except Exception as e:
self.logger.error(f"[NCAAFB Upcoming] Error updating upcoming games: {e}", exc_info=True) # Changed log prefix
# 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 NCAA FB game.""" # Updated docstring
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"[NCAAFB Upcoming] Failed to load logos for game: {game.get('id')}") # Changed log prefix
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"[NCAAFB Upcoming] Error displaying upcoming game: {e}", exc_info=True) # Changed log prefix
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("[NCAAFB Upcoming] No upcoming games found for favorite teams to display.") # Changed log prefix
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"[NCAAFB Upcoming] Switched to game index {self.current_game_index}") # Changed log prefix
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"[NCAAFB Upcoming] Error in display loop: {e}", exc_info=True) # Changed log prefix