import logging import pydantic import requests from typing import Optional from peewee import * from playhouse.shortcuts import model_to_dict from dataclasses import dataclass from helpers import SBA_SEASON, get_player_url from db_calls import get_team_by_owner, get_one_team, get_one_player db = SqliteDatabase( 'storage/gameplay.db', pragmas={ 'journal_mode': 'wal', 'cache_size': -1 * 64000, 'synchronous': 0 } ) PD_DB_URL = 'http://pd-database/api' PD_SEASON = 4 def param_char(other_params): if other_params: return '&' else: return '?' def pd_get_req_url(endpoint: str, api_ver: int = 1, object_id: int = None, params: list = None): req_url = f'{PD_DB_URL}/v{api_ver}/{endpoint}{"/" if object_id is not None else ""}' \ f'{object_id if object_id is not None else ""}' if params: other_params = False for x in params: req_url += f'{param_char(other_params)}{x[0]}={x[1]}' other_params = True logging.info(f'PD GET: {req_url}') return req_url def pd_db_get(endpoint: str, api_ver: int = 1, object_id: int = None, params: list = None, none_okay: bool = True): req_url = pd_get_req_url(endpoint, api_ver=api_ver, object_id=object_id, params=params) logging.info(f'get:\n{endpoint} id: {object_id} params: {params}') resp = requests.get(req_url, timeout=3) if resp.status_code == 200: data = resp.json() logging.info(f'return: {data}') return data elif none_okay: data = resp.json() logging.info(f'return: {data}') return None else: logging.warning(resp.text) raise ValueError(f'DB: {resp.text}') def pd_get_one_team(team_abbrev: str): team = pd_db_get('teams', params=[('abbrev', team_abbrev), ('season', PD_SEASON)], none_okay=False)['teams'][0] return team # req_url = pd_get_req_url('teams', ) # # resp = requests.get(req_url, timeout=3) # if resp.status_code == 200: # data = resp.json() # logging.info(f'return: {data}') # return data['teams'][0] # else: # logging.warning(resp.text) # raise ValueError(f'PD DB: {resp.text}') def pd_get_card_by_id(card_id: int): return pd_db_get('cards', object_id=card_id, none_okay=False) class BaseModel(Model): class Meta: database = db class ManagerAi(BaseModel): name = CharField(unique=True) @dataclass class StratManagerAi: name: str db.create_tables([ManagerAi]) class Game(BaseModel): away_team_id = IntegerField() home_team_id = IntegerField() week_num = IntegerField(null=True) game_num = IntegerField(null=True) channel_id = IntegerField() active = BooleanField(default=True) is_pd = BooleanField(default=False) opp_ai = ForeignKeyField(ManagerAi, null=True) # TODO: add get_away_team and get_home_team that deals with SBa/PD and returns Team object @dataclass class StratGame: id: int away_team_id: int home_team_id: int channel_id: int week_num: int = None game_num: int = None active: bool = True is_pd: bool = False opp_ai: StratManagerAi = None db.create_tables([Game]) # class GameModel(pydantic.BaseModel): # away_team_id: int # home_team_id: int # week_num: Optional[int] = None # game_num: Optional[int] = None # channel_id: int # active: Optional[bool] = True # is_pd: Optional[bool] = True def post_game(game_dict: dict): # game_model = GameModel.parse_obj(game_dict) # new_game = Game( # away_team_id=game_model.away_team_id, # home_team_id=game_model.home_team_id, # week_num=game_model.week_num, # game_num=game_model.game_num, # channel_id=game_model.channel_id, # active=game_model.active, # is_pd=game_model.is_pd # ) # new_game.save() new_game = Game.create(**game_dict) # return_game = model_to_dict(new_game) return_game = StratGame(new_game.id, new_game.away_team_id, new_game.home_team_id, new_game.channel_id, new_game.week_num, new_game.game_num, new_game.active, new_game.is_pd, new_game.opp_ai.id) db.close() return return_game def get_one_game(away_team_id=None, home_team_id=None, week_num=None, game_num=None, channel_id=None, active=None) \ -> Optional[StratGame]: this_game = Game.select() if away_team_id is not None: this_game = this_game.where(Game.away_team_id == away_team_id) if home_team_id is not None: this_game = this_game.where(Game.home_team_id == home_team_id) if week_num is not None: this_game = this_game.where(Game.week_num == week_num) if game_num is not None: this_game = this_game.where(Game.game_num == game_num) if channel_id is not None: this_game = this_game.where(Game.channel_id == channel_id) if active is not None: this_game = this_game.where(Game.active == active) if this_game.count() > 0: # return_game = model_to_dict(this_game[0]) r_game = this_game[0] return_game = StratGame(r_game.id, r_game.away_team_id, r_game.home_team_id, r_game.channel_id, r_game.week_num, r_game.game_num, r_game.active, r_game.is_pd, r_game.opp_ai.id if r_game.opp_ai else None) db.close() return return_game else: return None async def get_game_team(game: StratGame, gm_id: int = None, team_abbrev: str = None, team_id: int = None) -> dict: if not gm_id and not team_abbrev and not team_id: raise KeyError(f'get_game_team requires either one of gm_id, team_abbrev, or team_id to not be None') logging.info(f'getting game team for game {game.id} / gm_id: {gm_id} / ' f'tm_abbrev: {team_abbrev} / team_id: {team_id}') if game.is_pd: if gm_id: return pd_db_get('teams', params=[('season', PD_SEASON), ('gm_id', gm_id)])['teams'][0] elif team_id: return pd_db_get('teams', object_id=team_id) else: return pd_db_get('teams', params=[('season', PD_SEASON), ('abbrev', team_abbrev)])['teams'][0] else: if gm_id: return await get_team_by_owner(season=SBA_SEASON, owner_id=gm_id) elif team_id: return await get_one_team(team_id) else: return await get_one_team(team_abbrev, season=SBA_SEASON) def patch_game( game_id, away_team_id=None, home_team_id=None, week_num=None, game_num=None, channel_id=None, active=None): this_game = Game.get_by_id(game_id) if away_team_id is not None: this_game.away_team_id = away_team_id if home_team_id is not None: this_game.home_team_id = home_team_id if week_num is not None: this_game.week_num = week_num if game_num is not None: this_game.game_num = game_num if channel_id is not None: this_game.channel_id = channel_id if active is not None: this_game.active = active this_game.save() # return_game = model_to_dict(this_game) return_game = StratGame(this_game.id, this_game.away_team_id, this_game.home_team_id, this_game.channel_id, this_game.week_num, this_game.game_num, this_game.active, this_game.is_pd, this_game.opp_ai.id) db.close() return return_game class Lineup(BaseModel): game = ForeignKeyField(Game) team_id = IntegerField() player_id = IntegerField() card_id = IntegerField(null=True) # Only needed for Paper Dynasty games position = CharField() batting_order = IntegerField() after_play = IntegerField() active = BooleanField(default=True) @dataclass class StratLineup: id: int game: StratGame team_id: int player_id: int position: str batting_order: int after_play: int active: bool = True card_id: int = None def convert_stratlineup(lineup: Lineup) -> StratLineup: lineup_dict = model_to_dict(lineup) lineup_dict['game'] = StratGame(**lineup_dict['game']) return StratLineup(**lineup_dict) db.create_tables([Lineup]) def get_one_lineup( game_id: int, lineup_id: int = None, team_id: int = None, batting_order: int = None, position: str = None, active: bool = True, as_obj: bool = False) -> StratLineup: if not batting_order and not position and not lineup_id: raise KeyError(f'One of batting_order, position, or lineup_id must not be None') if lineup_id: this_lineup = Lineup.get_by_id(lineup_id) elif batting_order: this_lineup = Lineup.get_or_none( Lineup.game_id == game_id, Lineup.team_id == team_id, Lineup.batting_order == batting_order, Lineup.active == active ) else: this_lineup = Lineup.get_or_none( Lineup.game_id == game_id, Lineup.team_id == team_id, Lineup.position == position, Lineup.active == active ) logging.info(f'get_one_lineup / this_lineup: {this_lineup}') if as_obj: return this_lineup if not this_lineup: return None # return_value = model_to_dict(this_lineup) # return_value = StratLineup(**vars(this_lineup)["__data__"]) return_value = convert_stratlineup(this_lineup) # return_value = StratLineup( # game=StratGame(**model_to_dict(this_lineup.game)), # team_id=this_lineup.team_id, # player_id=this_lineup.player_id, # position=this_lineup.position, # batting_order=this_lineup.batting_order, # after_play=this_lineup.after_play, # active= # ) db.close() return return_value async def get_team_lineups(game_id: int, team_id: int, as_string: bool = True): all_lineups = Lineup.select().where( (Lineup.game_id == game_id) & (Lineup.team_id == team_id) & (Lineup.active == True) ).order_by(Lineup.batting_order) if all_lineups.count() == 0: return None if as_string: l_string = '' this_game = Game.get_by_id(game_id) for x in all_lineups: l_string += f'{x.batting_order}. {player_link(this_game, await get_player(this_game, x))} - {x.position}\n' db.close() return l_string return_lineups = [] for x in all_lineups: # return_lineups.append(model_to_dict(x)) return_lineups.append(convert_stratlineup(x)) db.close() return return_lineups def post_lineups(lineups: list): with db.atomic(): Lineup.insert_many(lineups).execute() db.close() def patch_lineup(lineup_id, active: Optional[bool] = None) -> StratLineup: this_lineup = Lineup.get_by_id(lineup_id) this_lineup.active = active this_lineup.save() # return_lineup = model_to_dict(this_lineup) return_value = convert_stratlineup(this_lineup) db.close() return return_value def make_sub(lineup_dict: dict): # Check for dupe player / card this_game = Game.get_by_id(lineup_dict['game_id']) if this_game.is_pd: player_conflict = Lineup.select().where( (Lineup.game_id == lineup_dict['game_id']) & (Lineup.team_id == lineup_dict['team_id']) & (Lineup.card_id == lineup_dict['card_id']) ) db_error = f'This card is already in the lineup' else: player_conflict = Lineup.select().where( (Lineup.game_id == lineup_dict['game_id']) & (Lineup.team_id == lineup_dict['team_id']) & (Lineup.player_id == lineup_dict['player_id']) ) db_error = f'This player is already in the lineup' if player_conflict.count(): db.close() raise DatabaseError(db_error) subbed_player = Lineup.get_or_none( Lineup.game_id == lineup_dict['game_id'], Lineup.team_id == lineup_dict['team_id'], Lineup.batting_order == lineup_dict['batting_order'], Lineup.active == True ) logging.info(f'subbed_player: {subbed_player}') if subbed_player: subbed_player = patch_lineup(subbed_player.id, active=False) new_lineup = Lineup.create(**lineup_dict) # return_lineup = model_to_dict(new_lineup) return_value = convert_stratlineup(new_lineup) curr_play = get_current_play(lineup_dict['game_id']) if curr_play.batter.id == subbed_player.id: patch_play(curr_play.id, batter_id=return_value.id) if curr_play.pitcher.id == subbed_player.id: patch_play(curr_play.id, pitcher_id=return_value.id) if curr_play.catcher is not None and curr_play.catcher.id == subbed_player.id: patch_play(curr_play.id, catcher_id=return_value.id) db.close() return return_value async def get_player(game, lineup_member) -> dict: if isinstance(game, Game): if game.is_pd: this_card = pd_get_card_by_id(lineup_member.card_id) player = this_card['player'] player['name'] = player['p_name'] player['team'] = this_card['team'] return player else: return await get_one_player( # lineup_member.player_id if isinstance(lineup_member, Lineup) else lineup_member['player_id'] lineup_member.player_id ) elif isinstance(game, StratGame): if game.is_pd: # card_id = lineup_member.card_id if isinstance(lineup_member, Lineup) else lineup_member['card_id'] card_id = lineup_member['card_id'] if isinstance(lineup_member, dict) else lineup_member.card_id this_card = pd_get_card_by_id(card_id) player = this_card['player'] player['name'] = player['p_name'] player['team'] = this_card['team'] logging.info(f'player: {player}') return player else: return await get_one_player( # lineup_member.player_id if isinstance(lineup_member, Lineup) else lineup_member['player_id'] lineup_member.player_id ) else: raise TypeError(f'Cannot get player; game is not a valid object') def player_link(game, player): if isinstance(game, Game): if game.is_pd: return f'[{player["p_name"]}]({player["image"]})' else: return get_player_url(player) elif isinstance(game, StratGame): if game.is_pd: return f'[{player["p_name"]}]({player["image"]})' else: return get_player_url(player) else: raise TypeError(f'Cannot get player link; game is not a valid object') class Play(BaseModel): game = ForeignKeyField(Game) play_num = IntegerField() batter = ForeignKeyField(Lineup) pitcher = ForeignKeyField(Lineup, null=True) on_base_code = IntegerField() inning_half = CharField() inning_num = IntegerField() batting_order = IntegerField() starting_outs = IntegerField() away_score = IntegerField() home_score = IntegerField() on_first = ForeignKeyField(Lineup, null=True) on_first_final = IntegerField(null=True) on_second = ForeignKeyField(Lineup, null=True) on_second_final = IntegerField(null=True) on_third = ForeignKeyField(Lineup, null=True) on_third_final = IntegerField(null=True) batter_final = IntegerField(null=True) pa = IntegerField(default=0) ab = IntegerField(default=0) # run = IntegerField(null=True) hit = IntegerField(default=0) rbi = IntegerField(default=0) double = IntegerField(default=0) triple = IntegerField(default=0) homerun = IntegerField(default=0) bb = IntegerField(default=0) so = IntegerField(default=0) hbp = IntegerField(default=0) sac = IntegerField(default=0) ibb = IntegerField(default=0) gidp = IntegerField(default=0) bphr = IntegerField(default=0) bpfo = IntegerField(default=0) bp1b = IntegerField(default=0) bplo = IntegerField(default=0) sb = IntegerField(default=0) cs = IntegerField(default=0) outs = IntegerField(default=0) catcher = ForeignKeyField(Lineup, null=True) defender = ForeignKeyField(Lineup, null=True) runner = ForeignKeyField(Lineup, null=True) check_pos = CharField(null=True) error = IntegerField(default=0) wild_pitch = IntegerField(default=0) passed_ball = IntegerField(default=0) pick_off = IntegerField(default=0) balk = IntegerField(default=0) complete = BooleanField(default=False) locked = BooleanField(default=False) @dataclass class StratPlay: id: int game: StratGame play_num: int batter: StratLineup pitcher: StratLineup on_base_code: int inning_half: str inning_num: int batting_order: int starting_outs: int away_score: int home_score: int on_first: StratLineup = None on_first_final: int = None on_second: StratLineup = None on_second_final: int = None on_third: StratLineup = None on_third_final: int = None batter_final: int = None pa: int = 0 ab: int = 0 # run: int = 0 hit: int = 0 rbi: int = 0 double: int = 0 triple: int = 0 homerun: int = 0 bb: int = 0 so: int = 0 hbp: int = 0 sac: int = 0 ibb: int = 0 gidp: int = 0 bphr: int = 0 bpfo: int = 0 bp1b: int = 0 bplo: int = 0 sb: int = 0 cs: int = 0 outs: int = 0 catcher: StratLineup = None defender: StratLineup = None runner: StratLineup = None check_pos: str = None error: int = 0 wild_pitch: int = 0 passed_ball: int = 0 pick_off: int = 0 balk: int = 0 complete: bool = False locked: bool = False def convert_stratplay(play: Play) -> StratPlay: play_dict = model_to_dict(play) play_dict['game'] = StratGame(**play_dict['game']) if play_dict['batter']: play_dict['batter'] = convert_stratlineup(play.batter) if play_dict['pitcher']: play_dict['pitcher'] = convert_stratlineup(play.pitcher) if play_dict['on_first']: play_dict['on_first'] = convert_stratlineup(play.on_first) if play_dict['on_second']: play_dict['on_second'] = convert_stratlineup(play.on_second) if play_dict['on_third']: play_dict['on_third'] = convert_stratlineup(play.on_third) if play_dict['catcher']: play_dict['catcher'] = convert_stratlineup(play.catcher) if play_dict['defender']: play_dict['defender'] = convert_stratlineup(play.defender) if play_dict['runner']: play_dict['runner'] = convert_stratlineup(play.runner) return StratPlay(**play_dict) db.create_tables([Play]) def post_play(play_dict: dict) -> StratPlay: logging.info(f'play_dict: {play_dict}') new_play = Play.create(**play_dict) # return_play = model_to_dict(new_play) return_play = convert_stratplay(new_play) db.close() return return_play def patch_play( play_id, batter_id: int = None, pitcher_id: int = None, catcher_id: int = None, locked: bool = None, pa: int = None, ab: int = None, hit: int = None, double: int = None, triple: int = None, homerun: int = None, outs: int = None, so: int = None, bp1b: int = None, bplo: int = None, bphr: int = None, bpfo: int = None, walk: int = None, hbp: int = None, ibb: int = None, sac: int = None, sb: int = None, cs: int = None, defender_id: int = None, check_pos: str = None, error: int = None, play_num: int = None, on_first_id: int = None, on_first_final: int = None, on_second_id: int = None, on_second_final: int = None, on_third_id: int = None, on_third_final: int = None, starting_outs: int = None, runner_id: int = None, complete: bool = None, rbi: int = None, wp: int = None, pb: int = None, pick: int = None, balk: int = None): this_play = Play.get_by_id(play_id) if batter_id is not None: this_play.batter_id = batter_id if pitcher_id is not None: this_play.pitcher_id = pitcher_id if catcher_id is not None: this_play.catcher_id = catcher_id if runner_id is not None: this_play.runner_id = runner_id if locked is not None: this_play.locked = locked if complete is not None: this_play.complete = complete if pa is not None: this_play.pa = pa if ab is not None: this_play.ab = ab if hit is not None: this_play.hit = hit if rbi is not None: this_play.rbi = rbi if double is not None: this_play.double = double if triple is not None: this_play.triple = triple if homerun is not None: this_play.homerun = homerun if starting_outs is not None: this_play.starting_outs = starting_outs if outs is not None: this_play.outs = outs if so is not None: this_play.so = so if bp1b is not None: this_play.bp1b = bp1b if bplo is not None: this_play.bplo = bplo if bphr is not None: this_play.bphr = bphr if bpfo is not None: this_play.bpfo = bpfo if walk is not None: this_play.bb = walk if hbp is not None: this_play.hbp = hbp if ibb is not None: this_play.ibb = ibb if sac is not None: this_play.sac = sac if sb is not None: this_play.sb = sb if cs is not None: this_play.cs = cs if wp is not None: this_play.wild_pitch = wp if pb is not None: this_play.passed_ball = pb if pick is not None: this_play.pick_off = pick if balk is not None: this_play.balk = balk if defender_id is not None: this_play.defender_id = defender_id if check_pos is not None: this_play.check_pos = check_pos if error is not None: this_play.error = error if play_num is not None: this_play.play_num = play_num if on_first_id is not None: if not on_first_id: this_play.on_first = None else: this_play.on_first_id = on_first_id if on_first_final is not None: if not on_first_final: this_play.on_first_final = None else: this_play.on_first_final = on_first_final if on_second_id is not None: this_play.on_second_id = on_second_id if not on_second_id: this_play.on_second = None else: this_play.on_second_id = on_second_id if on_second_final is not None: this_play.on_second_final = on_second_final if not on_second_final: this_play.on_second_final = None else: this_play.on_second_final = on_second_final if on_third_id is not None: this_play.on_third_id = on_third_id if not on_third_id: this_play.on_third = None else: this_play.on_third_id = on_third_id if on_third_final is not None: this_play.on_third_final = on_third_final if not on_third_final: this_play.on_third_final = None else: this_play.on_third_final = on_third_final this_play.save() # return_play = model_to_dict(this_play) return_play = convert_stratplay(this_play) db.close() return return_play # TODO: UPDATE ALL PLAY REFERENCES TO USE THE STRATPLAY def get_play_by_num(game_id, play_num: int): latest_play = Play.get_or_none(Play.game_id == game_id, Play.play_num == play_num) if not latest_play: return None # return_play = model_to_dict(latest_play) return_play = convert_stratplay(latest_play) db.close() return return_play def get_latest_play(game_id): latest_play = Play.select().where(Play.game_id == game_id).order_by(-Play.id).limit(1) if not latest_play: return None # return_play = model_to_dict(latest_play[0]) return_play = convert_stratplay(latest_play[0]) db.close() return return_play def undo_play(current_play_id): this_play = Play.get_by_id(current_play_id) last_play = Play.get_or_none(Play.game == this_play.game, Play.play_num == this_play.play_num - 1) count = 0 count += this_play.delete_instance() last_play.complete = False last_play.save() db.close() return count def get_current_play(game_id) -> Optional[StratPlay]: curr_play = Play.get_or_none(Play.game_id == game_id, Play.complete == False) if not curr_play: return None # return_play = model_to_dict(curr_play) return_play = convert_stratplay(curr_play) db.close() return return_play def get_last_inning_end_play(game_id, inning_half, inning_num): this_play = Play.select().where( (Play.game_id == game_id) & (Play.inning_half == inning_half) & (Play.inning_num == inning_num) ).order_by(-Play.id) # return_play = model_to_dict(this_play[0]) return_play = convert_stratplay(this_play[0]) db.close() return return_play def advance_runners(play_id: int, num_bases: int, is_error: bool = False, only_forced: bool = False): """ Advances runners and tallies RBIs """ this_play = Play.get_by_id(play_id) this_play.rbi = 0 if only_forced: if not this_play.on_first: if this_play.on_second: this_play.on_second_final = 2 if this_play.on_third: this_play.on_third_final = 3 this_play.save() db.close() return if this_play.on_second: if this_play.on_third: if num_bases > 0: this_play.on_third_final = 4 this_play.rbi += 1 if is_error else 0 if num_bases > 1: this_play.on_second_final = 4 this_play.rbi += 1 if is_error else 0 elif num_bases == 1: this_play.on_second_final = 3 else: this_play.on_second_final = 2 else: if this_play.on_third: this_play.on_third_final = 3 if num_bases > 2: this_play.on_first_final = 4 this_play.rbi += 1 if is_error else 0 elif num_bases == 2: this_play.on_first_final = 3 elif num_bases == 1: this_play.on_first_final = 2 else: this_play.on_first_final = 1 else: if this_play.on_third: if num_bases > 0: this_play.on_third_final = 4 this_play.rbi += 1 if is_error else 0 else: this_play.on_third_final = 3 if this_play.on_second: if num_bases > 1: this_play.on_second_final = 4 this_play.rbi += 1 if is_error else 0 elif num_bases == 1: this_play.on_second_final = 3 else: this_play.on_second_final = 2 if this_play.on_first: if num_bases > 2: this_play.on_first_final = 4 this_play.rbi += 1 if is_error else 0 elif num_bases == 2: this_play.on_first_final = 3 elif num_bases == 1: this_play.on_first_final = 2 else: this_play.on_first_final = 1 if num_bases == 4: this_play.rbi += 1 this_play.save() db.close() def advance_one_runner(play_id: int, from_base: int, num_bases: int): """ Advances runner and excludes RBIs """ this_play = Play.get_by_id(play_id) if from_base == 1: if num_bases > 2: this_play.on_first_final = 4 elif num_bases == 2: this_play.on_first_final = 3 elif num_bases == 1: this_play.on_first_final = 2 elif from_base == 2: if num_bases > 1: this_play.on_second_final = 4 elif num_bases == 1: this_play.on_second_final = 3 elif from_base == 3: if num_bases > 0: this_play.on_third_final = 4 if this_play.on_first and not this_play.on_first_final: this_play.on_first_final = 1 if this_play.on_second and not this_play.on_second_final: this_play.on_second_final = 2 if this_play.on_third and not this_play.on_third_final: this_play.on_third_final = 3 this_play.save() db.close() def complete_play(play_id, batter_to_base: int = None): """ Finalizes current play and sets base values for next play """ this_play = Play.get_by_id(play_id) this_play.locked = False this_play.complete = True this_play.save() logging.info(f'starting the inning calc') new_inning_half = this_play.inning_half new_inning_num = this_play.inning_num if this_play.runner or this_play.wild_pitch or this_play.passed_ball or this_play.pick_off or this_play.balk: new_batting_order = this_play.batting_order else: new_batting_order = this_play.batting_order + 1 if this_play.batting_order < 9 else 1 new_bteam_id = this_play.batter.team_id new_pteam_id = this_play.pitcher.team_id new_starting_outs = this_play.starting_outs + this_play.outs new_on_first = None new_on_second = None new_on_third = None score_increment = this_play.homerun if this_play.starting_outs + this_play.outs > 2: new_starting_outs = 0 new_obc = 0 if this_play.inning_half == 'Top': new_inning_half = 'Bot' new_bteam_id = this_play.game.home_team_id new_pteam_id = this_play.game.away_team_id else: new_inning_half = 'Top' new_inning_num += 1 new_bteam_id = this_play.game.away_team_id new_pteam_id = this_play.game.home_team_id if new_inning_num > 1: last_inning_play = get_last_inning_end_play(this_play.game.id, new_inning_half, new_inning_num - 1) new_batting_order = last_inning_play.batting_order + 1 if last_inning_play.batting_order < 9 else 1 else: new_batting_order = 1 # Not an inning-ending play else: logging.info(f'starting the obc calc') bases_occ = [False, False, False, False] # Set the occupied bases for the next play and lineup member occupying it for runner, base in [ (this_play.on_first, this_play.on_first_final), (this_play.on_second, this_play.on_second_final), (this_play.on_third, this_play.on_third_final) ]: if base: bases_occ[base - 1] = True if base == 1: new_on_first = runner elif base == 2: new_on_second = runner elif base == 3: new_on_third = runner elif base == 4: score_increment += 1 # Set the batter-runner occupied base for the next play if batter_to_base: if batter_to_base == 1: bases_occ[0] = True new_on_first = this_play.batter elif batter_to_base == 2: bases_occ[1] = True new_on_second = this_play.batter elif batter_to_base == 3: bases_occ[2] = True new_on_third = this_play.batter elif batter_to_base == 4: score_increment += 1 this_play.batter_final = batter_to_base this_play.save() # Set the OBC based on the bases occupied if bases_occ[2]: if bases_occ[1]: if bases_occ[0]: new_obc = 7 else: new_obc = 6 elif bases_occ[0]: new_obc = 5 else: new_obc = 3 elif bases_occ[1]: if bases_occ[0]: new_obc = 4 else: new_obc = 2 elif bases_occ[0]: new_obc = 1 else: new_obc = 0 if this_play.inning_half == 'Top': new_away_score = this_play.away_score + score_increment new_home_score = this_play.home_score else: new_away_score = this_play.away_score new_home_score = this_play.home_score + score_increment batter = get_one_lineup(this_play.game.id, team_id=new_bteam_id, batting_order=new_batting_order) batter_id = batter.id if batter else None pitcher = get_one_lineup(this_play.game_id, team_id=new_pteam_id, position='P') pitcher_id = pitcher.id if pitcher else None logging.info(f'done the obc calc') next_play = Play.create(**{ 'game_id': this_play.game.id, 'play_num': this_play.play_num + 1, 'batter_id': batter_id, 'pitcher_id': pitcher_id, 'on_base_code': new_obc, 'inning_half': new_inning_half, 'inning_num': new_inning_num, 'next_inning_num': new_inning_num, 'batting_order': new_batting_order, 'starting_outs': new_starting_outs, 'away_score': new_away_score, 'home_score': new_home_score, 'on_first': new_on_first, 'on_second': new_on_second, 'on_third': new_on_third }) # return_play = model_to_dict(next_play) return_play = convert_stratplay(next_play) db.close() return return_play def get_batting_stats(game_id, lineup_id: int = None, team_id: int = None): batting_stats = Play.select( Play.batter, fn.SUM(Play.pa).over(partition_by=[Play.batter_id]).alias('pl_pa'), fn.SUM(Play.ab).over(partition_by=[Play.batter_id]).alias('pl_ab'), fn.SUM(Play.hit).over(partition_by=[Play.batter_id]).alias('pl_hit'), fn.SUM(Play.rbi).over(partition_by=[Play.batter_id]).alias('pl_rbi'), fn.SUM(Play.double).over(partition_by=[Play.batter_id]).alias('pl_double'), fn.SUM(Play.triple).over(partition_by=[Play.batter_id]).alias('pl_triple'), fn.SUM(Play.homerun).over(partition_by=[Play.batter_id]).alias('pl_homerun'), fn.SUM(Play.bb).over(partition_by=[Play.batter_id]).alias('pl_bb'), fn.SUM(Play.so).over(partition_by=[Play.batter_id]).alias('pl_so'), fn.SUM(Play.hbp).over(partition_by=[Play.batter_id]).alias('pl_hbp'), fn.SUM(Play.sac).over(partition_by=[Play.batter_id]).alias('pl_sac'), fn.SUM(Play.ibb).over(partition_by=[Play.batter_id]).alias('pl_ibb'), fn.SUM(Play.gidp).over(partition_by=[Play.batter_id]).alias('pl_gidp'), fn.SUM(Play.sb).over(partition_by=[Play.batter_id]).alias('pl_sb'), fn.SUM(Play.cs).over(partition_by=[Play.batter_id]).alias('pl_cs'), fn.SUM(Play.bphr).over(partition_by=[Play.batter_id]).alias('pl_bphr'), fn.SUM(Play.bpfo).over(partition_by=[Play.batter_id]).alias('pl_bpfo'), fn.SUM(Play.bp1b).over(partition_by=[Play.batter_id]).alias('pl_bp1b'), fn.SUM(Play.bplo).over(partition_by=[Play.batter_id]).alias('pl_bplo'), fn.SUM(Play.pa).over(partition_by=[Play.batter.team_id]).alias('tm_pa'), fn.SUM(Play.ab).over(partition_by=[Play.batter.team_id]).alias('tm_ab'), fn.SUM(Play.hit).over(partition_by=[Play.batter.team_id]).alias('tm_hit'), fn.SUM(Play.rbi).over(partition_by=[Play.batter.team_id]).alias('tm_rbi'), fn.SUM(Play.double).over(partition_by=[Play.batter.team_id]).alias('tm_double'), fn.SUM(Play.triple).over(partition_by=[Play.batter.team_id]).alias('tm_triple'), fn.SUM(Play.homerun).over(partition_by=[Play.batter.team_id]).alias('tm_homerun'), fn.SUM(Play.bb).over(partition_by=[Play.batter.team_id]).alias('tm_bb'), fn.SUM(Play.so).over(partition_by=[Play.batter.team_id]).alias('tm_so'), fn.SUM(Play.hbp).over(partition_by=[Play.batter.team_id]).alias('tm_hbp'), fn.SUM(Play.sac).over(partition_by=[Play.batter.team_id]).alias('tm_sac'), fn.SUM(Play.ibb).over(partition_by=[Play.batter.team_id]).alias('tm_ibb'), fn.SUM(Play.gidp).over(partition_by=[Play.batter.team_id]).alias('tm_gidp'), fn.SUM(Play.sb).over(partition_by=[Play.batter.team_id]).alias('tm_sb'), fn.SUM(Play.cs).over(partition_by=[Play.batter.team_id]).alias('tm_ccs'), fn.SUM(Play.bphr).over(partition_by=[Play.batter.team_id]).alias('tm_bphr'), fn.SUM(Play.bpfo).over(partition_by=[Play.batter.team_id]).alias('tm_bpfo'), fn.SUM(Play.bp1b).over(partition_by=[Play.batter.team_id]).alias('tm_bp1b'), fn.SUM(Play.bplo).over(partition_by=[Play.batter.team_id]).alias('tm_bplo'), ).join(Lineup, on=Play.batter).where(Play.game_id == game_id) if lineup_id is not None: batting_stats = batting_stats.where(Play.batter_id == lineup_id) elif team_id is not None: batting_stats = batting_stats.where(Play.batter.team_id == team_id) done_batters = [] return_batters = [] for x in batting_stats: if x.batter.id not in done_batters: return_batters.append({ 'batter_id': x.batter_id, 'pl_pa': x.pl_pa, 'pl_ab': x.pl_ab, 'pl_hit': x.pl_hit, 'pl_rbi': x.pl_rbi, 'pl_double': x.pl_double, 'pl_triple': x.pl_triple, 'pl_homerun': x.pl_homerun, 'pl_bb': x.pl_bb, 'pl_so': x.pl_so, 'pl_hbp': x.pl_hbp, 'pl_sac': x.pl_sac, 'pl_ibb': x.pl_ibb, 'pl_gidp': x.pl_gidp, 'pl_sb': x.pl_sb, 'pl_cs': x.pl_cs, 'pl_bphr': x.pl_bphr, 'pl_bpfo': x.pl_bpfo, 'pl_bp1b': x.pl_bp1b, 'pl_bplo': x.pl_bplo, }) done_batters.append(x.batter.id) db.close() logging.info(f'batting stats: {return_batters}') return return_batters def get_pitching_stats(game_id, lineup_id: int = None, team_id: int = None): pitching_stats = Play.select( Play.pitcher, fn.SUM(Play.outs).over(partition_by=[Play.pitcher_id]).alias('pl_outs'), fn.SUM(Play.hit).over(partition_by=[Play.pitcher_id]).alias('pl_hit'), fn.COUNT(Play.on_first_final).filter( Play.on_first_final == 4).over(partition_by=[Play.pitcher_id]).alias('pl_run_first'), fn.COUNT(Play.on_second_final).filter( Play.on_second_final == 4).over(partition_by=[Play.pitcher_id]).alias('pl_run_second'), fn.COUNT(Play.on_third_final).filter( Play.on_third_final == 4).over(partition_by=[Play.pitcher_id]).alias('pl_run_third'), fn.COUNT(Play.batter_final).filter( Play.batter_final == 4).over(partition_by=[Play.pitcher_id]).alias('pl_run_batter'), fn.SUM(Play.so).over(partition_by=[Play.pitcher_id]).alias('pl_so'), fn.SUM(Play.bb).over(partition_by=[Play.pitcher_id]).alias('pl_bb'), fn.SUM(Play.hbp).over(partition_by=[Play.pitcher_id]).alias('pl_hbp'), fn.SUM(Play.wild_pitch).over(partition_by=[Play.pitcher_id]).alias('pl_wild_pitch'), fn.SUM(Play.balk).over(partition_by=[Play.pitcher_id]).alias('pl_balk'), fn.SUM(Play.homerun).over(partition_by=[Play.pitcher_id]).alias('pl_homerun'), fn.SUM(Play.outs).over(partition_by=[Play.pitcher.team_id]).alias('tm_outs'), fn.SUM(Play.hit).over(partition_by=[Play.pitcher.team_id]).alias('tm_hit'), fn.COUNT(Play.on_first_final).filter( Play.on_first_final == 4).over(partition_by=[Play.pitcher.team_id]).alias('tm_run_first'), fn.COUNT(Play.on_second_final).filter( Play.on_second_final == 4).over(partition_by=[Play.pitcher.team_id]).alias('tm_run_second'), fn.COUNT(Play.on_third_final).filter( Play.on_third_final == 4).over(partition_by=[Play.pitcher.team_id]).alias('tm_run_third'), fn.COUNT(Play.batter_final).filter( Play.batter_final == 4).over(partition_by=[Play.pitcher.team_id]).alias('tm_run_batter'), fn.SUM(Play.so).over(partition_by=[Play.pitcher.team_id]).alias('tm_so'), fn.SUM(Play.bb).over(partition_by=[Play.pitcher.team_id]).alias('tm_bb'), fn.SUM(Play.hbp).over(partition_by=[Play.pitcher.team_id]).alias('tm_hbp'), fn.SUM(Play.wild_pitch).over(partition_by=[Play.pitcher.team_id]).alias('tm_wild_pitch'), fn.SUM(Play.balk).over(partition_by=[Play.pitcher.team_id]).alias('tm_balk'), fn.SUM(Play.homerun).over(partition_by=[Play.pitcher.team_id]).alias('tm_homerun'), ).join(Lineup, on=Play.pitcher).where(Play.game_id == game_id) earned_runs_pl = Play.select().where( ((Play.on_first_final == 4) | (Play.on_second_final == 4) | (Play.on_third_final == 4) | (Play.batter_final == 4)) & (Play.error == 0) ).join(Lineup, on=Play.pitcher).where(Play.game_id == game_id) logging.info(f'earned_runs: {earned_runs_pl}') earned_runs_tm = Play.select().where( ((Play.on_first_final == 4) | (Play.on_second_final == 4) | (Play.on_third_final == 4) | (Play.batter_final == 4)) & (Play.error == 0) ).join(Lineup, on=Play.pitcher).where(Play.game_id == game_id) if lineup_id is not None: pitching_stats = pitching_stats.where(Play.pitcher_id == lineup_id) earned_runs_pl = earned_runs_pl.where(Play.pitcher_id == lineup_id) earned_runs_tm = earned_runs_tm.where(Play.pitcher_id == lineup_id) if team_id is not None: pitching_stats = pitching_stats.where(Play.pitcher.team_id == team_id) earned_runs_pl = earned_runs_pl.where(Play.pitcher.team_id == team_id) earned_runs_tm = earned_runs_tm.where(Play.pitcher.team_id == team_id) logging.info(f'query: {pitching_stats}') logging.info(f'count: {pitching_stats.count()}') logging.info(f'first entry: {pitching_stats[0]}') done_pitchers = [] return_pitchers = [] for x in pitching_stats: logging.info(f'x: {x}') if x.pitcher.id not in done_pitchers: return_pitchers.append({ 'pitcher_id': x.pitcher_id, 'pl_outs': x.pl_outs, 'pl_hit': x.pl_hit, 'pl_eruns': earned_runs_pl.count(), 'pl_runs': x.pl_run_first + x.pl_run_second + x.pl_run_third + x.pl_run_batter, 'pl_so': x.pl_so, 'pl_bb': x.pl_bb, 'pl_hbp': x.pl_hbp, 'pl_homerun': x.pl_homerun, 'pl_wild_pitch': x.pl_wild_pitch, 'pl_balk': x.pl_balk, 'tm_outs': x.tm_outs, 'tm_hit': x.tm_hit, 'tm_eruns': earned_runs_tm.count(), 'tm_runs': x.tm_run_first + x.tm_run_second + x.tm_run_third + x.tm_run_batter, 'tm_so': x.tm_so, 'tm_bb': x.tm_bb, 'tm_hbp': x.tm_hbp, 'tm_homerun': x.tm_homerun, 'tm_wild_pitch': x.tm_wild_pitch, 'tm_balk': x.tm_balk, }) done_pitchers.append(x.pitcher_id) db.close() logging.info(f'pitching stats: {return_pitchers}') return return_pitchers