mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-25 20:38:15 +00:00
ChuckandClaude Opus 5
6b74506695
* fix(sports): fetch odds for the games shown, not the whole window
SportsUpcoming.update() walked every upcoming game in the schedule
window and called _fetch_odds() on each one inside that collection
loop, narrowing to upcoming_games_to_show only afterwards. Each call is
a separate sequential ESPN request.
The comment sitting above it said odds were fetched "only for games that
will be displayed". The only narrowing it actually applied was
show_favorite_teams_only, which is not the default, so in the usual
configuration nothing narrowed it at all.
Measured on devpi, where the football plugin has the same shape:
467 odds requests in one 35s burst, 467 distinct events
315 NFL + 152 college-football -- roughly a whole season
plugin football-scoreboard operation timed out after 30.0s
The burst repeats each time the 1h odds TTL expires: 67 -> 327 -> 957 ->
1261 requests/hour across four consecutive hours. Between expiries the
cache works and the rate is zero, so this is a thundering herd on
expiry, not a caching failure.
The fetch now runs after selection, over team_games -- the list already
cut to upcoming_games_to_show. This mirrors the fix the football plugin
already carries; the shared base class never got it.
SportsLive is deliberately left as it is: it walks the raw event list
because it has to find which games are live, but only fetches odds for a
game that has already passed the is_live/is_halftime test, so its
fan-out is bounded by how many games are actually in progress. The test
pins that distinction rather than assuming it.
The test reads the AST rather than the source text, and asserts the full
set of call sites, so a new one has to be classified deliberately
instead of inheriting whichever behaviour it happens to land in. Writing
it that way is what turned up the SportsLive site, which I had missed.
Verified: reverting the fix fails the test with the offending iterable
named ("iterates over 'events'"). 525 passed, 9 skipped across the sports
and odds suites.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW
* test(sports): check the odds guard structurally, not by its text
Review caught that _guards_above() collected an `if` test even when the
call sat in that if's `else`, so moving _fetch_odds() into the else of
the is_live/is_halftime test would still pass -- while fetching odds for
exactly the non-live games the guard exists to exclude.
Verifying that turned up a wider hole in the same assertion. It matched
substrings of the *unparsed source*, so a negated condition satisfied it
too:
if not (details["is_live"] or details["is_halftime"]):
self._fetch_odds(details) # every non-live game
Both names still appear in that text, so `"is_live" in guards` held and
the test passed on code doing the opposite of what it claims to check.
The guard test is now structural. It walks the AST for an enclosing `if`
whose *body* (never its `else`) contains the call, and whose test
references both names without either sitting under a `not`.
Verified by mutation: fetching odds for non-live games now fails with
"does not sit in the true branch of a test requiring the game to be in
progress". Moving the call into the else of the *favourites* test still
passes, which is correct -- the game there is still live, so the
in-progress contract holds and the fan-out stays bounded by how many
games are actually in play.
525 passed, 9 skipped across the sports and odds suites.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW
---------
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
1098 lines
57 KiB
Python
1098 lines
57 KiB
Python
"""The three display modes layered on SportsCore: SportsUpcoming,
|
|
SportsRecent and SportsLive. Split out of the former
|
|
``src/base_classes/sports.py``; see docs/SPORTS_UNIFICATION.md.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
from abc import abstractmethod
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, List
|
|
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
from src.cache_manager import CacheManager
|
|
from src.display_manager import DisplayManager
|
|
|
|
from .core import SportsCore
|
|
|
|
|
|
class SportsUpcoming(SportsCore):
|
|
SKIN_MODE = "upcoming"
|
|
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
|
|
super().__init__(config, display_manager, cache_manager, logger, sport_key)
|
|
self.upcoming_games = [] # Store all fetched upcoming games initially
|
|
self.games_list = [] # Filtered list for display (favorite teams)
|
|
self.current_game_index = 0
|
|
self.last_update = 0
|
|
self.update_interval = self.mode_config.get("upcoming_update_interval", 3600) # Check for recent games every hour
|
|
self.last_log_time = 0
|
|
self.log_interval = 300
|
|
self.last_warning_time = 0
|
|
self.warning_cooldown = 300
|
|
self.last_game_switch = 0
|
|
self.game_display_duration = 15 # Display each upcoming game for 15 seconds
|
|
|
|
def _select_games_for_display(
|
|
self, processed_games: List[Dict], favorite_teams: List[str]
|
|
) -> List[Dict]:
|
|
"""
|
|
Single-pass game selection with proper deduplication and counting.
|
|
|
|
When a game involves two favorite teams, it counts toward BOTH teams' limits.
|
|
This prevents unexpected game counts from the multi-pass algorithm.
|
|
|
|
Team identity goes through the ``_favorite_key`` override point rather
|
|
than reading ``home_abbr``/``away_abbr`` directly, because abbreviations
|
|
are not unique in every league (NRL matches on team ID instead).
|
|
"""
|
|
sorted_games = sorted(
|
|
processed_games,
|
|
key=lambda g: g.get("start_time_utc")
|
|
or datetime.max.replace(tzinfo=timezone.utc),
|
|
)
|
|
|
|
if not favorite_teams:
|
|
return sorted_games
|
|
|
|
selected_games = []
|
|
selected_ids = set()
|
|
team_counts = {team: 0 for team in favorite_teams}
|
|
|
|
for game in sorted_games:
|
|
game_id = game.get("id")
|
|
if game_id in selected_ids:
|
|
continue
|
|
|
|
home = self._favorite_key(game, "home")
|
|
away = self._favorite_key(game, "away")
|
|
|
|
home_fav = home in favorite_teams
|
|
away_fav = away in favorite_teams
|
|
|
|
if not home_fav and not away_fav:
|
|
continue
|
|
|
|
home_needs = home_fav and team_counts[home] < self.upcoming_games_to_show
|
|
away_needs = away_fav and team_counts[away] < self.upcoming_games_to_show
|
|
|
|
if home_needs or away_needs:
|
|
selected_games.append(game)
|
|
selected_ids.add(game_id)
|
|
if home_fav:
|
|
team_counts[home] += 1
|
|
if away_fav:
|
|
team_counts[away] += 1
|
|
|
|
self.logger.debug(
|
|
f"Selected game {away}@{home}: team_counts={team_counts}"
|
|
)
|
|
|
|
if all(c >= self.upcoming_games_to_show for c in team_counts.values()):
|
|
self.logger.debug("All favorite teams satisfied, stopping selection")
|
|
break
|
|
|
|
self.logger.info(
|
|
f"Selected {len(selected_games)} games for {len(favorite_teams)} "
|
|
f"favorite teams: {team_counts}"
|
|
)
|
|
return selected_games
|
|
|
|
def update(self):
|
|
"""Update upcoming games data."""
|
|
if not self.is_enabled: return
|
|
current_time = time.time()
|
|
if current_time - self.last_update < self.update_interval:
|
|
return
|
|
|
|
self.last_update = current_time
|
|
|
|
# Fetch rankings if enabled
|
|
if self.show_ranking:
|
|
self._fetch_team_rankings()
|
|
|
|
try:
|
|
data = self._fetch_data() # Uses shared cache
|
|
if not data or 'events' not in data:
|
|
self.logger.warning("No events found in shared data.") # Changed log prefix
|
|
if not self.games_list: self.current_game = None
|
|
return
|
|
|
|
events = data['events']
|
|
# self.logger.info(f"Processing {len(events)} events from shared data.") # Changed log prefix
|
|
|
|
processed_games = []
|
|
favorite_games_found = 0
|
|
all_upcoming_games = 0 # Count all upcoming games regardless of favorites
|
|
|
|
for event in events:
|
|
game = self._extract_game_details(event)
|
|
# Count all upcoming games for debugging
|
|
if game and game['is_upcoming']:
|
|
all_upcoming_games += 1
|
|
|
|
# Filter criteria: must be upcoming ('pre' state)
|
|
if game and game['is_upcoming']:
|
|
# Only fetch odds for games that will be displayed
|
|
if self.show_favorite_teams_only:
|
|
if not self.favorite_teams:
|
|
continue
|
|
if game['home_abbr'] not in self.favorite_teams and game['away_abbr'] not in self.favorite_teams:
|
|
continue
|
|
processed_games.append(game)
|
|
# Count favorite team games for logging
|
|
if (game['home_abbr'] in self.favorite_teams or
|
|
game['away_abbr'] in self.favorite_teams):
|
|
favorite_games_found += 1
|
|
# Odds are NOT fetched here -- see after selection below.
|
|
|
|
# Enhanced logging for debugging
|
|
self.logger.info(f"Found {all_upcoming_games} total upcoming games in data")
|
|
self.logger.info(f"Found {len(processed_games)} upcoming games after filtering")
|
|
|
|
if processed_games:
|
|
for game in processed_games[:3]: # Show first 3
|
|
self.logger.info(f" {game['away_abbr']}@{game['home_abbr']} - {game['start_time_utc']}")
|
|
|
|
if self.favorite_teams and all_upcoming_games > 0:
|
|
self.logger.info(f"Favorite teams: {self.favorite_teams}")
|
|
self.logger.info(f"Found {favorite_games_found} favorite team upcoming games")
|
|
|
|
# Filter for favorite teams only if the config is set
|
|
if self.show_favorite_teams_only:
|
|
# Select N games per favorite team (where N = upcoming_games_to_show)
|
|
# Example: upcoming_games_to_show=2 with 3 favorite teams = up to 6 games total
|
|
team_games = []
|
|
for team in self.favorite_teams:
|
|
# Find games where this team is playing
|
|
if team_specific_games := [game for game in processed_games if game['home_abbr'] == team or game['away_abbr'] == team]:
|
|
# Sort by game time and take the earliest N games
|
|
team_specific_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
|
|
# Take up to upcoming_games_to_show games for this team
|
|
team_games.extend(team_specific_games[:self.upcoming_games_to_show])
|
|
|
|
# Sort the final list by game time (earliest first)
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
|
|
# Remove duplicates (in case a game involves multiple favorite teams)
|
|
seen_ids = set()
|
|
unique_team_games = []
|
|
for game in team_games:
|
|
if game['id'] not in seen_ids:
|
|
seen_ids.add(game['id'])
|
|
unique_team_games.append(game)
|
|
team_games = unique_team_games
|
|
else:
|
|
team_games = processed_games # Show all upcoming if no favorites
|
|
# Sort by game time, earliest first
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.max.replace(tzinfo=timezone.utc))
|
|
# Limit to the specified number of upcoming games
|
|
team_games = team_games[:self.upcoming_games_to_show]
|
|
|
|
# Odds are fetched here, for the games that survived selection,
|
|
# rather than in the loop that collects them. That loop walks every
|
|
# upcoming game in the schedule window, and for a college league
|
|
# the window is enormous -- a live rig logged 946 upcoming games in
|
|
# one cycle and displayed 1 of them. The comment up there claimed
|
|
# odds were fetched "only for games that will be displayed", but
|
|
# the only narrowing it applied was show_favorite_teams_only, which
|
|
# is not the default; in the usual case nothing narrowed it at all
|
|
# and every game cost a separate ESPN request on a Pi that is also
|
|
# driving the panel.
|
|
if self.show_odds:
|
|
for game in team_games:
|
|
self._fetch_odds(game)
|
|
|
|
# Log changes or periodically
|
|
should_log = (
|
|
current_time - self.last_log_time >= self.log_interval or
|
|
len(team_games) != len(self.games_list) or
|
|
any(g1['id'] != g2.get('id') for g1, g2 in zip(self.games_list, team_games)) or
|
|
(not self.games_list and team_games)
|
|
)
|
|
|
|
# Check if the list of games to display has changed
|
|
new_game_ids = {g['id'] for g in team_games}
|
|
current_game_ids = {g['id'] for g in self.games_list}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.logger.info(f"Found {len(team_games)} upcoming games within window for display.") # Changed log prefix
|
|
self.games_list = team_games
|
|
if not self.current_game or not self.games_list or self.current_game['id'] not in new_game_ids:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0] if self.games_list else None
|
|
self.last_game_switch = current_time
|
|
else:
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.games_list) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
except StopIteration:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0]
|
|
self.last_game_switch = current_time
|
|
|
|
elif self.games_list:
|
|
self.current_game = self.games_list[self.current_game_index] # Update data
|
|
|
|
if not self.games_list:
|
|
self.logger.info("No relevant upcoming games found to display.") # Changed log prefix
|
|
self.current_game = None
|
|
|
|
if should_log and not self.games_list:
|
|
# Log favorite teams only if no games are found and logging is needed
|
|
self.logger.debug(f"Favorite teams: {self.favorite_teams}") # Changed log prefix
|
|
self.logger.debug(f"Total upcoming games before filtering: {len(processed_games)}") # Changed log prefix
|
|
self.last_log_time = current_time
|
|
elif should_log:
|
|
self.last_log_time = current_time
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error updating upcoming games: {e}", exc_info=True) # Changed log prefix
|
|
# self.current_game = None # Decide if clear on error
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Draw the layout for an upcoming NCAA FB game.""" # Updated docstring
|
|
try:
|
|
main_img = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 255))
|
|
overlay = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 0))
|
|
draw_overlay = ImageDraw.Draw(overlay)
|
|
|
|
home_logo = self._load_and_resize_logo(game["home_id"], game["home_abbr"], game["home_logo_path"], game.get("home_logo_url"))
|
|
away_logo = self._load_and_resize_logo(game["away_id"], game["away_abbr"], game["away_logo_path"], game.get("away_logo_url"))
|
|
|
|
if not home_logo or not away_logo:
|
|
self.logger.error(f"Failed to load logos for game: {game.get('id')}") # Changed log prefix
|
|
draw_final = ImageDraw.Draw(main_img.convert('RGB'))
|
|
self._draw_text_with_outline(draw_final, "Logo Error", (5,5), self.fonts['status'])
|
|
self.display_manager.image.paste(main_img.convert('RGB'), (0, 0))
|
|
self.display_manager.update_display()
|
|
return
|
|
|
|
center_y = self.display_height // 2
|
|
|
|
# MLB-style logo positions
|
|
home_x = self.display_width - home_logo.width + 2
|
|
home_y = center_y - (home_logo.height // 2)
|
|
main_img.paste(home_logo, (home_x, home_y), home_logo)
|
|
|
|
away_x = -2
|
|
away_y = center_y - (away_logo.height // 2)
|
|
main_img.paste(away_logo, (away_x, away_y), away_logo)
|
|
|
|
# Draw Text Elements on Overlay
|
|
game_date = game.get("game_date", "")
|
|
game_time = game.get("game_time", "")
|
|
|
|
# Note: Rankings are now handled in the records/rankings section below
|
|
|
|
# "Next Game" at the top (use smaller status font)
|
|
status_font = self.fonts['status']
|
|
if self.display_width > 128:
|
|
status_font = self.fonts['time']
|
|
status_text = "Next Game"
|
|
status_width = draw_overlay.textlength(status_text, font=status_font)
|
|
status_x = (self.display_width - status_width) // 2
|
|
status_y = 1 # Changed from 2
|
|
self._draw_text_with_outline(draw_overlay, status_text, (status_x, status_y), status_font)
|
|
|
|
# Date text (centered, below "Next Game")
|
|
date_width = draw_overlay.textlength(game_date, font=self.fonts['time'])
|
|
date_x = (self.display_width - date_width) // 2
|
|
# Adjust Y position to stack date and time nicely
|
|
date_y = center_y - 7 # Raise date slightly
|
|
self._draw_text_with_outline(draw_overlay, game_date, (date_x, date_y), self.fonts['time'])
|
|
|
|
# Time text (centered, below Date)
|
|
time_width = draw_overlay.textlength(game_time, font=self.fonts['time'])
|
|
time_x = (self.display_width - time_width) // 2
|
|
time_y = date_y + 9 # Place time below date
|
|
self._draw_text_with_outline(draw_overlay, game_time, (time_x, time_y), self.fonts['time'])
|
|
|
|
# Draw odds if available
|
|
if 'odds' in game and game['odds']:
|
|
self._draw_dynamic_odds(draw_overlay, game['odds'], self.display_width, self.display_height)
|
|
|
|
# Draw records or rankings if enabled
|
|
if self.show_records or self.show_ranking:
|
|
record_font = self.fonts.get('detail', ImageFont.load_default())
|
|
|
|
# Get team abbreviations
|
|
away_abbr = game.get('away_abbr', '')
|
|
home_abbr = game.get('home_abbr', '')
|
|
|
|
record_bbox = draw_overlay.textbbox((0,0), "0-0", font=record_font)
|
|
record_height = record_bbox[3] - record_bbox[1]
|
|
record_y = self.display_height - record_height
|
|
self.logger.debug(f"Record positioning: height={record_height}, record_y={record_y}, display_height={self.display_height}")
|
|
|
|
# Display away team info
|
|
if away_abbr:
|
|
if self.show_ranking and self.show_records:
|
|
# When both rankings and records are enabled, rankings replace records completely
|
|
away_rank = self._team_rankings_cache.get(away_abbr, 0)
|
|
if away_rank > 0:
|
|
away_text = f"#{away_rank}"
|
|
else:
|
|
# Show nothing for unranked teams when rankings are prioritized
|
|
away_text = ''
|
|
elif self.show_ranking:
|
|
# Show ranking only if available
|
|
away_rank = self._team_rankings_cache.get(away_abbr, 0)
|
|
if away_rank > 0:
|
|
away_text = f"#{away_rank}"
|
|
else:
|
|
away_text = ''
|
|
elif self.show_records:
|
|
# Show record only when rankings are disabled
|
|
away_text = game.get('away_record', '')
|
|
else:
|
|
away_text = ''
|
|
|
|
if away_text:
|
|
away_record_x = 0
|
|
self.logger.debug(f"Drawing away ranking '{away_text}' at ({away_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
|
|
self._draw_text_with_outline(draw_overlay, away_text, (away_record_x, record_y), record_font)
|
|
|
|
# Display home team info
|
|
if home_abbr:
|
|
if self.show_ranking and self.show_records:
|
|
# When both rankings and records are enabled, rankings replace records completely
|
|
home_rank = self._team_rankings_cache.get(home_abbr, 0)
|
|
if home_rank > 0:
|
|
home_text = f"#{home_rank}"
|
|
else:
|
|
# Show nothing for unranked teams when rankings are prioritized
|
|
home_text = ''
|
|
elif self.show_ranking:
|
|
# Show ranking only if available
|
|
home_rank = self._team_rankings_cache.get(home_abbr, 0)
|
|
if home_rank > 0:
|
|
home_text = f"#{home_rank}"
|
|
else:
|
|
home_text = ''
|
|
elif self.show_records:
|
|
# Show record only when rankings are disabled
|
|
home_text = game.get('home_record', '')
|
|
else:
|
|
home_text = ''
|
|
|
|
if home_text:
|
|
home_record_bbox = draw_overlay.textbbox((0,0), home_text, font=record_font)
|
|
home_record_width = home_record_bbox[2] - home_record_bbox[0]
|
|
home_record_x = self.display_width - home_record_width
|
|
self.logger.debug(f"Drawing home ranking '{home_text}' at ({home_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
|
|
self._draw_text_with_outline(draw_overlay, home_text, (home_record_x, record_y), record_font)
|
|
|
|
# Composite and display
|
|
main_img = Image.alpha_composite(main_img, overlay)
|
|
main_img = main_img.convert('RGB')
|
|
self.display_manager.image.paste(main_img, (0, 0))
|
|
self.display_manager.update_display() # Update display here
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error displaying upcoming game: {e}", exc_info=True) # Changed log prefix
|
|
|
|
def display(self, force_clear=False) -> bool:
|
|
"""Display upcoming games, handling switching."""
|
|
if not self.is_enabled: return False
|
|
|
|
if not self.games_list:
|
|
if self.current_game: self.current_game = None # Clear state if list empty
|
|
current_time = time.time()
|
|
# Log warning periodically if no games found
|
|
if current_time - self.last_warning_time > self.warning_cooldown:
|
|
self.logger.info("No upcoming games found for favorite teams to display.") # Changed log prefix
|
|
self.last_warning_time = current_time
|
|
return False # Skip display update
|
|
|
|
try:
|
|
current_time = time.time()
|
|
|
|
# Check if it's time to switch games
|
|
if len(self.games_list) > 1 and current_time - self.last_game_switch >= self.game_display_duration:
|
|
self.current_game_index = (self.current_game_index + 1) % len(self.games_list)
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
self.last_game_switch = current_time
|
|
force_clear = True # Force redraw on switch
|
|
|
|
# Log team switching with sport prefix
|
|
if self.current_game:
|
|
away_abbr = self.current_game.get('away_abbr', 'UNK')
|
|
home_abbr = self.current_game.get('home_abbr', 'UNK')
|
|
sport_prefix = self.sport_key.upper() if hasattr(self, 'sport_key') else 'SPORT'
|
|
self.logger.info(f"[{sport_prefix} Upcoming] Showing {away_abbr} vs {home_abbr}")
|
|
else:
|
|
self.logger.debug(f"Switched to game index {self.current_game_index}")
|
|
|
|
if self.current_game:
|
|
self._render_game(self.current_game, force_clear)
|
|
return True
|
|
# update_display() is called within _draw_scorebug_layout for upcoming
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error in display loop: {e}", exc_info=True) # Changed log prefix
|
|
return False
|
|
|
|
|
|
class SportsRecent(SportsCore):
|
|
SKIN_MODE = "recent"
|
|
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
|
|
super().__init__(config, display_manager, cache_manager, logger, sport_key)
|
|
self.recent_games = [] # Store all fetched recent games initially
|
|
self.games_list = [] # Filtered list for display (favorite teams)
|
|
self.current_game_index = 0
|
|
self.last_update = 0
|
|
self.update_interval = self.mode_config.get("recent_update_interval", 3600) # Check for recent games every hour
|
|
self.last_game_switch = 0
|
|
self.game_display_duration = 15 # Display each recent game for 15 seconds
|
|
# Tracks when each game was first seen with an expired clock, keyed by
|
|
# game id. Promoted alongside the zero-clock helpers below; without it
|
|
# the first _get_zero_clock_duration() call raises AttributeError.
|
|
self._zero_clock_timestamps: Dict[str, float] = {} # Track games at 0:00
|
|
|
|
# -- Zero-clock tracking ------------------------------------------------
|
|
# Byte-identical in all nine plugin copies. Note that afl/nrl/soccer define
|
|
# these but never call them — their clocks count up, so 0:00 means kickoff
|
|
# rather than expiry (see CLOCK_COUNTS_DOWN on SportsLive). That makes the
|
|
# pair a future `CountdownClockMixin` candidate so it stops appearing in the
|
|
# MRO of sports that cannot use it — B2 work, not now.
|
|
|
|
def _get_zero_clock_duration(self, game_id: str) -> float:
|
|
"""Track how long a game has been at 0:00 clock."""
|
|
current_time = time.time()
|
|
if game_id not in self._zero_clock_timestamps:
|
|
self._zero_clock_timestamps[game_id] = current_time
|
|
return 0.0
|
|
return current_time - self._zero_clock_timestamps[game_id]
|
|
|
|
def _clear_zero_clock_tracking(self, game_id: str) -> None:
|
|
"""Clear tracking when game clock moves away from 0:00 or game ends."""
|
|
if game_id in self._zero_clock_timestamps:
|
|
del self._zero_clock_timestamps[game_id]
|
|
|
|
def _select_recent_games_for_display(
|
|
self, processed_games: List[Dict], favorite_teams: List[str]
|
|
) -> List[Dict]:
|
|
"""
|
|
Single-pass game selection for recent games with proper deduplication.
|
|
|
|
When a game involves two favorite teams, it counts toward BOTH teams' limits.
|
|
Games are sorted by most recent first.
|
|
|
|
Team identity goes through the ``_favorite_key`` override point rather
|
|
than reading ``home_abbr``/``away_abbr`` directly, because abbreviations
|
|
are not unique in every league (NRL matches on team ID instead).
|
|
"""
|
|
sorted_games = sorted(
|
|
processed_games,
|
|
key=lambda g: g.get("start_time_utc")
|
|
or datetime.min.replace(tzinfo=timezone.utc),
|
|
reverse=True,
|
|
)
|
|
|
|
if not favorite_teams:
|
|
return sorted_games
|
|
|
|
selected_games = []
|
|
selected_ids = set()
|
|
team_counts = {team: 0 for team in favorite_teams}
|
|
|
|
for game in sorted_games:
|
|
game_id = game.get("id")
|
|
if game_id in selected_ids:
|
|
continue
|
|
|
|
home = self._favorite_key(game, "home")
|
|
away = self._favorite_key(game, "away")
|
|
|
|
home_fav = home in favorite_teams
|
|
away_fav = away in favorite_teams
|
|
|
|
if not home_fav and not away_fav:
|
|
continue
|
|
|
|
home_needs = home_fav and team_counts[home] < self.recent_games_to_show
|
|
away_needs = away_fav and team_counts[away] < self.recent_games_to_show
|
|
|
|
if home_needs or away_needs:
|
|
selected_games.append(game)
|
|
selected_ids.add(game_id)
|
|
if home_fav:
|
|
team_counts[home] += 1
|
|
if away_fav:
|
|
team_counts[away] += 1
|
|
|
|
self.logger.debug(
|
|
f"Selected recent game {away}@{home}: team_counts={team_counts}"
|
|
)
|
|
|
|
if all(c >= self.recent_games_to_show for c in team_counts.values()):
|
|
self.logger.debug("All favorite teams satisfied, stopping selection")
|
|
break
|
|
|
|
self.logger.info(
|
|
f"Selected {len(selected_games)} recent games for {len(favorite_teams)} "
|
|
f"favorite teams: {team_counts}"
|
|
)
|
|
return selected_games
|
|
|
|
def update(self):
|
|
"""Update recent games data."""
|
|
if not self.is_enabled: return
|
|
current_time = time.time()
|
|
if current_time - self.last_update < self.update_interval:
|
|
return
|
|
|
|
self.last_update = current_time # Update time even if fetch fails
|
|
|
|
# Fetch rankings if enabled
|
|
if self.show_ranking:
|
|
self._fetch_team_rankings()
|
|
|
|
try:
|
|
data = self._fetch_data() # Uses shared cache
|
|
if not data or 'events' not in data:
|
|
self.logger.warning("No events found in shared data.") # Changed log prefix
|
|
if not self.games_list:
|
|
self.current_game = None # Clear display if no games were showing
|
|
return
|
|
|
|
events = data['events']
|
|
self.logger.info(f"Processing {len(events)} events from shared data.") # Changed log prefix
|
|
|
|
# Define date range for "recent" games (last 21 days to capture games from 3 weeks ago)
|
|
now = datetime.now(timezone.utc)
|
|
recent_cutoff = now - timedelta(days=21)
|
|
self.logger.info(f"Current time: {now}, Recent cutoff: {recent_cutoff} (21 days ago)")
|
|
|
|
# Process games and filter for final games, date range & favorite teams
|
|
processed_games = []
|
|
for event in events:
|
|
game = self._extract_game_details(event)
|
|
# Filter criteria: must be final AND within recent date range
|
|
if game and game['is_final']:
|
|
game_time = game.get('start_time_utc')
|
|
if game_time and game_time >= recent_cutoff:
|
|
processed_games.append(game)
|
|
# Filter for favorite teams only if the config is set
|
|
if self.show_favorite_teams_only:
|
|
# Get all games involving favorite teams
|
|
favorite_team_games = [game for game in processed_games
|
|
if game['home_abbr'] in self.favorite_teams or
|
|
game['away_abbr'] in self.favorite_teams]
|
|
self.logger.info(f"Found {len(favorite_team_games)} favorite team games out of {len(processed_games)} total final games within last 21 days")
|
|
|
|
# Select N games per favorite team (where N = recent_games_to_show)
|
|
# Example: recent_games_to_show=1 with 2 favorite teams = 2 games total
|
|
team_games = []
|
|
for team in self.favorite_teams:
|
|
# Find games where this team is playing
|
|
team_specific_games = [game for game in favorite_team_games
|
|
if game['home_abbr'] == team or game['away_abbr'] == team]
|
|
|
|
if team_specific_games:
|
|
# Sort by game time and take the most recent N games
|
|
team_specific_games.sort(key=lambda g: g.get('start_time_utc') or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
|
|
# Take up to recent_games_to_show games for this team
|
|
team_games.extend(team_specific_games[:self.recent_games_to_show])
|
|
|
|
# Sort the final list by game time (most recent first)
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
|
|
# Remove duplicates (in case a game involves multiple favorite teams)
|
|
seen_ids = set()
|
|
unique_team_games = []
|
|
for game in team_games:
|
|
if game['id'] not in seen_ids:
|
|
seen_ids.add(game['id'])
|
|
unique_team_games.append(game)
|
|
team_games = unique_team_games
|
|
|
|
# Debug: Show which games are selected for display
|
|
for i, game in enumerate(team_games):
|
|
self.logger.info(f"Game {i+1} for display: {game['away_abbr']} @ {game['home_abbr']} - {game.get('start_time_utc')} - Score: {game['away_score']}-{game['home_score']}")
|
|
else:
|
|
team_games = processed_games # Show all recent games if no favorites defined
|
|
self.logger.info(f"Found {len(processed_games)} total final games within last 21 days (no favorite teams filtering)")
|
|
# Sort games by start time, most recent first, and limit to recent_games_to_show
|
|
team_games.sort(key=lambda g: g.get('start_time_utc') or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
|
|
team_games = team_games[:self.recent_games_to_show]
|
|
|
|
# Check if the list of games to display has changed
|
|
new_game_ids = {g['id'] for g in team_games}
|
|
current_game_ids = {g['id'] for g in self.games_list}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.logger.info(f"Found {len(team_games)} final games within window for display.") # Changed log prefix
|
|
self.games_list = team_games
|
|
# Reset index if list changed or current game removed
|
|
if not self.current_game or not self.games_list or self.current_game['id'] not in new_game_ids:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0] if self.games_list else None
|
|
self.last_game_switch = current_time # Reset switch timer
|
|
else:
|
|
# Try to maintain position if possible
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.games_list) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.games_list[self.current_game_index] # Update data just in case
|
|
except StopIteration:
|
|
self.current_game_index = 0
|
|
self.current_game = self.games_list[0]
|
|
self.last_game_switch = current_time
|
|
|
|
elif self.games_list:
|
|
# List content is same, just update data for current game
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
|
|
if not self.games_list:
|
|
self.logger.info("No relevant recent games found to display.") # Changed log prefix
|
|
self.current_game = None # Ensure display clears if no games
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error updating recent games: {e}", exc_info=True) # Changed log prefix
|
|
# Don't clear current game on error, keep showing last known state
|
|
# self.current_game = None # Decide if we want to clear display on error
|
|
|
|
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
|
|
"""Draw the layout for a recently completed NCAA FB game.""" # Updated docstring
|
|
try:
|
|
main_img = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 255))
|
|
overlay = Image.new('RGBA', (self.display_width, self.display_height), (0, 0, 0, 0))
|
|
draw_overlay = ImageDraw.Draw(overlay)
|
|
|
|
home_logo = self._load_and_resize_logo(game["home_id"], game["home_abbr"], game["home_logo_path"], game.get("home_logo_url"))
|
|
away_logo = self._load_and_resize_logo(game["away_id"], game["away_abbr"], game["away_logo_path"], game.get("away_logo_url"))
|
|
|
|
if not home_logo or not away_logo:
|
|
self.logger.error(f"Failed to load logos for game: {game.get('id')}") # Changed log prefix
|
|
# Draw placeholder text if logos fail (similar to live)
|
|
draw_final = ImageDraw.Draw(main_img.convert('RGB'))
|
|
self._draw_text_with_outline(draw_final, "Logo Error", (5,5), self.fonts['status'])
|
|
self.display_manager.image.paste(main_img.convert('RGB'), (0, 0))
|
|
self.display_manager.update_display()
|
|
return
|
|
|
|
center_y = self.display_height // 2
|
|
|
|
# MLB-style logo positioning (closer to edges)
|
|
home_x = self.display_width - home_logo.width + 2
|
|
home_y = center_y - (home_logo.height // 2)
|
|
main_img.paste(home_logo, (home_x, home_y), home_logo)
|
|
|
|
away_x = -2
|
|
away_y = center_y - (away_logo.height // 2)
|
|
main_img.paste(away_logo, (away_x, away_y), away_logo)
|
|
|
|
# Draw Text Elements on Overlay
|
|
# Note: Rankings are now handled in the records/rankings section below
|
|
|
|
# Final Scores (Centered, same position as live)
|
|
home_score = str(game.get("home_score", "0"))
|
|
away_score = str(game.get("away_score", "0"))
|
|
score_text = f"{away_score}-{home_score}"
|
|
score_width = draw_overlay.textlength(score_text, font=self.fonts['score'])
|
|
score_x = (self.display_width - score_width) // 2
|
|
score_y = self.display_height - 14
|
|
self._draw_text_with_outline(draw_overlay, score_text, (score_x, score_y), self.fonts['score'])
|
|
|
|
# "Final" text (Top center)
|
|
status_text = game.get("period_text", "Final") # Use formatted period text (e.g., "Final/OT") or default "Final"
|
|
status_width = draw_overlay.textlength(status_text, font=self.fonts['time'])
|
|
status_x = (self.display_width - status_width) // 2
|
|
status_y = 1
|
|
self._draw_text_with_outline(draw_overlay, status_text, (status_x, status_y), self.fonts['time'])
|
|
|
|
# Draw odds if available
|
|
if 'odds' in game and game['odds']:
|
|
self._draw_dynamic_odds(draw_overlay, game['odds'], self.display_width, self.display_height)
|
|
|
|
# Draw records or rankings if enabled
|
|
if self.show_records or self.show_ranking:
|
|
record_font = self.fonts.get('detail', ImageFont.load_default())
|
|
|
|
# Get team abbreviations
|
|
away_abbr = game.get('away_abbr', '')
|
|
home_abbr = game.get('home_abbr', '')
|
|
|
|
record_bbox = draw_overlay.textbbox((0,0), "0-0", font=record_font)
|
|
record_height = record_bbox[3] - record_bbox[1]
|
|
record_y = self.display_height - record_height
|
|
self.logger.debug(f"Record positioning: height={record_height}, record_y={record_y}, display_height={self.display_height}")
|
|
|
|
# Display away team info
|
|
if away_abbr:
|
|
if self.show_ranking and self.show_records:
|
|
# When both rankings and records are enabled, rankings replace records completely
|
|
away_rank = self._team_rankings_cache.get(away_abbr, 0)
|
|
if away_rank > 0:
|
|
away_text = f"#{away_rank}"
|
|
else:
|
|
# Show nothing for unranked teams when rankings are prioritized
|
|
away_text = ''
|
|
elif self.show_ranking:
|
|
# Show ranking only if available
|
|
away_rank = self._team_rankings_cache.get(away_abbr, 0)
|
|
if away_rank > 0:
|
|
away_text = f"#{away_rank}"
|
|
else:
|
|
away_text = ''
|
|
elif self.show_records:
|
|
# Show record only when rankings are disabled
|
|
away_text = game.get('away_record', '')
|
|
else:
|
|
away_text = ''
|
|
|
|
if away_text:
|
|
away_record_x = 0
|
|
self.logger.debug(f"Drawing away ranking '{away_text}' at ({away_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
|
|
self._draw_text_with_outline(draw_overlay, away_text, (away_record_x, record_y), record_font)
|
|
|
|
# Display home team info
|
|
if home_abbr:
|
|
if self.show_ranking and self.show_records:
|
|
# When both rankings and records are enabled, rankings replace records completely
|
|
home_rank = self._team_rankings_cache.get(home_abbr, 0)
|
|
if home_rank > 0:
|
|
home_text = f"#{home_rank}"
|
|
else:
|
|
# Show nothing for unranked teams when rankings are prioritized
|
|
home_text = ''
|
|
elif self.show_ranking:
|
|
# Show ranking only if available
|
|
home_rank = self._team_rankings_cache.get(home_abbr, 0)
|
|
if home_rank > 0:
|
|
home_text = f"#{home_rank}"
|
|
else:
|
|
home_text = ''
|
|
elif self.show_records:
|
|
# Show record only when rankings are disabled
|
|
home_text = game.get('home_record', '')
|
|
else:
|
|
home_text = ''
|
|
|
|
if home_text:
|
|
home_record_bbox = draw_overlay.textbbox((0,0), home_text, font=record_font)
|
|
home_record_width = home_record_bbox[2] - home_record_bbox[0]
|
|
home_record_x = self.display_width - home_record_width
|
|
self.logger.debug(f"Drawing home ranking '{home_text}' at ({home_record_x}, {record_y}) with font size {record_font.size if hasattr(record_font, 'size') else 'unknown'}")
|
|
self._draw_text_with_outline(draw_overlay, home_text, (home_record_x, record_y), record_font)
|
|
|
|
self._custom_scorebug_layout(game, draw_overlay)
|
|
# Composite and display
|
|
main_img = Image.alpha_composite(main_img, overlay)
|
|
main_img = main_img.convert('RGB')
|
|
self.display_manager.image.paste(main_img, (0, 0))
|
|
self.display_manager.update_display() # Update display here
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error displaying recent game: {e}", exc_info=True) # Changed log prefix
|
|
|
|
def display(self, force_clear=False) -> bool:
|
|
"""Display recent games, handling switching."""
|
|
if not self.is_enabled or not self.games_list:
|
|
# If disabled or no games, ensure display might be cleared by main loop if needed
|
|
# Or potentially clear it here? For now, rely on main loop/other managers.
|
|
if not self.games_list and self.current_game:
|
|
self.current_game = None # Clear internal state if list becomes empty
|
|
return False
|
|
|
|
try:
|
|
current_time = time.time()
|
|
|
|
# Check if it's time to switch games
|
|
if len(self.games_list) > 1 and current_time - self.last_game_switch >= self.game_display_duration:
|
|
self.current_game_index = (self.current_game_index + 1) % len(self.games_list)
|
|
self.current_game = self.games_list[self.current_game_index]
|
|
self.last_game_switch = current_time
|
|
force_clear = True # Force redraw on switch
|
|
|
|
# Log team switching with sport prefix
|
|
if self.current_game:
|
|
away_abbr = self.current_game.get('away_abbr', 'UNK')
|
|
home_abbr = self.current_game.get('home_abbr', 'UNK')
|
|
sport_prefix = self.sport_key.upper() if hasattr(self, 'sport_key') else 'SPORT'
|
|
self.logger.info(f"[{sport_prefix} Recent] Showing {away_abbr} vs {home_abbr}")
|
|
else:
|
|
self.logger.debug(f"Switched to game index {self.current_game_index}")
|
|
|
|
if self.current_game:
|
|
self._render_game(self.current_game, force_clear)
|
|
return True
|
|
# update_display() is called within _draw_scorebug_layout for recent
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error in display loop: {e}", exc_info=True) # Changed log prefix
|
|
return False
|
|
|
|
class SportsLive(SportsCore):
|
|
# Per-sport constants for the "is this live game actually over?" check.
|
|
# These are values, not behavior, so they are class attributes rather than
|
|
# override points (see docs/SPORTS_UNIFICATION.md "Override points").
|
|
#
|
|
# FINAL_PERIOD: the period at/after which an expired clock can mean "over".
|
|
# 4 for four-quarter sports; hockey overrides to 3.
|
|
# CLOCK_COUNTS_DOWN: whether "0:00" means the clock expired. False for
|
|
# sports whose clock counts up (soccer/afl/nrl), where 0:00 is kickoff —
|
|
# running the expiry branch there would evict games that just started.
|
|
FINAL_PERIOD = 4
|
|
CLOCK_COUNTS_DOWN = True
|
|
|
|
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
|
|
super().__init__(config, display_manager, cache_manager, logger, sport_key)
|
|
self.update_interval = self.mode_config.get("live_update_interval", 15)
|
|
self.no_data_interval = 300
|
|
self.last_update = 0
|
|
self.live_games = []
|
|
self.current_game_index = 0
|
|
self.last_game_switch = 0 # Will be set to current_time when games are first loaded
|
|
self.game_display_duration = self.mode_config.get("live_game_duration", 20)
|
|
self.last_display_update = 0
|
|
self.last_log_time = 0
|
|
self.log_interval = 300
|
|
self.last_count_log_time = 0 # Track when we last logged count data
|
|
self.count_log_interval = 5 # Only log count data every 5 seconds
|
|
# Initialize test_mode - defaults to False (live mode)
|
|
self.test_mode = self.mode_config.get("test_mode", False)
|
|
# Freshness bookkeeping for _detect_stale_games(). The base class only
|
|
# *reads* this map; a subclass's update() stamps entries as it ingests a
|
|
# feed: {game_id: {"clock": ts, "score": ts, "last_seen": ts}}.
|
|
# Until a subclass writes "last_seen", the staleness branch of
|
|
# _detect_stale_games is inert and only the game-over check applies.
|
|
self.game_update_timestamps: Dict[str, Dict[str, float]] = {}
|
|
self.stale_game_timeout = self.mode_config.get("stale_game_timeout", 300) # 5 minutes default
|
|
|
|
@abstractmethod
|
|
def _test_mode_update(self) -> None:
|
|
return
|
|
|
|
def _is_game_really_over(self, game: Dict) -> bool:
|
|
"""Check if a game appears to be over even if API says it's live.
|
|
|
|
Two independent signals:
|
|
1. ``period_text`` says "final" — universal across every sport.
|
|
2. The clock has expired at/after :attr:`FINAL_PERIOD` — only meaningful
|
|
where :attr:`CLOCK_COUNTS_DOWN` is true.
|
|
|
|
Fails *safe*: anything ambiguous returns False and the game keeps being
|
|
displayed. The only caller, :meth:`_detect_stale_games`, removes games
|
|
on a True, so a false positive silently drops a live game.
|
|
"""
|
|
game_str = f"{game.get('away_abbr')}@{game.get('home_abbr')}"
|
|
|
|
# `period_text` may be present-but-None; `or ""` keeps that from raising
|
|
# AttributeError — the caller has no try/except around this call.
|
|
period_text = (game.get("period_text") or "").lower()
|
|
if "final" in period_text:
|
|
self.logger.debug(
|
|
f"_is_game_really_over({game_str}): "
|
|
f"returning True - 'final' in period_text='{period_text}'"
|
|
)
|
|
return True
|
|
|
|
if not self.CLOCK_COUNTS_DOWN:
|
|
# Count-up clock: 0:00 means the match has not started.
|
|
self.logger.debug(
|
|
f"_is_game_really_over({game_str}): returning False "
|
|
f"(count-up clock, period_text='{period_text}')"
|
|
)
|
|
return False
|
|
|
|
raw_clock = game.get("clock")
|
|
# `or 0` rather than a get() default: feeds routinely send an explicit
|
|
# null period, and `None >= FINAL_PERIOD` raises TypeError — which would
|
|
# take down the whole live-update pass, since the only caller
|
|
# (_detect_stale_games) has no try/except around it.
|
|
period = game.get("period") or 0
|
|
|
|
# Only check clock-based finish if we have a valid clock string. A
|
|
# missing or non-string clock is NOT coerced to "0:00": sports without a
|
|
# game clock (e.g. baseball, where `period` is the inning) would
|
|
# otherwise be declared over from the FINAL_PERIOD-th period onward.
|
|
if isinstance(raw_clock, str) and raw_clock.strip() and period >= self.FINAL_PERIOD:
|
|
clock = raw_clock
|
|
# Compare numerically rather than against a literal set: feeds spell
|
|
# an expired clock "0:00", ":00" and "00:00" depending on sport, and
|
|
# a membership test silently misses every spelling not listed.
|
|
clock_normalized = clock.replace(":", "").strip()
|
|
if clock_normalized.isdigit() and int(clock_normalized) == 0:
|
|
self.logger.debug(
|
|
f"_is_game_really_over({game_str}): "
|
|
f"returning True - clock at 0:00 (clock='{clock}', period={period})"
|
|
)
|
|
return True
|
|
|
|
self.logger.debug(
|
|
f"_is_game_really_over({game_str}): returning False"
|
|
)
|
|
return False
|
|
|
|
def _detect_stale_games(self, games: List[Dict]) -> None:
|
|
"""Remove games that appear stale or haven't updated.
|
|
|
|
Mutates ``games`` **in place** and returns None. Removal is by value
|
|
(``list.remove`` uses ``dict.__eq__``), so two structurally-equal game
|
|
dicts in the same list would drop the first occurrence.
|
|
"""
|
|
current_time = time.time()
|
|
|
|
for game in games[:]: # Copy list to iterate safely
|
|
game_id = game.get("id")
|
|
if not game_id:
|
|
continue
|
|
|
|
# Check if game data is stale
|
|
timestamps = self.game_update_timestamps.get(game_id, {})
|
|
last_seen = timestamps.get("last_seen", 0)
|
|
|
|
if last_seen > 0 and current_time - last_seen > self.stale_game_timeout:
|
|
self.logger.warning(
|
|
f"Removing stale game {game.get('away_abbr')}@{game.get('home_abbr')} "
|
|
f"(last seen {int(current_time - last_seen)}s ago)"
|
|
)
|
|
games.remove(game)
|
|
if game_id in self.game_update_timestamps:
|
|
del self.game_update_timestamps[game_id]
|
|
continue
|
|
|
|
# Also check if game appears to be over
|
|
if self._is_game_really_over(game):
|
|
self.logger.debug(
|
|
f"Removing game that appears over: {game.get('away_abbr')}@{game.get('home_abbr')} "
|
|
f"(clock={game.get('clock')}, period={game.get('period')}, period_text={game.get('period_text')})"
|
|
)
|
|
games.remove(game)
|
|
if game_id in self.game_update_timestamps:
|
|
del self.game_update_timestamps[game_id]
|
|
|
|
def update(self):
|
|
"""Update live game data and handle game switching."""
|
|
if not self.is_enabled:
|
|
return
|
|
|
|
# Define current_time and interval before the problematic line (originally line 455)
|
|
# Ensure 'import time' is present at the top of the file.
|
|
current_time = time.time()
|
|
|
|
# Define interval using a pattern similar to NFLLiveManager's update method.
|
|
# Uses getattr for robustness, assuming attributes for live_games, test_mode,
|
|
# no_data_interval, and update_interval are available on self.
|
|
_live_games_attr = self.live_games
|
|
_test_mode_attr = self.test_mode # test_mode is often from a base class or config
|
|
_no_data_interval_attr = self.no_data_interval # Default similar to NFLLiveManager
|
|
_update_interval_attr = self.update_interval # Default similar to NFLLiveManager
|
|
|
|
interval = _no_data_interval_attr if not _live_games_attr and not _test_mode_attr else _update_interval_attr
|
|
|
|
# Original line from traceback (line 455), now with variables defined:
|
|
if current_time - self.last_update >= interval:
|
|
self.last_update = current_time
|
|
|
|
# Fetch rankings if enabled
|
|
if self.show_ranking:
|
|
self._fetch_team_rankings()
|
|
|
|
if self.test_mode:
|
|
# Simulate clock running down in test mode
|
|
self._test_mode_update()
|
|
else:
|
|
# Fetch live game data
|
|
data = self._fetch_data()
|
|
new_live_games = []
|
|
if data and "events" in data:
|
|
for game in data["events"]:
|
|
details = self._extract_game_details(game)
|
|
if details and (details["is_live"] or details["is_halftime"]):
|
|
# If show_favorite_teams_only is true, only add if it's a favorite.
|
|
# Otherwise, add all games.
|
|
if self.show_all_live or not self.show_favorite_teams_only or (self.show_favorite_teams_only and (details["home_abbr"] in self.favorite_teams or details["away_abbr"] in self.favorite_teams)):
|
|
if self.show_odds:
|
|
self._fetch_odds(details)
|
|
new_live_games.append(details)
|
|
# Log changes or periodically
|
|
current_time_for_log = time.time() # Use a consistent time for logging comparison
|
|
should_log = (
|
|
current_time_for_log - self.last_log_time >= self.log_interval or
|
|
len(new_live_games) != len(self.live_games) or
|
|
any(g1['id'] != g2.get('id') for g1, g2 in zip(self.live_games, new_live_games)) or # Check if game IDs changed
|
|
(not self.live_games and new_live_games) # Log if games appeared
|
|
)
|
|
|
|
if should_log:
|
|
if new_live_games:
|
|
filter_text = "favorite teams" if self.show_favorite_teams_only or self.show_all_live else "all teams"
|
|
self.logger.info(f"Found {len(new_live_games)} live/halftime games for {filter_text}.")
|
|
for game_info in new_live_games: # Renamed game to game_info
|
|
self.logger.info(f" - {game_info['away_abbr']}@{game_info['home_abbr']} ({game_info.get('status_text', 'N/A')})")
|
|
else:
|
|
filter_text = "favorite teams" if self.show_favorite_teams_only or self.show_all_live else "criteria"
|
|
self.logger.info(f"No live/halftime games found for {filter_text}.")
|
|
self.last_log_time = current_time_for_log
|
|
|
|
|
|
# Update game list and current game
|
|
if new_live_games:
|
|
# Check if the games themselves changed, not just scores/time
|
|
new_game_ids = {g['id'] for g in new_live_games}
|
|
current_game_ids = {g['id'] for g in self.live_games}
|
|
|
|
if new_game_ids != current_game_ids:
|
|
self.live_games = sorted(new_live_games, key=lambda g: g.get('start_time_utc') or datetime.now(timezone.utc)) # Sort by start time
|
|
# Reset index if current game is gone or list is new
|
|
if not self.current_game or self.current_game['id'] not in new_game_ids:
|
|
self.current_game_index = 0
|
|
self.current_game = self.live_games[0] if self.live_games else None
|
|
self.last_game_switch = current_time
|
|
else:
|
|
# Find current game's new index if it still exists
|
|
try:
|
|
self.current_game_index = next(i for i, g in enumerate(self.live_games) if g['id'] == self.current_game['id'])
|
|
self.current_game = self.live_games[self.current_game_index] # Update current_game with fresh data
|
|
# Fix: Set last_game_switch if it's still 0 (initialized) to prevent immediate switching
|
|
if self.last_game_switch == 0:
|
|
self.last_game_switch = current_time
|
|
except StopIteration: # Should not happen if check above passed, but safety first
|
|
self.current_game_index = 0
|
|
self.current_game = self.live_games[0]
|
|
self.last_game_switch = current_time
|
|
|
|
else:
|
|
# Just update the data for the existing games
|
|
temp_game_dict = {g['id']: g for g in new_live_games}
|
|
self.live_games = [temp_game_dict.get(g['id'], g) for g in self.live_games] # Update in place
|
|
if self.current_game:
|
|
self.current_game = temp_game_dict.get(self.current_game['id'], self.current_game)
|
|
# Fix: Set last_game_switch if it's still 0 (initialized) to prevent immediate switching
|
|
# This handles the case where games were loaded previously but last_game_switch was never set
|
|
if self.last_game_switch == 0:
|
|
self.last_game_switch = current_time
|
|
|
|
# Display update handled by main loop based on interval
|
|
|
|
else:
|
|
# No live games found
|
|
if self.live_games: # Were there games before?
|
|
self.logger.info("Live games previously showing have ended or are no longer live.") # Changed log prefix
|
|
self.live_games = []
|
|
self.current_game = None
|
|
self.current_game_index = 0
|
|
|
|
else:
|
|
# Error fetching data or no events
|
|
if self.live_games: # Were there games before?
|
|
self.logger.warning("Could not fetch update; keeping existing live game data for now.") # Changed log prefix
|
|
else:
|
|
self.logger.warning("Could not fetch data and no existing live games.") # Changed log prefix
|
|
self.current_game = None # Clear current game if fetch fails and no games were active
|
|
|
|
# Handle game switching (outside test mode check)
|
|
# Fix: Don't check for switching if last_game_switch is still 0 (games haven't been loaded yet)
|
|
# This prevents immediate switching when the system has been running for a while before games load
|
|
if not self.test_mode and len(self.live_games) > 1 and self.last_game_switch > 0 and (current_time - self.last_game_switch) >= self.game_display_duration:
|
|
self.current_game_index = (self.current_game_index + 1) % len(self.live_games)
|
|
self.current_game = self.live_games[self.current_game_index]
|
|
self.last_game_switch = current_time
|
|
self.logger.info(f"Switched live view to: {self.current_game['away_abbr']}@{self.current_game['home_abbr']}") # Changed log prefix
|
|
# Force display update via flag or direct call if needed, but usually let main loop handle
|