pitch_leaderboard.py

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

from datetime import datetime
# from itertools import chain

from data import data_df
from stats import compute_pitch_stats, filter_data_by_date_and_game_kind
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

STATS = ['Count', 'Usage', 'Avg Velo', 'Max Velo', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', '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-Contact%', 'O-Contact%', '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-Contact%', 'O-Contact%', '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 != '-']
TEAMS  = [
    'Yomiuri',
    'Yakult',
    'DeNA',
    'Chunichi',
    'Hanshin',
    'Hiroshima',
    'Nipponham',
    'Rakuten',
    'Seibu',
    'Lotte',
    'ORIX',
    'SoftBank'
]
notes = '''**Limitations**
- [Column widths get messed up when filtering](https://github.com/gradio-app/gradio/issues/11564)

**To-do**
- Add post-season
'''


def gr_create_pitch_leaderboard(start_date, end_date, min_pitches, pitcher_lr='Both', batter_lr='Both', include_pitches=PITCH_TYPES, include_teams=None):
  assert pitcher_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')
    
  pitch_stats = (
        compute_pitch_stats(
          data, 
          player_type='pitcher', 
          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=True
        )
        .filter(pl.col('qualified') & (pl.col('ballKind').is_in(include_pitches)))
        .drop('pitId', 'ballKind_code', 'qualified')
        .rename({
          'pitcher_name': 'Pitcher', 
          'pitcher_team_name_short': 'Team', 
          'count': 'Count', 
          'usage': 'Usage', 
          'ballKind': 'Pitch', 
          'general_ballKind': 'Pitch (General)', 
          'Avg MPH': 'Avg Velo', 
          'Max MPH': 'Max Velo',
          'Avg MPH_pctl': 'Avg Velo_pctl', 
          'Max MPH_pctl': 'Max Velo_pctl'
        })
        .with_columns(
          pl.col(stat).mul(100)
          for stat in PCT_STATS
        )
  )

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

  styling = []
  for i, row in enumerate(pitch_stats[COLUMNS].iter_rows()):
    styling_row = []
    for col, item in zip(pitch_stats[COLUMNS].columns, row):
      # _styling = 'font-size: 0.75em; '
      if col in STATS_WITH_PCTLS:
        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[COLUMNS].iter_rows():
    display_value_row = []
    for col, item in zip(COLUMNS, row):
      if col in PCT_STATS:
        display_value_row.append(f'{item:.1f}%')
      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[COLUMNS].rows(),
      'headers': COLUMNS,
      'metadata': {
          'styling': styling,
          'display_value': display_value,
      }
  }
  
  return value


def create_pitch_leaderboard():
  now = datetime.now()
  start_datetime_init = datetime(now.year, 1, 1)
  end_datetime_init = now
  with gr.Blocks() as app:
    gr.Markdown('# 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):
        min_pitches = gr.Number(100, label='Min. Pitches', precision=0, minimum=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.Button('Pin columns')
    pin_columns = gr.Checkbox(True, label='Pin columns')
    leaderboard = gr.DataFrame(
      pl.DataFrame({'Pitcher': [], 'Pitch': []}),
      column_widths=[125, 75, 60, 125, 125] + [max(50, 10*len(stat)) for stat in STATS],
      show_copy_button=True,
      show_search='filter',
      pinned_columns=1,
      elem_id='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=leaderboard)
    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_columns : gr.update(pinned_columns=None if _pin_columns else 3),
      # inputs=pin_columns,
      # outputs=leaderboard
    # )
    pin_columns.input(
      lambda pin: gr.DataFrame(pinned_columns=1 if pin else None),
      inputs=pin_columns,
      outputs=leaderboard
    )
    
  return app

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