From bdbc55a42558404306bb937d8618fc4705d0ff2a Mon Sep 17 00:00:00 2001 From: death916 Date: Fri, 31 Oct 2025 04:15:26 -0700 Subject: [PATCH] nfl stats --- deathclock/deathclock.py | 379 +++++++++++++++++++++++---------------- utils/scores.py | 197 +++++++++++++++----- 2 files changed, 381 insertions(+), 195 deletions(-) diff --git a/deathclock/deathclock.py b/deathclock/deathclock.py index 4ee905c..74a4179 100755 --- a/deathclock/deathclock.py +++ b/deathclock/deathclock.py @@ -1,50 +1,55 @@ -# deathclock.py (or your main app file name) +# deathclock.py +import asyncio + +# from utils.alarm import Alarm # Commented out import +import logging +import time +from datetime import datetime, timezone + +# --- Import typing for hints --- +from typing import Any, Dict, List import reflex as rx -from datetime import datetime, timezone -import asyncio -import time -# --- Import typing for hints --- -from typing import List, Dict, Any -from rxconfig import config + +from utils.news import News +from utils.scores import NBAScores, mlbScores + # --- Import your Weather utility --- from utils.weather import Weather -from utils.scores import NBAScores, mlbScores -from utils.news import News -# from utils.alarm import Alarm # Commented out import -import logging - # --- Constants --- -WEATHER_IMAGE_PATH = "/weather.jpg" # Web path in assets folder +WEATHER_IMAGE_PATH = "/weather.jpg" # Web path in assets folder WEATHER_FETCH_INTERVAL = 360 -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) + class State(rx.State): - # --- State Variables --- - current_time: str = "" # Note: rx.moment replaces the need for this if used for display + current_time: str = ( + "" # Note: rx.moment replaces the need for this if used for display + ) alarm_time: str = "" alarms: list = [] news: List[Dict[str, Any]] = [] nba_scores: List[Dict[str, Any]] = [] mlb_scores: List[Dict[str, Any]] = [] - _news_client: News | None = None # This will be set in the constructor + nfl_scores: List[Dict[str, Any]] = [] + _news_client: News | None = None # This will be set in the constructor last_weather_update: str = "Never" weather_img: str = WEATHER_IMAGE_PATH - _weather_client: Weather | None = None # This will be set in the constructor + _weather_client: Weather | None = None # This will be set in the constructor _mlb_client: mlbScores | None = None _nba_client: NBAScores | None = None last_sports_update: float = 0.0 - # --- Initialize Utility Client --- def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Initialize background clients try: - self._weather_client = Weather() self._news_client = News() self._mlb_client = mlbScores() @@ -52,9 +57,9 @@ class State(rx.State): logging.info("Weather client initialized successfully.") except Exception as e: logging.error(f"Failed to initialize Weather client: {e}", exc_info=True) - self._weather_client = None # Mark as unusable + self._weather_client = None # Set error state if needed - self.weather_img = "/error_placeholder.png" # Provide a placeholder error image + self.weather_img = "/error_placeholder.png" self.last_weather_update = "Client Init Error" self.mlb_scores = "" self.nba_scores = "" @@ -63,7 +68,9 @@ class State(rx.State): # --- on_load Handler --- async def start_background_tasks(self): """Starts the weather background task when the page loads.""" - rx.remove_local_storage("chakra-ui-color-mode") # trying to test themes remove after + rx.remove_local_storage( + "chakra-ui-color-mode" + ) # trying to test themes remove after logging.info("Triggering background tasks: Weather") # Return a list containing the handler references return [State.fetch_weather, State.fetch_sports] @@ -72,15 +79,19 @@ class State(rx.State): @rx.event(background=True) async def fetch_sports(self): - # Fetches sports scores periodically while True: try: logging.info("Fetching sports scores...") # Fetch MLB and NBA scores # check if sports has updated in last 5 minutes if so skip - if self.last_sports_update and (time.time() - self.last_sports_update) < 300: - logging.info("Sports scores already updated within the last 5 minutes. Skipping fetch.") + if ( + self.last_sports_update + and (time.time() - self.last_sports_update) < 300 + ): + logging.info( + "Sports scores already updated within the last 5 minutes. Skipping fetch." + ) await asyncio.sleep(300) continue @@ -105,19 +116,25 @@ class State(rx.State): async with self: self.mlb_scores = mlb_scores self.nba_scores = nba_scores - self.last_sports_update = time.time() # Update last sports update time - logging.info(f"Fetched {len(mlb_scores)} MLB scores and {len(nba_scores)} NBA scores.") + self.last_sports_update = ( + time.time() + ) # Update last sports update time + logging.info( + f"Fetched {len(mlb_scores)} MLB scores and {len(nba_scores)} NBA scores." + ) yield # Update frontend except Exception as e: - logging.error(f"Error in fetch_sports background task: {e}", exc_info=True) + logging.error( + f"Error in fetch_sports background task: {e}", exc_info=True + ) await asyncio.sleep(500) - # (Commented out news fetcher) - """ + # (Commented out news fetcher) + @rx.event(background=True) async def fetch_news(self): - #Fetches news periodically + # Fetches news periodically # Placeholder for the actual news fetching logic while True: try: @@ -134,25 +151,26 @@ class State(rx.State): self.news = news_items yield - except Exception as e: - logging.error(f"Error in fetch_news background task: {e}", exc_info=True) + logging.error( + f"Error in fetch_news background task: {e}", exc_info=True + ) await asyncio.sleep(500) - - """ # --- Weather Background Task --- @rx.event(background=True) async def fetch_weather(self): """Fetches the weather screenshot periodically.""" # Check if the client initialized correctly - if not hasattr(self, '_weather_client') or self._weather_client is None: - logging.warning("Weather client not initialized. Stopping fetch_weather task.") + if not hasattr(self, "_weather_client") or self._weather_client is None: + logging.warning( + "Weather client not initialized. Stopping fetch_weather task." + ) async with self: self.last_weather_update = "Error: Weather client unavailable" yield - return # Exit the task permanently if client init failed + return # Exit the task permanently if client init failed while True: try: @@ -162,132 +180,187 @@ class State(rx.State): if img_web_path: async with self: - timestamp = int(time.time()) # Unused timestamp, kept as per instruction + timestamp = int( + time.time() + ) # Unused timestamp, kept as per instruction self.weather_img = f"{img_web_path}" - self.last_weather_update = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') - logging.info(f"State.weather_img updated to: {self.weather_img}") - yield # Update frontend + self.last_weather_update = datetime.now(timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S UTC" + ) + logging.info( + f"State.weather_img updated to: {self.weather_img}" + ) + yield # Update frontend else: - logging.warning("get_weather_screenshot returned None. State not updated.") + logging.warning( + "get_weather_screenshot returned None. State not updated." + ) # Optionally update status async with self: self.last_weather_update = f"Failed fetch @ {datetime.now(timezone.utc).strftime('%H:%M:%S UTC')}" yield except Exception as e: - logging.error(f"Error in fetch_weather background task: {e}", exc_info=True) - async with self: # Update state to show error - self.last_weather_update = f"Error @ {datetime.now(timezone.utc).strftime('%H:%M:%S UTC')}" + logging.error( + f"Error in fetch_weather background task: {e}", exc_info=True + ) + async with self: # Update state to show error + self.last_weather_update = ( + f"Error @ {datetime.now(timezone.utc).strftime('%H:%M:%S UTC')}" + ) yield await asyncio.sleep(WEATHER_FETCH_INTERVAL) + def index() -> rx.Component: - - return rx.container( - rx.theme_panel(default_open=False), - - rx.center( - rx.vstack( - - rx.button( - rx.moment(interval=1000, format="HH:mm:ss"), - font_size="4xl", - font_weight="bold", - color="white", - background_color="#6f42c1", - ), - - - rx.flex( - rx.card( - rx.box( - rx.text("NBA Scores"), - rx.foreach( - State.nba_scores, - lambda score: rx.vstack( - rx.card( - rx.text(f"{score['away_team']} {score['away_score']} @ " - f"{score['home_team']} {score['home_score']} " - f"(Status: {score['status']})"), - ), - spacing="1", - padding="2", - ), - - ), - ), - ), - - rx.card( - rx.vstack( - rx.heading("Weather", size="4"), - - rx.image( - - src=State.weather_img, - alt="Current weather conditions for Sacramento", - width="100%", - height="auto", - - object_fit="contain", - border_radius="var(--radius-3)", - ), - rx.text( - f"Last Update: {State.last_weather_update}", - size="1", - color_scheme="gray", - padding_top="0.5em" - ), - align="center", - spacing="2", - ) - ), - - rx.card( - rx.box( - rx.text("MLB Scores"), - # Add rx.vstack here to control spacing of foreach items - rx.vstack( - rx.foreach( - State.mlb_scores, - # Lambda now returns the styled card directly - lambda score: rx.card( - rx.text(f"{score.get('away_team','?')} {score.get('away_score','-')} @ " # Use .get() for safety - f"{score.get('home_team','?')} {score.get('home_score','-')} " - f"({score.get('status','?')})", - size="1", - ), - size="1", - padding="1", - variant="surface", - width="100%", - ), - ), - spacing="1", - align_items="stretch", - width="100%", - ), - ), - ), - - spacing="3", - width="100%", - justify="center", - align="stretch", - ), - - align="center", - spacing="4", + # Build NBA scores list (safe access with .get where appropriate) + nba_scores_list = rx.vstack( + rx.foreach( + State.nba_scores, + lambda score: rx.card( + rx.text( + f"{score.get('away_team', '?')} {score.get('away_score', '-')} @ " + f"{score.get('home_team', '?')} {score.get('home_score', '-')} " + f"(Status: {score.get('status', '?')})" + ) ), ), + spacing="1", + padding="2", + align_items="stretch", + width="100%", + ) - padding="2rem", - max_width="1200px", - margin="0 auto", - ), -''' # Commented out style block + nba_card = rx.card( + rx.box( + rx.text("NBA Scores"), + nba_scores_list, + ) + ) + + # Weather card + weather_card = rx.card( + rx.vstack( + rx.heading("Weather", size="4"), + rx.image( + src=State.weather_img, + alt="Current weather conditions for Sacramento", + width="100%", + height="auto", + object_fit="contain", + border_radius="var(--radius-3)", + ), + rx.text( + f"Last Update: {State.last_weather_update}", + size="1", + color_scheme="gray", + padding_top="0.5em", + ), + align="center", + spacing="2", + ) + ) + + # MLB scores list + mlb_scores_list = rx.vstack( + rx.foreach( + State.mlb_scores, + lambda score: rx.card( + rx.text( + f"{score.get('away_team', '?')} {score.get('away_score', '-')} @ " + f"{score.get('home_team', '?')} {score.get('home_score', '-')} " + f"({score.get('status', '?')})", + size="1", + ), + size="1", + padding="1", + variant="surface", + width="100%", + ), + ), + spacing="1", + align_items="stretch", + width="100%", + ) + + mlb_card = rx.card( + rx.box( + rx.text("MLB Scores"), + mlb_scores_list, + ) + ) + + nfl_scores_list = rx.vstack( + rx.foreach( + State.nfl_scores, + lambda score: rx.card( + rx.text( + f"{score.get('away_team', '?')} {score.get('away_score', '-')} @ " + f"{score.get('home_team', '?')} {score.get('home_score', '-')} " + f"({score.get('status', '?')})", + size="1", + ), + size="1", + padding="1", + variant="surface", + width="100%", + ), + ), + spacing="1", + align_items="stretch", + width="100%", + ) + + nfl_card = rx.card( + rx.box( + rx.text("NFL Scores"), + nfl_scores_list, + ) + ) + + # Main flexible content area + main_flex = rx.flex( + nba_card, + weather_card, + mlb_card, + nfl_card, + spacing="3", + width="100%", + justify="center", + align="stretch", + ) + + # Top clock button + clock_button = rx.button( + rx.moment(interval=1000, format="HH:mm:ss"), + font_size="4xl", + font_weight="bold", + color="white", + background_color="#6f42c1", + ) + + # Compose the page + page = rx.container( # pyright: ignore[reportReturnType] + rx.theme_panel(default_open=False), + rx.center( + rx.vstack( + clock_button, + main_flex, + align="center", + spacing="4", + ) + ), + padding="2rem", + max_width="1200px", + margin="0 auto", + ) + + return page + + +""" # Commented out style block style = { "background_color": "black", @@ -299,7 +372,7 @@ style = { "color": "#ffffff", }, } -''' # Commented out style block +""" app = rx.App( theme=rx.theme( appearance="dark", @@ -309,14 +382,12 @@ app = rx.App( gray_color="mauve", has_background=True, ), - #style=style # using theme instead + # style=style # using theme instead ) - - app.add_page( index, - title="DeathClock", # Example title - on_load=State.start_background_tasks # Trigger tasks when this page loads + title="DeathClock", # Example title + on_load=State.start_background_tasks, # Trigger tasks when this page loads ) diff --git a/utils/scores.py b/utils/scores.py index b75e3dc..9e7cdfd 100755 --- a/utils/scores.py +++ b/utils/scores.py @@ -3,26 +3,29 @@ from nba_api.live.nba.endpoints import scoreboard from datetime import datetime, timedelta import statsapi import reflex as rx -import logging # Use logging for consistency +import logging # Use logging for consistency + + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class NBAScores(rx.Base): - - async def get_scores(self): # Make async to match usage in State + async def get_scores(self): # Make async to match usage in State """Fetches NBA scores and returns them as a list of dicts.""" try: # Get scoreboard data - # Consider running blocking IO in a thread pool if it becomes an issue + board = scoreboard.ScoreBoard() data = board.get_dict() # Check if we have a valid scoreboard response - if 'scoreboard' not in data: + if "scoreboard" not in data: logging.warning("No NBA scoreboard data found in response") return [] - games = data['scoreboard'].get('games', []) + games = data["scoreboard"].get("games", []) if not games: logging.info("No active NBA games found in scoreboard") return [] @@ -31,87 +34,199 @@ class NBAScores(rx.Base): for game in games: try: game_data = { - 'home_team': game['homeTeam']['teamTricode'], - 'home_score': game['homeTeam']['score'], - 'away_team': game['awayTeam']['teamTricode'], - 'away_score': game['awayTeam']['score'], - 'period': game['period'], - 'status': game['gameStatusText'] + "home_team": game["homeTeam"]["teamTricode"], + "home_score": game["homeTeam"]["score"], + "away_team": game["awayTeam"]["teamTricode"], + "away_score": game["awayTeam"]["score"], + "period": game["period"], + "status": game["gameStatusText"], } scores_list.append(game_data) except KeyError as e: - logging.error(f"Error processing NBA game data: {e} for game: {game.get('gameId', 'N/A')}") - continue # Skip this game + logging.error( + f"Error processing NBA game data: {e} for game: {game.get('gameId', 'N/A')}" + ) + continue # Skip this game - # No need to store in self._scores if this is just a utility return scores_list except Exception as e: logging.error(f"Error fetching NBA scores: {e}", exc_info=True) - return [] # Return empty list on error + return [] # Return empty list on error + class mlbScores(rx.Base): - - async def get_games(self): # Make async + async def get_games(self): # Make async """Fetches MLB games data.""" try: - games = statsapi.schedule() # Assuming sync for now + games = statsapi.schedule() # Assuming sync for now return games except Exception as e: logging.error(f"Error fetching MLB games: {e}", exc_info=True) return [] - async def get_scores(self): # Make async to match usage in State + async def get_scores(self): # Make async to match usage in State """Fetches and formats MLB scores.""" - games = await self.get_games() # Await the async get_games + games = await self.get_games() # Await the async get_games scores_list = [] if not games: - logging.info("No MLB games found today.") - return [] + logging.info("No MLB games found today.") + return [] for game in games: try: # Ensure keys exist, provide defaults if necessary game_data = { - 'home_team': game.get('home_name', 'N/A'), - 'home_score': game.get('home_score', '-'), - 'away_team': game.get('away_name', 'N/A'), - 'away_score': game.get('away_score', '-'), - 'status': game.get('status', 'Scheduled') # Provide default status + "home_team": game.get("home_name", "N/A"), + "home_score": game.get("home_score", "-"), + "away_team": game.get("away_name", "N/A"), + "away_score": game.get("away_score", "-"), + "status": game.get("status", "Scheduled"), # Provide default status } scores_list.append(game_data) except KeyError as e: # This block might be less necessary with .get() above - logging.error(f"Error processing MLB game data: {e} for game: {game.get('game_id', 'N/A')}") - continue # Skip this game - return scores_list # RETURN THE LIST + logging.error( + f"Error processing MLB game data: {e} for game: {game.get('game_id', 'N/A')}" + ) + continue # Skip this game + return scores_list # RETURN THE LIST + + +class nflScores: + async def get_scores(self): + # get nfl scores from espn + nfl_scores = [] + try: + import aiohttp + + # use current local date in YYYYMMDD format + date_str = datetime.now().strftime("%Y%m%d") + url = f"https://site.api.espn.com/apis/site/v2/sports/football/nfl/scoreboard?dates={date_str}" + + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=15) as resp: + if resp.status != 200: + logging.error( + f"ESPN NFL scoreboard returned status {resp.status}" + ) + return [] + data = await resp.json() + + events = data.get("events", []) + if not events: + logging.info("No NFL games found for date %s", date_str) + return [] + + nfl_scores = [] + for ev in events: + try: + competitions = ev.get("competitions", []) + if not competitions: + logging.warning( + "No competitions found for event: %s", ev.get("id") + ) + continue + comp = competitions[0] + competitors = comp.get("competitors", []) + home_team = away_team = "N/A" + home_score = away_score = "-" + + for c in competitors: + team = c.get("team", {}) or {} + abbr = ( + team.get("abbreviation") + or team.get("displayName") + or team.get("shortDisplayName") + or "N/A" + ) + score = ( + c.get("score", "-") if c.get("score") is not None else "-" + ) + homeAway = c.get("homeAway", "").lower() + if homeAway == "home": + home_team = abbr + home_score = score + else: + away_team = abbr + away_score = score + + status = comp.get("status", {}) or {} + status_type = status.get("type", {}) or {} + status_text = ( + status_type.get("shortDetail") + or status_type.get("description") + or status_type.get("state") + or status.get("type") + or status.get("detail") + or "Unknown" + ) + + game_data = { + "home_team": home_team, + "home_score": home_score, + "away_team": away_team, + "away_score": away_score, + "status": status_text, + } + nfl_scores.append(game_data) + except Exception as e: + logging.error( + f"Error processing NFL event: {e} for event {ev.get('id')}", + exc_info=True, + ) + continue + print(nfl_scores) + return nfl_scores + + except Exception as e: + logging.error(f"Error fetching NFL scores: {e}", exc_info=True) + return [] + -# Keep the __main__ block for testing if desired, but update calls to be async if needed if __name__ == "__main__": import asyncio async def test_scores(): nba_client = NBAScores() - nba_results = await nba_client.get_scores() # await async method + nba_results = await nba_client.get_scores() # await async method print("\nNBA Scores:") if nba_results: for game in nba_results: - print(f"{game['away_team']} {game['away_score']} @ " - f"{game['home_team']} {game['home_score']} " - f"(Status: {game['status']})") + print( + f"{game['away_team']} {game['away_score']} @ " + f"{game['home_team']} {game['home_score']} " + f"(Status: {game['status']})" + ) else: print("No NBA games/scores available") mlb_client = mlbScores() - mlb_results = await mlb_client.get_scores() # await async method + mlb_results = await mlb_client.get_scores() # await async method print("\nMLB Scores:") if mlb_results: for game in mlb_results: - print(f"{game['away_team']} {game['away_score']} @ " - f"{game['home_team']} {game['home_score']} " - f"(Status: {game['status']})") + print( + f"{game['away_team']} {game['away_score']} @ " + f"{game['home_team']} {game['home_score']} " + f"(Status: {game['status']})" + ) else: print("No MLB games/scores available") + nfl_client = nflScores() + nfl_results = await nfl_client.get_scores() # await async method + + print("\nNFL Scores:") + if nfl_results: + for game in nfl_results: + print( + f"{game['away_team']} {game['away_score']} @ " + f"{game['home_team']} {game['home_score']} " + f"(Status: {game['status']})" + ) + else: + print("No NFL games/scores available") + asyncio.run(test_scores())