import json
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo
from pathlib import Path
import gradio as gr
import pandas as pd
import numpy as np
# Optional pybaseball
HAVE_PYBASEBALL = True
try:
from pybaseball import statcast_pitcher
except Exception:
HAVE_PYBASEBALL = False
statcast_pitcher = None
# HTTP helper
try:
import requests
def http_get_json(url, params=None, timeout=10):
r = requests.get(url, params=params, timeout=timeout, headers={"User-Agent": "zone-live-mlb/1.0"})
r.raise_for_status(); return r.json()
except Exception:
from urllib.parse import urlencode
from urllib.request import urlopen, Request
def http_get_json(url, params=None, timeout=10):
if params: url = f"{url}?{urlencode(params)}"
req = Request(url, headers={"User-Agent": "zone-live-mlb/1.0"})
with urlopen(req, timeout=timeout) as resp: return json.load(resp)
API_SCHEDULE = "https://statsapi.mlb.com/api/v1/schedule"
API_BOXSCORE = "https://statsapi.mlb.com/api/v1/game/{gamePk}/boxscore"
API_LIVEFEED = "https://statsapi.mlb.com/api/v1.1/game/{gamePk}/feed/live"
API_PBP = "https://statsapi.mlb.com/api/v1/game/{gamePk}/playByPlay"
IN_PROGRESS_STATES = {"In Progress","Mid Inning","End Inning","Warmup","Game Delayed","Delayed Start","Manager Challenge","Manager Review","Pitching Change"}
EXCLUDE_STATES = {"Final","Game Over","Completed Early","Postponed","Cancelled","Suspended"}
# ===== Helpers =====
def _team_label(team_side: dict) -> str:
t = (team_side or {}).get("team") or {}
return t.get("name") or t.get("teamName") or t.get("clubName") or t.get("shortName") or t.get("abbreviation") or "UNKNOWN"
def _candidate_dates():
eastern_today = datetime.now(ZoneInfo("America/New_York")).date(); utc_today = date.today()
return sorted({d.isoformat() for d in {eastern_today, utc_today, eastern_today - timedelta(days=1), utc_today - timedelta(days=1)}})
_DEF_FEET_CAP = 3.0
def _to_inches(val):
if val is None: return None
try: v = float(val)
except Exception: return None
return v * 12.0 if abs(v) <= _DEF_FEET_CAP else v
def _get_live_boxscore(game_pk: int):
feed = http_get_json(API_LIVEFEED.format(gamePk=int(game_pk)), {"language":"en"})
live = (feed.get("liveData") or {}); box = (live.get("boxscore") or {}); plays = (live.get("plays") or {}).get("allPlays", []) or []
return box, plays
def _get_pbp(game_pk: int):
try:
data = http_get_json(API_PBP.format(gamePk=int(game_pk)), {"language": "en"}); return (data or {}).get("allPlays", []) or []
except Exception: return []
# ===== Live data =====
def get_games(inprogress_only: bool, include_finals_today: bool):
games, eastern_today_iso = {}, datetime.now(ZoneInfo("America/New_York")).date().isoformat()
for d in _candidate_dates():
try: data = http_get_json(API_SCHEDULE, {"sportId": 1, "date": d})
except Exception: continue
for day in data.get("dates", []):
day_date = day.get("date")
for g in (day.get("games") or []):
status = (g.get("status") or {}).get("detailedState") or ""
if inprogress_only and status not in IN_PROGRESS_STATES: continue
if not inprogress_only:
is_final_like = status in EXCLUDE_STATES
if is_final_like and not (include_finals_today and day_date == eastern_today_iso and status in {"Final","Game Over","Completed Early"}):
continue
game_pk = g.get("gamePk")
if not game_pk: continue
home_team = _team_label((g.get("teams") or {}).get("home"))
away_team = _team_label((g.get("teams") or {}).get("away"))
label = f"{away_team} @ {home_team} | {status} | game_pk={game_pk}"
games[int(game_pk)] = {"game_pk": int(game_pk), "label": label}
return list(games.values())
# swing/whiff flags
def _swing_whiff_flags_from_text(desc: str, is_in_play_flag: bool, type_code: str):
d = (desc or "").lower(); is_in_play = bool(is_in_play_flag) or (type_code == "X")
is_foul = "foul" in d; is_foul_tip = "foul tip" in d; is_swinging_strike = "swinging strike" in d; is_missed_bunt = "missed bunt" in d
swing = is_in_play or is_foul or is_foul_tip or is_swinging_strike or is_missed_bunt; whiff = is_swinging_strike or is_missed_bunt
return swing, whiff, is_in_play
# Pitch name mapping/canonicalization
PITCH_NAME_MAP = {"FF":"Four-seam Fastball","FA":"Fastball","FT":"Two-seam Fastball","SI":"Sinker","FC":"Cutter","SL":"Slider","CH":"Changeup","CU":"Curveball","KC":"Knuckle Curve","KN":"Knuckleball","FS":"Splitter","ST":"Sweeper","SC":"Screwball","EP":"Eephus","FO":"Forkball","GY":"Gyroball"}
CODE_TO_NAME = {"FF":"Four-seam Fastball","FA":"Fastball","FT":"Two-seam Fastball","SI":"Sinker","FC":"Cutter","SL":"Slider","ST":"Sweeper","CH":"Changeup","CU":"Curveball","KC":"Knuckle Curve","FS":"Splitter"}
NAME_TO_CODE = {"four-seam fastball":"FF","four seam fastball":"FF","4-seam fastball":"FF","fastball":"FA","two-seam fastball":"FT","2-seam fastball":"FT","two seam fastball":"FT","sinker":"SI","cutter":"FC","slider":"SL","sweeper":"ST","changeup":"CH","change-up":"CH","curveball":"CU","knuckle curve":"KC","knuckle-curve":"KC","splitter":"FS"}
def _canon_pitch_code(x: str) -> str:
if x is None: return "UNK"
s = str(x).strip(); u = s.upper()
return u if u in CODE_TO_NAME else NAME_TO_CODE.get(s.lower(), s)
def _friendly_pitch(code: str) -> str: return CODE_TO_NAME.get(str(code).upper(), str(code))
def _pitch_label(code_or_name: str) -> str:
if not code_or_name: return "UNK"
s = str(code_or_name).strip(); key = s.upper(); return PITCH_NAME_MAP.get(key, s if len(s) > 3 else s.title())
# Styling & HTML tables
def _style_pos_neg(val, invert=False):
try:
if pd.isna(val): return ""
v = float(val)
except Exception: return ""
pos_color, neg_color = ("red","green") if invert else ("green","red")
color = pos_color if v>0 else (neg_color if v<0 else ""); bold = "font-weight:600;" if color else ""; color_css = f"color:{color};" if color else ""
return f'{v:.2f}'
def _style_compare_live_vs_season(live, season):
if pd.isna(live): return ""
try: l = float(live)
except Exception: return str(live)
s = None if pd.isna(season) else float(season)
if s is None: return f"{l:.2f}"
if l > s: return f'{l:.2f}'
if l < s: return f'{l:.2f}'
return f"{l:.2f}"
def _table_html(df, highlight_specs=None, caption=None, compare_cols=None):
df = df.copy(); highlight_specs = highlight_specs or []
headers = ''.join(f'| {c} | ' if c!="Pitch type" else f'{c} | ' for c in df.columns)
rows_html = []
for _, row in df.iterrows():
cells = []
for col in df.columns:
val = row[col]
if col == "Pitch type": cells.append(f'{val} | '); continue
if compare_cols and col == compare_cols[0]:
html = _style_compare_live_vs_season(val, row.get(compare_cols[1])); cells.append(f'{html} | '); continue
spec = next((inv for c, inv in highlight_specs if c == col), None)
is_num = False
try:
if pd.isna(val): is_num = True
else: _ = float(val); is_num = True
except Exception: is_num = False
if spec is not None:
html = _style_pos_neg(val, invert=spec); cells.append(f'{html} | ')
else:
txt = "" if pd.isna(val) else (f"{float(val):.2f}" if is_num else str(val)); cells.append(f'{txt} | ')
rows_html.append(f"{''.join(cells)}
")
caption_html = f'{caption}' if caption else ''
return f'''{caption_html}{headers}
{''.join(rows_html)}
'''
# ===== Pitch data extraction =====
def _extract_pitch_type(ev):
det = ev.get("details") or {}; code = (det.get("type") or {}).get("code"); name = det.get("type") or {}
return code or name or (det.get("description") or "")
def _extract_spin_break(ev):
pdict = ev.get("pitchData") or {}; spin = pdict.get("breaks", {}).get("spinRate") or pdict.get("spinRate")
coords = pdict.get("coordinates", {})
hmov = _to_inches(coords.get("pfxX")); vmov = _to_inches(coords.get("pfxZ"))
try: hmov = float(hmov) if hmov is not None else None
except Exception: hmov = None
try: vmov = float(vmov) if vmov is not None else None
except Exception: vmov = None
return spin, hmov, vmov
# ===== Live per-pitch + totals =====
def live_per_pitch_and_totals(game_pk: int, pitcher_id: int):
try: feed = http_get_json(API_LIVEFEED.format(gamePk=int(game_pk)), {"language": "en"})
except Exception: feed = {}
plays = (((feed.get("liveData") or {}).get("plays") or {}).get("allPlays")) or []
if not plays: plays = _get_pbp(game_pk)
rows, exit_velos, bip_count, total_pitches, strike_like_count = [], [], 0, 0, 0
for play in plays:
play_pitcher = (((play.get("matchup") or {}).get("pitcher") or {}).get("id"))
for ev in (play.get("playEvents") or []):
if not ev.get("isPitch"): continue
ev_pitcher = (((ev.get("matchup") or {}).get("pitcher") or {}).get("id")); pid = ev_pitcher or play_pitcher
if pid != int(pitcher_id): continue
pdict = ev.get("pitchData") or {}; spd = pdict.get("startSpeed");
if spd is None: continue
det = ev.get("details") or {}; desc = det.get("description") or (det.get("call") or {}).get("description") or ""; type_code = (det.get("type") or {}).get("code")
swing, whiff, is_in_play = _swing_whiff_flags_from_text(desc, det.get("isInPlay"), type_code)
is_ball = (type_code == "B"); is_strike_like = (type_code == "S") or (not is_ball and (is_in_play or "called strike" in desc.lower() or "swinging strike" in desc.lower() or "foul" in desc.lower() or "foul tip" in desc.lower() or "missed bunt" in desc.lower()))
total_pitches += 1; strike_like_count += 1 if is_strike_like else 0
ptype = _canon_pitch_code(_extract_pitch_type(ev)); spin, hmov, vmov = _extract_spin_break(ev)
ev_hit = ev.get("hitData") or {}; ev_ev = ev_hit.get("launchSpeed")
if ev_ev is not None:
try: exit_velos.append(float(ev_ev))
except Exception: pass
if is_in_play: bip_count += 1
rows.append({"pitch_type": ptype, "speed": float(spd), "spin": spin, "hmov": hmov, "vmov": vmov, "swing": 1 if swing else 0, "whiff": 1 if whiff else 0})
if not rows:
return pd.DataFrame(columns=["pitch_type","avg_speed","avg_spin","avg_hmov","avg_vmov","swings","whiffs","whiff_pct","num_pitches"]), pd.DataFrame(columns=["avg_exit_velo","swings","whiffs","whiff_pct","bip","pitches","strike_pct"])
df = pd.DataFrame(rows)
per_pitch = (df.groupby("pitch_type").agg(avg_speed=("speed","mean"), avg_spin=("spin","mean"), avg_hmov=("hmov","mean"), avg_vmov=("vmov","mean"), swings=("swing","sum"), whiffs=("whiff","sum"), num_pitches=("speed","size")).reset_index())
per_pitch["pitch_type"] = per_pitch["pitch_type"].astype(str).str.upper()
per_pitch["whiff_pct"] = (100.0 * per_pitch["whiffs"] / per_pitch["swings"].where(per_pitch["swings"] > 0, pd.NA)).round(2)
for c in ["avg_speed","avg_spin","avg_hmov","avg_vmov"]: per_pitch[c] = per_pitch[c].round(2)
total_swings = int(df["swing"].sum()); total_whiffs = int(df["whiff"].sum())
totals = pd.DataFrame([{ "avg_exit_velo": (round(sum(exit_velos) / len(exit_velos), 2) if exit_velos else None), "swings": total_swings, "whiffs": total_whiffs, "whiff_pct": (round(100.0 * total_whiffs / total_swings, 2) if total_swings > 0 else None), "bip": int(bip_count), "pitches": int(total_pitches), "strike_pct": (round(100.0 * strike_like_count / total_pitches, 2) if total_pitches > 0 else None)}])
return per_pitch.sort_values(["num_pitches","pitch_type"], ascending=[False, True]).reset_index(drop=True), totals
# ===== Season aggregates =====
def season_year_now_eastern(): return datetime.now(ZoneInfo("America/New_York")).year
def season_per_pitch_and_totals(pitcher_id: int, season_year: int):
if not HAVE_PYBASEBALL: return pd.DataFrame(), pd.DataFrame()
try:
start, end = f"{season_year}-03-01", date.today().isoformat(); df = statcast_pitcher(start_dt=start, end_dt=end, player_id=pitcher_id)
if df is None or df.empty: return pd.DataFrame(), pd.DataFrame()
except Exception: return pd.DataFrame(), pd.DataFrame()
df = df.dropna(subset=["pitch_type","release_speed"]).copy(); df["pitch_type"] = df["pitch_type"].astype(str).map(_canon_pitch_code).str.upper()
def _safe_to_inches(x):
try: return _to_inches(float(x))
except Exception: return None
df["pfx_x_in"] = df["pfx_x"].apply(_safe_to_inches) if "pfx_x" in df.columns else np.nan
df["pfx_z_in"] = df["pfx_z"].apply(_safe_to_inches) if "pfx_z" in df.columns else np.nan
mph_to_fps, default_extension_ft = 1.4666667, 5.5
t_plate = pd.to_numeric(df.get("plate_time", pd.Series(dtype=float)), errors="coerce")
need_est = t_plate.isna()
if need_est.any():
speed_fps = pd.to_numeric(df["release_speed"], errors="coerce") * mph_to_fps; speed_fps = speed_fps.replace([0, np.inf, -np.inf], np.nan)
ext_ft = pd.to_numeric(df.get("release_extension", default_extension_ft), errors="coerce").fillna(default_extension_ft)
dist_ft = 60.5 - ext_ft; t_plate = t_plate.where(~need_est, dist_ft / speed_fps)
G_IN_S2 = 386.089; grav_drop_in = 0.5 * G_IN_S2 * (t_plate ** 2)
pfx_z_in_num = pd.to_numeric(df["pfx_z_in"], errors="coerce"); df["vmov_with_grav_in"] = pfx_z_in_num - grav_drop_in
def _flags(row):
desc = str(row.get("description") or row.get("des") or "").lower(); type_code = str(row.get("type") or "")
is_in_play = "in play" in desc or type_code == "X"; is_foul = "foul" in desc; is_foul_tip = "foul tip" in desc
is_swinging_strike = "swinging_strike" in desc or "swinging strike" in desc; is_missed_bunt = "missed_bunt" in desc or "missed bunt" in desc
swing = is_in_play or is_foul or is_foul_tip or is_swinging_strike or is_missed_bunt; whiff = is_swinging_strike or is_missed_bunt
is_ball = (type_code == "B"); is_strike_like = (type_code == "S") or (not is_ball and (is_in_play or "called strike" in desc or "swinging strike" in desc or "foul" in desc or "foul tip" in desc or "missed bunt" in desc))
return pd.Series({"swing": 1 if swing else 0, "whiff": 1 if whiff else 0, "strike_like": 1 if is_strike_like else 0})
df = pd.concat([df, df.apply(_flags, axis=1)], axis=1)
per_pitch = (df.groupby("pitch_type").agg(avg_speed=("release_speed","mean"), avg_spin=("release_spin_rate","mean"), avg_hmov=("pfx_x_in","mean"), avg_vmov=("vmov_with_grav_in","mean"), swings=("swing","sum"), whiffs=("whiff","sum"), num_pitches=("release_speed","size")).reset_index())
per_pitch["avg_hmov"] = -per_pitch["avg_hmov"]; per_pitch["whiff_pct"] = (100.0 * per_pitch["whiffs"] / per_pitch["swings"].where(per_pitch["swings"] > 0, pd.NA)).round(2)
for c in ["avg_speed","avg_spin","avg_hmov","avg_vmov"]: per_pitch[c] = per_pitch[c].round(2)
per_pitch = per_pitch.sort_values(["num_pitches","pitch_type"], ascending=[False, True]).reset_index(drop=True)
total_swings, total_whiffs = int(df["swing"].sum()), int(df["whiff"].sum())
avg_exit_velo = round(pd.to_numeric(df.get("launch_speed", pd.Series(dtype=float)), errors="coerce").dropna().astype(float).mean(), 2) if "launch_speed" in df.columns else None
total_pitches = int(len(df)); strike_like = int(df["strike_like"].sum()) if "strike_like" in df.columns else 0
totals = pd.DataFrame([{ "avg_exit_velo": avg_exit_velo, "swings": total_swings, "whiffs": total_whiffs, "whiff_pct": (round(100.0 * total_whiffs / total_swings, 2) if total_swings>0 else None), "pitches": total_pitches, "strike_pct": (round(100.0 * strike_like / total_pitches, 2) if total_pitches>0 else None)}])
return per_pitch, totals
# ===== Compare (four tables) =====
def compare_live_vs_season(live_per, live_totals, season_per, season_totals):
if season_per is None or season_per.empty:
m = live_per.copy().rename(columns={"avg_speed":"avg_speed_live","avg_spin":"avg_spin_live","avg_hmov":"avg_hmov_live","avg_vmov":"avg_vmov_live","num_pitches":"num_pitches_live"})
for c in ["avg_speed_season","avg_spin_season","avg_hmov_season","avg_vmov_season","num_pitches_season"]: m[c] = pd.NA
else:
sp = season_per.rename(columns={"num_pitches":"num_pitches_season"})
m = pd.merge(live_per, sp[["pitch_type","avg_speed","avg_spin","avg_hmov","avg_vmov","num_pitches_season"]], on="pitch_type", how="outer", suffixes=("_live","_season"))
if "num_pitches_live" not in m.columns and "num_pitches" in live_per.columns:
m = pd.merge(m, live_per[["pitch_type","num_pitches"]].rename(columns={"num_pitches":"num_pitches_live"}), on="pitch_type", how="left")
for base in ["avg_speed","avg_spin","avg_hmov","avg_vmov"]:
m[f"{base}_diff"] = pd.to_numeric(m.get(f"{base}_live"), errors="coerce") - pd.to_numeric(m.get(f"{base}_season"), errors="coerce")
table1 = m[["pitch_type","avg_speed_live","avg_speed_season","avg_spin_live","avg_spin_season","avg_hmov_live","avg_hmov_season","avg_vmov_live","avg_vmov_season"]].copy().rename(columns={"pitch_type":"Pitch type","avg_speed_live":"avg speed live","avg_speed_season":"avg speed season","avg_spin_live":"avg spin live","avg_spin_season":"avg spin season","avg_hmov_live":"avg hmov live","avg_hmov_season":"avg hmov season","avg_vmov_live":"avg vmov live","avg_vmov_season":"avg vmov season"}).sort_values(["avg speed live","Pitch type"], ascending=[False, True]).reset_index(drop=True)
order_codes = table1["Pitch type"].tolist(); t2 = m[["pitch_type","avg_speed_diff","avg_spin_diff","avg_hmov_diff","avg_vmov_diff","num_pitches_live","num_pitches_season"]].copy().rename(columns={"pitch_type":"Pitch type","avg_speed_diff":"avg speed diff","avg_spin_diff":"avg spin diff","avg_hmov_diff":"avg hmov diff","avg_vmov_diff":"avg vmov diff","num_pitches_live":"num pitches","num_pitches_season":"num pitches (year)"}).set_index("Pitch type")
in_order = [p for p in order_codes if p in t2.index]; rest = [p for p in t2.index if p not in in_order]; table2 = pd.concat([t2.loc[in_order], t2.loc[rest]]).reset_index()
totals_live = live_totals.iloc[0] if not live_totals.empty else pd.Series(dtype="float"); totals_season = season_totals.iloc[0] if not season_totals.empty else pd.Series(dtype="float")
table3 = pd.DataFrame([{ "total pitches live": totals_live.get("pitches", pd.NA), "strike% live": totals_live.get("strike_pct", pd.NA), "strike% season": totals_season.get("strike_pct", pd.NA), "balls in play live": totals_live.get("bip", pd.NA), "swings live": totals_live.get("swings", pd.NA)}])
avg_ev_live, avg_ev_season = totals_live.get("avg_exit_velo", pd.NA), totals_season.get("avg_exit_velo", pd.NA)
wh_live, wh_season = totals_live.get("whiff_pct", pd.NA), totals_season.get("whiff_pct", pd.NA)
ev_diff = ((pd.to_numeric(pd.Series([avg_ev_live])) - pd.to_numeric(pd.Series([avg_ev_season]))).round(2).iloc[0] if pd.notna(avg_ev_live) and pd.notna(avg_ev_season) else pd.NA)
wh_diff = ((pd.to_numeric(pd.Series([wh_live])) - pd.to_numeric(pd.Series([wh_season]))).round(2).iloc[0] if pd.notna(wh_live) and pd.notna(wh_season) else pd.NA)
table4 = pd.DataFrame([{ "avg exit velo live": avg_ev_live, "avg exit velo season": avg_ev_season, "exit velo diff": ev_diff, "whiff percent live": wh_live, "whiff percent season": wh_season, "whiff percent diff": wh_diff }])
table1["Pitch type"] = table1["Pitch type"].apply(_friendly_pitch); table2["Pitch type"] = table2["Pitch type"].apply(_friendly_pitch)
for df_ in [table1, table2, table3, table4]:
for col in df_.columns:
if pd.api.types.is_numeric_dtype(df_[col]): df_[col] = pd.to_numeric(df_[col], errors="coerce").round(2)
return table1.reset_index(drop=True), table2.reset_index(drop=True), table3.reset_index(drop=True), table4.reset_index(drop=True)
# ===== Savant predictive metrics (savantfunctions) =====
try:
# Use functions that actually exist in savantfunctions.py
from savantfunctions import load_metrics, list_feature_columns, compute_correlations, METRICS
HAVE_SAVANT = True
except Exception:
HAVE_SAVANT = False
load_metrics = list_feature_columns = compute_correlations = None
METRICS = ("whiff_pct","strike_pct","chase_pct","exit_velo","fb_velo")
def _pick_first(paths):
for p in paths:
if Path(p).exists():
return p
return None
PERFORMANCE_PATH = _pick_first([
"data_f3_metrics.csv", "data/data_f3_metrics.csv", "outputs/data_f3_metrics.csv",
"data_f3_metrics.parquet", "data/data_f3_metrics.parquet", "outputs/data_f3_metrics.parquet",
])
def compute_savant_tables(pitcher_name: str, top_n: int = 25, k: int = 25, ci: float = 0.90):
"""
Build three small tables of top correlations (|r|) vs `game_woba` for
the feature bands: thru1, thru2, thru3. Uses savantfunctions helpers.
"""
if not HAVE_SAVANT:
return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
"savantfunctions.py not importable. Ensure it's in repo root.")
if PERFORMANCE_PATH is None:
return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
"Could not locate data_f3_metrics.* in repo root, data/, or outputs/.")
try:
df = load_metrics(PERFORMANCE_PATH)
def _top_corr_df(band: str) -> pd.DataFrame:
feats = list_feature_columns(time_bands=[band], metrics=METRICS)
pairs = compute_correlations(df, features=feats, target="game_woba") # sorted by |r|
rows = [{"metric": m, "weighted_r": (np.nan if pd.isna(r) else float(r))}
for m, r in pairs if not pd.isna(r)]
out = pd.DataFrame(rows)
return out.head(top_n) if not out.empty else pd.DataFrame(columns=["metric","weighted_r"])
t1 = _top_corr_df("thru1") # “End of 1st”
t12 = _top_corr_df("thru2") # “End of 2nd”
t13 = _top_corr_df("thru3") # “End of 3rd”
diag = (f"Top |r| vs game_woba from data_f3_metrics for bands thru1/thru2/thru3. "
f"Rows: {len(df):,}. Showing up to {top_n} metrics per band.")
return t1, t12, t13, diag
except Exception as e:
return pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), (
f"savantfunctions error: {e}\n"
"Check that columns like 'thru1_whiff_pct' exist in data_f3_metrics."
)
# ===== UI callbacks =====
def refresh_games(inprogress_only: bool, include_finals_today: bool):
games = get_games(inprogress_only=inprogress_only, include_finals_today=include_finals_today)
if not games:
scope = "in-progress" if inprogress_only else ("active/pregame + finals(today)" if include_finals_today else "active/pregame")
return (gr.update(choices=[], value=None), gr.update(choices=[], value=None), {}, f"No {scope} MLB games found across Eastern/UTC (today/yesterday).")
labels = [g["label"] for g in games]; mapping = {g["label"]: g["game_pk"] for g in games}
return (gr.update(choices=labels, value=None), gr.update(choices=[], value=None), mapping, f"Found {len(labels)} game(s). Pick a game, then a pitcher.")
def on_game_change(game_label, mapping, include_not_appeared):
if not (game_label and mapping): return gr.update(choices=[], value=None), "No game selected."
game_pk = mapping[game_label]
pitchers = get_all_roster_pitchers(game_pk) if include_not_appeared else get_pitchers_in_game(game_pk)
if not pitchers:
return gr.update(choices=[], value=None), f"Selected game_pk={game_pk}. No pitchers found {'(roster)' if include_not_appeared else '(appeared)'}."
labels = [f"{name} (id={pid})" for pid, name in pitchers]
return gr.update(choices=labels, value=None), f"Selected game_pk={game_pk}. Choose a pitcher."
def on_pitcher_change(game_label, pitcher_label, mapping):
if not (game_label and pitcher_label and mapping):
return (pd.DataFrame(), "", "", "", "", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "Make sure a game and pitcher are selected.")
game_pk = mapping[game_label]
try: pitcher_id = int(pitcher_label.split("id=")[-1].strip(")"))
except Exception: return (pd.DataFrame(), "", "", "", "", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "Could not parse pitcher id.")
live_per, live_totals = live_per_pitch_and_totals(game_pk, pitcher_id)
season_per = season_totals = pd.DataFrame(); season_year = season_year_now_eastern()
if HAVE_PYBASEBALL: season_per, season_totals = season_per_pitch_and_totals(pitcher_id, season_year)
table1_abs, table2_diff, table3_counts, table4_totals = compare_live_vs_season(live_per, live_totals, season_per, season_totals)
t2_html = _table_html(table2_diff, highlight_specs=[("avg speed diff", False)], caption="(2) Per-pitch: Deltas (Live − Season) + Volume")
t3_html = _table_html(table3_counts, compare_cols=("strike% live", "strike% season"), caption="(3) Live Totals: Pitches, Strike% (live vs season), BIP, Swings")
t4_html = _table_html(table4_totals, highlight_specs=[("exit velo diff", True), ("whiff percent diff", False)], caption="(4) Totals: Exit Velo & Whiff%")
name = pitcher_label.split(" (id=")[0]
pred1, pred2, pred3, diag = compute_savant_tables(name)
note = (f"Live vs {season_year} season metrics for {name}. Deltas = live − season. BIP Live = balls with isInPlay=true." if HAVE_PYBASEBALL and not (season_per.empty and season_totals.empty) else f"Live metrics for {name}{'' if HAVE_PYBASEBALL else ' (install `pybaseball` to enable season averages)'}. Deltas only shown when season data is available.")
print("LIVE codes:", sorted(live_per.get("pitch_type", pd.Series(dtype=str)).astype(str).unique().tolist() if not live_per.empty else []))
print("SEASON codes:", sorted(season_per.get("pitch_type", pd.Series(dtype=str)).astype(str).unique().tolist() if not season_per.empty else []))
return table1_abs, t2_html, t3_html, t4_html, note, pred1, pred2, pred3, diag
# Persist last good results
def _is_bad_result(t1_df, note_text):
if isinstance(note_text, str) and "Make sure a game and pitcher are selected." in note_text: return True
try:
if hasattr(t1_df, "empty") and t1_df.empty: return True
except Exception: pass
return False
def safe_on_pitcher_tick(game_label, pitcher_label, mapping, last_state):
last_state = last_state or {}
last = {k: last_state.get(k, pd.DataFrame() if k in {"t1","p1","p2","p3"} else ("" if k in {"t2","t3","t4","note","diag"} else None)) for k in ["t1","t2","t3","t4","note","p1","p2","p3","diag"]}
t1, h2, h3, h4, note, p1, p2, p3, diag = on_pitcher_change(game_label, pitcher_label, mapping)
if _is_bad_result(t1, note):
return last["t1"], last["t2"], last["t3"], last["t4"], last["note"], last["p1"], last["p2"], last["p3"], last["diag"], last
new_state = {"t1": t1, "t2": h2, "t3": h3, "t4": h4, "note": note, "p1": p1, "p2": p2, "p3": p3, "diag": diag}
return t1, h2, h3, h4, note, p1, p2, p3, diag, new_state
def _set_timer(active: bool, secs: float):
try: s = int(secs)
except Exception: s = 15
s = max(5, min(300, s)); return gr.update(value=float(s), active=bool(active))
# ===== Pitcher lists (appeared vs roster) =====
def get_pitchers_in_game(game_pk: int):
try: box, plays = _get_live_boxscore(game_pk)
except Exception: return []
pitchers = set(); teams = (box.get("teams") or {})
for side in ("home","away"):
team = teams.get(side) or {}; ids = team.get("pitchers") or []; players = team.get("players") or {}
for pid in ids:
info = players.get(f"ID{pid}", {}); name = ((info.get("person") or {}).get("fullName")) or None
if name: pitchers.add((int(pid), name))
if not pitchers and plays:
for p in plays:
pid = (((p.get("matchup") or {}).get("pitcher") or {}).get("id"));
if not pid: continue
name = None
for side in ("home","away"):
players = ((teams.get(side) or {}).get("players") or {}); info = players.get(f"ID{pid}", {}); cand = ((info.get("person") or {}).get("fullName"))
if cand: name = cand; break
if name: pitchers.add((int(pid), name))
return sorted(pitchers, key=lambda x: x[1])
def get_all_roster_pitchers(game_pk: int):
try: box, _ = _get_live_boxscore(game_pk)
except Exception: return []
pitchers = set(); teams = (box.get("teams") or {})
for side in ("home","away"):
team = teams.get(side) or {}; players = team.get("players") or {}
for _, info in players.items():
person = info.get("person") or {}; pos = (info.get("position") or {}).get("abbreviation") or (person.get("primaryPosition") or {}).get("abbreviation")
if pos == "P": pid, name = person.get("id"), person.get("fullName");
if pos == "P" and pid and name: pitchers.add((int(pid), name))
return sorted(pitchers, key=lambda x: x[1])
# ===== Gradio App =====
with gr.Blocks(title="MLB Live Pitcher Dashboard — Zone Style + Savant Targets") as app:
gr.Markdown("# MLB Live Pitcher Dashboard — Zone Style\nLive vs season tables with optional auto-refresh, plus savantfunctions predictive metrics tabs.")
with gr.Row():
inprog_only = gr.Checkbox(label="Only show in-progress games (hide pregame)", value=True)
include_finals_today = gr.Checkbox(label="Include finished games from today", value=False)
include_roster = gr.Checkbox(label="Include pitchers who haven't appeared", value=False)
refresh_btn = gr.Button("Refresh games", variant="primary"); status = gr.Markdown(visible=True)
game_map_state = gr.State({})
game_dd = gr.Dropdown(label="Games (today/yesterday, Eastern & UTC)", choices=[], value=None)
pitcher_dd = gr.Dropdown(label="Pitcher", choices=[], value=None)
table1_abs_df = gr.Dataframe(label="(1) Per-pitch: Live & Season Averages")
table2_diff_ht = gr.HTML(label="(2) Per-pitch: Deltas (Live − Season) + Volume")
table3_cnt_ht = gr.HTML(label="(3) Live Totals")
table4_tot_ht = gr.HTML(label="(4) Totals: Exit Velo & Whiff%")
note = gr.Markdown(visible=True)
with gr.Tabs():
with gr.TabItem("Top Predictors — End of 1st"): sav_t1_df = gr.Dataframe(interactive=False)
with gr.TabItem("Top Predictors — End of 2nd"): sav_t12_df = gr.Dataframe(interactive=False)
with gr.TabItem("Top Predictors — End of 3rd"): sav_t13_df = gr.Dataframe(interactive=False)
sav_diag = gr.Markdown(visible=True)
last_good = gr.State({"t1": pd.DataFrame(), "t2": "", "t3": "", "t4": "", "note": "", "p1": pd.DataFrame(), "p2": pd.DataFrame(), "p3": pd.DataFrame(), "diag": ""})
with gr.Row():
auto_refresh = gr.Checkbox(label="Auto-refresh", value=True)
refresh_secs = gr.Slider(minimum=5, maximum=120, value=15, step=1, label="Interval (seconds)")
timer = gr.Timer(value=15.0, active=True)
timer.tick(fn=safe_on_pitcher_tick, inputs=[game_dd, pitcher_dd, game_map_state, last_good], outputs=[table1_abs_df, table2_diff_ht, table3_cnt_ht, table4_tot_ht, note, sav_t1_df, sav_t12_df, sav_t13_df, sav_diag, last_good])
auto_refresh.change(_set_timer, [auto_refresh, refresh_secs], [timer])
refresh_secs.change(_set_timer, [auto_refresh, refresh_secs], [timer])
refresh_btn.click(fn=refresh_games, inputs=[inprog_only, include_finals_today], outputs=[game_dd, pitcher_dd, game_map_state, status])
game_dd.change(fn=on_game_change, inputs=[game_dd, game_map_state, include_roster], outputs=[pitcher_dd, status])
include_roster.change(fn=on_game_change, inputs=[game_dd, game_map_state, include_roster], outputs=[pitcher_dd, status])
pitcher_dd.change(fn=on_pitcher_change, inputs=[game_dd, pitcher_dd, game_map_state], outputs=[table1_abs_df, table2_diff_ht, table3_cnt_ht, table4_tot_ht, note, sav_t1_df, sav_t12_df, sav_t13_df, sav_diag])
if __name__ == "__main__":
app.launch()