rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-04-10 13:02:59 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
91211d5c86846fab43228285094341cb6a39f55d
LEDMatrix/src/ncaa_fb_managers.py
Chuck 91211d5c86 fix upcoming game logic
2025-09-15 18:22:25 -04:00

1687 lines
90 KiB
Python
Raw Blame History

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
from src.config_manager import ConfigManager
from src.odds_manager import OddsManager
from src.logo_downloader import download_missing_logo
import pytz
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 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'
)
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
_processed_games_cache = {} # Cache for processed game data
_processed_games_timestamp = 0
logger = logging.getLogger('NCAAFB') # Changed logger name
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager):
self.display_manager = display_manager
self.config = config
self.cache_manager = cache_manager
self.config_manager = self.cache_manager.config_manager
self.odds_manager = OddsManager(self.cache_manager, self.config_manager)
self.ncaa_fb_config = config.get("ncaa_fb_scoreboard", {}) # Changed config key
self.is_enabled = self.ncaa_fb_config.get("enabled", False)
self.show_odds = self.ncaa_fb_config.get("show_odds", 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.show_records = self.ncaa_fb_config.get('show_records', False)
self.show_ranking = self.ncaa_fb_config.get('show_ranking', False)
self.season_cache_duration = self.ncaa_fb_config.get("season_cache_duration_seconds", 86400) # 24 hours default
# Number of games to show (instead of time-based windows)
self.recent_games_to_show = self.ncaa_fb_config.get("recent_games_to_show", 5) # Show last 5 games
self.upcoming_games_to_show = self.ncaa_fb_config.get("upcoming_games_to_show", 10) # Show next 10 games
# Set up session with retry logic
self.session = requests.Session()
retry_strategy = Retry(
total=5, # increased number of retries
backoff_factor=1, # increased backoff factor
status_forcelist=[429, 500, 502, 503, 504], # added 429 to retry list
allowed_methods=["GET", "HEAD", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)
# Set up headers
self.headers = {
'User-Agent': 'LEDMatrix/1.0 (https://github.com/yourusername/LEDMatrix; contact@example.com)',
'Accept': 'application/json',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive'
}
self.last_update = 0
self.current_game = None
self.fonts = self._load_fonts()
self.favorite_teams = self.ncaa_fb_config.get("favorite_teams", [])
# Check display modes to determine what data to fetch
display_modes = self.ncaa_fb_config.get("display_modes", {})
self.recent_enabled = display_modes.get("ncaa_fb_recent", False)
self.upcoming_enabled = display_modes.get("ncaa_fb_upcoming", False)
self.live_enabled = display_modes.get("ncaa_fb_live", False)
self.logger.setLevel(logging.INFO)
self.display_width = self.display_manager.matrix.width
self.display_height = self.display_manager.matrix.height
self._logo_cache = {}
# Initialize team rankings cache
self._team_rankings_cache = {}
self._rankings_cache_timestamp = 0
self._rankings_cache_duration = 3600 # Cache rankings for 1 hour
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}")
self.logger.info(f"Display modes - Recent: {self.recent_enabled}, Upcoming: {self.upcoming_enabled}, Live: {self.live_enabled}")
def _fetch_team_rankings(self) -> Dict[str, int]:
"""Fetch current team rankings from ESPN API."""
current_time = time.time()
# Check if we have cached rankings that are still valid
if (self._team_rankings_cache and
current_time - self._rankings_cache_timestamp < self._rankings_cache_duration):
return self._team_rankings_cache
try:
rankings_url = "https://site.api.espn.com/apis/site/v2/sports/football/college-football/rankings"
response = self.session.get(rankings_url, headers=self.headers, timeout=30)
response.raise_for_status()
data = response.json()
rankings = {}
rankings_data = data.get('rankings', [])
if rankings_data:
# Use the first ranking (usually AP Top 25)
first_ranking = rankings_data[0]
teams = first_ranking.get('ranks', [])
for team_data in teams:
team_info = team_data.get('team', {})
team_abbr = team_info.get('abbreviation', '')
current_rank = team_data.get('current', 0)
if team_abbr and current_rank > 0:
rankings[team_abbr] = current_rank
# Cache the results
self._team_rankings_cache = rankings
self._rankings_cache_timestamp = current_time
self.logger.debug(f"Fetched rankings for {len(rankings)} teams")
return rankings
except Exception as e:
self.logger.error(f"Error fetching team rankings: {e}")
return {}
def _get_timezone(self):
try:
timezone_str = self.config.get('timezone', 'UTC')
return pytz.timezone(timezone_str)
except pytz.UnknownTimeZoneError:
return pytz.utc
def _should_log(self, warning_type: str, cooldown: int = 60) -> bool:
"""Check if we should log a warning based on cooldown period."""
current_time = time.time()
if current_time - self._last_warning_time > cooldown:
self._last_warning_time = current_time
return True
return False
def _fetch_odds(self, game: Dict) -> None:
"""Fetch odds for a specific game if conditions are met."""
# Check if odds should be shown for this sport
if not self.show_odds:
return
# Check if we should only fetch for favorite teams
is_favorites_only = self.ncaa_fb_config.get("show_favorite_teams_only", False)
if is_favorites_only:
home_abbr = game.get('home_abbr')
away_abbr = game.get('away_abbr')
if not (home_abbr in self.favorite_teams or away_abbr in self.favorite_teams):
self.logger.debug(f"Skipping odds fetch for non-favorite game in favorites-only mode: {away_abbr}@{home_abbr}")
return
self.logger.debug(f"Proceeding with odds fetch for game: {game.get('id', 'N/A')}")
# Fetch odds using OddsManager (ESPN API)
try:
# Determine update interval based on game state
is_live = game.get('status', '').lower() == 'in'
update_interval = self.ncaa_fb_config.get("live_odds_update_interval", 60) if is_live \
else self.ncaa_fb_config.get("odds_update_interval", 3600)
odds_data = self.odds_manager.get_odds(
sport="football",
league="ncaa_fb",
event_id=game['id'],
update_interval_seconds=update_interval
)
if odds_data:
game['odds'] = odds_data
self.logger.debug(f"Successfully fetched and attached odds for game {game['id']}")
else:
self.logger.debug(f"No odds data returned for game {game['id']}")
except Exception as e:
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
def _fetch_ncaa_fb_api_data(self, use_cache: bool = True) -> Optional[Dict]:
"""
Fetches the full season schedule for NCAAFB using week-by-week approach to ensure
we get all games, then caches the complete dataset.
This method now uses background threading to prevent blocking the display.
"""
now = datetime.now(pytz.utc)
current_year = now.year
years_to_check = [current_year]
if now.month < 8:
years_to_check.append(current_year - 1)
all_events = []
for year in years_to_check:
cache_key = f"ncaafb_schedule_{year}"
if use_cache:
cached_data = self.cache_manager.get(cache_key, max_age=self.season_cache_duration)
if cached_data:
self.logger.info(f"[NCAAFB] Using cached schedule for {year}")
all_events.extend(cached_data)
continue
# Check if we're already fetching this year's data in background
if hasattr(self, '_background_fetching') and year in self._background_fetching:
self.logger.info(f"[NCAAFB] Background fetch already in progress for {year}, using partial data")
# Return partial data if available, or trigger background fetch
partial_data = self._get_partial_schedule_data(year)
if partial_data:
all_events.extend(partial_data)
continue
self.logger.info(f"[NCAAFB] Fetching full {year} season schedule from ESPN API...")
# Start background fetch for complete data
self._start_background_schedule_fetch(year)
# For immediate response, fetch current/recent games only
year_events = self._fetch_immediate_games(year)
all_events.extend(year_events)
if not all_events:
self.logger.warning("[NCAAFB] No events found in schedule data.")
return None
return {'events': all_events}
def _fetch_immediate_games(self, year: int) -> List[Dict]:
"""Fetch immediate games (current week + next few days) for quick display."""
immediate_events = []
try:
# Fetch current week and next few days for immediate display
now = datetime.now(pytz.utc)
for days_offset in range(-1, 7): # Yesterday through next 6 days
check_date = now + timedelta(days=days_offset)
date_str = check_date.strftime('%Y%m%d')
url = f"https://site.api.espn.com/apis/site/v2/sports/football/college-football/scoreboard?dates={date_str}"
response = self.session.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
data = response.json()
date_events = data.get('events', [])
immediate_events.extend(date_events)
if days_offset == 0: # Today
self.logger.debug(f"[NCAAFB] Immediate fetch - Current date ({date_str}): {len(date_events)} events")
except requests.exceptions.RequestException as e:
self.logger.warning(f"[NCAAFB] Error fetching immediate games for {year}: {e}")
return immediate_events
def _start_background_schedule_fetch(self, year: int):
"""Start background thread to fetch complete season schedule."""
import threading
if not hasattr(self, '_background_fetching'):
self._background_fetching = set()
if year in self._background_fetching:
return # Already fetching
self._background_fetching.add(year)
def background_fetch():
try:
start_time = time.time()
self.logger.info(f"[NCAAFB] Starting background fetch for {year} season...")
year_events = []
# Fetch week by week to ensure we get complete season data
for week in range(1, 16):
# Add timeout check to prevent infinite background fetching
if time.time() - start_time > 300: # 5 minute timeout
self.logger.warning(f"[NCAAFB] Background fetch timeout after 5 minutes for {year}")
break
try:
url = f"https://site.api.espn.com/apis/site/v2/sports/football/college-football/scoreboard?seasontype=2&week={week}"
response = self.session.get(url, headers=self.headers, timeout=15)
response.raise_for_status()
data = response.json()
week_events = data.get('events', [])
year_events.extend(week_events)
# Log progress for first few weeks
if week <= 3:
self.logger.debug(f"[NCAAFB] Background - Week {week}: fetched {len(week_events)} events")
# If no events found for this week, we might be past the season
if not week_events and week > 10:
self.logger.debug(f"[NCAAFB] Background - No events found for week {week}, ending season fetch")
break
except requests.exceptions.RequestException as e:
self.logger.warning(f"[NCAAFB] Background - Error fetching week {week} for {year}: {e}")
continue
# Also fetch postseason games (bowl games, playoffs) if we haven't timed out
if time.time() - start_time < 300:
try:
url = f"https://site.api.espn.com/apis/site/v2/sports/football/college-football/scoreboard?seasontype=3"
response = self.session.get(url, headers=self.headers, timeout=15)
response.raise_for_status()
data = response.json()
postseason_events = data.get('events', [])
year_events.extend(postseason_events)
self.logger.debug(f"[NCAAFB] Background - Postseason: fetched {len(postseason_events)} events")
except requests.exceptions.RequestException as e:
self.logger.warning(f"[NCAAFB] Background - Error fetching postseason for {year}: {e}")
# Cache the complete data
cache_key = f"ncaafb_schedule_{year}"
self.cache_manager.set(cache_key, year_events)
elapsed_time = time.time() - start_time
self.logger.info(f"[NCAAFB] Background fetch completed for {year}: {len(year_events)} events cached in {elapsed_time:.1f}s")
except Exception as e:
self.logger.error(f"[NCAAFB] Background fetch failed for {year}: {e}")
finally:
self._background_fetching.discard(year)
# Start background thread
fetch_thread = threading.Thread(target=background_fetch, daemon=True)
fetch_thread.start()
def _get_partial_schedule_data(self, year: int) -> List[Dict]:
"""Get partial schedule data if available from cache or previous fetch."""
cache_key = f"ncaafb_schedule_{year}"
cached_data = self.cache_manager.get(cache_key, max_age=self.season_cache_duration * 2) # Allow older data
if cached_data:
self.logger.debug(f"[NCAAFB] Using partial cached data for {year}: {len(cached_data)} events")
return cached_data
return []
def _fetch_data(self, date_str: str = None) -> Optional[Dict]:
"""Fetch data using shared data mechanism or direct fetch for live."""
if isinstance(self, NCAAFBLiveManager):
return self._fetch_ncaa_fb_api_data(use_cache=False)
else:
return self._fetch_ncaa_fb_api_data(use_cache=True)
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_dynamic_odds(self, draw: ImageDraw.Draw, odds: Dict[str, Any], width: int, height: int) -> None:
"""Draw odds with dynamic positioning - only show negative spread and position O/U based on favored team."""
home_team_odds = odds.get('home_team_odds', {})
away_team_odds = odds.get('away_team_odds', {})
home_spread = home_team_odds.get('spread_odds')
away_spread = away_team_odds.get('spread_odds')
# Get top-level spread as fallback
top_level_spread = odds.get('spread')
# If we have a top-level spread and the individual spreads are None or 0, use the top-level
if top_level_spread is not None:
if home_spread is None or home_spread == 0.0:
home_spread = top_level_spread
if away_spread is None:
away_spread = -top_level_spread
# Determine which team is favored (has negative spread)
home_favored = home_spread is not None and home_spread < 0
away_favored = away_spread is not None and away_spread < 0
# Only show the negative spread (favored team)
favored_spread = None
favored_side = None
if home_favored:
favored_spread = home_spread
favored_side = 'home'
self.logger.debug(f"Home team favored with spread: {favored_spread}")
elif away_favored:
favored_spread = away_spread
favored_side = 'away'
self.logger.debug(f"Away team favored with spread: {favored_spread}")
else:
self.logger.debug("No clear favorite - spreads: home={home_spread}, away={away_spread}")
# Show the negative spread on the appropriate side
if favored_spread is not None:
spread_text = str(favored_spread)
font = self.fonts['detail'] # Use detail font for odds
if favored_side == 'home':
# Home team is favored, show spread on right side
spread_width = draw.textlength(spread_text, font=font)
spread_x = width - spread_width # Top right
spread_y = 0
self._draw_text_with_outline(draw, spread_text, (spread_x, spread_y), font, fill=(0, 255, 0))
self.logger.debug(f"Showing home spread '{spread_text}' on right side")
else:
# Away team is favored, show spread on left side
spread_x = 0 # Top left
spread_y = 0
self._draw_text_with_outline(draw, spread_text, (spread_x, spread_y), font, fill=(0, 255, 0))
self.logger.debug(f"Showing away spread '{spread_text}' on left side")
# Show over/under on the opposite side of the favored team
over_under = odds.get('over_under')
if over_under is not None:
ou_text = f"O/U: {over_under}"
font = self.fonts['detail'] # Use detail font for odds
ou_width = draw.textlength(ou_text, font=font)
if favored_side == 'home':
# Home team is favored, show O/U on left side (opposite of spread)
ou_x = 0 # Top left
ou_y = 0
self.logger.debug(f"Showing O/U '{ou_text}' on left side (home favored)")
elif favored_side == 'away':
# Away team is favored, show O/U on right side (opposite of spread)
ou_x = width - ou_width # Top right
ou_y = 0
self.logger.debug(f"Showing O/U '{ou_text}' on right side (away favored)")
else:
# No clear favorite, show O/U in center
ou_x = (width - ou_width) // 2
ou_y = 0
self.logger.debug(f"Showing O/U '{ou_text}' in center (no clear favorite)")
self._draw_text_with_outline(draw, ou_text, (ou_x, ou_y), font, fill=(0, 255, 0))
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, team_name: str = None) -> Optional[Image.Image]:
"""Load and resize a team logo, with caching and automatic download if missing."""
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:
# Try to download missing logo first
if not os.path.exists(logo_path):
self.logger.info(f"Logo not found for {team_abbrev} at {logo_path}. Attempting to download.")
# Try to download the logo from ESPN API
success = download_missing_logo(team_abbrev, 'ncaa_fb', team_name)
if not success:
# Create placeholder if download fails
self.logger.warning(f"Failed to download logo for {team_abbrev}. 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
home_abbr = home_team["team"]["abbreviation"]
away_abbr = away_team["team"]["abbreviation"]
# Check if this is a favorite team game BEFORE doing expensive logging
is_favorite_game = (home_abbr in self.favorite_teams or away_abbr in self.favorite_teams)
# Only log debug info for favorite team games
if is_favorite_game:
self.logger.debug(f"[NCAAFB] Processing favorite team game: {game_event.get('id')}")
self.logger.debug(f"[NCAAFB] Found teams: {away_abbr}@{home_abbr}, Status: {status['type']['name']}, State: {status['type']['state']}")
home_record = home_team.get('records', [{}])[0].get('summary', '') if home_team.get('records') else ''
away_record = away_team.get('records', [{}])[0].get('summary', '') if away_team.get('records') else ''
# Don't show "0-0" records - set to blank instead
if home_record == "0-0":
home_record = ''
if away_record == "0-0":
away_record = ''
# Remove early filtering - let individual managers handle their own filtering
# This allows shared data to contain all games, and each manager can filter as needed
game_time, game_date = "", ""
if start_time_utc:
local_time = start_time_utc.astimezone(self._get_timezone())
game_time = local_time.strftime("%I:%M%p").lstrip('0')
# Check date format from config
use_short_date_format = self.config.get('display', {}).get('use_short_date_format', False)
if use_short_date_format:
game_date = local_time.strftime("%-m/%-d")
else:
game_date = self.display_manager.format_date_with_ordinal(local_time)
# --- Football Specific Details (Likely same for NFL/NCAAFB) ---
situation = competition.get("situation")
down_distance_text = ""
possession_indicator = None # Default to None
scoring_event = "" # Track scoring events
if situation and status["type"]["state"] == "in":
down = situation.get("down")
distance = situation.get("distance")
# Validate down and distance values before formatting
if (down is not None and isinstance(down, int) and 1 <= down <= 4 and
distance is not None and isinstance(distance, int) and distance >= 0):
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
# Detect scoring events from status detail
status_detail = status["type"].get("detail", "").lower()
status_short = status["type"].get("shortDetail", "").lower()
# Check for scoring events in status text
if any(keyword in status_detail for keyword in ["touchdown", "td"]):
scoring_event = "TOUCHDOWN"
elif any(keyword in status_detail for keyword in ["field goal", "fg"]):
scoring_event = "FIELD GOAL"
elif any(keyword in status_detail for keyword in ["extra point", "pat", "point after"]):
scoring_event = "PAT"
elif any(keyword in status_short for keyword in ["touchdown", "td"]):
scoring_event = "TOUCHDOWN"
elif any(keyword in status_short for keyword in ["field goal", "fg"]):
scoring_event = "FIELD GOAL"
elif any(keyword in status_short for keyword in ["extra point", "pat"]):
scoring_event = "PAT"
# Determine possession based on team ID
possession_team_id = situation.get("possession")
if possession_team_id:
if possession_team_id == home_team.get("id"):
possession_indicator = "home"
elif possession_team_id == away_team.get("id"):
possession_indicator = "away"
# 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 = "Q3" # Fixed: period 3 is 3rd quarter, not halftime
elif period == 4: period_text = "Q4"
elif period > 4: period_text = "OT" # OT starts after Q4
elif status["type"]["state"] == "halftime" or status["type"]["name"] == "STATUS_HALFTIME": # Check explicit halftime state
period_text = "HALF"
elif status["type"]["state"] == "post":
if period > 4 : 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
# For upcoming games, we'll show based on number of games, not time window
# For recent games, we'll show based on number of games, not time window
is_within_window = True # Always include games, let the managers filter by count
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" or
status["type"]["name"].lower() in ['scheduled', 'pre-game', 'status_scheduled']),
"is_halftime": status["type"]["state"] == "halftime" or status["type"]["name"] == "STATUS_HALFTIME", # Added halftime check
"home_abbr": home_abbr,
"home_score": home_team.get("score", "0"),
"home_record": home_record,
"home_logo_path": os.path.join(self.logo_dir, f"{home_abbr}.png"),
"home_timeouts": home_timeouts,
"away_abbr": away_abbr,
"away_score": away_team.get("score", "0"),
"away_record": away_record,
"away_logo_path": os.path.join(self.logo_dir, f"{away_abbr}.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
"possession_indicator": possession_indicator, # Added for easy home/away check
"scoring_event": scoring_event, # Track scoring events (TOUCHDOWN, FIELD GOAL, PAT)
"is_within_window": is_within_window, # Whether game is within display window
}
# 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, cache_manager: CacheManager):
super().__init__(config, display_manager, cache_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", # Placeholder ID for home team
"possession_indicator": "home", # Explicitly set for test
"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
def update(self):
"""Update live game data and handle game switching."""
if not self.is_enabled:
return
# Define current_time and interval before the problematic line (originally line 455)
# Ensure 'import time' is present at the top of the file.
current_time = time.time()
# Define interval using a pattern similar to NFLLiveManager's update method.
# Uses getattr for robustness, assuming attributes for live_games, test_mode,
# no_data_interval, and update_interval are available on self.
_live_games_attr = getattr(self, 'live_games', [])
_test_mode_attr = getattr(self, 'test_mode', False) # test_mode is often from a base class or config
_no_data_interval_attr = getattr(self, 'no_data_interval', 300) # Default similar to NFLLiveManager
_update_interval_attr = getattr(self, 'update_interval', 15) # Default similar to NFLLiveManager
interval = _no_data_interval_attr if not _live_games_attr and not _test_mode_attr else _update_interval_attr
# Original line from traceback (line 455), now with variables defined:
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"] < 4: # Q4 is period 4
self.current_game["period"] += 1
# Update period_text based on new period
if self.current_game["period"] == 1: self.current_game["period_text"] = "Q1"
elif self.current_game["period"] == 2: self.current_game["period_text"] = "Q2"
elif self.current_game["period"] == 3: self.current_game["period_text"] = "Q3"
elif self.current_game["period"] == 4: 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"]):
# If show_favorite_teams_only is true, only add if it's a favorite.
# Otherwise, add all games.
if self.ncaa_fb_config.get("show_favorite_teams_only", False):
if details["home_abbr"] in self.favorite_teams or details["away_abbr"] in self.favorite_teams:
if self.show_odds:
self._fetch_odds(details)
new_live_games.append(details)
else:
if self.show_odds:
self._fetch_odds(details)
new_live_games.append(details)
# Log changes or periodically
current_time_for_log = time.time() # Use a consistent time for logging comparison
should_log = (
current_time_for_log - 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:
filter_text = "favorite teams" if self.ncaa_fb_config.get("show_favorite_teams_only", False) else "all teams"
self.logger.info(f"[NCAAFB] Found {len(new_live_games)} live/halftime games for {filter_text}.")
for game_info in new_live_games: # Renamed game to game_info
self.logger.info(f" - {game_info['away_abbr']}@{game_info['home_abbr']} ({game_info.get('status_text', 'N/A')})")
else:
filter_text = "favorite teams" if self.ncaa_fb_config.get("show_favorite_teams_only", False) else "criteria"
self.logger.info(f"[NCAAFB] No live/halftime games found for {filter_text}.")
self.last_log_time = current_time_for_log
# 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"], game.get("home_team_name"))
away_logo = self._load_and_resize_logo(game["away_abbr"], game.get("away_team_name"))
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 or Scoring Event (Below Period/Clock)
scoring_event = game.get("scoring_event", "")
down_distance = game.get("down_distance_text", "")
# Show scoring event if detected, otherwise show down & distance
if scoring_event and game.get("is_live"):
# Display scoring event with special formatting
event_width = draw_overlay.textlength(scoring_event, font=self.fonts['detail'])
event_x = (self.display_width - event_width) // 2
event_y = (self.display_height) - 7
# Color coding for different scoring events
if scoring_event == "TOUCHDOWN":
event_color = (255, 215, 0) # Gold
elif scoring_event == "FIELD GOAL":
event_color = (0, 255, 0) # Green
elif scoring_event == "PAT":
event_color = (255, 165, 0) # Orange
else:
event_color = (255, 255, 255) # White
self._draw_text_with_outline(draw_overlay, scoring_event, (event_x, event_y), self.fonts['detail'], fill=event_color)
elif 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 # Top of D&D text
self._draw_text_with_outline(draw_overlay, down_distance, (dd_x, dd_y), self.fonts['detail'], fill=(200, 200, 0)) # Yellowish text
# Possession Indicator (small football icon)
possession = game.get("possession_indicator")
if possession: # Only draw if possession is known
ball_radius_x = 3 # Wider for football shape
ball_radius_y = 2 # Shorter for football shape
ball_color = (139, 69, 19) # Brown color for the football
lace_color = (255, 255, 255) # White for laces
# Approximate height of the detail font (4x6 font at size 6 is roughly 6px tall)
detail_font_height_approx = 6
ball_y_center = dd_y + (detail_font_height_approx // 2) # Center ball vertically with D&D text
possession_ball_padding = 3 # Pixels between D&D text and ball
if possession == "away":
# Position ball to the left of D&D text
ball_x_center = dd_x - possession_ball_padding - ball_radius_x
elif possession == "home":
# Position ball to the right of D&D text
ball_x_center = dd_x + dd_width + possession_ball_padding + ball_radius_x
else:
ball_x_center = 0 # Should not happen / no indicator
if ball_x_center > 0: # Draw if position is valid
# Draw the football shape (ellipse)
draw_overlay.ellipse(
(ball_x_center - ball_radius_x, ball_y_center - ball_radius_y, # x0, y0
ball_x_center + ball_radius_x, ball_y_center + ball_radius_y), # x1, y1
fill=ball_color, outline=(0,0,0)
)
# Draw a simple horizontal lace
draw_overlay.line(
(ball_x_center - 1, ball_y_center, ball_x_center + 1, ball_y_center),
fill=lace_color, width=1
)
# 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))
# Draw odds if available
if 'odds' in game and game['odds']:
self._draw_dynamic_odds(draw_overlay, game['odds'], self.display_width, self.display_height)
# 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, cache_manager: CacheManager):
super().__init__(config, display_manager, cache_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
# Define date range for "recent" games (last 21 days to capture games from 3 weeks ago)
now = datetime.now(timezone.utc)
recent_cutoff = now - timedelta(days=21)
self.logger.info(f"[NCAAFB Recent DEBUG] Current time: {now}, Recent cutoff: {recent_cutoff} (21 days ago)")
# Process games and filter for final games, date range & favorite teams
processed_games = []
favorite_games_found = 0
for event in events:
game = self._extract_game_details(event)
# Filter criteria: must be final AND within recent date range
if game and game['is_final']:
game_time = game.get('start_time_utc')
if game_time and game_time >= recent_cutoff:
processed_games.append(game)
# Count favorite team games for logging
if (game['home_abbr'] in self.favorite_teams or
game['away_abbr'] in self.favorite_teams):
favorite_games_found += 1
# Special check for Tennessee game in recent games
if (game['home_abbr'] == 'TENN' and game['away_abbr'] == 'UGA') or (game['home_abbr'] == 'UGA' and game['away_abbr'] == 'TENN'):
self.logger.info(f"[NCAAFB Recent DEBUG] Found Tennessee game in recent: {game['away_abbr']} @ {game['home_abbr']} - {game.get('start_time_utc')} - Score: {game['away_score']}-{game['home_score']}")
# 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]
self.logger.info(f"[NCAAFB Recent] Found {favorite_games_found} favorite team games out of {len(processed_games)} total final games within last 21 days")
# Debug: Show which games are selected for display
for i, game in enumerate(team_games):
self.logger.info(f"[NCAAFB Recent DEBUG] Game {i+1} for display: {game['away_abbr']} @ {game['home_abbr']} - {game.get('start_time_utc')} - Score: {game['away_score']}-{game['home_score']}")
else:
team_games = processed_games # Show all recent games if no favorites defined
self.logger.info(f"[NCAAFB Recent] Found {len(processed_games)} total final games within last 21 days (no favorite teams configured)")
# 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)
# Limit to the specified number of recent games
team_games = team_games[:self.recent_games_to_show]
# 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"], game.get("home_team_name"))
away_logo = self._load_and_resize_logo(game["away_abbr"], game.get("away_team_name"))
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
# MLB-style logo positioning (closer to edges)
home_x = self.display_width - home_logo.width + 2
home_y = center_y - (home_logo.height // 2)
main_img.paste(home_logo, (home_x, home_y), home_logo)
away_x = -2
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'])
# Draw odds if available
if 'odds' in game and game['odds']:
self._draw_dynamic_odds(draw_overlay, game['odds'], self.display_width, self.display_height)
# Draw records or rankings if enabled
if self.show_records or self.show_ranking:
try:
record_font = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6)
except IOError:
record_font = ImageFont.load_default()
# Get team abbreviations
away_abbr = game.get('away_abbr', '')
home_abbr = game.get('home_abbr', '')
record_bbox = draw_overlay.textbbox((0,0), "0-0", font=record_font)
record_height = record_bbox[3] - record_bbox[1]
record_y = self.display_height - record_height
# Display away team info
if away_abbr:
if self.show_ranking:
# Show ranking if available
rankings = self._fetch_team_rankings()
away_rank = rankings.get(away_abbr, 0)
if away_rank > 0:
away_text = f"#{away_rank}"
elif self.show_records:
# Only show record if show_records is enabled
away_text = game.get('away_record', '')
else:
# Show nothing if show_records is false and team is unranked
away_text = ''
else:
# Show record only if show_records is enabled
if self.show_records:
away_text = game.get('away_record', '')
else:
away_text = ''
if away_text:
away_record_x = 0
self._draw_text_with_outline(draw_overlay, away_text, (away_record_x, record_y), record_font)
# Display home team info
if home_abbr:
if self.show_ranking:
# Show ranking if available
rankings = self._fetch_team_rankings()
home_rank = rankings.get(home_abbr, 0)
if home_rank > 0:
home_text = f"#{home_rank}"
elif self.show_records:
# Only show record if show_records is enabled
home_text = game.get('home_record', '')
else:
# Show nothing if show_records is false and team is unranked
home_text = ''
else:
# Show record only if show_records is enabled
if self.show_records:
home_text = game.get('home_record', '')
else:
home_text = ''
if home_text:
home_record_bbox = draw_overlay.textbbox((0,0), home_text, font=record_font)
home_record_width = home_record_bbox[2] - home_record_bbox[0]
home_record_x = self.display_width - home_record_width
self._draw_text_with_outline(draw_overlay, home_text, (home_record_x, record_y), record_font)
# 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, cache_manager: CacheManager):
super().__init__(config, display_manager, cache_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 = []
favorite_games_found = 0
all_upcoming_games = 0 # Count all upcoming games regardless of favorites
for event in events:
game = self._extract_game_details(event)
# Count all upcoming games for debugging
if game and game['is_upcoming']:
all_upcoming_games += 1
# Filter criteria: must be upcoming ('pre' state)
if game and game['is_upcoming']:
# Only fetch odds for games that will be displayed
if self.ncaa_fb_config.get("show_favorite_teams_only", False):
if not self.favorite_teams:
continue
if game['home_abbr'] not in self.favorite_teams and game['away_abbr'] not in self.favorite_teams:
continue
processed_games.append(game)
# Count favorite team games for logging
if (game['home_abbr'] in self.favorite_teams or
game['away_abbr'] in self.favorite_teams):
favorite_games_found += 1
if self.show_odds:
self._fetch_odds(game)
# Enhanced logging for debugging
self.logger.info(f"[NCAAFB Upcoming] Found {all_upcoming_games} total upcoming games in data")
self.logger.info(f"[NCAAFB Upcoming] Found {len(processed_games)} upcoming games after filtering")
# Debug: Check what statuses we're seeing
status_counts = {}
status_names = {} # Track actual status names from ESPN
favorite_team_games = []
for event in events:
game = self._extract_game_details(event)
if game:
status = "upcoming" if game['is_upcoming'] else "final" if game['is_final'] else "live" if game['is_live'] else "other"
status_counts[status] = status_counts.get(status, 0) + 1
# Track actual ESPN status names
actual_status = event.get('competitions', [{}])[0].get('status', {}).get('type', {})
status_name = actual_status.get('name', 'Unknown')
status_state = actual_status.get('state', 'Unknown')
status_names[f"{status_name} ({status_state})"] = status_names.get(f"{status_name} ({status_state})", 0) + 1
# Check for favorite team games regardless of status
if (game['home_abbr'] in self.favorite_teams or game['away_abbr'] in self.favorite_teams):
favorite_team_games.append({
'teams': f"{game['away_abbr']} @ {game['home_abbr']}",
'status': status,
'date': game.get('start_time_utc', 'Unknown'),
'espn_status': f"{status_name} ({status_state})"
})
# Special check for Tennessee game (Georgia @ Tennessee)
if (game['home_abbr'] == 'TENN' and game['away_abbr'] == 'UGA') or (game['home_abbr'] == 'UGA' and game['away_abbr'] == 'TENN'):
self.logger.info(f"[NCAAFB DEBUG] Found Tennessee game: {game['away_abbr']} @ {game['home_abbr']} - {status} - {game.get('start_time_utc')} - ESPN: {status_name} ({status_state})")
self.logger.info(f"[NCAAFB Upcoming] Status breakdown: {status_counts}")
self.logger.info(f"[NCAAFB Upcoming] ESPN status names: {status_names}")
if favorite_team_games:
self.logger.info(f"[NCAAFB Upcoming] Favorite team games found: {len(favorite_team_games)}")
for game in favorite_team_games[:3]: # Show first 3
self.logger.info(f"[NCAAFB Upcoming] {game['teams']} - {game['status']} - {game['date']} - ESPN: {game['espn_status']}")
if self.favorite_teams and all_upcoming_games > 0:
self.logger.info(f"[NCAAFB Upcoming] Favorite teams: {self.favorite_teams}")
self.logger.info(f"[NCAAFB Upcoming] Found {favorite_games_found} favorite team upcoming games")
# Filter for favorite teams only if the config is set
if self.ncaa_fb_config.get("show_favorite_teams_only", False):
# Get all games involving favorite teams
favorite_team_games = [game for game in processed_games
if game['home_abbr'] in self.favorite_teams or
game['away_abbr'] in self.favorite_teams]
# Select one game per favorite team (earliest upcoming game for each team)
team_games = []
for team in self.favorite_teams:
# Find games where this team is playing
team_specific_games = [game for game in favorite_team_games
if game['home_abbr'] == team or game['away_abbr'] == team]
if team_specific_games:
# Sort by game time and take the earliest
team_specific_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
team_games.append(team_specific_games[0])
# Sort the final list by game time
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
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))
# Limit to the specified number of upcoming games
team_games = team_games[:self.upcoming_games_to_show]
# 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"], game.get("home_team_name"))
away_logo = self._load_and_resize_logo(game["away_abbr"], game.get("away_team_name"))
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
# MLB-style logo positions
home_x = self.display_width - home_logo.width + 2
home_y = center_y - (home_logo.height // 2)
main_img.paste(home_logo, (home_x, home_y), home_logo)
away_x = -2
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'])
# Draw odds if available
if 'odds' in game and game['odds']:
self._draw_dynamic_odds(draw_overlay, game['odds'], self.display_width, self.display_height)
# Draw records or rankings if enabled
if self.show_records or self.show_ranking:
try:
record_font = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6)
except IOError:
record_font = ImageFont.load_default()
# Get team abbreviations
away_abbr = game.get('away_abbr', '')
home_abbr = game.get('home_abbr', '')
record_bbox = draw_overlay.textbbox((0,0), "0-0", font=record_font)
record_height = record_bbox[3] - record_bbox[1]
record_y = self.display_height - record_height
# Display away team info
if away_abbr:
if self.show_ranking:
# Show ranking if available
rankings = self._fetch_team_rankings()
away_rank = rankings.get(away_abbr, 0)
if away_rank > 0:
away_text = f"#{away_rank}"
elif self.show_records:
# Only show record if show_records is enabled
away_text = game.get('away_record', '')
else:
# Show nothing if show_records is false and team is unranked
away_text = ''
else:
# Show record only if show_records is enabled
if self.show_records:
away_text = game.get('away_record', '')
else:
away_text = ''
if away_text:
away_record_x = 0
self._draw_text_with_outline(draw_overlay, away_text, (away_record_x, record_y), record_font)
# Display home team info
if home_abbr:
if self.show_ranking:
# Show ranking if available
rankings = self._fetch_team_rankings()
home_rank = rankings.get(home_abbr, 0)
if home_rank > 0:
home_text = f"#{home_rank}"
elif self.show_records:
# Only show record if show_records is enabled
home_text = game.get('home_record', '')
else:
# Show nothing if show_records is false and team is unranked
home_text = ''
else:
# Show record only if show_records is enabled
if self.show_records:
home_text = game.get('home_record', '')
else:
home_text = ''
if home_text:
home_record_bbox = draw_overlay.textbbox((0,0), home_text, font=record_font)
home_record_width = home_record_bbox[2] - home_record_bbox[0]
home_record_x = self.display_width - home_record_width
self._draw_text_with_outline(draw_overlay, home_text, (home_record_x, record_y), record_font)
# 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
Reference in New Issue View Git Blame Copy Permalink