pitch_leaderboard.py

import gradio as gr
import polars as pl
import numpy as np

from datetime import datetime
# from itertools import chain
from functools import partial
from typing import Optional

from data import data_df
from stats import compute_pitch_stats, filter_data_by_date_and_game_kind, get_stat_val
from convert import ball_kind, ball_kind_to_color, get_text_color_from_color, team_names_short_to_color, get_text_color_from_team
from plotting import stat_cmap
from teams import TEAMS
from utils import get_col_width

# STATS = ['Count', 'Usage', 'Avg Velo', 'Max Velo', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Con%', 'O-Con%', 'SwStr%', 'Whiff%', 'CSW%', 'Strike%', 'Ball%', 'F-Str%', 'PAR%', 'GB%', 'FB%', 'LD%', 'OFFB%', 'IFFB%', 'AIR%', 'Zone%', 'Arm%', 'Glove%', 'High%', 'Low%', 'MM%', 'Behind%']
# PCT_STATS = ['Usage', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Con%', 'O-Con%', 'SwStr%', 'Whiff%', 'CSW%', 'Strike%', 'Ball%', 'F-Str%', 'PAR%', 'GB%', 'FB%', 'LD%', 'OFFB%', 'IFFB%', 'AIR%', 'Zone%', 'Arm%', 'Glove%', 'High%', 'Low%', 'MM%', 'Behind%']
# STATS_WITH_PCTLS = ['Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Con%', 'O-Con%', 'SwStr%', 'Whiff%', 'CSW%', 'Strike%', 'Ball%', 'F-Str%', 'PAR%', 'GB%', 'FB%', 'LD%', 'OFFB%', 'IFFB%', 'AIR%', 'Zone%']
# COLUMNS = ['Pitcher', 'Team', 'Throws', 'Pitch', 'Pitch (General)'] + STATS

PITCH_TYPES = [pitch_type for pitch_type in ball_kind.values() if pitch_type != '-']

notes = '''**Limitations**
- [Column widths get messed up when filtering](https://github.com/gradio-app/gradio/issues/11564)

[**To-do**](https://docs.google.com/document/d/1fS-lba94FjczcNI1AXRwtfJMytBQY3jGeQCT_B4tQZ8)
'''

def create_pitch_leaderboard(player_team_type):
  '''Creates entire pitch leaderboard Gradio app given a player/team type'''

  # pitcher/batter, player/team
  pitching = player_team_type in ('pitcher', 'team pitching')
  team = 'team' in player_team_type

  # stats
  if pitching:
    prefix_cols = ['Pitcher', 'Team', 'Throws', 'Pitch', 'Pitch (General)', 'Count', 'Usage', 'Avg Velo', 'Max Velo']
    if team:
      prefix_cols = [col for col in prefix_cols if col not in ('Pitcher', 'Throws')]
    theme_to_cols = {
      'Plate Discipline': ['Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Con%', 'O-Con%', 'SwStr%', 'Whiff%', 'CSW%', 'Strike%', 'Ball%', 'F-Str%', 'PAR%', 'PLUS%'],
      'Batted Ball': ['GB%', 'FB%', 'LD%', 'OFFB%', 'IFFB%', 'AIR%', 'HR/FB', 'HR/OFFB'],
      'Approach': ['Zone%', 'Arm%', 'Glove%', 'High%', 'Low%', 'MM%', 'Behind%']
    }
    
  else:
    prefix_cols = ['Batter', 'Team', 'Bats', 'Pitch', 'Pitch (General)', 'Count', 'Usage', 'Avg Velo', 'Max Velo']
    if team:
      prefix_cols = [col for col in prefix_cols if col not in ('Batter', 'Bats')]
    theme_to_cols = {
      'Plate Discipline': ['Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Con%', 'O-Con%', 'SwStr%', 'Whiff%', 'CSW%', 'PLUS%'],
      'Batted Ball': ['GB%', 'FB%', 'LD%', 'OFFB%', 'IFFB%', 'AIR%', 'HR/FB', 'HR/OFFB']
    }

  all_cols = prefix_cols + sum(list(theme_to_cols.values()), [])
  for theme, cols in theme_to_cols.items():
    theme_to_cols[theme] = prefix_cols + cols

  # col names
  player_type = 'pitcher' if pitching else 'batter'
  id_col = f'{player_type[:3].lower()}Id' if not team else f'{player_type}_team_name_short'

  def gr_create_pitch_leaderboard(
    start_date: datetime, 
    end_date: datetime, 
    min_pitches: int, 
    pitcher_lr: str = 'Both', 
    batter_lr: str = 'Both', 
    include_pitches: list[str] = PITCH_TYPES, 
    include_teams : Optional[list[str]] = None
  ):
    assert pitcher_lr in ['Both', 'Left', 'Right']
    assert batter_lr in ['Both', 'Left', 'Right']

    data = data_df.filter(pl.col('ballKind_code') != '-')
    
    data = filter_data_by_date_and_game_kind(data, start_date=start_date, end_date=end_date, game_kind='Regular Season')

    rename = {
      f'{player_type}_team_name_short': 'Team',
      'count': 'Count', 
      # 'usage': 'Usage',
      'ballKind': 'Pitch', 
      'general_ballKind': 'Pitch (General)', 
    }
    if not team:
      rename[f'{player_type}_name'] = player_type.title()
          
    pitch_stats = (
          compute_pitch_stats(
            data, 
            player_type=player_team_type, 
            min_pitches=min_pitches, 
            pitch_class_type='specific', 
            pitcher_lr='both' if pitcher_lr=='Both' else pitcher_lr[0].lower(), 
            batter_lr='both' if batter_lr == 'Both' else batter_lr[0].lower(), 
            group_by_team=not team
          )
          .filter(pl.col('qualified') & (pl.col('ballKind').is_in(include_pitches)))
          .drop(([id_col] if not team else []) + ['ballKind_code', 'qualified'])
          .rename(rename)
          .with_columns(
            pl.col(stat).mul(100)
            for stat in all_cols if get_stat_val(stat, 'percent', False)
          )
    )

    if include_teams is not None:
      pitch_stats = pitch_stats.filter(pl.col('Team').is_in(include_teams))

    def create_df_value(cols):
      styling = []
      for i, row in enumerate(pitch_stats[cols].iter_rows()):
        styling_row = []
        for col, item in zip(pitch_stats[cols].columns, row):
          # _styling = 'font-size: 0.75em; '
          if get_stat_val(col, 'percentile', False):
            r, g, b = (stat_cmap([pitch_stats[f'{col}_pctl'][i]])[0, :3]*255).astype(np.uint8)
            styling_row.append(f'color: black; background-color: rgba({r}, {g}, {b})')
          elif col == 'Team':
            styling_row.append(f'color: {get_text_color_from_team(item)}; background-color: {team_names_short_to_color[item]}')
          elif col in ['Pitch', 'Pitch (General)']:
            color = ball_kind_to_color[item]
            styling_row.append(f'color: {get_text_color_from_color(color)}; background-color: {color}')
          else:
            styling_row.append('')
        styling.append(styling_row)

      display_value = []
      for row in pitch_stats[cols].iter_rows():
        display_value_row = []
        for col, item in zip(all_cols, row):
          if get_stat_val(col, 'percent', False):
            display_value_row.append(f'{item:.1f}%')
          elif col in ['OBP']:
            # TO-DO: implement this in stat registry
            display_value_row.append(f'{item:.3f}')
          elif isinstance(item, float):
            display_value_row.append(f'{item:.1f}')
          else:
            display_value_row.append(item)
        display_value.append(display_value_row)

      value = {
          'data': pitch_stats[cols].rows(),
          'headers': cols,
          'metadata': {
              'styling': styling,
              'display_value': display_value,
          }
      }
      
      return value

    return [create_df_value(cols) for cols in theme_to_cols.values()]


  now = datetime.now()
  start_datetime_init = datetime(now.year, 1, 1)
  end_datetime_init = now
  with gr.Blocks() as app:
    gr.Markdown(f'# {player_team_type.title()} Pitch Leaderboard')
    with gr.Row():
      start_date = gr.DateTime(start_datetime_init, include_time=False, type='datetime', label='Start')
      end_date = gr.DateTime(end_datetime_init, include_time=False, type='datetime', label='End')
    with gr.Row():
      with gr.Column(scale=3):
        include_pitches = gr.CheckboxGroup(PITCH_TYPES, value=PITCH_TYPES, label='Pitches')
        all_pitches = gr.Button('Select/Deselect all pitches')
      with gr.Column(scale=1):
        if not team:
          min_pitches = gr.Number(100, label='Min. Pitches', precision=0, minimum=0)
        else:
          min_pitches = gr.State(0)
        pitcher_lr = gr.Radio(['Both', 'Left', 'Right'], value='Both', label='Pitcher handedness') 
        batter_lr = gr.Radio(['Both', 'Left', 'Right'], value='Both', label='Batter handedness')
    with gr.Row():
      include_teams = gr.CheckboxGroup(TEAMS, value=TEAMS, label='Teams', scale=3)
      all_teams = gr.Button('Select/Deselect all teams')
        
    search = gr.Button('Search')
    pin_columns = gr.Checkbox(True, label='Pin columns')
    leaderboards = []
    for theme, cols in theme_to_cols.items():
      with gr.Tab(theme):
        leaderboard = gr.DataFrame(
          pl.DataFrame({'Pitcher': [], 'Pitch': []}),
          column_widths=[get_col_width(col, player_team_type) for col in cols],
          show_copy_button=True,
          show_search='filter',
          pinned_columns=1,
          elem_id='leaderboard'
        )
        leaderboards.append(leaderboard)

    gr.Markdown(notes)

    search.click(gr_create_pitch_leaderboard, inputs=[start_date, end_date, min_pitches, pitcher_lr, batter_lr, include_pitches, include_teams], outputs=leaderboards)
    all_pitches.click(lambda _pitch_types : [] if _pitch_types == PITCH_TYPES else PITCH_TYPES, inputs=include_pitches, outputs=include_pitches)
    all_teams.click(lambda _teams : [] if _teams == TEAMS else TEAMS, inputs=include_teams, outputs=include_teams)
    pin_columns.input(
      lambda pin: [gr.DataFrame(pinned_columns=1 if pin else None)]*len(theme_to_cols),
      inputs=pin_columns,
      outputs=leaderboards
    )
    
  return app

create_pitcher_pitch_leaderboard = partial(
  create_pitch_leaderboard,
  player_team_type='pitcher'
)

create_team_pitching_pitch_leaderboard = partial(
  create_pitch_leaderboard,
  player_team_type='team pitching'
)

create_batter_pitch_leaderboard = partial(
  create_pitch_leaderboard,
  player_team_type='batter'
)

create_team_batting_pitch_leaderboard = partial(
  create_pitch_leaderboard,
  player_team_type='team batting'
)

if __name__ == '__main__':
  app = create_pitch_leaderboard()
  app.launch()