mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-10 21:03:01 +00:00
gambling updates
This commit is contained in:
@@ -122,8 +122,10 @@ class CacheManager:
|
||||
self._memory_cache[key] = data
|
||||
self._memory_cache_timestamps[key] = time.time()
|
||||
|
||||
except Exception:
|
||||
pass # Silently fail if cache save fails
|
||||
except (IOError, OSError) as e:
|
||||
self.logger.error(f"Failed to save cache for key '{key}': {e}")
|
||||
except Exception as e:
|
||||
self.logger.error(f"An unexpected error occurred while saving cache for key '{key}': {e}")
|
||||
|
||||
def load_cache(self, key: str) -> Optional[Dict[str, Any]]:
|
||||
"""Load data from cache with memory caching."""
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
from typing import Dict, Any
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
class ConfigManager:
|
||||
def __init__(self, config_path: str = None, secrets_path: str = None):
|
||||
@@ -93,6 +93,18 @@ class ConfigManager:
|
||||
print(f"An unexpected error occurred while saving configuration: {str(e)}")
|
||||
raise
|
||||
|
||||
def get_secret(self, key: str) -> Optional[Any]:
|
||||
"""Get a secret value by key."""
|
||||
try:
|
||||
if not os.path.exists(self.secrets_path):
|
||||
return None
|
||||
with open(self.secrets_path, 'r') as f:
|
||||
secrets = json.load(f)
|
||||
return secrets.get(key)
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
print(f"Error reading secrets file: {e}")
|
||||
return None
|
||||
|
||||
def _deep_merge(self, target: Dict, source: Dict) -> None:
|
||||
"""Deep merge source dict into target dict."""
|
||||
for key, value in source.items():
|
||||
|
||||
@@ -11,6 +11,7 @@ from .cache_manager import CacheManager
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util.retry import Retry
|
||||
import pytz
|
||||
from src.odds_manager import OddsManager
|
||||
|
||||
# Get logger
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -21,8 +22,10 @@ class BaseMLBManager:
|
||||
self.config = config
|
||||
self.display_manager = display_manager
|
||||
self.mlb_config = config.get('mlb', {})
|
||||
self.show_odds = self.mlb_config.get("show_odds", False)
|
||||
self.favorite_teams = self.mlb_config.get('favorite_teams', [])
|
||||
self.cache_manager = CacheManager()
|
||||
self.odds_manager = OddsManager(self.cache_manager, self.config)
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.logger.setLevel(logging.INFO) # Set logger level to INFO
|
||||
|
||||
@@ -50,6 +53,22 @@ class BaseMLBManager:
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
||||
}
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a game and attach it to the game dictionary."""
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
try:
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="baseball",
|
||||
league="mlb",
|
||||
event_id=game["id"]
|
||||
)
|
||||
if odds_data:
|
||||
game['odds'] = odds_data
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
|
||||
|
||||
def _get_team_logo(self, team_abbr: str) -> Optional[Image.Image]:
|
||||
"""Get team logo from the configured directory."""
|
||||
try:
|
||||
@@ -241,6 +260,40 @@ class BaseMLBManager:
|
||||
# draw.text((score_x, score_y), score_text, font=score_font, fill=(255, 255, 255))
|
||||
self._draw_text_with_outline(draw, score_text, (score_x, score_y), score_font)
|
||||
|
||||
# Draw betting odds if available and enabled
|
||||
if self.show_odds and 'odds' in game_data:
|
||||
odds_details = game_data['odds'].get('details', 'N/A')
|
||||
home_team_odds = game_data['odds'].get('home_team_odds', {})
|
||||
away_team_odds = game_data['odds'].get('away_team_odds', {})
|
||||
|
||||
# Extract spread and format it
|
||||
home_spread = home_team_odds.get('point_spread', 'N/A')
|
||||
away_spread = away_team_odds.get('point_spread', 'N/A')
|
||||
|
||||
# Add a plus sign to positive spreads
|
||||
if isinstance(home_spread, (int, float)) and home_spread > 0:
|
||||
home_spread = f"+{home_spread}"
|
||||
|
||||
if isinstance(away_spread, (int, float)) and away_spread > 0:
|
||||
away_spread = f"+{away_spread}"
|
||||
|
||||
# Define colors for odds text
|
||||
odds_font = self.display_manager.status_font
|
||||
odds_color = (255, 0, 0) # Red text
|
||||
outline_color = (0, 0, 0) # Black outline
|
||||
|
||||
# Draw away team odds
|
||||
if away_spread != 'N/A':
|
||||
away_odds_x = 5 # Adjust as needed
|
||||
away_odds_y = height - 10 # Adjust as needed
|
||||
self._draw_text_with_outline(draw, str(away_spread), (away_odds_x, away_odds_y), odds_font, odds_color, outline_color)
|
||||
|
||||
# Draw home team odds
|
||||
if home_spread != 'N/A':
|
||||
home_odds_x = width - 30 # Adjust as needed
|
||||
home_odds_y = height - 10 # Adjust as needed
|
||||
self._draw_text_with_outline(draw, str(home_spread), (home_odds_x, home_odds_y), odds_font, odds_color, outline_color)
|
||||
|
||||
return image
|
||||
|
||||
def _format_game_time(self, game_time: str) -> str:
|
||||
@@ -568,6 +621,7 @@ class MLBLiveManager(BaseMLBManager):
|
||||
game['home_team'] in self.favorite_teams or
|
||||
game['away_team'] in self.favorite_teams
|
||||
):
|
||||
self._fetch_odds(game)
|
||||
# Ensure scores are valid numbers
|
||||
try:
|
||||
game['home_score'] = int(game['home_score'])
|
||||
@@ -942,6 +996,7 @@ class MLBRecentManager(BaseMLBManager):
|
||||
|
||||
# Only add favorite team games that are final and within time window
|
||||
if is_favorite_game and is_final and is_within_time:
|
||||
self._fetch_odds(game)
|
||||
new_recent_games.append(game)
|
||||
logger.info(f"[MLB] Added favorite team game to recent list: {game['away_team']} @ {game['home_team']}")
|
||||
|
||||
@@ -1056,6 +1111,7 @@ class MLBUpcomingManager(BaseMLBManager):
|
||||
logger.info(f"Status state not final: {game['status_state'] not in ['post', 'final', 'completed']}")
|
||||
|
||||
if is_upcoming:
|
||||
self._fetch_odds(game)
|
||||
new_upcoming_games.append(game)
|
||||
logger.info(f"Added favorite team game to upcoming list: {game['away_team']} @ {game['home_team']}")
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -32,6 +33,7 @@ class BaseNBAManager:
|
||||
_shared_data = None
|
||||
_last_shared_update = 0
|
||||
cache_manager = CacheManager() # Make cache_manager a class attribute
|
||||
odds_manager = OddsManager(cache_manager)
|
||||
logger = logging.getLogger('NBA') # Make logger a class attribute
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
@@ -40,6 +42,7 @@ class BaseNBAManager:
|
||||
self.config = config
|
||||
self.nba_config = config.get("nba_scoreboard", {})
|
||||
self.is_enabled = self.nba_config.get("enabled", False)
|
||||
self.show_odds = self.nba_config.get("show_odds", False)
|
||||
self.test_mode = self.nba_config.get("test_mode", False)
|
||||
self.logo_dir = self.nba_config.get("logo_dir", "assets/sports/nba_logos")
|
||||
self.update_interval = self.nba_config.get("update_interval_seconds", 300)
|
||||
|
||||
@@ -10,6 +10,7 @@ import numpy as np
|
||||
from .cache_manager import CacheManager
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util.retry import Retry
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Get logger
|
||||
@@ -24,8 +25,10 @@ class BaseNCAABaseballManager:
|
||||
self.config = config
|
||||
self.display_manager = display_manager
|
||||
self.ncaa_baseball_config = config.get('ncaa_baseball_scoreboard', {})
|
||||
self.show_odds = self.ncaa_baseball_config.get('show_odds', False)
|
||||
self.favorite_teams = self.ncaa_baseball_config.get('favorite_teams', [])
|
||||
self.cache_manager = CacheManager()
|
||||
self.odds_manager = OddsManager(self.cache_manager, self.config)
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.logger.setLevel(logging.DEBUG) # Set logger level to DEBUG
|
||||
|
||||
@@ -53,6 +56,22 @@ class BaseNCAABaseballManager:
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
||||
}
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a game and attach it to the game dictionary."""
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
try:
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="baseball",
|
||||
league="college-baseball",
|
||||
event_id=game["id"]
|
||||
)
|
||||
if odds_data:
|
||||
game['odds'] = odds_data
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
|
||||
|
||||
def _get_team_logo(self, team_abbr: str) -> Optional[Image.Image]:
|
||||
"""Get team logo from the configured directory or generate a fallback."""
|
||||
try:
|
||||
@@ -256,6 +275,40 @@ class BaseNCAABaseballManager:
|
||||
score_x = (width - score_width) // 2
|
||||
score_y = height - score_font.getmetrics()[0] - 2 # Adjusted for font metrics
|
||||
self._draw_text_with_outline(draw, score_text, (score_x, score_y), score_font)
|
||||
|
||||
# Draw betting odds if available and enabled
|
||||
if self.show_odds and 'odds' in game_data:
|
||||
odds_details = game_data['odds'].get('details', 'N/A')
|
||||
home_team_odds = game_data['odds'].get('home_team_odds', {})
|
||||
away_team_odds = game_data['odds'].get('away_team_odds', {})
|
||||
|
||||
# Extract spread and format it
|
||||
home_spread = home_team_odds.get('point_spread', 'N/A')
|
||||
away_spread = away_team_odds.get('point_spread', 'N/A')
|
||||
|
||||
# Add a plus sign to positive spreads
|
||||
if isinstance(home_spread, (int, float)) and home_spread > 0:
|
||||
home_spread = f"+{home_spread}"
|
||||
|
||||
if isinstance(away_spread, (int, float)) and away_spread > 0:
|
||||
away_spread = f"+{away_spread}"
|
||||
|
||||
# Define colors for odds text
|
||||
odds_font = self.display_manager.status_font
|
||||
odds_color = (255, 0, 0) # Red text
|
||||
outline_color = (0, 0, 0) # Black outline
|
||||
|
||||
# Draw away team odds
|
||||
if away_spread != 'N/A':
|
||||
away_odds_x = 5
|
||||
away_odds_y = height - 10
|
||||
self._draw_text_with_outline(draw, str(away_spread), (away_odds_x, away_odds_y), odds_font, odds_color, outline_color)
|
||||
|
||||
# Draw home team odds
|
||||
if home_spread != 'N/A':
|
||||
home_odds_x = width - 30
|
||||
home_odds_y = height - 10
|
||||
self._draw_text_with_outline(draw, str(home_spread), (home_odds_x, home_odds_y), odds_font, odds_color, outline_color)
|
||||
|
||||
return image
|
||||
|
||||
@@ -498,6 +551,7 @@ class NCAABaseballLiveManager(BaseNCAABaseballManager):
|
||||
try:
|
||||
game['home_score'] = int(game['home_score'])
|
||||
game['away_score'] = int(game['away_score'])
|
||||
self._fetch_odds(game)
|
||||
new_live_games.append(game)
|
||||
except (ValueError, TypeError):
|
||||
self.logger.warning(f"[NCAABaseball] Invalid score format for game {game['away_team']} @ {game['home_team']}")
|
||||
@@ -777,6 +831,7 @@ class NCAABaseballRecentManager(BaseNCAABaseballManager):
|
||||
logger.info(f"[NCAABaseball] Is within time window: {is_within_time}")
|
||||
|
||||
if is_final and is_within_time:
|
||||
self._fetch_odds(game)
|
||||
new_recent_games.append(game)
|
||||
logger.info(f"[NCAABaseball] Added favorite team game to recent list: {game['away_team']} @ {game['home_team']}")
|
||||
|
||||
@@ -869,6 +924,7 @@ class NCAABaseballUpcomingManager(BaseNCAABaseballManager):
|
||||
logger.info(f"[NCAABaseball] Is upcoming state: {is_upcoming_state}")
|
||||
|
||||
if is_within_time and is_upcoming_state:
|
||||
self._fetch_odds(game)
|
||||
new_upcoming_games.append(game)
|
||||
logger.info(f"[NCAABaseball] Added favorite team game to upcoming list: {game['away_team']} @ {game['home_team']}")
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager # Keep CacheManager import
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -83,6 +84,7 @@ class BaseNCAAFBManager: # Renamed class
|
||||
_shared_data = None
|
||||
_last_shared_update = 0
|
||||
cache_manager = CacheManager()
|
||||
odds_manager = OddsManager(cache_manager)
|
||||
logger = logging.getLogger('NCAAFB') # Changed logger name
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
@@ -91,6 +93,7 @@ class BaseNCAAFBManager: # Renamed class
|
||||
self.config = config
|
||||
self.ncaa_fb_config = config.get("ncaa_fb_scoreboard", {}) # Changed config key
|
||||
self.is_enabled = self.ncaa_fb_config.get("enabled", False)
|
||||
self.show_odds = self.ncaa_fb_config.get("show_odds", False)
|
||||
self.test_mode = self.ncaa_fb_config.get("test_mode", False)
|
||||
self.logo_dir = self.ncaa_fb_config.get("logo_dir", "assets/sports/ncaa_fbs_logos") # Changed logo dir
|
||||
self.update_interval = self.ncaa_fb_config.get("update_interval_seconds", 60)
|
||||
@@ -121,6 +124,44 @@ class BaseNCAAFBManager: # Renamed class
|
||||
except pytz.UnknownTimeZoneError:
|
||||
return pytz.utc
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a specific game if conditions are met."""
|
||||
self.logger.debug(f"Checking odds for game: {game.get('id', 'N/A')}")
|
||||
|
||||
# Ensure the API key is set in the secrets config
|
||||
if not self.config_manager.get_secret("the_odds_api_key"):
|
||||
if self._should_log('no_api_key', cooldown=3600): # Log once per hour
|
||||
self.logger.warning("Odds API key not found. Skipping odds fetch.")
|
||||
return
|
||||
|
||||
# Check if odds should be shown for this sport
|
||||
if not self.show_odds:
|
||||
self.logger.debug("Odds display is disabled for NCAAFB.")
|
||||
return
|
||||
|
||||
# Fetch odds using OddsManager
|
||||
try:
|
||||
# Determine update interval based on game state
|
||||
is_live = game.get('status', '').lower() == 'in'
|
||||
update_interval = self.ncaa_fb_config.get("live_odds_update_interval", 60) if is_live \
|
||||
else self.ncaa_fb_config.get("odds_update_interval", 3600)
|
||||
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="football",
|
||||
league="college-football",
|
||||
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}")
|
||||
|
||||
@classmethod
|
||||
def _fetch_shared_data(cls, past_days: int, future_days: int, date_str: str = None) -> Optional[Dict]:
|
||||
"""Fetch and cache data for all managers to share."""
|
||||
@@ -552,6 +593,9 @@ class NCAAFBLiveManager(BaseNCAAFBManager): # Renamed class
|
||||
details["home_abbr"] in self.favorite_teams or
|
||||
details["away_abbr"] in self.favorite_teams
|
||||
):
|
||||
# Fetch odds if enabled
|
||||
if self.show_odds:
|
||||
self._fetch_odds(details)
|
||||
new_live_games.append(details)
|
||||
|
||||
# Log changes or periodically
|
||||
|
||||
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -32,6 +33,7 @@ class BaseNCAAMBasketballManager:
|
||||
_shared_data = None
|
||||
_last_shared_update = 0
|
||||
cache_manager = CacheManager() # Make cache_manager a class attribute
|
||||
odds_manager = OddsManager(cache_manager, ConfigManager())
|
||||
logger = logging.getLogger('NCAAMBasketball') # Make logger a class attribute
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
@@ -40,6 +42,7 @@ class BaseNCAAMBasketballManager:
|
||||
self.config = config
|
||||
self.ncaam_basketball_config = config.get("ncaam_basketball_scoreboard", {})
|
||||
self.is_enabled = self.ncaam_basketball_config.get("enabled", False)
|
||||
self.show_odds = self.ncaam_basketball_config.get("show_odds", False)
|
||||
self.test_mode = self.ncaam_basketball_config.get("test_mode", False)
|
||||
self.logo_dir = self.ncaam_basketball_config.get("logo_dir", "assets/sports/ncaam_logos")
|
||||
self.update_interval = self.ncaam_basketball_config.get("update_interval_seconds", 300)
|
||||
@@ -66,6 +69,22 @@ class BaseNCAAMBasketballManager:
|
||||
self.logger.info(f"Initialized NCAAMBasketball manager with display dimensions: {self.display_width}x{self.display_height}")
|
||||
self.logger.info(f"Logo directory: {self.logo_dir}")
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a game and attach it to the game dictionary."""
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
try:
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="basketball",
|
||||
league="mens-college-basketball",
|
||||
event_id=game["id"]
|
||||
)
|
||||
if odds_data:
|
||||
game['odds'] = odds_data
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
|
||||
|
||||
def _get_timezone(self):
|
||||
try:
|
||||
return pytz.timezone(self.config_manager.get_timezone())
|
||||
@@ -436,7 +455,8 @@ class BaseNCAAMBasketballManager:
|
||||
"away_score": away_team.get("score", "0"),
|
||||
"away_logo_path": os.path.join(self.logo_dir, f"{away_team['team']['abbreviation']}.png"),
|
||||
"game_time": game_time,
|
||||
"game_date": game_date
|
||||
"game_date": game_date,
|
||||
"id": game_event.get("id")
|
||||
}
|
||||
|
||||
# Log game details for debugging
|
||||
@@ -568,6 +588,25 @@ class BaseNCAAMBasketballManager:
|
||||
clock_y = period_y + 10 # Position below period
|
||||
self._draw_text_with_outline(draw, clock, (clock_x, clock_y), self.fonts['time'])
|
||||
|
||||
# Display odds if available
|
||||
if 'odds' in game:
|
||||
odds = game['odds']
|
||||
spread = odds.get('spread', {}).get('point', None)
|
||||
if spread is not None:
|
||||
# Format spread text
|
||||
spread_text = f"{spread:+.1f}" if spread > 0 else f"{spread:.1f}"
|
||||
|
||||
# Choose color and position based on which team has the spread
|
||||
if odds.get('spread', {}).get('team') == game['home_abbr']:
|
||||
text_color = (255, 100, 100) # Reddish
|
||||
spread_x = self.display_width - draw.textlength(spread_text, font=self.fonts['status']) - 2
|
||||
else:
|
||||
text_color = (100, 255, 100) # Greenish
|
||||
spread_x = 2
|
||||
|
||||
spread_y = self.display_height - 8
|
||||
self._draw_text_with_outline(draw, spread_text, (spread_x, spread_y), self.fonts['status'], fill=text_color)
|
||||
|
||||
# Display the image
|
||||
self.display_manager.image.paste(main_img, (0, 0))
|
||||
self.display_manager.update_display()
|
||||
@@ -677,6 +716,7 @@ class NCAAMBasketballLiveManager(BaseNCAAMBasketballManager):
|
||||
details["home_abbr"] in self.favorite_teams or
|
||||
details["away_abbr"] in self.favorite_teams
|
||||
):
|
||||
self._fetch_odds(details)
|
||||
new_live_games.append(details)
|
||||
if self.favorite_teams and (
|
||||
details["home_abbr"] in self.favorite_teams or
|
||||
@@ -810,6 +850,7 @@ class NCAAMBasketballRecentManager(BaseNCAAMBasketballManager):
|
||||
game = self._extract_game_details(event)
|
||||
# Filter for recent games: must be final and within the time window
|
||||
if game and game['is_final'] and game['is_within_window']:
|
||||
self._fetch_odds(game)
|
||||
new_recent_games.append(game)
|
||||
|
||||
# Filter for favorite teams
|
||||
@@ -940,6 +981,7 @@ class NCAAMBasketballUpcomingManager(BaseNCAAMBasketballManager):
|
||||
for event in events:
|
||||
game = self._extract_game_details(event)
|
||||
if game and game['is_upcoming']:
|
||||
self._fetch_odds(game)
|
||||
new_upcoming_games.append(game)
|
||||
self.logger.debug(f"Processing upcoming game: {game['away_abbr']} vs {game['home_abbr']}")
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -83,6 +84,7 @@ class BaseNFLManager: # Renamed class
|
||||
_shared_data = None
|
||||
_last_shared_update = 0
|
||||
cache_manager = CacheManager()
|
||||
odds_manager = OddsManager(cache_manager)
|
||||
logger = logging.getLogger('NFL') # Changed logger name
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
@@ -91,6 +93,7 @@ class BaseNFLManager: # Renamed class
|
||||
self.config = config
|
||||
self.nfl_config = config.get("nfl_scoreboard", {}) # Changed config key
|
||||
self.is_enabled = self.nfl_config.get("enabled", False)
|
||||
self.show_odds = self.nfl_config.get("show_odds", False)
|
||||
self.test_mode = self.nfl_config.get("test_mode", False)
|
||||
self.logo_dir = self.nfl_config.get("logo_dir", "assets/sports/nfl_logos") # Changed logo dir
|
||||
self.update_interval = self.nfl_config.get("update_interval_seconds", 60)
|
||||
@@ -121,6 +124,44 @@ class BaseNFLManager: # Renamed class
|
||||
except pytz.UnknownTimeZoneError:
|
||||
return pytz.utc
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a specific game if conditions are met."""
|
||||
self.logger.debug(f"Checking odds for game: {game.get('id', 'N/A')}")
|
||||
|
||||
# Ensure the API key is set in the secrets config
|
||||
if not self.config_manager.get_secret("the_odds_api_key"):
|
||||
if self._should_log('no_api_key', cooldown=3600): # Log once per hour
|
||||
self.logger.warning("Odds API key not found. Skipping odds fetch.")
|
||||
return
|
||||
|
||||
# Check if odds should be shown for this sport
|
||||
if not self.show_odds:
|
||||
self.logger.debug("Odds display is disabled for NFL.")
|
||||
return
|
||||
|
||||
# Fetch odds using OddsManager
|
||||
try:
|
||||
# Determine update interval based on game state
|
||||
is_live = game.get('status', '').lower() == 'in'
|
||||
update_interval = self.nfl_config.get("live_odds_update_interval", 60) if is_live \
|
||||
else self.nfl_config.get("odds_update_interval", 3600)
|
||||
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="football",
|
||||
league="nfl",
|
||||
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}")
|
||||
|
||||
@classmethod
|
||||
def _fetch_shared_data(cls, past_days: int, future_days: int, date_str: str = None) -> Optional[Dict]:
|
||||
"""Fetch and cache data for all managers to share."""
|
||||
|
||||
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -81,6 +82,7 @@ class BaseNHLManager:
|
||||
_shared_data = None
|
||||
_last_shared_update = 0
|
||||
cache_manager = CacheManager() # Make cache_manager a class attribute
|
||||
odds_manager = OddsManager(cache_manager, ConfigManager())
|
||||
logger = logging.getLogger('NHL') # Make logger a class attribute
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
@@ -89,6 +91,7 @@ class BaseNHLManager:
|
||||
self.config = config
|
||||
self.nhl_config = config.get("nhl_scoreboard", {})
|
||||
self.is_enabled = self.nhl_config.get("enabled", False)
|
||||
self.show_odds = self.nhl_config.get("show_odds", False)
|
||||
self.test_mode = self.nhl_config.get("test_mode", False) # Use test_mode from config
|
||||
self.logo_dir = self.nhl_config.get("logo_dir", "assets/sports/nhl_logos")
|
||||
self.update_interval = self.nhl_config.get("update_interval_seconds", 60)
|
||||
@@ -115,6 +118,22 @@ class BaseNHLManager:
|
||||
self.logger.info(f"Initialized NHL manager with display dimensions: {self.display_width}x{self.display_height}")
|
||||
self.logger.info(f"Logo directory: {self.logo_dir}")
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a game and attach it to the game dictionary."""
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
try:
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="hockey",
|
||||
league="nhl",
|
||||
event_id=game["id"]
|
||||
)
|
||||
if odds_data:
|
||||
game['odds'] = odds_data
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
|
||||
|
||||
def _get_timezone(self):
|
||||
try:
|
||||
return pytz.timezone(self.config_manager.get_timezone())
|
||||
@@ -387,7 +406,8 @@ class BaseNHLManager:
|
||||
"away_score": away_team.get("score", "0"),
|
||||
"away_logo_path": os.path.join(self.logo_dir, f"{away_team['team']['abbreviation']}.png"),
|
||||
"game_time": game_time,
|
||||
"game_date": game_date
|
||||
"game_date": game_date,
|
||||
"id": game_event.get("id")
|
||||
}
|
||||
|
||||
# Log game details for debugging
|
||||
@@ -513,6 +533,25 @@ class BaseNHLManager:
|
||||
status_y = 5
|
||||
self._draw_text_with_outline(draw, status_text, (status_x, status_y), self.fonts['time'])
|
||||
|
||||
# Display odds if available
|
||||
if 'odds' in game:
|
||||
odds = game['odds']
|
||||
spread = odds.get('spread', {}).get('point', None)
|
||||
if spread is not None:
|
||||
# Format spread text
|
||||
spread_text = f"{spread:+.1f}" if spread > 0 else f"{spread:.1f}"
|
||||
|
||||
# Choose color and position based on which team has the spread
|
||||
if odds.get('spread', {}).get('team') == game['home_abbr']:
|
||||
text_color = (255, 100, 100) # Reddish
|
||||
spread_x = self.display_width - draw.textlength(spread_text, font=self.fonts['status']) - 2
|
||||
else:
|
||||
text_color = (100, 255, 100) # Greenish
|
||||
spread_x = 2
|
||||
|
||||
spread_y = self.display_height - 8
|
||||
self._draw_text_with_outline(draw, spread_text, (spread_x, spread_y), self.fonts['status'], fill=text_color)
|
||||
|
||||
# Display the image
|
||||
self.display_manager.image.paste(main_img, (0, 0))
|
||||
self.display_manager.update_display()
|
||||
@@ -608,6 +647,7 @@ class NHLLiveManager(BaseNHLManager):
|
||||
details["home_abbr"] in self.favorite_teams or
|
||||
details["away_abbr"] in self.favorite_teams
|
||||
):
|
||||
self._fetch_odds(details)
|
||||
new_live_games.append(details)
|
||||
|
||||
# Only log if there's a change in games or enough time has passed
|
||||
@@ -707,6 +747,9 @@ class NHLRecentManager(BaseNHLManager):
|
||||
for event in events:
|
||||
game = self._extract_game_details(event)
|
||||
if game:
|
||||
# Fetch odds if enabled
|
||||
if self.show_odds:
|
||||
self._fetch_odds(game)
|
||||
self.recent_games.append(game)
|
||||
self.logger.debug(f"Processing game: {game['away_abbr']} vs {game['home_abbr']} - Final: {game['is_final']}, Within window: {game['is_within_window']}")
|
||||
|
||||
@@ -796,6 +839,9 @@ class NHLUpcomingManager(BaseNHLManager):
|
||||
self.logger.debug(f"[NHL] Game time: {game['start_time_utc']}")
|
||||
|
||||
if not game['is_final'] and game['is_within_window']:
|
||||
# Fetch odds if enabled
|
||||
if self.show_odds:
|
||||
self._fetch_odds(game)
|
||||
new_upcoming_games.append(game)
|
||||
self.logger.debug(f"[NHL] Added to upcoming games: {game['away_abbr']} vs {game['home_abbr']}")
|
||||
|
||||
|
||||
65
src/odds_manager.py
Normal file
65
src/odds_manager.py
Normal file
@@ -0,0 +1,65 @@
|
||||
import requests
|
||||
import logging
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from typing import Optional, List, Dict, Any
|
||||
|
||||
class OddsManager:
|
||||
def __init__(self, cache_manager: CacheManager, config_manager: ConfigManager):
|
||||
self.cache_manager = cache_manager
|
||||
self.config_manager = config_manager
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.base_url = "https://sports.core.api.espn.com/v2/sports"
|
||||
|
||||
def get_odds(self, sport: str, league: str, event_id: str, update_interval_seconds=3600):
|
||||
cache_key = f"odds_espn_{sport}_{league}_{event_id}"
|
||||
|
||||
cached_data = self.cache_manager.get_cached_data(cache_key, max_age=update_interval_seconds)
|
||||
if cached_data:
|
||||
self.logger.info(f"Using cached odds from ESPN for {cache_key}")
|
||||
return cached_data
|
||||
|
||||
self.logger.info(f"Fetching fresh odds from ESPN for {cache_key}")
|
||||
|
||||
try:
|
||||
url = f"{self.base_url}/{sport}/leagues/{league}/events/{event_id}/competitions/{event_id}/odds"
|
||||
response = requests.get(url)
|
||||
response.raise_for_status()
|
||||
raw_data = response.json()
|
||||
|
||||
odds_data = self._extract_espn_data(raw_data)
|
||||
|
||||
if odds_data:
|
||||
self.cache_manager.save_cache(cache_key, odds_data)
|
||||
|
||||
return odds_data
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
self.logger.error(f"Error fetching odds from ESPN API for {cache_key}: {e}")
|
||||
except json.JSONDecodeError:
|
||||
self.logger.error(f"Error decoding JSON response from ESPN API for {cache_key}.")
|
||||
|
||||
return self.cache_manager.load_cache(cache_key)
|
||||
|
||||
def _extract_espn_data(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
if "items" in data and data["items"]:
|
||||
item = data["items"][0]
|
||||
# Find the desired bookmaker, e.g., 'fanduel'
|
||||
provider = next((p for p in item.get('providers', []) if p.get('name', '').lower() == 'fanduel'), item['providers'][0] if item.get('providers') else {})
|
||||
|
||||
return {
|
||||
"details": provider.get("details"),
|
||||
"over_under": provider.get("overUnder"),
|
||||
"spread": provider.get("spread"),
|
||||
"home_team_odds": {
|
||||
"money_line": provider.get("homeTeamOdds", {}).get("moneyLine"),
|
||||
"spread_odds": provider.get("homeTeamOdds", {}).get("spreadOdds")
|
||||
},
|
||||
"away_team_odds": {
|
||||
"money_line": provider.get("awayTeamOdds", {}).get("moneyLine"),
|
||||
"spread_odds": provider.get("awayTeamOdds", {}).get("spreadOdds")
|
||||
}
|
||||
}
|
||||
return None
|
||||
@@ -11,6 +11,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
import pytz
|
||||
|
||||
# Constants
|
||||
@@ -95,6 +96,7 @@ class BaseSoccerManager:
|
||||
_shared_data = {} # Dictionary to hold shared data per league/date
|
||||
_last_shared_update = {} # Dictionary for update times per league/date
|
||||
cache_manager = CacheManager()
|
||||
odds_manager = OddsManager(cache_manager, ConfigManager())
|
||||
logger = logging.getLogger('Soccer') # Use 'Soccer' logger
|
||||
|
||||
# Class attribute to store soccer_config for shared access
|
||||
@@ -109,6 +111,7 @@ class BaseSoccerManager:
|
||||
BaseSoccerManager._soccer_config_shared = self.soccer_config # Store for class methods
|
||||
|
||||
self.is_enabled = self.soccer_config.get("enabled", False)
|
||||
self.show_odds = self.soccer_config.get("show_odds", False)
|
||||
self.test_mode = self.soccer_config.get("test_mode", False)
|
||||
self.logo_dir = self.soccer_config.get("logo_dir", "assets/sports/soccer_logos") # Soccer logos
|
||||
self.update_interval = self.soccer_config.get("update_interval_seconds", 60) # General fallback
|
||||
@@ -147,6 +150,22 @@ class BaseSoccerManager:
|
||||
|
||||
self.config_manager = ConfigManager(config)
|
||||
|
||||
def _fetch_odds(self, game: Dict) -> None:
|
||||
"""Fetch odds for a game and attach it to the game dictionary."""
|
||||
if not self.show_odds:
|
||||
return
|
||||
|
||||
try:
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport="soccer",
|
||||
league=game["league_slug"],
|
||||
event_id=game["id"]
|
||||
)
|
||||
if odds_data:
|
||||
game['odds'] = odds_data
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error fetching odds for game {game.get('id', 'N/A')}: {e}")
|
||||
|
||||
# --- Team League Map Management ---
|
||||
@classmethod
|
||||
def _load_team_league_map(cls) -> None:
|
||||
@@ -576,7 +595,8 @@ class BaseSoccerManager:
|
||||
"away_logo": self._load_and_resize_logo(away_team["team"]["abbreviation"]),
|
||||
"game_time": game_time, # Formatted local time (e.g., 2:30pm)
|
||||
"game_date": game_date, # Formatted local date (e.g., 7/21)
|
||||
"league": league_name
|
||||
"league": league_name,
|
||||
"league_slug": league_slug
|
||||
}
|
||||
|
||||
self.logger.debug(f"[Soccer] Extracted game: {details['away_abbr']} {details['away_score']} @ {details['home_abbr']} {details['home_score']} ({details['game_clock_display']}) - League: {details['league']} - Final: {details['is_final']}, Upcoming: {details['is_upcoming']}, Live: {details['is_live']}, Within Window: {details['is_within_window']}")
|
||||
@@ -693,6 +713,25 @@ class BaseSoccerManager:
|
||||
status_y_top = 1 # Original Y position for live/final status
|
||||
self._draw_text_with_outline(draw, status_text, (status_x, status_y_top), status_font_top)
|
||||
|
||||
# Display odds if available
|
||||
if 'odds' in game:
|
||||
odds = game['odds']
|
||||
spread = odds.get('spread', {}).get('point', None)
|
||||
if spread is not None:
|
||||
# Format spread text
|
||||
spread_text = f"{spread:+.1f}" if spread > 0 else f"{spread:.1f}"
|
||||
|
||||
# Choose color and position based on which team has the spread
|
||||
if odds.get('spread', {}).get('team') == game['home_abbr']:
|
||||
text_color = (255, 100, 100) # Reddish
|
||||
spread_x = self.display_width - draw.textlength(spread_text, font=self.fonts['status']) - 2
|
||||
else:
|
||||
text_color = (100, 255, 100) # Greenish
|
||||
spread_x = 2
|
||||
|
||||
spread_y = self.display_height - 8
|
||||
self._draw_text_with_outline(draw, spread_text, (spread_x, spread_y), self.fonts['status'], fill=text_color)
|
||||
|
||||
# --- Display Image ---
|
||||
self.display_manager.image.paste(main_img, (0, 0))
|
||||
self.display_manager.update_display()
|
||||
@@ -786,6 +825,7 @@ class SoccerLiveManager(BaseSoccerManager):
|
||||
details["home_abbr"] in self.favorite_teams or
|
||||
details["away_abbr"] in self.favorite_teams
|
||||
):
|
||||
self._fetch_odds(details)
|
||||
new_live_games.append(details)
|
||||
|
||||
# Logging
|
||||
@@ -914,6 +954,7 @@ class SoccerRecentManager(BaseSoccerManager):
|
||||
if game and game['is_final'] and game['start_time_utc'] and game['start_time_utc'] >= cutoff_time:
|
||||
# Check favorite teams if list is provided
|
||||
if not self.favorite_teams or (game['home_abbr'] in self.favorite_teams or game['away_abbr'] in self.favorite_teams):
|
||||
self._fetch_odds(game)
|
||||
new_recent_games.append(game)
|
||||
|
||||
# Sort games by start time, most recent first
|
||||
@@ -1016,6 +1057,7 @@ class SoccerUpcomingManager(BaseSoccerManager):
|
||||
game['start_time_utc'] >= now_utc and game['start_time_utc'] <= cutoff_time:
|
||||
# Check favorite teams if list is provided
|
||||
if not self.favorite_teams or (game['home_abbr'] in self.favorite_teams or game['away_abbr'] in self.favorite_teams):
|
||||
self._fetch_odds(game)
|
||||
new_upcoming_games.append(game)
|
||||
|
||||
# Sort games by start time, soonest first
|
||||
|
||||
Reference in New Issue
Block a user