mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-11 05:13:01 +00:00
* Fix leaderboard scrolling performance after PR #39 merge - Restore leaderboard background updates that were accidentally removed - Fix duration method call from get_dynamic_duration() back to get_duration() - Restore proper fallback duration (600s instead of 60s) for leaderboard - Add back sports manager updates that feed data to leaderboard - Fix leaderboard defer_update priority to prevent scrolling lag These changes restore the leaderboard's dynamic duration calculation and ensure it gets proper background updates for smooth scrolling. * Apply PR #60 leaderboard performance optimizations - Change scroll_delay from 0.05s to 0.01s (100fps instead of 20fps) - Remove conditional scrolling logic - scroll every frame for smooth animation - Add FPS tracking and logging for performance monitoring - Restore high-framerate scrolling that was working before PR #39 merge These changes restore the smooth leaderboard scrolling performance that was achieved in PR #60 but was lost during the PR #39 merge. * Fix critical bugs identified in PR #39 review - Fix record filtering logic bug: change away_record == set to away_record in set - Fix incorrect sport specification: change 'nfl' to 'ncaa_fb' for NCAA Football data requests - These bugs were causing incorrect data display and wrong sport data fetching Addresses issues found by cursor bot in PR #39 review: - Record filtering was always evaluating to False - NCAA Football was fetching NFL data instead of college football data * Enhance cache clearing implementation from PR #39 - Add detailed logging to cache clearing process for better visibility - Log cache clearing statistics (memory entries and file count) - Improve startup logging to show cache clearing and data refetch process - Addresses legoguy1000's comment about preventing stale data issues This enhances the cache clearing implementation that was added in PR #39 to help prevent legacy cache issues and stale data problems. * continuing on base_classes - added baseball and api extractor since we don't use ESPN api for all sports * tests * fix missing duration * ensure milb, mlb, ncaa bb are all using new baseball base class properly * cursor rule to help with PR creation * fix image call * fix _scoreboard suffix on milb, MLB
This commit is contained in:
Chuck
363
src/base_classes/api_extractors.py
Normal file
363
src/base_classes/api_extractors.py
Normal file
@@ -0,0 +1,363 @@
|
||||
"""
|
||||
Abstract API Data Extraction Layer
|
||||
|
||||
This module provides a pluggable system for extracting game data from different
|
||||
sports APIs. Each sport can have its own extractor that handles sport-specific
|
||||
fields and data structures.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Dict, Any, Optional, List
|
||||
import logging
|
||||
from datetime import datetime
|
||||
import pytz
|
||||
|
||||
class APIDataExtractor(ABC):
|
||||
"""Abstract base class for API data extraction."""
|
||||
|
||||
def __init__(self, logger: logging.Logger):
|
||||
self.logger = logger
|
||||
|
||||
@abstractmethod
|
||||
def extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
||||
"""Extract common game details from raw API data."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_sport_specific_fields(self, game_event: Dict) -> Dict:
|
||||
"""Extract sport-specific fields (downs, innings, periods, etc.)."""
|
||||
pass
|
||||
|
||||
def _extract_common_details(self, game_event: Dict) -> tuple[Dict | None, Dict | None, Dict | None, Dict | None, Dict | None]:
|
||||
"""Extract common game details that work across all sports."""
|
||||
if not game_event:
|
||||
return None, None, None, None, None
|
||||
|
||||
try:
|
||||
competition = game_event["competitions"][0]
|
||||
status = competition["status"]
|
||||
competitors = competition["competitors"]
|
||||
game_date_str = game_event["date"]
|
||||
situation = competition.get("situation")
|
||||
|
||||
# Parse game time
|
||||
start_time_utc = None
|
||||
try:
|
||||
start_time_utc = datetime.fromisoformat(game_date_str.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
self.logger.warning(f"Could not parse game date: {game_date_str}")
|
||||
|
||||
# Extract teams
|
||||
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"Could not find home or away team in event: {game_event.get('id')}")
|
||||
return None, None, None, None, None
|
||||
|
||||
return {
|
||||
"game_event": game_event,
|
||||
"competition": competition,
|
||||
"status": status,
|
||||
"situation": situation,
|
||||
"start_time_utc": start_time_utc,
|
||||
"home_team": home_team,
|
||||
"away_team": away_team
|
||||
}, home_team, away_team, status, situation
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting common details: {e}")
|
||||
return None, None, None, None, None
|
||||
|
||||
|
||||
class ESPNFootballExtractor(APIDataExtractor):
|
||||
"""ESPN API extractor for football (NFL/NCAA)."""
|
||||
|
||||
def extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
||||
"""Extract football game details from ESPN API."""
|
||||
common_data, home_team, away_team, status, situation = self._extract_common_details(game_event)
|
||||
if not common_data:
|
||||
return None
|
||||
|
||||
try:
|
||||
# Extract basic team info
|
||||
home_abbr = home_team["team"]["abbreviation"]
|
||||
away_abbr = away_team["team"]["abbreviation"]
|
||||
home_score = home_team.get("score", "0")
|
||||
away_score = away_team.get("score", "0")
|
||||
|
||||
# Extract sport-specific fields
|
||||
sport_fields = self.get_sport_specific_fields(game_event)
|
||||
|
||||
# Build game details
|
||||
details = {
|
||||
"id": game_event.get("id"),
|
||||
"home_abbr": home_abbr,
|
||||
"away_abbr": away_abbr,
|
||||
"home_score": str(home_score),
|
||||
"away_score": str(away_score),
|
||||
"home_team_name": home_team["team"].get("displayName", ""),
|
||||
"away_team_name": away_team["team"].get("displayName", ""),
|
||||
"status_text": status["type"].get("shortDetail", ""),
|
||||
"is_live": status["type"]["state"] == "in",
|
||||
"is_final": status["type"]["state"] == "post",
|
||||
"is_upcoming": status["type"]["state"] == "pre",
|
||||
**sport_fields # Add sport-specific fields
|
||||
}
|
||||
|
||||
return details
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting football game details: {e}")
|
||||
return None
|
||||
|
||||
def get_sport_specific_fields(self, game_event: Dict) -> Dict:
|
||||
"""Extract football-specific fields."""
|
||||
try:
|
||||
competition = game_event["competitions"][0]
|
||||
status = competition["status"]
|
||||
situation = competition.get("situation", {})
|
||||
|
||||
sport_fields = {
|
||||
"down": "",
|
||||
"distance": "",
|
||||
"possession": "",
|
||||
"is_redzone": False,
|
||||
"home_timeouts": 0,
|
||||
"away_timeouts": 0,
|
||||
"scoring_event": ""
|
||||
}
|
||||
|
||||
if situation and status["type"]["state"] == "in":
|
||||
sport_fields.update({
|
||||
"down": situation.get("down", ""),
|
||||
"distance": situation.get("distance", ""),
|
||||
"possession": situation.get("possession", ""),
|
||||
"is_redzone": situation.get("isRedZone", False),
|
||||
"home_timeouts": situation.get("homeTimeouts", 0),
|
||||
"away_timeouts": situation.get("awayTimeouts", 0)
|
||||
})
|
||||
|
||||
# Detect scoring events
|
||||
status_detail = status["type"].get("detail", "").lower()
|
||||
if "touchdown" in status_detail or "field goal" in status_detail:
|
||||
sport_fields["scoring_event"] = status_detail
|
||||
|
||||
return sport_fields
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting football-specific fields: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
class ESPNBaseballExtractor(APIDataExtractor):
|
||||
"""ESPN API extractor for baseball (MLB)."""
|
||||
|
||||
def extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
||||
"""Extract baseball game details from ESPN API."""
|
||||
common_data, home_team, away_team, status, situation = self._extract_common_details(game_event)
|
||||
if not common_data:
|
||||
return None
|
||||
|
||||
try:
|
||||
# Extract basic team info
|
||||
home_abbr = home_team["team"]["abbreviation"]
|
||||
away_abbr = away_team["team"]["abbreviation"]
|
||||
home_score = home_team.get("score", "0")
|
||||
away_score = away_team.get("score", "0")
|
||||
|
||||
# Extract sport-specific fields
|
||||
sport_fields = self.get_sport_specific_fields(game_event)
|
||||
|
||||
# Build game details
|
||||
details = {
|
||||
"id": game_event.get("id"),
|
||||
"home_abbr": home_abbr,
|
||||
"away_abbr": away_abbr,
|
||||
"home_score": str(home_score),
|
||||
"away_score": str(away_score),
|
||||
"home_team_name": home_team["team"].get("displayName", ""),
|
||||
"away_team_name": away_team["team"].get("displayName", ""),
|
||||
"status_text": status["type"].get("shortDetail", ""),
|
||||
"is_live": status["type"]["state"] == "in",
|
||||
"is_final": status["type"]["state"] == "post",
|
||||
"is_upcoming": status["type"]["state"] == "pre",
|
||||
**sport_fields # Add sport-specific fields
|
||||
}
|
||||
|
||||
return details
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting baseball game details: {e}")
|
||||
return None
|
||||
|
||||
def get_sport_specific_fields(self, game_event: Dict) -> Dict:
|
||||
"""Extract baseball-specific fields."""
|
||||
try:
|
||||
competition = game_event["competitions"][0]
|
||||
status = competition["status"]
|
||||
situation = competition.get("situation", {})
|
||||
|
||||
sport_fields = {
|
||||
"inning": "",
|
||||
"outs": 0,
|
||||
"bases": "",
|
||||
"strikes": 0,
|
||||
"balls": 0,
|
||||
"pitcher": "",
|
||||
"batter": ""
|
||||
}
|
||||
|
||||
if situation and status["type"]["state"] == "in":
|
||||
sport_fields.update({
|
||||
"inning": situation.get("inning", ""),
|
||||
"outs": situation.get("outs", 0),
|
||||
"bases": situation.get("bases", ""),
|
||||
"strikes": situation.get("strikes", 0),
|
||||
"balls": situation.get("balls", 0),
|
||||
"pitcher": situation.get("pitcher", ""),
|
||||
"batter": situation.get("batter", "")
|
||||
})
|
||||
|
||||
return sport_fields
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting baseball-specific fields: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
class ESPNHockeyExtractor(APIDataExtractor):
|
||||
"""ESPN API extractor for hockey (NHL/NCAA)."""
|
||||
|
||||
def extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
||||
"""Extract hockey game details from ESPN API."""
|
||||
common_data, home_team, away_team, status, situation = self._extract_common_details(game_event)
|
||||
if not common_data:
|
||||
return None
|
||||
|
||||
try:
|
||||
# Extract basic team info
|
||||
home_abbr = home_team["team"]["abbreviation"]
|
||||
away_abbr = away_team["team"]["abbreviation"]
|
||||
home_score = home_team.get("score", "0")
|
||||
away_score = away_team.get("score", "0")
|
||||
|
||||
# Extract sport-specific fields
|
||||
sport_fields = self.get_sport_specific_fields(game_event)
|
||||
|
||||
# Build game details
|
||||
details = {
|
||||
"id": game_event.get("id"),
|
||||
"home_abbr": home_abbr,
|
||||
"away_abbr": away_abbr,
|
||||
"home_score": str(home_score),
|
||||
"away_score": str(away_score),
|
||||
"home_team_name": home_team["team"].get("displayName", ""),
|
||||
"away_team_name": away_team["team"].get("displayName", ""),
|
||||
"status_text": status["type"].get("shortDetail", ""),
|
||||
"is_live": status["type"]["state"] == "in",
|
||||
"is_final": status["type"]["state"] == "post",
|
||||
"is_upcoming": status["type"]["state"] == "pre",
|
||||
**sport_fields # Add sport-specific fields
|
||||
}
|
||||
|
||||
return details
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting hockey game details: {e}")
|
||||
return None
|
||||
|
||||
def get_sport_specific_fields(self, game_event: Dict) -> Dict:
|
||||
"""Extract hockey-specific fields."""
|
||||
try:
|
||||
competition = game_event["competitions"][0]
|
||||
status = competition["status"]
|
||||
situation = competition.get("situation", {})
|
||||
|
||||
sport_fields = {
|
||||
"period": "",
|
||||
"period_text": "",
|
||||
"power_play": False,
|
||||
"penalties": "",
|
||||
"shots_on_goal": {"home": 0, "away": 0}
|
||||
}
|
||||
|
||||
if situation and status["type"]["state"] == "in":
|
||||
period = status.get("period", 0)
|
||||
period_text = ""
|
||||
if period == 1:
|
||||
period_text = "P1"
|
||||
elif period == 2:
|
||||
period_text = "P2"
|
||||
elif period == 3:
|
||||
period_text = "P3"
|
||||
elif period > 3:
|
||||
period_text = f"OT{period-3}"
|
||||
|
||||
sport_fields.update({
|
||||
"period": str(period),
|
||||
"period_text": period_text,
|
||||
"power_play": situation.get("isPowerPlay", False),
|
||||
"penalties": situation.get("penalties", ""),
|
||||
"shots_on_goal": {
|
||||
"home": situation.get("homeShots", 0),
|
||||
"away": situation.get("awayShots", 0)
|
||||
}
|
||||
})
|
||||
|
||||
return sport_fields
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting hockey-specific fields: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
class SoccerAPIExtractor(APIDataExtractor):
|
||||
"""Generic extractor for soccer APIs (different structure than ESPN)."""
|
||||
|
||||
def extract_game_details(self, game_event: Dict) -> Optional[Dict]:
|
||||
"""Extract soccer game details from various soccer APIs."""
|
||||
# This would need to be adapted based on the specific soccer API being used
|
||||
# For now, return a basic structure
|
||||
try:
|
||||
return {
|
||||
"id": game_event.get("id"),
|
||||
"home_abbr": game_event.get("home_team", {}).get("abbreviation", ""),
|
||||
"away_abbr": game_event.get("away_team", {}).get("abbreviation", ""),
|
||||
"home_score": str(game_event.get("home_score", "0")),
|
||||
"away_score": str(game_event.get("away_score", "0")),
|
||||
"home_team_name": game_event.get("home_team", {}).get("name", ""),
|
||||
"away_team_name": game_event.get("away_team", {}).get("name", ""),
|
||||
"status_text": game_event.get("status", ""),
|
||||
"is_live": game_event.get("is_live", False),
|
||||
"is_final": game_event.get("is_final", False),
|
||||
"is_upcoming": game_event.get("is_upcoming", False),
|
||||
**self.get_sport_specific_fields(game_event)
|
||||
}
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting soccer game details: {e}")
|
||||
return None
|
||||
|
||||
def get_sport_specific_fields(self, game_event: Dict) -> Dict:
|
||||
"""Extract soccer-specific fields."""
|
||||
try:
|
||||
return {
|
||||
"half": game_event.get("half", ""),
|
||||
"stoppage_time": game_event.get("stoppage_time", ""),
|
||||
"cards": {
|
||||
"home_yellow": game_event.get("home_yellow_cards", 0),
|
||||
"away_yellow": game_event.get("away_yellow_cards", 0),
|
||||
"home_red": game_event.get("home_red_cards", 0),
|
||||
"away_red": game_event.get("away_red_cards", 0)
|
||||
},
|
||||
"possession": {
|
||||
"home": game_event.get("home_possession", 0),
|
||||
"away": game_event.get("away_possession", 0)
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error extracting soccer-specific fields: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
# Factory function removed - sport classes now instantiate extractors directly
|
||||
165
src/base_classes/baseball.py
Normal file
165
src/base_classes/baseball.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
Baseball Base Classes
|
||||
|
||||
This module provides baseball-specific base classes that extend the core sports functionality
|
||||
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
|
||||
|
||||
class Baseball(SportsCore):
|
||||
"""Base class for baseball sports with common functionality."""
|
||||
|
||||
# Baseball sport configuration (moved from sport_configs.py)
|
||||
SPORT_CONFIG = {
|
||||
'update_cadence': 'daily',
|
||||
'season_length': 162,
|
||||
'games_per_week': 6,
|
||||
'api_endpoints': ['scoreboard', 'standings', 'stats'],
|
||||
'sport_specific_fields': ['inning', 'outs', 'bases', 'strikes', 'balls', 'pitcher', 'batter'],
|
||||
'update_interval_seconds': 30,
|
||||
'logo_dir': 'assets/sports/mlb_logos',
|
||||
'show_records': True,
|
||||
'show_ranking': True,
|
||||
'show_odds': True,
|
||||
'data_source_type': 'espn', # Can be overridden for MLB API
|
||||
'api_base_url': 'https://site.api.espn.com/apis/site/v2/sports/baseball'
|
||||
}
|
||||
|
||||
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.sport_config = self.get_sport_config()
|
||||
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)
|
||||
|
||||
def get_sport_config(self) -> Dict[str, Any]:
|
||||
"""Get baseball sport configuration."""
|
||||
return self.SPORT_CONFIG.copy()
|
||||
|
||||
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', '')
|
||||
if inning:
|
||||
display_parts.append(f"Inning: {inning}")
|
||||
|
||||
# Outs information
|
||||
if self.show_outs:
|
||||
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', '')
|
||||
if bases:
|
||||
display_parts.append(f"Bases: {bases}")
|
||||
|
||||
# Count information
|
||||
if self.show_count:
|
||||
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', '')
|
||||
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)
|
||||
if is_live:
|
||||
return True
|
||||
|
||||
# Check inning to determine if game is active
|
||||
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', '')
|
||||
|
||||
if self._is_baseball_game_live(game):
|
||||
if inning:
|
||||
return f"Live - {inning}"
|
||||
else:
|
||||
return "Live"
|
||||
elif game.get('is_final', False):
|
||||
return "Final"
|
||||
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."""
|
||||
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)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error checking if baseball game should be shown: {e}")
|
||||
return False
|
||||
|
||||
|
||||
288
src/base_classes/data_sources.py
Normal file
288
src/base_classes/data_sources.py
Normal file
@@ -0,0 +1,288 @@
|
||||
"""
|
||||
Pluggable Data Source Architecture
|
||||
|
||||
This module provides abstract data sources that can be plugged into the sports system
|
||||
to support different APIs and data providers.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Dict, Any, Optional, List
|
||||
import requests
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
import time
|
||||
|
||||
class DataSource(ABC):
|
||||
"""Abstract base class for data sources."""
|
||||
|
||||
def __init__(self, logger: logging.Logger):
|
||||
self.logger = logger
|
||||
self.session = requests.Session()
|
||||
|
||||
# Configure retry strategy
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util.retry import Retry
|
||||
|
||||
retry_strategy = Retry(
|
||||
total=5,
|
||||
backoff_factor=1,
|
||||
status_forcelist=[429, 500, 502, 503, 504],
|
||||
)
|
||||
adapter = HTTPAdapter(max_retries=retry_strategy)
|
||||
self.session.mount("http://", adapter)
|
||||
self.session.mount("https://", adapter)
|
||||
|
||||
@abstractmethod
|
||||
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
||||
"""Fetch live games for a sport/league."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
||||
"""Fetch schedule for a sport/league within date range."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def fetch_standings(self, sport: str, league: str) -> Dict:
|
||||
"""Fetch standings for a sport/league."""
|
||||
pass
|
||||
|
||||
def get_headers(self) -> Dict[str, str]:
|
||||
"""Get headers for API requests."""
|
||||
return {
|
||||
'User-Agent': 'LEDMatrix/1.0',
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
|
||||
class ESPNDataSource(DataSource):
|
||||
"""ESPN API data source."""
|
||||
|
||||
def __init__(self, logger: logging.Logger):
|
||||
super().__init__(logger)
|
||||
self.base_url = "https://site.api.espn.com/apis/site/v2/sports"
|
||||
|
||||
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
||||
"""Fetch live games from ESPN API."""
|
||||
try:
|
||||
url = f"{self.base_url}/{sport}/{league}/scoreboard"
|
||||
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
events = data.get('events', [])
|
||||
|
||||
# Filter for live games
|
||||
live_events = [event for event in events
|
||||
if event.get('competitions', [{}])[0].get('status', {}).get('type', {}).get('state') == 'in']
|
||||
|
||||
self.logger.debug(f"Fetched {len(live_events)} live games for {sport}/{league}")
|
||||
return live_events
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching live games from ESPN: {e}")
|
||||
return []
|
||||
|
||||
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
||||
"""Fetch schedule from ESPN API."""
|
||||
try:
|
||||
start_date, end_date = date_range
|
||||
url = f"{self.base_url}/{sport}/{league}/scoreboard"
|
||||
|
||||
params = {
|
||||
'dates': f"{start_date.strftime('%Y%m%d')}-{end_date.strftime('%Y%m%d')}"
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
events = data.get('events', [])
|
||||
|
||||
self.logger.debug(f"Fetched {len(events)} scheduled games for {sport}/{league}")
|
||||
return events
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching schedule from ESPN: {e}")
|
||||
return []
|
||||
|
||||
def fetch_standings(self, sport: str, league: str) -> Dict:
|
||||
"""Fetch standings from ESPN API."""
|
||||
try:
|
||||
url = f"{self.base_url}/{sport}/{league}/standings"
|
||||
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
self.logger.debug(f"Fetched standings for {sport}/{league}")
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching standings from ESPN: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
class MLBAPIDataSource(DataSource):
|
||||
"""MLB API data source."""
|
||||
|
||||
def __init__(self, logger: logging.Logger):
|
||||
super().__init__(logger)
|
||||
self.base_url = "https://statsapi.mlb.com/api/v1"
|
||||
|
||||
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
||||
"""Fetch live games from MLB API."""
|
||||
try:
|
||||
url = f"{self.base_url}/schedule"
|
||||
params = {
|
||||
'sportId': 1, # MLB
|
||||
'date': datetime.now().strftime('%Y-%m-%d'),
|
||||
'hydrate': 'game,team,venue,weather'
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
games = data.get('dates', [{}])[0].get('games', [])
|
||||
|
||||
# Filter for live games
|
||||
live_games = [game for game in games
|
||||
if game.get('status', {}).get('abstractGameState') == 'Live']
|
||||
|
||||
self.logger.debug(f"Fetched {len(live_games)} live games from MLB API")
|
||||
return live_games
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching live games from MLB API: {e}")
|
||||
return []
|
||||
|
||||
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
||||
"""Fetch schedule from MLB API."""
|
||||
try:
|
||||
start_date, end_date = date_range
|
||||
url = f"{self.base_url}/schedule"
|
||||
|
||||
params = {
|
||||
'sportId': 1, # MLB
|
||||
'startDate': start_date.strftime('%Y-%m-%d'),
|
||||
'endDate': end_date.strftime('%Y-%m-%d'),
|
||||
'hydrate': 'game,team,venue'
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
all_games = []
|
||||
for date_data in data.get('dates', []):
|
||||
all_games.extend(date_data.get('games', []))
|
||||
|
||||
self.logger.debug(f"Fetched {len(all_games)} scheduled games from MLB API")
|
||||
return all_games
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching schedule from MLB API: {e}")
|
||||
return []
|
||||
|
||||
def fetch_standings(self, sport: str, league: str) -> Dict:
|
||||
"""Fetch standings from MLB API."""
|
||||
try:
|
||||
url = f"{self.base_url}/standings"
|
||||
params = {
|
||||
'leagueId': 103, # American League
|
||||
'season': datetime.now().year,
|
||||
'standingsType': 'regularSeason'
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
self.logger.debug(f"Fetched standings from MLB API")
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching standings from MLB API: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
class SoccerAPIDataSource(DataSource):
|
||||
"""Soccer API data source (generic structure)."""
|
||||
|
||||
def __init__(self, logger: logging.Logger, api_key: str = None):
|
||||
super().__init__(logger)
|
||||
self.api_key = api_key
|
||||
self.base_url = "https://api.football-data.org/v4" # Example API
|
||||
|
||||
def get_headers(self) -> Dict[str, str]:
|
||||
"""Get headers with API key for soccer API."""
|
||||
headers = super().get_headers()
|
||||
if self.api_key:
|
||||
headers['X-Auth-Token'] = self.api_key
|
||||
return headers
|
||||
|
||||
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
||||
"""Fetch live games from soccer API."""
|
||||
try:
|
||||
# This would need to be adapted based on the specific soccer API
|
||||
url = f"{self.base_url}/matches"
|
||||
params = {
|
||||
'status': 'LIVE',
|
||||
'competition': league
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
matches = data.get('matches', [])
|
||||
|
||||
self.logger.debug(f"Fetched {len(matches)} live games from soccer API")
|
||||
return matches
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching live games from soccer API: {e}")
|
||||
return []
|
||||
|
||||
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
||||
"""Fetch schedule from soccer API."""
|
||||
try:
|
||||
start_date, end_date = date_range
|
||||
url = f"{self.base_url}/matches"
|
||||
|
||||
params = {
|
||||
'competition': league,
|
||||
'dateFrom': start_date.strftime('%Y-%m-%d'),
|
||||
'dateTo': end_date.strftime('%Y-%m-%d')
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
matches = data.get('matches', [])
|
||||
|
||||
self.logger.debug(f"Fetched {len(matches)} scheduled games from soccer API")
|
||||
return matches
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching schedule from soccer API: {e}")
|
||||
return []
|
||||
|
||||
def fetch_standings(self, sport: str, league: str) -> Dict:
|
||||
"""Fetch standings from soccer API."""
|
||||
try:
|
||||
url = f"{self.base_url}/competitions/{league}/standings"
|
||||
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
self.logger.debug(f"Fetched standings from soccer API")
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching standings from soccer API: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
# Factory function removed - sport classes now instantiate data sources directly
|
||||
@@ -7,11 +7,40 @@ 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
|
||||
|
||||
class Football(SportsCore):
|
||||
"""Base class for football sports with common functionality."""
|
||||
|
||||
# Football sport configuration (moved from sport_configs.py)
|
||||
SPORT_CONFIG = {
|
||||
'update_cadence': 'weekly',
|
||||
'season_length': 17, # NFL default
|
||||
'games_per_week': 1,
|
||||
'api_endpoints': ['scoreboard', 'standings'],
|
||||
'sport_specific_fields': ['down', 'distance', 'possession', 'timeouts', 'is_redzone'],
|
||||
'update_interval_seconds': 60,
|
||||
'logo_dir': 'assets/sports/nfl_logos',
|
||||
'show_records': True,
|
||||
'show_ranking': True,
|
||||
'show_odds': True,
|
||||
'data_source_type': 'espn',
|
||||
'api_base_url': 'https://site.api.espn.com/apis/site/v2/sports/football'
|
||||
}
|
||||
|
||||
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.sport_config = self.get_sport_config()
|
||||
self.api_extractor = ESPNFootballExtractor(logger)
|
||||
self.data_source = ESPNDataSource(logger)
|
||||
|
||||
def get_sport_config(self) -> Dict[str, Any]:
|
||||
"""Get football sport configuration."""
|
||||
return self.SPORT_CONFIG.copy()
|
||||
|
||||
def _fetch_game_odds(self, _: Dict) -> None:
|
||||
pass
|
||||
|
||||
@@ -6,10 +6,39 @@ 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 src.base_classes.data_sources import ESPNDataSource
|
||||
|
||||
class Hockey(SportsCore):
|
||||
"""Base class for hockey sports with common functionality."""
|
||||
|
||||
# Hockey sport configuration (moved from sport_configs.py)
|
||||
SPORT_CONFIG = {
|
||||
'update_cadence': 'daily',
|
||||
'season_length': 82, # NHL default
|
||||
'games_per_week': 3,
|
||||
'api_endpoints': ['scoreboard', 'standings'],
|
||||
'sport_specific_fields': ['period', 'power_play', 'penalties', 'shots_on_goal'],
|
||||
'update_interval_seconds': 30,
|
||||
'logo_dir': 'assets/sports/nhl_logos',
|
||||
'show_records': True,
|
||||
'show_ranking': True,
|
||||
'show_odds': True,
|
||||
'data_source_type': 'espn',
|
||||
'api_base_url': 'https://site.api.espn.com/apis/site/v2/sports/hockey'
|
||||
}
|
||||
|
||||
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.sport_config = self.get_sport_config()
|
||||
self.api_extractor = ESPNHockeyExtractor(logger)
|
||||
self.data_source = ESPNDataSource(logger)
|
||||
|
||||
def get_sport_config(self) -> Dict[str, Any]:
|
||||
"""Get hockey sport configuration."""
|
||||
return self.SPORT_CONFIG.copy()
|
||||
|
||||
def _fetch_odds(self, game: Dict, league: str) -> None:
|
||||
super()._fetch_odds(game, "hockey", league)
|
||||
|
||||
@@ -15,6 +15,10 @@ 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 .api_extractors import ESPNFootballExtractor, ESPNBaseballExtractor, ESPNHockeyExtractor
|
||||
from .data_sources import ESPNDataSource, MLBAPIDataSource
|
||||
|
||||
class SportsCore:
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
|
||||
self.logger = logger
|
||||
@@ -28,6 +32,11 @@ class SportsCore:
|
||||
self.display_height = self.display_manager.matrix.height
|
||||
|
||||
self.sport_key = sport_key
|
||||
|
||||
# Initialize new architecture components (will be overridden by sport-specific classes)
|
||||
self.sport_config = None
|
||||
self.api_extractor = None
|
||||
self.data_source = None
|
||||
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)
|
||||
@@ -267,20 +276,143 @@ class SportsCore:
|
||||
return None
|
||||
|
||||
def _fetch_data(self) -> Optional[Dict]:
|
||||
"""Override this from the sports class"""
|
||||
pass
|
||||
"""Fetch data using the new architecture components."""
|
||||
try:
|
||||
# Use the data source to fetch live games
|
||||
live_games = self.data_source.fetch_live_games(self.sport_key, self.sport_key)
|
||||
|
||||
if not live_games:
|
||||
self.logger.debug(f"No live games found for {self.sport_key}")
|
||||
return None
|
||||
|
||||
# Use the API extractor to process each game
|
||||
processed_games = []
|
||||
for game_event in live_games:
|
||||
game_details = self.api_extractor.extract_game_details(game_event)
|
||||
if game_details:
|
||||
# Add sport-specific fields
|
||||
sport_fields = self.api_extractor.get_sport_specific_fields(game_event)
|
||||
game_details.update(sport_fields)
|
||||
|
||||
# Fetch odds if enabled
|
||||
if self.show_odds:
|
||||
self._fetch_odds(game_details, self.sport_key, self.sport_key)
|
||||
|
||||
processed_games.append(game_details)
|
||||
|
||||
if processed_games:
|
||||
self.logger.debug(f"Successfully processed {len(processed_games)} games for {self.sport_key}")
|
||||
return {
|
||||
'games': processed_games,
|
||||
'sport': self.sport_key,
|
||||
'timestamp': time.time()
|
||||
}
|
||||
else:
|
||||
self.logger.debug(f"No valid games processed for {self.sport_key}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching data for {self.sport_key}: {e}")
|
||||
return None
|
||||
|
||||
def _get_partial_schedule_data(self, year: int) -> List[Dict]:
|
||||
"""Override this from the sports class"""
|
||||
return []
|
||||
"""Get schedule data using the new architecture components."""
|
||||
try:
|
||||
# Calculate date range for the year
|
||||
start_date = datetime(year, 1, 1)
|
||||
end_date = datetime(year, 12, 31)
|
||||
|
||||
# Use the data source to fetch schedule
|
||||
schedule_games = self.data_source.fetch_schedule(
|
||||
self.sport_key,
|
||||
self.sport_key,
|
||||
(start_date, end_date)
|
||||
)
|
||||
|
||||
if not schedule_games:
|
||||
self.logger.debug(f"No schedule data found for {self.sport_key} in {year}")
|
||||
return []
|
||||
|
||||
# Use the API extractor to process each game
|
||||
processed_games = []
|
||||
for game_event in schedule_games:
|
||||
game_details = self.api_extractor.extract_game_details(game_event)
|
||||
if game_details:
|
||||
# Add sport-specific fields
|
||||
sport_fields = self.api_extractor.get_sport_specific_fields(game_event)
|
||||
game_details.update(sport_fields)
|
||||
processed_games.append(game_details)
|
||||
|
||||
self.logger.debug(f"Successfully processed {len(processed_games)} schedule games for {self.sport_key} in {year}")
|
||||
return processed_games
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching schedule data for {self.sport_key} in {year}: {e}")
|
||||
return []
|
||||
|
||||
def _fetch_immediate_games(self) -> List[Dict]:
|
||||
"""Override this from the sports class"""
|
||||
return []
|
||||
"""Fetch immediate games using the new architecture components."""
|
||||
try:
|
||||
# Use the data source to fetch live games
|
||||
live_games = self.data_source.fetch_live_games(self.sport_key, self.sport_key)
|
||||
|
||||
if not live_games:
|
||||
self.logger.debug(f"No immediate games found for {self.sport_key}")
|
||||
return []
|
||||
|
||||
# Use the API extractor to process each game
|
||||
processed_games = []
|
||||
for game_event in live_games:
|
||||
game_details = self.api_extractor.extract_game_details(game_event)
|
||||
if game_details:
|
||||
# Add sport-specific fields
|
||||
sport_fields = self.api_extractor.get_sport_specific_fields(game_event)
|
||||
game_details.update(sport_fields)
|
||||
processed_games.append(game_details)
|
||||
|
||||
self.logger.debug(f"Successfully processed {len(processed_games)} immediate games for {self.sport_key}")
|
||||
return processed_games
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching immediate games for {self.sport_key}: {e}")
|
||||
return []
|
||||
|
||||
def _fetch_game_odds(self, _: Dict) -> None:
|
||||
"""Override this from the sports class"""
|
||||
pass
|
||||
def _fetch_game_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a specific game using the new architecture."""
|
||||
try:
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
# Check if we should only fetch for favorite teams
|
||||
is_favorites_only = self.mode_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
|
||||
|
||||
# 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 \
|
||||
else self.mode_config.get("odds_update_interval", 3600)
|
||||
|
||||
# Fetch odds using OddsManager
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport=self.sport_key,
|
||||
league=self.sport_key,
|
||||
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_odds(self, game: Dict, sport: str, league: str) -> None:
|
||||
"""Fetch odds for a specific game if conditions are met."""
|
||||
@@ -339,7 +471,26 @@ class SportsCore:
|
||||
return False
|
||||
|
||||
def _fetch_team_rankings(self) -> Dict[str, int]:
|
||||
return {}
|
||||
"""Fetch team rankings using the new architecture components."""
|
||||
try:
|
||||
# Use the data source to fetch standings
|
||||
standings_data = self.data_source.fetch_standings(self.sport_key, self.sport_key)
|
||||
|
||||
if not standings_data:
|
||||
self.logger.debug(f"No standings data found for {self.sport_key}")
|
||||
return {}
|
||||
|
||||
# Extract rankings from standings data
|
||||
rankings = {}
|
||||
# This would need to be implemented based on the specific data structure
|
||||
# returned by each data source
|
||||
|
||||
self.logger.debug(f"Successfully fetched rankings for {self.sport_key}")
|
||||
return rankings
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching team rankings for {self.sport_key}: {e}")
|
||||
return {}
|
||||
|
||||
def _extract_game_details_common(self, game_event: Dict) -> tuple[Dict | None, Dict | None, Dict | None, Dict | None, Dict | None]:
|
||||
if not game_event:
|
||||
@@ -393,7 +544,7 @@ class SportsCore:
|
||||
# Don't show "0-0" records - set to blank instead
|
||||
if home_record in {"0-0", "0-0-0"}:
|
||||
home_record = ''
|
||||
if away_record == {"0-0", "0-0-0"}:
|
||||
if away_record in {"0-0", "0-0-0"}:
|
||||
away_record = ''
|
||||
|
||||
details = {
|
||||
|
||||
Reference in New Issue
Block a user