rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-04-10 21:03:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Consolidate Baseball leagues to completely use Baseball class (#84)

Browse Source 
* Consolidate MLB to completely use Baseball class

* typos

* add OT period number

* Add new live class and abstracts

* NCAA BB is consolidated

* MLB Working

* NCAA Hockey and NHL working

* didn't need wrapper function

* Add hockey shots on goal

* cleanup

---------

Co-authored-by: Alex Resnick <adr8282@gmail.com>
This commit is contained in:
Alex Resnick
2025-09-30 16:55:45 -05:00
committed by GitHub
parent e584026bda
commit 2d6c238ea0
9 changed files with 1480 additions and 3838 deletions
Download Patch File Download Diff File Expand all files Collapse all files

662
src/base_classes/baseball.py
View File

@@ -5,142 +5,676 @@ This module provides baseball-specific base classes that extend the core sports
with baseball-specific logic for innings, outs, bases, strikes, balls, etc.
"""
from typing import Dict, Any, Optional, List
from src.base_classes.sports import SportsCore
from src.base_classes.api_extractors import ESPNBaseballExtractor
from src.base_classes.data_sources import ESPNDataSource, MLBAPIDataSource
import logging
import random
import time
from typing import Any, Dict, Optional
from PIL import Image, ImageDraw
from src.base_classes.data_sources import ESPNDataSource
from src.base_classes.sports import SportsCore, SportsLive
class Baseball(SportsCore):
"""Base class for baseball sports with common functionality."""
def __init__(self, config: Dict[str, Any], display_manager, cache_manager, logger: logging.Logger, sport_key: str):
def __init__(
self,
config: Dict[str, Any],
display_manager,
cache_manager,
logger: logging.Logger,
sport_key: str,
):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
# Initialize baseball-specific architecture components
self.api_extractor = ESPNBaseballExtractor(logger)
# Choose data source based on sport (MLB uses MLB API, others use ESPN)
if sport_key == 'mlb':
self.data_source = MLBAPIDataSource(logger)
else:
self.data_source = ESPNDataSource(logger)
# Baseball-specific configuration
self.show_innings = self.mode_config.get("show_innings", True)
self.show_outs = self.mode_config.get("show_outs", True)
self.show_bases = self.mode_config.get("show_bases", True)
self.show_count = self.mode_config.get("show_count", True)
self.show_pitcher_batter = self.mode_config.get("show_pitcher_batter", False)
self.data_source = ESPNDataSource(logger)
self.sport = "baseball"
def _get_baseball_display_text(self, game: Dict) -> str:
"""Get baseball-specific display text."""
try:
display_parts = []
# Inning information
if self.show_innings:
inning = game.get('inning', '')
inning = game.get("inning", "")
if inning:
display_parts.append(f"Inning: {inning}")
# Outs information
if self.show_outs:
outs = game.get('outs', 0)
outs = game.get("outs", 0)
if outs is not None:
display_parts.append(f"Outs: {outs}")
# Bases information
if self.show_bases:
bases = game.get('bases', '')
bases = game.get("bases", "")
if bases:
display_parts.append(f"Bases: {bases}")
# Count information
if self.show_count:
strikes = game.get('strikes', 0)
balls = game.get('balls', 0)
strikes = game.get("strikes", 0)
balls = game.get("balls", 0)
if strikes is not None and balls is not None:
display_parts.append(f"Count: {balls}-{strikes}")
# Pitcher/Batter information
if self.show_pitcher_batter:
pitcher = game.get('pitcher', '')
batter = game.get('batter', '')
pitcher = game.get("pitcher", "")
batter = game.get("batter", "")
if pitcher:
display_parts.append(f"Pitcher: {pitcher}")
if batter:
display_parts.append(f"Batter: {batter}")
return " | ".join(display_parts) if display_parts else ""
except Exception as e:
self.logger.error(f"Error getting baseball display text: {e}")
return ""
def _is_baseball_game_live(self, game: Dict) -> bool:
"""Check if a baseball game is currently live."""
try:
# Check if game is marked as live
is_live = game.get('is_live', False)
is_live = game.get("is_live", False)
if is_live:
return True
# Check inning to determine if game is active
inning = game.get('inning', '')
if inning and inning != 'Final':
inning = game.get("inning", "")
if inning and inning != "Final":
return True
return False
except Exception as e:
self.logger.error(f"Error checking if baseball game is live: {e}")
return False
def _get_baseball_game_status(self, game: Dict) -> str:
"""Get baseball-specific game status."""
try:
status = game.get('status_text', '')
inning = game.get('inning', '')
status = game.get("status_text", "")
inning = game.get("inning", "")
if self._is_baseball_game_live(game):
if inning:
return f"Live - {inning}"
else:
return "Live"
elif game.get('is_final', False):
elif game.get("is_final", False):
return "Final"
elif game.get('is_upcoming', False):
elif game.get("is_upcoming", False):
return "Upcoming"
else:
return status
except Exception as e:
self.logger.error(f"Error getting baseball game status: {e}")
return ""
class BaseballLive(Baseball):
"""Base class for live baseball games."""
def __init__(self, config: Dict[str, Any], display_manager, cache_manager, logger: logging.Logger, sport_key: str):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
self.logger.info(f"{sport_key.upper()} Live Manager initialized")
def _should_show_baseball_game(self, game: Dict) -> bool:
"""Determine if a baseball game should be shown."""
def _extract_game_details(self, game_event: Dict) -> Optional[Dict]:
"""Extract relevant game details from ESPN NCAA FB API response."""
details, home_team, away_team, status, situation = (
self._extract_game_details_common(game_event)
)
if details is None or home_team is None or away_team is None or status is None:
return
try:
# Only show live games
if not self._is_baseball_game_live(game):
return False
# Check if game meets display criteria
return self._should_show_game(game)
# print(status["type"]["state"])
# exit()
game_status = status["type"]["name"].lower()
status_state = status["type"]["state"].lower()
# Get team abbreviations
home_abbr = home_team["team"]["abbreviation"]
away_abbr = away_team["team"]["abbreviation"]
# Check if this is a favorite team game
is_favorite_game = (
home_abbr in self.favorite_teams or away_abbr in self.favorite_teams
)
# Log all teams found for debugging
self.logger.debug(
f"Found game: {away_abbr} @ {home_abbr} (Status: {game_status}, State: {status_state})"
)
# Only log detailed information for favorite teams
if is_favorite_game:
self.logger.debug(f"Full status data: {game_event['status']}")
self.logger.debug(f"Status type: {game_status}, State: {status_state}")
self.logger.debug(f"Status detail: {status['type'].get('detail', '')}")
self.logger.debug(
f"Status shortDetail: {status['type'].get('shortDetail', '')}"
)
# Get game state information
if status_state == "in":
# For live games, get detailed state
inning = game_event["status"].get(
"period", 1
) # Get inning from status period
# Get inning information from status
status_detail = status["type"].get("detail", "").lower()
status_short = status["type"].get("shortDetail", "").lower()
if is_favorite_game:
self.logger.debug(
f"Raw status detail: {status['type'].get('detail')}"
)
self.logger.debug(
f"Raw status short: {status['type'].get('shortDetail')}"
)
# Determine inning half from status information
inning_half = "top" # Default
# Handle end of inning: next inning is top
if "end" in status_detail or "end" in status_short:
inning_half = "top"
inning = (
game_event["status"].get("period", 1) + 1
) # Use period and increment for next inning
if is_favorite_game:
self.logger.debug(
f"Detected end of inning. Setting to Top {inning}"
)
# Handle middle of inning: next is bottom of current inning
elif "mid" in status_detail or "mid" in status_short:
inning_half = "bottom"
if is_favorite_game:
self.logger.debug(
f"Detected middle of inning. Setting to Bottom {inning}"
)
# Handle bottom of inning
elif (
"bottom" in status_detail
or "bot" in status_detail
or "bottom" in status_short
or "bot" in status_short
):
inning_half = "bottom"
if is_favorite_game:
self.logger.debug(f"Detected bottom of inning: {inning}")
# Handle top of inning
elif "top" in status_detail or "top" in status_short:
inning_half = "top"
if is_favorite_game:
self.logger.debug(f"Detected top of inning: {inning}")
if is_favorite_game:
self.logger.debug(f"Status detail: {status_detail}")
self.logger.debug(f"Status short: {status_short}")
self.logger.debug(f"Determined inning: {inning_half} {inning}")
# Get count and bases from situation
situation = game_event["competitions"][0].get("situation", {})
if is_favorite_game:
self.logger.debug(f"Full situation data: {situation}")
# Get count from the correct location in the API response
count = situation.get("count", {})
balls = count.get("balls", 0)
strikes = count.get("strikes", 0)
outs = situation.get("outs", 0)
# Add detailed logging for favorite team games
if is_favorite_game:
self.logger.debug(f"Full situation data: {situation}")
self.logger.debug(f"Count object: {count}")
self.logger.debug(
f"Raw count values - balls: {balls}, strikes: {strikes}"
)
self.logger.debug(f"Raw outs value: {outs}")
# Try alternative locations for count data
if balls == 0 and strikes == 0:
# First try the summary field
if "summary" in situation:
try:
count_summary = situation["summary"]
balls, strikes = map(int, count_summary.split("-"))
if is_favorite_game:
self.logger.debug(
f"Using summary count: {count_summary}"
)
except (ValueError, AttributeError):
if is_favorite_game:
self.logger.debug("Could not parse summary count")
else:
# Check if count is directly in situation
balls = situation.get("balls", 0)
strikes = situation.get("strikes", 0)
if is_favorite_game:
self.logger.debug(
f"Using direct situation count: balls={balls}, strikes={strikes}"
)
self.logger.debug(
f"Full situation keys: {list(situation.keys())}"
)
if is_favorite_game:
self.logger.debug(f"Final count: balls={balls}, strikes={strikes}")
# Get base runners
bases_occupied = [
situation.get("onFirst", False),
situation.get("onSecond", False),
situation.get("onThird", False),
]
if is_favorite_game:
self.logger.debug(f"Bases occupied: {bases_occupied}")
else:
# Default values for non-live games
inning = 1
inning_half = "top"
balls = 0
strikes = 0
outs = 0
bases_occupied = [False, False, False]
details.update(
{
"status": game_status,
"status_state": status_state,
"inning": inning,
"inning_half": inning_half,
"balls": balls,
"strikes": strikes,
"outs": outs,
"bases_occupied": bases_occupied,
"start_time": game_event["date"],
}
)
# Basic validation (can be expanded)
if not details["home_abbr"] or not details["away_abbr"]:
self.logger.warning(
f"Missing team abbreviation in event: {details['id']}"
)
return None
self.logger.debug(
f"Extracted: {details['away_abbr']}@{details['home_abbr']}, Status: {status['type']['name']}, Live: {details['is_live']}, Final: {details['is_final']}, Upcoming: {details['is_upcoming']}"
)
return details
except Exception as e:
self.logger.error(f"Error checking if baseball game should be shown: {e}")
return False
# Log the problematic event structure if possible
self.logger.error(
f"Error extracting game details: {e} from event: {game_event.get('id')}",
exc_info=True,
)
return None
class BaseballLive(Baseball, SportsLive):
"""Base class for live baseball games."""
def __init__(
self,
config: Dict[str, Any],
display_manager,
cache_manager,
logger: logging.Logger,
sport_key: str,
):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
def _test_mode_update(self):
if self.current_game and self.current_game["is_live"]:
# self.current_game["bases_occupied"] = [
# random.choice([True, False]) for _ in range(3)
# ]
# self.current_game["balls"] = random.choice([1, 2, 3])
# self.current_game["strikes"] = random.choice([1, 2])
# self.current_game["outs"] = random.choice([1, 2])
if self.current_game["inning_half"] == "top": self.current_game["inning_half"] = "bottom"
else: self.current_game["inning_half"] = "top"; self.current_game["inning"] += 1
self.current_game["balls"] = (self.current_game["balls"] + 1) % 4
self.current_game["strikes"] = (self.current_game["strikes"] + 1) % 3
self.current_game["outs"] = (self.current_game["outs"] + 1) % 3
self.current_game["bases_occupied"] = [not b for b in self.current_game["bases_occupied"]]
if self.current_game["inning"] % 2 == 0: self.current_game["home_score"] = str(int(self.current_game["home_score"]) + 1)
else: self.current_game["away_score"] = str(int(self.current_game["away_score"]) + 1)
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_id"],
game["home_abbr"],
game["home_logo_path"],
game.get("home_logo_url"),
)
away_logo = self._load_and_resize_logo(
game["away_id"],
game["away_abbr"],
game["away_logo_path"],
game.get("away_logo_url"),
)
if not home_logo or not away_logo:
self.logger.error(
f"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)
# --- Live Game Specific Elements ---
# Define default text color
text_color = (255, 255, 255)
# Draw Inning (Top Center)
inning_half = game["inning_half"]
inning_num = game["inning"]
if game["is_final"]:
inning_text = "FINAL"
else:
inning_half_indicator = (
"▲" if game["inning_half"].lower() == "top" else "▼"
)
inning_num = game["inning"]
inning_text = f"{inning_half_indicator}{inning_num}"
inning_bbox = draw_overlay.textbbox(
(0, 0), inning_text, font=self.display_manager.font
)
inning_width = inning_bbox[2] - inning_bbox[0]
inning_x = (self.display_width - inning_width) // 2
inning_y = 1 # Position near top center
# draw_overlay.text((inning_x, inning_y), inning_text, fill=(255, 255, 255), font=self.display_manager.font)
self._draw_text_with_outline(
draw_overlay,
inning_text,
(inning_x, inning_y),
self.display_manager.font,
)
# --- REVISED BASES AND OUTS DRAWING ---
bases_occupied = game["bases_occupied"] # [1st, 2nd, 3rd]
outs = game.get("outs", 0)
inning_half = game["inning_half"]
# Define geometry
base_diamond_size = 7
out_circle_diameter = 3
out_vertical_spacing = 2 # Space between out circles
spacing_between_bases_outs = (
3 # Horizontal space between base cluster and out column
)
base_vert_spacing = 1 # Internal vertical space in base cluster
base_horiz_spacing = 1 # Internal horizontal space in base cluster
# Calculate cluster dimensions
base_cluster_height = (
base_diamond_size + base_vert_spacing + base_diamond_size
)
base_cluster_width = (
base_diamond_size + base_horiz_spacing + base_diamond_size
)
out_cluster_height = 3 * out_circle_diameter + 2 * out_vertical_spacing
out_cluster_width = out_circle_diameter
# Calculate overall start positions
overall_start_y = (
inning_bbox[3] + 0
) # Start immediately below inning text (moved up 3 pixels)
# Center the BASE cluster horizontally
bases_origin_x = (self.display_width - base_cluster_width) // 2
# Determine relative positions for outs based on inning half
if inning_half == "top": # Away batting, outs on left
outs_column_x = (
bases_origin_x - spacing_between_bases_outs - out_cluster_width
)
else: # Home batting, outs on right
outs_column_x = (
bases_origin_x + base_cluster_width + spacing_between_bases_outs
)
# Calculate vertical alignment offset for outs column (center align with bases cluster)
outs_column_start_y = (
overall_start_y + (base_cluster_height // 2) - (out_cluster_height // 2)
)
# --- Draw Bases (Diamonds) ---
base_color_occupied = (255, 255, 255)
base_color_empty = (255, 255, 255) # Outline color
h_d = base_diamond_size // 2
# 2nd Base (Top center relative to bases_origin_x)
c2x = bases_origin_x + base_cluster_width // 2
c2y = overall_start_y + h_d
poly2 = [
(c2x, overall_start_y),
(c2x + h_d, c2y),
(c2x, c2y + h_d),
(c2x - h_d, c2y),
]
if bases_occupied[1]:
draw_overlay.polygon(poly2, fill=base_color_occupied)
else:
draw_overlay.polygon(poly2, outline=base_color_empty)
base_bottom_y = c2y + h_d # Bottom Y of 2nd base diamond
# 3rd Base (Bottom left relative to bases_origin_x)
c3x = bases_origin_x + h_d
c3y = base_bottom_y + base_vert_spacing + h_d
poly3 = [
(c3x, base_bottom_y + base_vert_spacing),
(c3x + h_d, c3y),
(c3x, c3y + h_d),
(c3x - h_d, c3y),
]
if bases_occupied[2]:
draw_overlay.polygon(poly3, fill=base_color_occupied)
else:
draw_overlay.polygon(poly3, outline=base_color_empty)
# 1st Base (Bottom right relative to bases_origin_x)
c1x = bases_origin_x + base_cluster_width - h_d
c1y = base_bottom_y + base_vert_spacing + h_d
poly1 = [
(c1x, base_bottom_y + base_vert_spacing),
(c1x + h_d, c1y),
(c1x, c1y + h_d),
(c1x - h_d, c1y),
]
if bases_occupied[0]:
draw_overlay.polygon(poly1, fill=base_color_occupied)
else:
draw_overlay.polygon(poly1, outline=base_color_empty)
# --- Draw Outs (Vertical Circles) ---
circle_color_out = (255, 255, 255)
circle_color_empty_outline = (100, 100, 100)
for i in range(3):
cx = outs_column_x
cy = outs_column_start_y + i * (
out_circle_diameter + out_vertical_spacing
)
coords = [cx, cy, cx + out_circle_diameter, cy + out_circle_diameter]
if i < outs:
draw_overlay.ellipse(coords, fill=circle_color_out)
else:
draw_overlay.ellipse(coords, outline=circle_color_empty_outline)
# --- Draw Balls-Strikes Count (BDF Font) ---
balls = game.get("balls", 0)
strikes = game.get("strikes", 0)
# Add debug logging for count with cooldown
current_time = time.time()
if (
game["home_abbr"] in self.favorite_teams
or game["away_abbr"] in self.favorite_teams
) and current_time - self.last_count_log_time >= self.count_log_interval:
self.logger.debug(f"Displaying count: {balls}-{strikes}")
self.logger.debug(
f"Raw count data: balls={game.get('balls')}, strikes={game.get('strikes')}"
)
self.last_count_log_time = current_time
count_text = f"{balls}-{strikes}"
bdf_font = self.display_manager.calendar_font
bdf_font.set_char_size(height=7 * 64) # Set 7px height
count_text_width = self.display_manager.get_text_width(count_text, bdf_font)
# Position below the base/out cluster
cluster_bottom_y = (
overall_start_y + base_cluster_height
) # Find the bottom of the taller part (bases)
count_y = cluster_bottom_y + 2 # Start 2 pixels below cluster
# Center horizontally within the BASE cluster width
count_x = bases_origin_x + (base_cluster_width - count_text_width) // 2
# Ensure draw object is set and draw text
self.display_manager.draw = draw_overlay
# self.display_manager._draw_bdf_text(count_text, count_x, count_y, text_color, font=bdf_font)
# Use _draw_text_with_outline for count text
# self._draw_text_with_outline(draw, count_text, (count_x, count_y), bdf_font, fill=text_color)
# Draw Balls-Strikes Count with outline using BDF font
# Define outline color (consistent with _draw_text_with_outline default)
outline_color_for_bdf = (0, 0, 0)
# Draw outline
for dx_offset, dy_offset in [
(-1, -1),
(-1, 0),
(-1, 1),
(0, -1),
(0, 1),
(1, -1),
(1, 0),
(1, 1),
]:
self.display_manager._draw_bdf_text(
count_text,
count_x + dx_offset,
count_y + dy_offset,
color=outline_color_for_bdf,
font=bdf_font,
)
# Draw main text
self.display_manager._draw_bdf_text(
count_text, count_x, count_y, color=text_color, font=bdf_font
)
# Draw Team:Score at the bottom (matching main branch format)
score_font = self.display_manager.font # Use PressStart2P
outline_color = (0, 0, 0)
score_text_color = (
255,
255,
255,
) # Use a specific name for score text color
# Helper function for outlined text
def draw_bottom_outlined_text(x, y, text):
self._draw_text_with_outline(
draw_overlay,
text,
(x, y),
score_font,
fill=score_text_color,
outline_color=outline_color,
)
away_abbr = game["away_abbr"]
home_abbr = game["home_abbr"]
away_score_str = str(game["away_score"])
home_score_str = str(game["home_score"])
away_text = f"{away_abbr}:{away_score_str}"
home_text = f"{home_abbr}:{home_score_str}"
# Calculate Y position (bottom edge)
# Get font height (approximate or precise)
try:
font_height = score_font.getbbox("A")[3] - score_font.getbbox("A")[1]
except AttributeError:
font_height = 8 # Fallback for default font
score_y = (
self.display_height - font_height - 2
) # 2 pixels padding from bottom
# Away Team:Score (Bottom Left)
away_score_x = 2 # 2 pixels padding from left
draw_bottom_outlined_text(away_score_x, score_y, away_text)
# Home Team:Score (Bottom Right)
home_text_bbox = draw_overlay.textbbox((0, 0), home_text, font=score_font)
home_text_width = home_text_bbox[2] - home_text_bbox[0]
home_score_x = (
self.display_width - home_text_width - 2
) # 2 pixels padding from right
draw_bottom_outlined_text(home_score_x, score_y, home_text)
# Draw gambling 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 Football game: {e}", exc_info=True
) # Changed log prefix

223
src/base_classes/football.py
View File

@@ -5,20 +5,14 @@ from datetime import datetime, timezone, timedelta
import logging
from PIL import Image, ImageDraw, ImageFont
import time
import pytz
from src.base_classes.sports import SportsCore
from src.base_classes.api_extractors import ESPNFootballExtractor
from src.base_classes.data_sources import ESPNDataSource
import requests
from src.base_classes.sports import SportsCore, SportsLive
class Football(SportsCore):
"""Base class for football sports with common functionality."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
# Initialize football-specific architecture components
self.api_extractor = ESPNFootballExtractor(logger)
self.data_source = ESPNDataSource(logger)
self.sport = "football"
@@ -82,12 +76,12 @@ class Football(SportsCore):
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
if period == 0:
period_text = "Start" # Before kickoff
elif period >= 1 and period <= 4:
period_text = f"Q{period}" # OT starts after Q4
elif period > 4:
period_text = f"OT{period - 4}" # 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":
@@ -122,179 +116,48 @@ class Football(SportsCore):
logging.error(f"Error extracting game details: {e} from event: {game_event.get('id')}", exc_info=True)
return None
class FootballLive(Football):
class FootballLive(Football, SportsLive):
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
self.update_interval = self.mode_config.get("live_update_interval", 15)
self.no_data_interval = 300
self.last_update = 0
self.live_games = []
self.current_game_index = 0
self.last_game_switch = 0
self.game_display_duration = self.mode_config.get("live_game_duration", 20)
self.last_display_update = 0
self.last_log_time = 0
self.log_interval = 300
def update(self):
"""Update live game data and handle game switching."""
if not self.is_enabled:
return
def _test_mode_update(self):
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']}"
# 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()
# Display update handled by main loop or explicit call if needed immediately
# self.display(force_clear=True) # Only if immediate update is desired here
# 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
except ValueError:
self.logger.warning("Test mode: Could not parse clock") # Changed log prefix
# No actual display call here, let main loop handle it
# Original line from traceback (line 455), now with variables defined:
if current_time - self.last_update >= interval:
self.last_update = current_time
# Fetch rankings if enabled
if self.show_ranking:
self._fetch_team_rankings()
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("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 game in data["events"]:
details = self._extract_game_details(game)
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.show_favorite_teams_only:
if details["home_abbr"] in self.favorite_teams or details["away_abbr"] in self.favorite_teams:
new_live_games.append(details)
else:
new_live_games.append(details)
for game in new_live_games:
if self.show_odds:
self._fetch_odds(game)
# 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.show_favorite_teams_only else "all teams"
self.logger.info(f"Found {len(new_live_games)} live/halftime games for {filter_text}.")
for game_info in new_live_games: # Renamed game to game_info
self.logger.info(f" - {game_info['away_abbr']}@{game_info['home_abbr']} ({game_info.get('status_text', 'N/A')})")
else:
filter_text = "favorite teams" if self.show_favorite_teams_only else "criteria"
self.logger.info(f"No live/halftime games found for {filter_text}.")
self.last_log_time = current_time_for_log
# 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("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("Could not fetch update; keeping existing live game data for now.") # Changed log prefix
else:
self.logger.warning("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"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

452
src/base_classes/hockey.py
View File

@@ -1,30 +1,37 @@
from typing import Dict, Any, Optional
from src.display_manager import DisplayManager
from src.cache_manager import CacheManager
from datetime import datetime, timezone
import logging
from PIL import Image, ImageDraw, ImageFont
import time
from src.base_classes.sports import SportsCore
from src.base_classes.api_extractors import ESPNHockeyExtractor
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from PIL import Image, ImageDraw, ImageFont
from src.base_classes.data_sources import ESPNDataSource
from src.base_classes.sports import SportsCore, SportsLive
from src.cache_manager import CacheManager
from src.display_manager import DisplayManager
class Hockey(SportsCore):
"""Base class for hockey sports with common functionality."""
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
def __init__(
self,
config: Dict[str, Any],
display_manager: DisplayManager,
cache_manager: CacheManager,
logger: logging.Logger,
sport_key: str,
):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
# Initialize hockey-specific architecture components
self.api_extractor = ESPNHockeyExtractor(logger)
self.data_source = ESPNDataSource(logger)
self.sport = "hockey"
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 ---
details, home_team, away_team, status, situation = self._extract_game_details_common(game_event)
details, home_team, away_team, status, situation = (
self._extract_game_details_common(game_event)
)
if details is None or home_team is None or away_team is None or status is None:
return
try:
@@ -32,7 +39,46 @@ class Hockey(SportsCore):
status = competition["status"]
powerplay = False
penalties = ""
shots_on_goal = {"home": 0, "away": 0}
home_team_saves = next(
(
int(c["displayValue"])
for c in home_team["statistics"]
if c.get("name") == "saves"
),
0,
)
home_team_saves_per = next(
(
float(c["displayValue"])
for c in home_team["statistics"]
if c.get("name") == "savePct"
),
0.0,
)
away_team_saves = next(
(
int(c["displayValue"])
for c in away_team["statistics"]
if c.get("name") == "saves"
),
0,
)
away_team_saves_per = next(
(
float(c["displayValue"])
for c in away_team["statistics"]
if c.get("name") == "savePct"
),
0.0,
)
home_shots = 0
away_shots = 0
if home_team_saves_per > 0:
away_shots = round(home_team_saves / home_team_saves_per)
if away_team_saves_per > 0:
home_shots = round(away_team_saves / away_team_saves_per)
if situation and status["type"]["state"] == "in":
# Detect scoring events from status detail
@@ -40,222 +86,192 @@ class Hockey(SportsCore):
status_short = status["type"].get("shortDetail", "").lower()
powerplay = situation.get("isPowerPlay", False)
penalties = situation.get("penalties", "")
shots_on_goal = {
"home": situation.get("homeShots", 0),
"away": situation.get("awayShots", 0)
}
# 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 = "P1"
elif period == 2: period_text = "P2"
elif period == 3: period_text = "P3" # Fixed: period 3 is 3rd quarter, not halftime
elif period > 3: period_text = f"OT {period - 3}" # OT starts after P3
if period == 0:
period_text = "Start" # Before kickoff
elif period >= 1 and period <= 3:
period_text = f"P{period}" # OT starts after Q4
elif period > 3:
period_text = f"OT{period - 3}" # OT starts after Q4
elif status["type"]["state"] == "post":
if period > 3 :
period_text = "Final/OT"
else:
period_text = "Final"
if period > 3:
period_text = "Final/OT"
else:
period_text = "Final"
elif status["type"]["state"] == "pre":
period_text = details.get("game_time", "") # Show time for upcoming
period_text = details.get("game_time", "") # Show time for upcoming
details.update({
"period": period,
"period_text": period_text, # Formatted quarter/status
"clock": status.get("displayClock", "0:00"),
"power_play": powerplay,
"penalties": penalties,
"shots_on_goal": shots_on_goal
})
details.update(
{
"period": period,
"period_text": period_text, # Formatted quarter/status
"clock": status.get("displayClock", "0:00"),
"power_play": powerplay,
"penalties": penalties,
"home_shots": home_shots,
"away_shots": away_shots,
}
)
# Basic validation (can be expanded)
if not details['home_abbr'] or not details['away_abbr']:
self.logger.warning(f"Missing team abbreviation in event: {details['id']}")
return None
if not details["home_abbr"] or not details["away_abbr"]:
self.logger.warning(
f"Missing team abbreviation in event: {details['id']}"
)
return None
self.logger.debug(f"Extracted: {details['away_abbr']}@{details['home_abbr']}, Status: {status['type']['name']}, Live: {details['is_live']}, Final: {details['is_final']}, Upcoming: {details['is_upcoming']}")
self.logger.debug(
f"Extracted: {details['away_abbr']}@{details['home_abbr']}, Status: {status['type']['name']}, Live: {details['is_live']}, Final: {details['is_final']}, Upcoming: {details['is_upcoming']}"
)
return details
except Exception as e:
# Log the problematic event structure if possible
logging.error(f"Error extracting game details: {e} from event: {game_event.get('id')}", exc_info=True)
self.logger.error(
f"Error extracting game details: {e} from event: {game_event.get('id')}",
exc_info=True,
)
return None
class HockeyLive(Hockey):
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
class HockeyLive(Hockey, SportsLive):
def __init__(
self,
config: Dict[str, Any],
display_manager: DisplayManager,
cache_manager: CacheManager,
logger: logging.Logger,
sport_key: str,
):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
self.update_interval = self.mode_config.get("live_update_interval", 15)
self.no_data_interval = 300
self.last_update = 0
self.live_games = []
self.current_game_index = 0
self.last_game_switch = 0
self.game_display_duration = self.mode_config.get("live_game_duration", 20)
self.last_display_update = 0
self.last_log_time = 0
self.log_interval = 300
def update(self):
"""Update live game data."""
if not self.is_enabled: return
current_time = time.time()
interval = self.no_data_interval if not self.live_games else self.update_interval
if current_time - self.last_update >= interval:
self.last_update = current_time
if self.show_ranking:
self._fetch_team_rankings()
if self.test_mode:
# For testing, we'll just update the clock to show it's working
if self.current_game:
minutes = int(self.current_game["clock"].split(":")[0])
seconds = int(self.current_game["clock"].split(":")[1])
seconds -= 1
if seconds < 0:
seconds = 59
minutes -= 1
if minutes < 0:
minutes = 19
if self.current_game["period"] < 3:
self.current_game["period"] += 1
else:
self.current_game["period"] = 1
self.current_game["clock"] = f"{minutes:02d}:{seconds:02d}"
# Always update display in test mode
self.display(force_clear=True)
else:
# Fetch live game data from ESPN API
data = self._fetch_data()
if data and "events" in data:
# Find all live games involving favorite teams
new_live_games = []
for event in data["events"]:
details = self._extract_game_details(event)
if details and details["is_live"]:
self._fetch_odds(details)
new_live_games.append(details)
# Filter for favorite teams only if the config is set
if self.show_favorite_teams_only:
new_live_games = [game for game in new_live_games
if game['home_abbr'] in self.favorite_teams or
game['away_abbr'] in self.favorite_teams]
# Only log if there's a change in games or enough time has passed
should_log = (
current_time - self.last_log_time >= self.log_interval or
len(new_live_games) != len(self.live_games) or
not self.live_games # Log if we had no games before
)
if should_log:
if new_live_games:
filter_text = "favorite teams" if self.show_favorite_teams_only else "all teams"
self.logger.info(f"[NCAAMH] Found {len(new_live_games)} live games involving {filter_text}")
for game in new_live_games:
self.logger.info(f"[NCAAMH] Live game: {game['away_abbr']} vs {game['home_abbr']} - Period {game['period']}, {game['clock']}")
else:
filter_text = "favorite teams" if self.show_favorite_teams_only else "criteria"
self.logger.info(f"[NCAAMH] No live games found matching {filter_text}")
self.last_log_time = current_time
if new_live_games:
# Update the current game with the latest data
for new_game in new_live_games:
if self.current_game and (
(new_game["home_abbr"] == self.current_game["home_abbr"] and
new_game["away_abbr"] == self.current_game["away_abbr"]) or
(new_game["home_abbr"] == self.current_game["away_abbr"] and
new_game["away_abbr"] == self.current_game["home_abbr"])
):
self.current_game = new_game
break
# Only update the games list if we have new games
if not self.live_games or set(game["away_abbr"] + game["home_abbr"] for game in new_live_games) != set(game["away_abbr"] + game["home_abbr"] for game in self.live_games):
self.live_games = new_live_games
# If we don't have a current game or it's not in the new list, start from the beginning
if not self.current_game or self.current_game not in self.live_games:
self.current_game_index = 0
self.current_game = self.live_games[0]
self.last_game_switch = current_time
# Update display if data changed, limit rate
if current_time - self.last_display_update >= 1.0:
# self.display(force_clear=True) # REMOVED: DisplayController handles this
self.last_display_update = current_time
def _test_mode_update(self):
if self.current_game and self.current_game["is_live"]:
# For testing, we'll just update the clock to show it's working
minutes = int(self.current_game["clock"].split(":")[0])
seconds = int(self.current_game["clock"].split(":")[1])
seconds -= 1
if seconds < 0:
seconds = 59
minutes -= 1
if minutes < 0:
minutes = 19
if self.current_game["period"] < 3:
self.current_game["period"] += 1
else:
# No live games found
self.live_games = []
self.current_game = None
# Check if it's time to switch games
if len(self.live_games) > 1 and (current_time - self.last_game_switch) >= self.game_display_duration:
self.current_game_index = (self.current_game_index + 1) % len(self.live_games)
self.current_game = self.live_games[self.current_game_index]
self.last_game_switch = current_time
# self.display(force_clear=True) # REMOVED: DisplayController handles this
self.last_display_update = current_time # Track time for potential display update
self.current_game["period"] = 1
self.current_game["clock"] = f"{minutes:02d}:{seconds:02d}"
# Always update display in test mode
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
"""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_id"], game["home_abbr"], game["home_logo_path"], game.get("home_logo_url"))
away_logo = self._load_and_resize_logo(game["away_id"], game["away_abbr"], game["away_logo_path"], game.get("away_logo_url"))
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_id"],
game["home_abbr"],
game["home_logo_path"],
game.get("home_logo_url"),
)
away_logo = self._load_and_resize_logo(
game["away_id"],
game["away_abbr"],
game["away_logo_path"],
game.get("away_logo_url"),
)
if not home_logo or not away_logo:
self.logger.error(f"Failed to load logos for live game: {game.get('id')}") # Changed log prefix
self.logger.error(
f"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))
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_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_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 ---
# Note: Rankings are now handled in the records/rankings section below
# 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"],
)
# 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_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'])
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'])
# Shots on Goal
shots_font = ImageFont.truetype("assets/fonts/4x6-font.ttf", 6)
home_shots = str(game.get("home_shots", "0"))
away_shots = str(game.get("away_shots", "0"))
shots_text = f"{away_shots} SHOTS {home_shots}"
shots_width = draw_overlay.textlength(shots_text, font=shots_font)
shots_x = (self.display_width - shots_width) // 2
shots_y = (
self.display_height - 10
) # centered #from 14 # Position score higher
self._draw_text_with_outline(
draw_overlay, shots_text, (shots_x, shots_y), shots_font
)
# 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)
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:
@@ -264,16 +280,20 @@ class HockeyLive(Hockey):
self.logger.debug(f"Loaded 6px record font successfully")
except IOError:
record_font = ImageFont.load_default()
self.logger.warning(f"Failed to load 6px font, using default font (size: {record_font.size})")
self.logger.warning(
f"Failed to load 6px font, using default font (size: {record_font.size})"
)
# 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)
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 - 4
self.logger.debug(f"Record positioning: height={record_height}, record_y={record_y}, display_height={self.display_height}")
self.logger.debug(
f"Record positioning: height={record_height}, record_y={record_y}, display_height={self.display_height}"
)
# Display away team info
if away_abbr:
@@ -284,24 +304,31 @@ class HockeyLive(Hockey):
away_text = f"#{away_rank}"
else:
# Show nothing for unranked teams when rankings are prioritized
away_text = ''
away_text = ""
elif self.show_ranking:
# Show ranking only if available
away_rank = self._team_rankings_cache.get(away_abbr, 0)
if away_rank > 0:
away_text = f"#{away_rank}"
else:
away_text = ''
away_text = ""
elif self.show_records:
# Show record only when rankings are disabled
away_text = game.get('away_record', '')
away_text = game.get("away_record", "")
else:
away_text = ''
away_text = ""
if away_text:
away_record_x = 3
self.logger.debug(f"Drawing away ranking '{away_text}' at ({away_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
self._draw_text_with_outline(draw_overlay, away_text, (away_record_x, record_y), record_font)
self.logger.debug(
f"Drawing away ranking '{away_text}' at ({away_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}"
)
self._draw_text_with_outline(
draw_overlay,
away_text,
(away_record_x, record_y),
record_font,
)
# Display home team info
if home_abbr:
@@ -312,34 +339,45 @@ class HockeyLive(Hockey):
home_text = f"#{home_rank}"
else:
# Show nothing for unranked teams when rankings are prioritized
home_text = ''
home_text = ""
elif self.show_ranking:
# Show ranking only if available
home_rank = self._team_rankings_cache.get(home_abbr, 0)
if home_rank > 0:
home_text = f"#{home_rank}"
else:
home_text = ''
home_text = ""
elif self.show_records:
# Show record only when rankings are disabled
home_text = game.get('home_record', '')
home_text = game.get("home_record", "")
else:
home_text = ''
home_text = ""
if home_text:
home_record_bbox = draw_overlay.textbbox((0,0), home_text, font=record_font)
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 - 3
self.logger.debug(f"Drawing home ranking '{home_text}' at ({home_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
self._draw_text_with_outline(draw_overlay, home_text, (home_record_x, record_y), record_font)
self.logger.debug(
f"Drawing home ranking '{home_text}' at ({home_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}"
)
self._draw_text_with_outline(
draw_overlay,
home_text,
(home_record_x, record_y),
record_font,
)
# 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
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
self.display_manager.update_display() # Update display here for live
except Exception as e:
self.logger.error(f"Error displaying live Hockey game: {e}", exc_info=True) # Changed log prefix
self.logger.error(
f"Error displaying live Hockey game: {e}", exc_info=True
) # Changed log prefix

260
src/base_classes/sports.py
View File

@@ -1,26 +1,30 @@
from typing import Dict, Any, Optional, List
from src.display_manager import DisplayManager
from src.cache_manager import CacheManager
from datetime import datetime, timedelta, timezone
import logging
import os
from src.odds_manager import OddsManager
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Callable
import pytz
import requests
from PIL import Image, ImageDraw, ImageFont
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from PIL import Image, ImageDraw, ImageFont
import pytz
import time
from abc import ABC, abstractmethod
from src.background_data_service import get_background_service
from src.logo_downloader import download_missing_logo, LogoDownloader
from pathlib import Path
# Import new architecture components (individual classes will import what they need)
from src.base_classes.api_extractors import APIDataExtractor
from src.base_classes.data_sources import DataSource
from src.cache_manager import CacheManager
from src.display_manager import DisplayManager
from src.dynamic_team_resolver import DynamicTeamResolver
from src.logo_downloader import LogoDownloader, download_missing_logo
from src.odds_manager import OddsManager
class SportsCore:
class SportsCore(ABC):
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
self.logger = logger
self.config = config
@@ -41,20 +45,21 @@ class SportsCore:
self.api_extractor: APIDataExtractor
self.data_source: DataSource
self.mode_config = config.get(f"{sport_key}_scoreboard", {}) # Changed config key
self.is_enabled = self.mode_config.get("enabled", False)
self.show_odds = self.mode_config.get("show_odds", False)
self.test_mode = self.mode_config.get("test_mode", False)
self.is_enabled: bool = self.mode_config.get("enabled", False)
self.show_odds: bool = self.mode_config.get("show_odds", False)
self.test_mode: bool = self.mode_config.get("test_mode", False)
self.logo_dir = Path(self.mode_config.get("logo_dir", "assets/sports/ncaa_logos")) # Changed logo dir
self.update_interval = self.mode_config.get(
self.update_interval: int = self.mode_config.get(
"update_interval_seconds", 60)
self.show_records = self.mode_config.get('show_records', False)
self.show_ranking = self.mode_config.get('show_ranking', False)
self.show_records: bool = self.mode_config.get('show_records', False)
self.show_ranking: bool = self.mode_config.get('show_ranking', False)
# Number of games to show (instead of time-based windows)
self.recent_games_to_show = self.mode_config.get(
self.recent_games_to_show: int = self.mode_config.get(
"recent_games_to_show", 5) # Show last 5 games
self.upcoming_games_to_show = self.mode_config.get(
self.upcoming_games_to_show: int = self.mode_config.get(
"upcoming_games_to_show", 10) # Show next 10 games
self.show_favorite_teams_only = self.mode_config.get("show_favorite_teams_only", False)
self.show_favorite_teams_only: bool = self.mode_config.get("show_favorite_teams_only", False)
self.show_all_live: bool = self.mode_config.get("show_all_live", False)
self.session = requests.Session()
retry_strategy = Retry(
@@ -114,6 +119,9 @@ class SportsCore:
self.background_enabled = False
self.logger.info("Background service disabled")
def _get_season_schedule_dates(self) -> tuple[str, str]:
return "", ""
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
@@ -314,14 +322,6 @@ class SportsCore:
if not self.show_odds:
return
# Check if we should only fetch for favorite teams
if self.show_favorite_teams_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
# Determine update interval based on game state
is_live = game.get('is_live', False)
update_interval = self.mode_config.get("live_odds_update_interval", 60) if is_live \
@@ -485,16 +485,12 @@ class SportsCore:
logging.error(f"Error extracting game details: {e} from event: {game_event.get('id')}", exc_info=True)
return None, None, None, None, None
@abstractmethod
def _extract_game_details(self, game_event: dict) -> dict | None:
details, _, _, _, _ = self._extract_game_details_common(game_event)
return details
# def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
# pass
# def display(self, force_clear=False):
# pass
@abstractmethod
def _fetch_data(self) -> Optional[Dict]:
pass
@@ -610,62 +606,22 @@ class SportsUpcoming(SportsCore):
# Enhanced logging for debugging
self.logger.info(f"Found {all_upcoming_games} total upcoming games in data")
self.logger.info(f"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"Found Tennessee game: {game['away_abbr']} @ {game['home_abbr']} - {status} - {game.get('start_time_utc')} - ESPN: {status_name} ({status_state})")
self.logger.info(f"Status breakdown: {status_counts}")
self.logger.info(f"ESPN status names: {status_names}")
if favorite_team_games:
self.logger.info(f"Favorite team games found: {len(favorite_team_games)}")
for game in favorite_team_games[:3]: # Show first 3
self.logger.info(f" {game['teams']} - {game['status']} - {game['date']} - ESPN: {game['espn_status']}")
if processed_games:
for game in processed_games[:3]: # Show first 3
self.logger.info(f" {game['away_abbr']}@{game['home_abbr']} - {game['start_time_utc']}")
if self.favorite_teams and all_upcoming_games > 0:
self.logger.info(f"Favorite teams: {self.favorite_teams}")
self.logger.info(f"Found {favorite_games_found} favorite team upcoming games")
# Filter for favorite teams only if the config is set
if self.show_favorite_teams_only:
# Get all games involving favorite teams
favorite_team_games = [game for game in processed_games
if game['home_abbr'] in self.favorite_teams or
game['away_abbr'] in self.favorite_teams]
if self.show_favorite_teams_only:
# 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:
# Find games where this team is playing
if team_specific_games := [game for game in processed_games if game['home_abbr'] == team or game['away_abbr'] == team]:
# 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])
@@ -1200,4 +1156,144 @@ class SportsRecent(SportsCore):
# update_display() is called within _draw_scorebug_layout for recent
except Exception as e:
self.logger.error(f"Error in display loop: {e}", exc_info=True) # Changed log prefix
self.logger.error(f"Error in display loop: {e}", exc_info=True) # Changed log prefix
class SportsLive(SportsCore):
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
self.update_interval = self.mode_config.get("live_update_interval", 15)
self.no_data_interval = 300
self.last_update = 0
self.live_games = []
self.current_game_index = 0
self.last_game_switch = 0
self.game_display_duration = self.mode_config.get("live_game_duration", 20)
self.last_display_update = 0
self.last_log_time = 0
self.log_interval = 300
@abstractmethod
def _test_mode_update(self) -> None:
return
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 = self.live_games
_test_mode_attr = self.test_mode # test_mode is often from a base class or config
_no_data_interval_attr = self.no_data_interval # Default similar to NFLLiveManager
_update_interval_attr = self.update_interval # 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
# Fetch rankings if enabled
if self.show_ranking:
self._fetch_team_rankings()
if self.test_mode:
# Simulate clock running down in test mode
self._test_mode_update()
else:
# Fetch live game data
data = self._fetch_data()
new_live_games = []
if data and "events" in data:
for game in data["events"]:
details = self._extract_game_details(game)
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.show_all_live or not self.show_favorite_teams_only or (self.show_favorite_teams_only and (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)
# 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.show_favorite_teams_only or self.show_all_live else "all teams"
self.logger.info(f"Found {len(new_live_games)} live/halftime games for {filter_text}.")
for game_info in new_live_games: # Renamed game to game_info
self.logger.info(f" - {game_info['away_abbr']}@{game_info['home_abbr']} ({game_info.get('status_text', 'N/A')})")
else:
filter_text = "favorite teams" if self.show_favorite_teams_only or self.show_all_live else "criteria"
self.logger.info(f"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("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("Could not fetch update; keeping existing live game data for now.") # Changed log prefix
else:
self.logger.warning("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"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

1512
src/mlb_manager.py
View File

File diff suppressed because it is too large Load Diff

1135
src/ncaa_baseball_managers.py
View File

File diff suppressed because it is too large Load Diff

146
src/ncaam_hockey_managers.py
View File

@@ -47,69 +47,116 @@ class BaseNCAAMHockeyManager(Hockey): # Renamed class
self.logger.info(f"Initialized NCAAMHockey 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 _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_ncaa_fb_api_data(self, use_cache: bool = True) -> Optional[Dict]:
def _fetch_ncaa_hockey_api_data(self, use_cache: bool = True) -> Optional[Dict]:
"""
Fetches the full season schedule for NCAAMH, caches it, and then filters
for relevant games based on the current configuration.
"""
now = datetime.now(pytz.utc)
current_year = now.year
years_to_check = [current_year]
season_year = now.year
if now.month < 8:
years_to_check.append(current_year - 1)
season_year = now.year - 1
datestring = f"{season_year}0901-{season_year+1}0501"
cache_key = f"ncaa_mens_hockey_schedule_{season_year}"
all_events = []
for year in years_to_check:
cache_key = f"ncaamh_schedule_{year}"
if use_cache:
cached_data = self.cache_manager.get(cache_key)
if cached_data:
self.logger.info(f"[NCAAMH] Using cached schedule for {year}")
all_events.extend(cached_data)
continue
self.logger.info(f"[NCAAMH] Fetching full {year} season schedule from ESPN API...")
try:
response = self.session.get(ESPN_NCAAMH_SCOREBOARD_URL, params={"dates": year,"limit":1000},headers=self.headers, timeout=15)
response.raise_for_status()
data = response.json()
events = data.get('events', [])
if use_cache:
self.cache_manager.set(cache_key, events)
self.logger.info(f"[NCAAMH] Successfully fetched and cached {len(events)} events for {year} season.")
all_events.extend(events)
except requests.exceptions.RequestException as e:
self.logger.error(f"[NCAAMH] API error fetching full schedule for {year}: {e}")
continue
if use_cache:
cached_data = self.cache_manager.get(cache_key)
if cached_data:
# Validate cached data structure
if isinstance(cached_data, dict) and 'events' in cached_data:
self.logger.info(f"Using cached schedule for {season_year}")
return cached_data
elif isinstance(cached_data, list):
# Handle old cache format (list of events)
self.logger.info(f"Using cached schedule for {season_year} (legacy format)")
return {'events': cached_data}
else:
self.logger.warning(f"Invalid cached data format for {season_year}: {type(cached_data)}")
# Clear invalid cache
self.cache_manager.clear_cache(cache_key)
if not all_events:
self.logger.warning("[NCAAMH] No events found in schedule data.")
# If background service is disabled, fall back to synchronous fetch
if not self.background_enabled or not self.background_service:
return self._fetch_ncaa_api_data_sync(use_cache)
self.logger.info(f"Fetching full {season_year} season schedule from ESPN API...")
# Start background fetch
self.logger.info(f"Starting background fetch for {season_year} season schedule...")
def fetch_callback(result):
"""Callback when background fetch completes."""
if result.success:
self.logger.info(f"Background fetch completed for {season_year}: {len(result.data.get('events'))} events")
else:
self.logger.error(f"Background fetch failed for {season_year}: {result.error}")
# Clean up request tracking
if season_year in self.background_fetch_requests:
del self.background_fetch_requests[season_year]
# Get background service configuration
background_config = self.mode_config.get("background_service", {})
timeout = background_config.get("request_timeout", 30)
max_retries = background_config.get("max_retries", 3)
priority = background_config.get("priority", 2)
# Submit background fetch request
request_id = self.background_service.submit_fetch_request(
sport="ncaa_mens_hockey",
year=season_year,
url=ESPN_NCAAMH_SCOREBOARD_URL,
cache_key=cache_key,
params={"dates": datestring, "limit": 1000},
headers=self.headers,
timeout=timeout,
max_retries=max_retries,
priority=priority,
callback=fetch_callback
)
# Track the request
self.background_fetch_requests[season_year] = request_id
# For immediate response, try to get partial data
partial_data = self._get_weeks_data()
if partial_data:
return partial_data
return None
def _fetch_ncaa_api_data_sync(self, use_cache: bool = True) -> Optional[Dict]:
"""
Synchronous fallback for fetching NCAA Mens Hockey data when background service is disabled.
"""
now = datetime.now(pytz.utc)
current_year = now.year
cache_key = f"ncaa_mens_hockey_schedule_{current_year}"
self.logger.info(f"Fetching full {current_year} season schedule from ESPN API (sync mode)...")
try:
response = self.session.get(ESPN_NCAAMH_SCOREBOARD_URL, params={"dates": current_year, "limit":1000}, headers=self.headers, timeout=15)
response.raise_for_status()
data = response.json()
events = data.get('events', [])
if use_cache:
self.cache_manager.set(cache_key, events)
self.logger.info(f"Successfully fetched {len(events)} events for the {current_year} season.")
return {'events': events}
except requests.exceptions.RequestException as e:
self.logger.error(f"[API error fetching full schedule: {e}")
return None
return {'events': all_events}
def _fetch_data(self) -> Optional[Dict]:
"""Fetch data using shared data mechanism or direct fetch for live."""
if isinstance(self, NCAAMHockeyLiveManager):
return self._fetch_ncaa_fb_api_data(use_cache=False)
return self._fetch_todays_games()
else:
return self._fetch_ncaa_fb_api_data(use_cache=True)
return self._fetch_ncaa_hockey_api_data(use_cache=True)
class NCAAMHockeyLiveManager(BaseNCAAMHockeyManager, HockeyLive): # Renamed class
"""Manager for live NCAA Mens Hockey games."""
@@ -133,7 +180,8 @@ class NCAAMHockeyLiveManager(BaseNCAAMHockeyManager, HockeyLive): # Renamed clas
"home_logo_path": Path(self.logo_dir, "RIT.png"),
"away_logo_path": Path(self.logo_dir, "CLAR .png"),
"game_time": "7:30 PM",
"game_date": "Apr 17"
"game_date": "Apr 17",
"is_live": True, "is_final": False, "is_upcoming": False,
}
self.live_games = [self.current_game]
self.logger.info("Initialized NCAAMHockeyLiveManager with test game: RIT vs CLAR ")

32
src/nfl_managers.py
View File

@@ -1,14 +1,16 @@
import os
import logging
import requests
from typing import Dict, Any, Optional, List
from pathlib import Path
import os
from datetime import datetime, timedelta
from src.display_manager import DisplayManager
from src.cache_manager import CacheManager
from pathlib import Path
from typing import Any, Dict, List, Optional
import pytz
from src.base_classes.sports import SportsRecent, SportsUpcoming
import requests
from src.base_classes.football import Football, FootballLive
from src.base_classes.sports import SportsRecent, SportsUpcoming
from src.cache_manager import CacheManager
from src.display_manager import DisplayManager
# Constants
ESPN_NFL_SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/football/nfl/scoreboard"
@@ -50,7 +52,7 @@ class BaseNFLManager(Football): # Renamed class
if now.month < 8:
season_year = now.year - 1
datestring = f"{season_year}0801-{season_year+1}0301"
cache_key = f"nfl_schedule_{season_year}"
cache_key = f"{self.sport_key}_schedule_{season_year}"
# Check cache first
if use_cache:
@@ -58,14 +60,14 @@ class BaseNFLManager(Football): # Renamed class
if cached_data:
# Validate cached data structure
if isinstance(cached_data, dict) and 'events' in cached_data:
self.logger.info(f"[NFL] Using cached schedule for {season_year}")
self.logger.info(f"Using cached schedule for {season_year}")
return cached_data
elif isinstance(cached_data, list):
# Handle old cache format (list of events)
self.logger.info(f"[NFL] Using cached schedule for {season_year} (legacy format)")
self.logger.info(f"Using cached schedule for {season_year} (legacy format)")
return {'events': cached_data}
else:
self.logger.warning(f"[NFL] Invalid cached data format for {season_year}: {type(cached_data)}")
self.logger.warning(f"Invalid cached data format for {season_year}: {type(cached_data)}")
# Clear invalid cache
self.cache_manager.clear_cache(cache_key)
@@ -74,7 +76,7 @@ class BaseNFLManager(Football): # Renamed class
return self._fetch_nfl_api_data_sync(use_cache)
# Start background fetch
self.logger.info(f"[NFL] Starting background fetch for {season_year} season schedule...")
self.logger.info(f"Starting background fetch for {season_year} season schedule...")
def fetch_callback(result):
"""Callback when background fetch completes."""
@@ -125,7 +127,7 @@ class BaseNFLManager(Football): # Renamed class
current_year = now.year
cache_key = f"nfl_schedule_{current_year}"
self.logger.info(f"[NFL] Fetching full {current_year} season schedule from ESPN API (sync mode)...")
self.logger.info(f"Fetching full {current_year} season schedule from ESPN API (sync mode)...")
try:
response = self.session.get(ESPN_NFL_SCOREBOARD_URL, params={"dates": current_year, "limit":1000}, headers=self.headers, timeout=15)
response.raise_for_status()
@@ -135,10 +137,10 @@ class BaseNFLManager(Football): # Renamed class
if use_cache:
self.cache_manager.set(cache_key, events)
self.logger.info(f"[NFL] Successfully fetched {len(events)} events for the {current_year} season.")
self.logger.info(f"Successfully fetched {len(events)} events for the {current_year} season.")
return {'events': events}
except requests.exceptions.RequestException as e:
self.logger.error(f"[NFL] API error fetching full schedule: {e}")
self.logger.error(f"API error fetching full schedule: {e}")
return None
def _fetch_data(self) -> Optional[Dict]:

896
src/nhl_managers.py
View File

File diff suppressed because it is too large Load Diff