mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-10 21:03:01 +00:00
format gambling displays for all sports, add gambling ticker
This commit is contained in:
Chuck
406
src/odds_ticker_manager.py
Normal file
406
src/odds_ticker_manager.py
Normal file
@@ -0,0 +1,406 @@
|
||||
import time
|
||||
import logging
|
||||
import requests
|
||||
import json
|
||||
from typing import Dict, Any, List, Optional
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import os
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
import pytz
|
||||
from src.display_manager import DisplayManager
|
||||
from src.cache_manager import CacheManager
|
||||
from src.config_manager import ConfigManager
|
||||
from src.odds_manager import OddsManager
|
||||
|
||||
# Get logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class OddsTickerManager:
|
||||
"""Manager for displaying scrolling odds ticker for multiple sports leagues."""
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
|
||||
self.config = config
|
||||
self.display_manager = display_manager
|
||||
self.odds_ticker_config = config.get('odds_ticker', {})
|
||||
self.is_enabled = self.odds_ticker_config.get('enabled', False)
|
||||
self.show_favorite_teams_only = self.odds_ticker_config.get('show_favorite_teams_only', False)
|
||||
self.enabled_leagues = self.odds_ticker_config.get('enabled_leagues', ['nfl', 'nba', 'mlb'])
|
||||
self.update_interval = self.odds_ticker_config.get('update_interval', 3600)
|
||||
self.scroll_speed = self.odds_ticker_config.get('scroll_speed', 2)
|
||||
self.scroll_delay = self.odds_ticker_config.get('scroll_delay', 0.05)
|
||||
self.display_duration = self.odds_ticker_config.get('display_duration', 30)
|
||||
|
||||
# Initialize managers
|
||||
self.cache_manager = CacheManager()
|
||||
self.odds_manager = OddsManager(self.cache_manager, ConfigManager())
|
||||
|
||||
# State variables
|
||||
self.last_update = 0
|
||||
self.current_position = 0
|
||||
self.last_scroll_time = 0
|
||||
self.games_data = []
|
||||
self.current_game_index = 0
|
||||
self.current_image = None
|
||||
|
||||
# Font setup
|
||||
self.fonts = self._load_fonts()
|
||||
|
||||
# League configurations
|
||||
self.league_configs = {
|
||||
'nfl': {
|
||||
'sport': 'football',
|
||||
'league': 'nfl',
|
||||
'logo_dir': 'assets/sports/nfl_logos',
|
||||
'favorite_teams': config.get('nfl_scoreboard', {}).get('favorite_teams', []),
|
||||
'enabled': config.get('nfl_scoreboard', {}).get('enabled', False)
|
||||
},
|
||||
'nba': {
|
||||
'sport': 'basketball',
|
||||
'league': 'nba',
|
||||
'logo_dir': 'assets/sports/nba_logos',
|
||||
'favorite_teams': config.get('nba_scoreboard', {}).get('favorite_teams', []),
|
||||
'enabled': config.get('nba_scoreboard', {}).get('enabled', False)
|
||||
},
|
||||
'mlb': {
|
||||
'sport': 'baseball',
|
||||
'league': 'mlb',
|
||||
'logo_dir': 'assets/sports/mlb_logos',
|
||||
'favorite_teams': config.get('mlb', {}).get('favorite_teams', []),
|
||||
'enabled': config.get('mlb', {}).get('enabled', False)
|
||||
},
|
||||
'ncaa_fb': {
|
||||
'sport': 'football',
|
||||
'league': 'college-football',
|
||||
'logo_dir': 'assets/sports/ncaa_fbs_logos',
|
||||
'favorite_teams': config.get('ncaa_fb_scoreboard', {}).get('favorite_teams', []),
|
||||
'enabled': config.get('ncaa_fb_scoreboard', {}).get('enabled', False)
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(f"OddsTickerManager initialized with enabled leagues: {self.enabled_leagues}")
|
||||
logger.info(f"Show favorite teams only: {self.show_favorite_teams_only}")
|
||||
|
||||
def _load_fonts(self) -> Dict[str, ImageFont.FreeTypeFont]:
|
||||
"""Load fonts for the ticker display."""
|
||||
try:
|
||||
return {
|
||||
'small': ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 6),
|
||||
'medium': ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 8),
|
||||
'large': ImageFont.truetype("assets/fonts/PressStart2P-Regular.ttf", 10)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading fonts: {e}")
|
||||
return {
|
||||
'small': ImageFont.load_default(),
|
||||
'medium': ImageFont.load_default(),
|
||||
'large': ImageFont.load_default()
|
||||
}
|
||||
|
||||
def _get_team_logo(self, team_abbr: str, logo_dir: str) -> Optional[Image.Image]:
|
||||
"""Get team logo from the configured directory."""
|
||||
try:
|
||||
logo_path = os.path.join(logo_dir, f"{team_abbr}.png")
|
||||
if os.path.exists(logo_path):
|
||||
logo = Image.open(logo_path)
|
||||
return logo
|
||||
else:
|
||||
logger.debug(f"Logo not found: {logo_path}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading logo for {team_abbr}: {e}")
|
||||
return None
|
||||
|
||||
def _fetch_upcoming_games(self) -> List[Dict[str, Any]]:
|
||||
"""Fetch upcoming games with odds for all enabled leagues."""
|
||||
games_data = []
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
for league_key in self.enabled_leagues:
|
||||
if league_key not in self.league_configs:
|
||||
logger.warning(f"Unknown league: {league_key}")
|
||||
continue
|
||||
|
||||
league_config = self.league_configs[league_key]
|
||||
if not league_config['enabled']:
|
||||
logger.debug(f"League {league_key} is disabled, skipping")
|
||||
continue
|
||||
|
||||
try:
|
||||
# Fetch upcoming games for this league
|
||||
games = self._fetch_league_games(league_config, now)
|
||||
games_data.extend(games)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching games for {league_key}: {e}")
|
||||
|
||||
# Sort games by start time
|
||||
games_data.sort(key=lambda x: x.get('start_time', datetime.max))
|
||||
|
||||
# Filter for favorite teams if enabled
|
||||
if self.show_favorite_teams_only:
|
||||
all_favorite_teams = []
|
||||
for league_config in self.league_configs.values():
|
||||
all_favorite_teams.extend(league_config.get('favorite_teams', []))
|
||||
|
||||
games_data = [
|
||||
game for game in games_data
|
||||
if (game.get('home_team') in all_favorite_teams or
|
||||
game.get('away_team') in all_favorite_teams)
|
||||
]
|
||||
|
||||
logger.info(f"Fetched {len(games_data)} upcoming games for odds ticker")
|
||||
return games_data
|
||||
|
||||
def _fetch_league_games(self, league_config: Dict[str, Any], now: datetime) -> List[Dict[str, Any]]:
|
||||
"""Fetch upcoming games for a specific league."""
|
||||
games = []
|
||||
|
||||
# Get dates for API request (today and next 7 days)
|
||||
dates = []
|
||||
for i in range(8): # Today + 7 days
|
||||
date = now + timedelta(days=i)
|
||||
dates.append(date.strftime("%Y%m%d"))
|
||||
|
||||
for date in dates:
|
||||
try:
|
||||
# ESPN API endpoint for games with date parameter
|
||||
sport = league_config['sport']
|
||||
league = league_config['league']
|
||||
url = f"https://site.api.espn.com/apis/site/v2/sports/{sport}/{league}/scoreboard?dates={date}"
|
||||
|
||||
logger.debug(f"Fetching {league} games from ESPN API for date: {date}")
|
||||
response = requests.get(url, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
for event in data.get('events', []):
|
||||
game_id = event['id']
|
||||
status = event['status']['type']['name'].lower()
|
||||
|
||||
# Only include upcoming games
|
||||
if status in ['scheduled', 'pre-game']:
|
||||
game_time = datetime.fromisoformat(event['date'].replace('Z', '+00:00'))
|
||||
|
||||
# Only include games in the next 7 days
|
||||
if now <= game_time <= now + timedelta(days=7):
|
||||
# Get team information
|
||||
competitors = event['competitions'][0]['competitors']
|
||||
home_team = next(c for c in competitors if c['homeAway'] == 'home')
|
||||
away_team = next(c for c in competitors if c['homeAway'] == 'away')
|
||||
|
||||
home_abbr = home_team['team']['abbreviation']
|
||||
away_abbr = away_team['team']['abbreviation']
|
||||
|
||||
# Fetch odds for this game
|
||||
odds_data = self.odds_manager.get_odds(
|
||||
sport=sport,
|
||||
league=league,
|
||||
event_id=game_id
|
||||
)
|
||||
|
||||
game_data = {
|
||||
'id': game_id,
|
||||
'league': league_config['league'],
|
||||
'league_key': league_config['sport'],
|
||||
'home_team': home_abbr,
|
||||
'away_team': away_abbr,
|
||||
'start_time': game_time,
|
||||
'odds': odds_data,
|
||||
'logo_dir': league_config['logo_dir']
|
||||
}
|
||||
|
||||
games.append(game_data)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching {league_config['league']} games for date {date}: {e}")
|
||||
|
||||
return games
|
||||
|
||||
def _format_odds_text(self, game: Dict[str, Any]) -> str:
|
||||
"""Format the odds text for display."""
|
||||
odds = game.get('odds', {})
|
||||
if not odds:
|
||||
return f"{game['away_team']} vs {game['home_team']}"
|
||||
|
||||
# Extract odds data
|
||||
home_team_odds = odds.get('home_team_odds', {})
|
||||
away_team_odds = odds.get('away_team_odds', {})
|
||||
|
||||
home_spread = home_team_odds.get('spread_odds')
|
||||
away_spread = away_team_odds.get('spread_odds')
|
||||
home_ml = home_team_odds.get('money_line')
|
||||
away_ml = away_team_odds.get('money_line')
|
||||
over_under = odds.get('over_under')
|
||||
|
||||
# Format time
|
||||
game_time = game['start_time']
|
||||
timezone_str = self.config.get('timezone', 'UTC')
|
||||
try:
|
||||
tz = pytz.timezone(timezone_str)
|
||||
except pytz.exceptions.UnknownTimeZoneError:
|
||||
tz = pytz.UTC
|
||||
|
||||
if game_time.tzinfo is None:
|
||||
game_time = game_time.replace(tzinfo=pytz.UTC)
|
||||
local_time = game_time.astimezone(tz)
|
||||
time_str = local_time.strftime("%I:%M %p")
|
||||
|
||||
# Build odds string
|
||||
odds_parts = [f"[{time_str}]"]
|
||||
|
||||
# Add away team and odds
|
||||
odds_parts.append(game['away_team'])
|
||||
if away_spread is not None:
|
||||
spread_str = f"{away_spread:+.1f}" if away_spread > 0 else f"{away_spread:.1f}"
|
||||
odds_parts.append(spread_str)
|
||||
if away_ml is not None:
|
||||
ml_str = f"ML {away_ml:+d}" if away_ml > 0 else f"ML {away_ml}"
|
||||
odds_parts.append(ml_str)
|
||||
|
||||
odds_parts.append("vs")
|
||||
|
||||
# Add home team and odds
|
||||
odds_parts.append(game['home_team'])
|
||||
if home_spread is not None:
|
||||
spread_str = f"{home_spread:+.1f}" if home_spread > 0 else f"{home_spread:.1f}"
|
||||
odds_parts.append(spread_str)
|
||||
if home_ml is not None:
|
||||
ml_str = f"ML {home_ml:+d}" if home_ml > 0 else f"ML {home_ml}"
|
||||
odds_parts.append(ml_str)
|
||||
|
||||
# Add over/under
|
||||
if over_under is not None:
|
||||
odds_parts.append(f"O/U {over_under}")
|
||||
|
||||
return " ".join(odds_parts)
|
||||
|
||||
def _create_ticker_image(self, game: Dict[str, Any]) -> Image.Image:
|
||||
"""Create a scrolling ticker image for a game."""
|
||||
width = self.display_manager.matrix.width
|
||||
height = self.display_manager.matrix.height
|
||||
|
||||
# Create a wider image for scrolling
|
||||
scroll_width = width * 3 # 3x width for smooth scrolling
|
||||
image = Image.new('RGB', (scroll_width, height), color=(0, 0, 0))
|
||||
draw = ImageDraw.Draw(image)
|
||||
|
||||
# Format the odds text
|
||||
odds_text = self._format_odds_text(game)
|
||||
|
||||
# Load team logos
|
||||
home_logo = self._get_team_logo(game['home_team'], game['logo_dir'])
|
||||
away_logo = self._get_team_logo(game['away_team'], game['logo_dir'])
|
||||
|
||||
# Calculate text position (start off-screen to the right)
|
||||
text_width = draw.textlength(odds_text, font=self.fonts['medium'])
|
||||
text_x = scroll_width - text_width - 10 # Start off-screen right
|
||||
text_y = (height - self.fonts['medium'].size) // 2
|
||||
|
||||
# Draw the text
|
||||
self._draw_text_with_outline(draw, odds_text, (text_x, text_y), self.fonts['medium'])
|
||||
|
||||
# Add team logos if available
|
||||
logo_size = 16
|
||||
logo_y = (height - logo_size) // 2
|
||||
|
||||
if away_logo:
|
||||
away_logo.thumbnail((logo_size, logo_size), Image.Resampling.LANCZOS)
|
||||
away_x = text_x - logo_size - 5
|
||||
image.paste(away_logo, (away_x, logo_y), away_logo)
|
||||
|
||||
if home_logo:
|
||||
home_logo.thumbnail((logo_size, logo_size), Image.Resampling.LANCZOS)
|
||||
home_x = text_x + text_width + 5
|
||||
image.paste(home_logo, (home_x, logo_y), home_logo)
|
||||
|
||||
return image
|
||||
|
||||
def _draw_text_with_outline(self, draw: ImageDraw.Draw, text: str, position: tuple, font: ImageFont.FreeTypeFont,
|
||||
fill: tuple = (255, 255, 255), outline_color: tuple = (0, 0, 0)) -> None:
|
||||
"""Draw text with a black outline for better readability."""
|
||||
x, y = position
|
||||
# Draw outline
|
||||
for dx, dy in [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 1), (1, -1), (1, 0), (1, 1)]:
|
||||
draw.text((x + dx, y + dy), text, font=font, fill=outline_color)
|
||||
# Draw main text
|
||||
draw.text((x, y), text, font=font, fill=fill)
|
||||
|
||||
def update(self):
|
||||
"""Update odds ticker data."""
|
||||
if not self.is_enabled:
|
||||
return
|
||||
|
||||
current_time = time.time()
|
||||
if current_time - self.last_update < self.update_interval:
|
||||
return
|
||||
|
||||
try:
|
||||
logger.info("Updating odds ticker data")
|
||||
self.games_data = self._fetch_upcoming_games()
|
||||
self.last_update = current_time
|
||||
self.current_position = 0
|
||||
self.current_game_index = 0
|
||||
|
||||
if self.games_data:
|
||||
logger.info(f"Updated odds ticker with {len(self.games_data)} games")
|
||||
else:
|
||||
logger.warning("No games found for odds ticker")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating odds ticker: {e}", exc_info=True)
|
||||
|
||||
def display(self, force_clear: bool = False):
|
||||
"""Display the odds ticker."""
|
||||
if not self.is_enabled or not self.games_data:
|
||||
return
|
||||
|
||||
try:
|
||||
current_time = time.time()
|
||||
|
||||
# Check if it's time to switch games
|
||||
if current_time - self.last_update >= self.display_duration:
|
||||
self.current_game_index = (self.current_game_index + 1) % len(self.games_data)
|
||||
self.current_position = 0
|
||||
self.last_update = current_time
|
||||
force_clear = True
|
||||
|
||||
# Get current game
|
||||
current_game = self.games_data[self.current_game_index]
|
||||
|
||||
# Create ticker image if needed
|
||||
if force_clear or self.current_image is None:
|
||||
self.current_image = self._create_ticker_image(current_game)
|
||||
|
||||
# Scroll the image
|
||||
if current_time - self.last_scroll_time >= self.scroll_delay:
|
||||
self.current_position += self.scroll_speed
|
||||
self.last_scroll_time = current_time
|
||||
|
||||
# Calculate crop region
|
||||
width = self.display_manager.matrix.width
|
||||
height = self.display_manager.matrix.height
|
||||
|
||||
# Reset position when we've scrolled past the end
|
||||
if self.current_position >= self.current_image.width - width:
|
||||
self.current_position = 0
|
||||
|
||||
# Crop the scrolling region
|
||||
crop_x = self.current_position
|
||||
crop_y = 0
|
||||
crop_width = width
|
||||
crop_height = height
|
||||
|
||||
# Ensure we don't go out of bounds
|
||||
if crop_x + crop_width > self.current_image.width:
|
||||
crop_x = self.current_image.width - crop_width
|
||||
|
||||
cropped_image = self.current_image.crop((crop_x, crop_y, crop_x + crop_width, crop_y + crop_height))
|
||||
|
||||
# Display the cropped image
|
||||
self.display_manager.image = cropped_image
|
||||
self.display_manager.draw = ImageDraw.Draw(self.display_manager.image)
|
||||
self.display_manager.update_display()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error displaying odds ticker: {e}", exc_info=True)
|
||||
Reference in New Issue
Block a user