import math from typing import Optional import pydantic from peewee import * from playhouse.shortcuts import model_to_dict import logging db = SqliteDatabase( 'storage/gameday.db', pragmas={ 'journal_mode': 'wal', 'cache_size': -1 * 64000, 'synchronous': 0 } ) class BaseModel(Model): class Meta: database = db 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) 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() return_game = model_to_dict(new_game) db.close() new_gs = post_game_state({'game_id': new_game.id}) return return_game def get_game(away_team_id=None, home_team_id=None, week_num=None, game_num=None, channel_id=None, active=None): 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]) db.close() return return_game else: raise DatabaseError('No game found') 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) db.close() return return_game class AtBat(BaseModel): game = ForeignKeyField(Game) batter_id = IntegerField() pitcher_id = IntegerField() on_base_code = IntegerField() inning = IntegerField() pa = IntegerField() ab = IntegerField() run = IntegerField() hit = IntegerField() rbi = IntegerField() double = IntegerField() triple = IntegerField() homerun = IntegerField() bb = IntegerField() so = IntegerField() hbp = IntegerField() sac = IntegerField() ibb = IntegerField() gidp = IntegerField() bphr = IntegerField() bpfo = IntegerField() bp1b = IntegerField() bplo = IntegerField() error = IntegerField() class AtBatModel(pydantic.BaseModel): game_id: int batter_id: int pitcher_id: int on_base_code: int inning: int pa: Optional[int] = 1 ab: Optional[int] = 0 run: Optional[int] = 0 hit: Optional[int] = 0 rbi: Optional[int] = 0 double: Optional[int] = 0 triple: Optional[int] = 0 homerun: Optional[int] = 0 bb: Optional[int] = 0 so: Optional[int] = 0 hbp: Optional[int] = 0 sac: Optional[int] = 0 ibb: Optional[int] = 0 gidp: Optional[int] = 0 bphr: Optional[int] = 0 bpfo: Optional[int] = 0 bp1b: Optional[int] = 0 bplo: Optional[int] = 0 error: Optional[int] = False def post_atbat(ab_dict: dict): this_ab = AtBatModel.parse_obj(ab_dict) this_game = Game.get_by_id(this_ab.game_id) new_ab = AtBat( game=this_game, batter_id=this_ab.batter_id, pitcher_id=this_ab.pitcher_id, on_base_code=this_ab.on_base_code, inning=this_ab.inning, pa=this_ab.pa, ab=this_ab.ab, run=this_ab.run, hit=this_ab.hit, rbi=this_ab.rbi, double=this_ab.double, triple=this_ab.triple, homerun=this_ab.homerun, bb=this_ab.bb, so=this_ab.so, hbp=this_ab.hbp, sac=this_ab.sac, ibb=this_ab.ibb, gidp=this_ab.gidp, bphr=this_ab.bphr, bpfo=this_ab.bpfo, bp1b=this_ab.bp1b, bplo=this_ab.bplo, error=this_ab.error ) new_ab.save() return_ab = model_to_dict(new_ab) db.close() return return_ab def get_atbat(game_id=None, batter_id=None, pitcher_id=None, on_base_code=None, inning=None, error=None): this_ab = AtBat.select() if game_id is not None: this_game = Game.get_by_id(game_id) this_ab = this_ab.where(AtBat.game == this_game) if batter_id is not None: this_ab = this_ab.where(AtBat.batter_id == batter_id) if pitcher_id is not None: this_ab = this_ab.where(AtBat.pitcher_id == pitcher_id) if on_base_code is not None: this_ab = this_ab.where(AtBat.on_base_code == on_base_code) if inning is not None: this_ab = this_ab.where(AtBat.inning == inning) if inning is not None: this_ab = this_ab.where(AtBat.error == error) return_data = {'atbats': []} for x in this_ab: return_data['atbats'].append(model_to_dict(x)) db.close() return return_data def get_one_atbat(game_id=None, batter_id=None, pitcher_id=None, on_base_code=None, inning=None, error=None): all_abs = get_atbat(game_id, batter_id, pitcher_id, on_base_code, inning, error) if len(all_abs['atbats']) == 0: raise KeyError('At Bat not found') else: return all_abs['atbats'][len(all_abs['atbats']) - 1] def get_atbat_totals(lineup_member_id): this_member = LineupMember.get_by_id(lineup_member_id) all_abs = AtBat.select( fn.COUNT(AtBat.ab).alias('pas'), fn.SUM(AtBat.ab).alias('abs'), fn.SUM(AtBat.run).alias('runs'), fn.SUM(AtBat.hit).alias('hits'), fn.SUM(AtBat.rbi).alias('rbis'), fn.SUM(AtBat.double).alias('doubles'), fn.SUM(AtBat.triple).alias('triples'), fn.SUM(AtBat.homerun).alias('hrs'), fn.SUM(AtBat.bb).alias('bbs'), fn.SUM(AtBat.so).alias('sos'), fn.SUM(AtBat.hbp).alias('hbps'), fn.SUM(AtBat.sac).alias('sacs'), fn.SUM(AtBat.ibb).alias('ibbs'), fn.SUM(AtBat.gidp).alias('gidps'), fn.SUM(AtBat.bphr).alias('bphrs'), fn.SUM(AtBat.bpfo).alias('bpfos'), fn.SUM(AtBat.bp1b).alias('bp1bs'), fn.SUM(AtBat.bplo).alias('bplos'), ).where((AtBat.batter_id == this_member.player_id) & (AtBat.game == this_member.game) & (AtBat.pa > 0)) if all_abs.count() > 0: this_line = { 'empty': False, 'pa': all_abs[0].pas if all_abs[0].pas else '', 'ab': all_abs[0].abs if all_abs[0].abs else '', 'run': all_abs[0].runs if all_abs[0].runs else '', 'hit': all_abs[0].hits if all_abs[0].hits else '', 'rbi': all_abs[0].rbis if all_abs[0].rbis else '', 'double': all_abs[0].doubles if all_abs[0].doubles else '', 'triple': all_abs[0].triples if all_abs[0].triples else '', 'homerun': all_abs[0].hrs if all_abs[0].hrs else '', 'bb': all_abs[0].bbs if all_abs[0].bbs else '', 'so': all_abs[0].sos if all_abs[0].sos else '', 'hbp': all_abs[0].hbps if all_abs[0].hbps else '', 'sac': all_abs[0].sacs if all_abs[0].sacs else '', 'ibb': all_abs[0].ibbs if all_abs[0].ibbs else '', 'gidp': all_abs[0].gidps if all_abs[0].gidps else '', 'bphr': all_abs[0].bphrs if all_abs[0].bphrs else '', 'bpfo': all_abs[0].bpfos if all_abs[0].bpfos else '', 'bp1b': all_abs[0].bp1bs if all_abs[0].bp1bs else '', 'bplo': all_abs[0].bplos if all_abs[0].bplos else '' } else: this_line = { 'empty': True, 'pa': '', 'ab': '', 'run': '', 'hit': '', 'rbi': '', 'double': '', 'triple': '', 'homerun': '', 'bb': '', 'so': '', 'hbp': '', 'sac': '', 'ibb': '', 'gidp': '', 'bphr': '', 'bpfo': '', 'bp1b': '', 'bplo': '' } return this_line def get_atbat_pitching_totals(pitcher_member_id): this_member = LineupMember.get_by_id(pitcher_member_id) totals = AtBat.select( fn.SUM(AtBat.hit).alias('hits'), fn.SUM(AtBat.run).alias('runs'), fn.SUM(AtBat.so).alias('sos'), fn.SUM(AtBat.bb).alias('bbs'), fn.SUM(AtBat.ibb).alias('ibbs'), fn.SUM(AtBat.hbp).alias('hbps'), fn.SUM(AtBat.homerun).alias('hrs'), fn.SUM(AtBat.gidp).alias('gidp'), ).where( (AtBat.pitcher_id == this_member.player_id) & (AtBat.game == this_member.game) ) earned_runs = AtBat.select( fn.Sum(AtBat.run).alias('eruns') ).where( (AtBat.pitcher_id == this_member.player_id) & (AtBat.game == this_member.game) & (AtBat.error == 0) ) outs_recorded = AtBat.select().where( (AtBat.pitcher_id == this_member.player_id) & (AtBat.game == this_member.game) & (AtBat.hit == 0) & (AtBat.bb == 0) & (AtBat.hbp == 0) & (AtBat.error == 0) ) if outs_recorded.count() > 0: complete_inn = math.floor((outs_recorded.count() + totals[0].gidp) / 3) extra_outs = (outs_recorded.count() + totals[0].gidp) % 3 logging.info(f'comp_in: {complete_inn} / extra_outs: {extra_outs}') ip = complete_inn + (extra_outs / 3.0) else: ip = 0.0 if totals.count() > 0: walks = totals[0].bbs + totals[0].ibbs logging.info(f'bbs: {totals[0].bbs} / ibbs: {totals[0].ibbs} / walks: {walks}') this_line = { 'empty': False, 'ip': ip, 'hit': totals[0].hits if totals[0].hits else '', 'run': totals[0].runs if totals[0].runs else '', 'erun': earned_runs[0].eruns if earned_runs[0].eruns else '', 'so': totals[0].sos if totals[0].sos else '', 'bb': walks if walks else '', 'hbp': totals[0].hbps if totals[0].hbps else '', 'hr': totals[0].hrs if totals[0].hrs else '', } else: this_line = { 'empty': True, 'ip': '', 'hit': '', 'run': '', 'erun': '', 'so': '', 'bb': '', 'hbp': '', 'hr': '', } return this_line def patch_atbat(atbat_id, game_id=None, batter_id=None, pitcher_id=None, on_base_code=None, pa=None, ab=None, run=None, hit=None, rbi=None, double=None, triple=None, homerun=None, bb=None, so=None, hbp=None, sac=None, ibb=None, gidp=None, bphr=None, bpfo=None, bp1b=None, bplo=None, inning=None, error=None): this_ab = AtBat.get_by_id(atbat_id) if game_id is not None: this_ab.game_id = game_id if batter_id is not None: this_ab.batter_id = batter_id if pitcher_id is not None: this_ab.pitcher_id = pitcher_id if on_base_code is not None: this_ab.on_base_code = on_base_code if pa is not None: this_ab.pa = pa if ab is not None: this_ab.ab = ab if run is not None: this_ab.run = run if hit is not None: this_ab.hit = hit if rbi is not None: this_ab.rbi = rbi if double is not None: this_ab.double = double if triple is not None: this_ab.triple = triple if homerun is not None: this_ab.homerun = homerun if bb is not None: this_ab.bb = bb if so is not None: this_ab.so = so if hbp is not None: this_ab.hbp = hbp if sac is not None: this_ab.sac = sac if ibb is not None: this_ab.ibb = ibb if gidp is not None: this_ab.gidp = gidp if bphr is not None: this_ab.bphr = bphr if bpfo is not None: this_ab.bpfo = bpfo if bp1b is not None: this_ab.bp1b = bp1b if bplo is not None: this_ab.bplo = bplo if inning is not None: this_ab.inning = inning if error is not None: this_ab.error = error this_ab.save() return_ab = model_to_dict(this_ab) db.close() return return_ab class Running(BaseModel): game = ForeignKeyField(Game) runner_id = IntegerField() stolen_base = IntegerField() caught_stealing = IntegerField() extra_base_attempt = IntegerField() extra_base_taken = IntegerField() class RunningModel(pydantic.BaseModel): game_id: int runner_id: int stolen_base: Optional[int] = 0 caught_stealing: Optional[int] = 0 extra_base_attempt: Optional[int] = 0 extra_base_taken: Optional[int] = 0 def post_running(running_dict: dict): running_model = RunningModel.parse_obj(running_dict) new_running = Running( game=Game.get_by_id(running_model.game_id), runner_id=running_model.runner_id, stolen_base=running_model.stolen_base, caught_stealing=running_model.caught_stealing, extra_base_attempt=running_model.extra_base_attempt, extra_base_taken=running_model.extra_base_taken, ) new_running.save() return_running = model_to_dict(new_running) db.close() return return_running def get_running( game_id=None, runner_id=None, stolen_base=None, caught_stealing=None, extra_base_attempt=None, extra_base_taken=None): this_running = Running.select() if game_id: this_running = this_running.where(Running.game == Game.get_by_id(game_id)) if runner_id: this_running = this_running.where(Running.runner_id == runner_id) if stolen_base: this_running = this_running.where(Running.stolen_base == stolen_base) if caught_stealing: this_running = this_running.where(Running.caught_stealing == caught_stealing) if extra_base_attempt: this_running = this_running.where(Running.extra_base_attempt == extra_base_attempt) if extra_base_taken: this_running = this_running.where(Running.extra_base_taken == extra_base_taken) return_data = {'running': []} for x in this_running: return_data['running'].append(model_to_dict(x)) db.close() return return_data def get_running_totals(lineup_member_id): this_member = LineupMember.get_by_id(lineup_member_id) all_running = Running.select( fn.SUM(Running.stolen_base).alias('sb'), fn.SUM(Running.caught_stealing).alias('cs'), fn.SUM(Running.extra_base_attempt).alias('xba'), fn.SUM(Running.extra_base_taken).alias('xbt'), ).where( (Running.runner_id == this_member.player_id) & (Running.game == this_member.game) ) if all_running.count() > 0: this_line = { 'empty': False, 'sb': all_running[0].sb if all_running[0].sb else '', 'cs': all_running[0].cs if all_running[0].cs else '', 'xba': all_running[0].xba if all_running[0].xba else '', 'xbt': all_running[0].xbt if all_running[0].xbt else '', } else: this_line = { 'empty': True, 'sb': '', 'cs': '', 'xba': '', 'xbt': '', } return this_line def patch_running( running_id, game_id=None, runner_id=None, stolen_base=None, caught_stealing=None, extra_base_attempt=None, extra_base_taken=None): this_running = Running.get_by_id(running_id) if game_id is not None: this_running.game = Game.get_by_id(game_id) if runner_id is not None: this_running.runner_id = runner_id if stolen_base is not None: this_running.stolen_base = stolen_base if caught_stealing is not None: this_running.caught_stealing = caught_stealing if extra_base_attempt is not None: this_running.extra_base_attempt = extra_base_attempt if extra_base_taken is not None: this_running.extra_base_taken = extra_base_taken this_running.save() return_running = model_to_dict(this_running) db.close() return return_running class Defense(BaseModel): game = ForeignKeyField(Game) at_bat = ForeignKeyField(AtBat, null=True) player_id = IntegerField() x_check = BooleanField() position = CharField() hit = BooleanField() error = BooleanField() stolen_base_attempt = BooleanField() caught_stealing = BooleanField() rob_attempt = BooleanField() rob_success = BooleanField() runner_adv_att = BooleanField() runner_throw_out = BooleanField() class DefenseModel(pydantic.BaseModel): game_id: int at_bat_id: Optional[int] = 0 player_id: int x_check: Optional[int] = 0 position: str hit: Optional[int] = 0 error: Optional[int] = 0 stolen_base_attempt: Optional[int] = 0 caught_stealing: Optional[int] = 0 rob_attempt: Optional[int] = 0 rob_success: Optional[int] = 0 runner_adv_att: Optional[int] = 0 runner_throw_out: Optional[int] = 0 def post_defense(defense_dict: dict): defense_model = DefenseModel.parse_obj(defense_dict) new_defense = Defense( game=Game.get_by_id(defense_model.game_id), at_bat=AtBat.get_by_id(defense_model.at_bat_id) if defense_model.at_bat_id else None, player_id=defense_model.player_id, x_check=defense_model.x_check, position=defense_model.position, hit=defense_model.hit, error=defense_model.error, stolen_base_attempt=defense_model.stolen_base_attempt, caught_stealing=defense_model.caught_stealing, rob_attempt=defense_model.rob_attempt, rob_success=defense_model.rob_success, runner_adv_att=defense_model.runner_adv_att, runner_throw_out=defense_model.runner_throw_out, ) new_defense.save() return_defense = model_to_dict(new_defense) db.close() return return_defense def get_defense( game_id=None, at_bat_id=None, player_id=None, x_check=None, position=None, hit=None, error=None, sba=None, cs=None, roba=None, robs=None, raa=None, rto=None): this_defense = Defense.select() if game_id: this_defense = this_defense.where(Defense.game == Game.get_by_id(game_id)) if at_bat_id: this_defense = this_defense.where(Defense.at_bat == AtBat.get_by_id(at_bat_id)) if player_id: this_defense = this_defense.where(Defense.player_id == player_id) if x_check: this_defense = this_defense.where(Defense.x_check == x_check) if position: this_defense = this_defense.where(Defense.position == position) if hit: this_defense = this_defense.where(Defense.hit == hit) if error: this_defense = this_defense.where(Defense.error == error) if sba: this_defense = this_defense.where(Defense.sba == sba) if cs: this_defense = this_defense.where(Defense.cs == cs) if roba: this_defense = this_defense.where(Defense.roba == roba) if robs: this_defense = this_defense.where(Defense.robs == robs) if raa: this_defense = this_defense.where(Defense.raa == raa) if rto: this_defense = this_defense.where(Defense.rto == rto) return_data = {'defense': []} for x in this_defense: return_data['defense'].append(model_to_dict(x)) db.close() return return_data def get_defense_totals(lineup_member_id): this_member = LineupMember.get_by_id(lineup_member_id) all_pos = Defense.select(Defense.position).where( (Defense.player_id == this_member.player_id) & (Defense.game == this_member.game) ) all_lines = [] for x in all_pos: all_defense = Defense.select( fn.SUM(Defense.x_check).alias('xch'), fn.SUM(Defense.hit).alias('xhit'), fn.SUM(Defense.error).alias('errors'), fn.SUM(Defense.stolen_base_attempt).alias('sba'), fn.SUM(Defense.caught_stealing).alias('csc'), fn.SUM(Defense.rob_attempt).alias('roba'), fn.SUM(Defense.rob_success).alias('robs'), fn.SUM(Defense.runner_adv_att).alias('raa'), fn.SUM(Defense.runner_throw_out).alias('rto'), ).where( (Defense.player_id == this_member.player_id) & (Defense.game == this_member.game) & (Defense.position == x.position) ) all_lines.append({ 'pos': x.position, 'xch': all_defense[0].xch if all_defense[0].xch else '', 'xhit': all_defense[0].xhit if all_defense[0].xhit else '', 'errors': all_defense[0].errors if all_defense[0].errors else '', 'sba': all_defense[0].sba if all_defense[0].sba else '', 'csc': all_defense[0].csc if all_defense[0].csc else '', 'roba': all_defense[0].roba if all_defense[0].roba else '', 'robs': all_defense[0].robs if all_defense[0].robs else '', 'raa': all_defense[0].raa if all_defense[0].raa else '', 'rto': all_defense[0].rto if all_defense[0].rto else '', }) db.close() return {'defense': all_lines} def patch_defense( defense_id, game_id=None, at_bat_id=None, player_id=None, x_check=None, position=None, hit=None, error=None, sba=None, cs=None, roba=None, robs=None, raa=None, rto=None): this_defense = Defense.get_by_id(defense_id) if game_id is not None: this_defense.game = Game.get_by_id(game_id) if at_bat_id is not None: this_defense.at_bat = AtBat.get_by_id(at_bat_id) if player_id is not None: this_defense.x_check = x_check if position is not None: this_defense.position = position if hit is not None: this_defense.hit = hit if error is not None: this_defense.error = error if sba is not None: this_defense.sba = sba if cs is not None: this_defense.cs = cs if roba is not None: this_defense.roba = roba if robs is not None: this_defense.robs = robs if raa is not None: this_defense.raa = raa if rto is not None: this_defense.rto = rto this_defense.save() return_defense = model_to_dict(this_defense) db.close() return return_defense class Chaos(BaseModel): game = ForeignKeyField(Game) pitcher_id = IntegerField() catcher_id = IntegerField() wild_pitch = IntegerField() passed_ball = IntegerField() pick_off = IntegerField() balk = IntegerField() class ChaosModel(pydantic.BaseModel): game_id: int pitcher_id: int catcher_id: int wild_pitch: int = 0 passed_ball: int = 0 pick_off: int = 0 balk: int = 0 def post_chaos(chaos_dict: dict): chaos_model = ChaosModel.parse_obj(chaos_dict) new_chaos = Chaos( game=Game.get_by_id(chaos_model.game_id), pitcher_id=chaos_model.pitcher_id, catcher_id=chaos_model.catcher_id, wild_pitch=chaos_model.wild_pitch, passed_ball=chaos_model.passed_ball, pick_off=chaos_model.pick_off, balk=chaos_model.balk, ) new_chaos.save() return_chaos = model_to_dict(new_chaos) db.close() return return_chaos def get_chaos( game_id=None, pitcher_id=None, catcher_id=None, wild_pitch=None, passed_ball=None, pick_off=None, balk=None): this_chaos = Chaos.select() if game_id: this_chaos = this_chaos.where(Chaos.game == Game.get_by_id(game_id)) if pitcher_id: this_chaos = this_chaos.where(Chaos.pitcher_id == pitcher_id) if catcher_id: this_chaos = this_chaos.where(Chaos.catcher_id == catcher_id) if wild_pitch: this_chaos = this_chaos.where(Chaos.wild_pitch == wild_pitch) if passed_ball: this_chaos = this_chaos.where(Chaos.passed_ball == passed_ball) if pick_off: this_chaos = this_chaos.where(Chaos.pick_off == pick_off) if balk: this_chaos = this_chaos.where(Chaos.balk == balk) return_data = {'chaos': []} for x in this_chaos: return_data['chaos'].append(model_to_dict(x)) db.close() return return_data def get_chaos_totals(lineup_member_id): this_member = LineupMember.get_by_id(lineup_member_id) all_chaos = Chaos.select( fn.SUM(Chaos.wild_pitch).alias('wp'), fn.SUM(Chaos.passed_ball).alias('pb'), fn.SUM(Chaos.pick_off).alias('po'), fn.SUM(Chaos.balk).alias('bk'), ).where( ((Chaos.pitcher_id == this_member.player_id) | (Chaos.catcher_id == this_member.player_id)) & (Chaos.game == this_member.game) ) if all_chaos.count() > 0: this_line = { 'empty': False, 'wp': all_chaos[0].wp if all_chaos[0].wp else '', 'pb': all_chaos[0].pb if all_chaos[0].pb else '', 'po': all_chaos[0].po if all_chaos[0].po else '', 'bk': all_chaos[0].bk if all_chaos[0].bk else '', } else: this_line = { 'empty': True, 'wp': '', 'pb': '', 'po': '', 'bk': '', } return this_line def patch_chaos( chaos_id, pitcher_id=None, catcher_id=None, wild_pitch=None, passed_ball=None, pick_off=None, balk=None): this_chaos = Chaos.get_by_id(chaos_id) if pitcher_id is not None: this_chaos.pitcher_id = pitcher_id if catcher_id is not None: this_chaos.catcher_id = catcher_id if wild_pitch is not None: this_chaos.wild_pitch = wild_pitch if passed_ball is not None: this_chaos.passed_ball = passed_ball if pick_off is not None: this_chaos.pick_off = pick_off if balk is not None: this_chaos.balk = balk this_chaos.save() return_chaos = model_to_dict(this_chaos) db.close() return return_chaos class LineupMember(BaseModel): game = ForeignKeyField(Game) team_id = IntegerField() player_id = IntegerField() position = CharField() batting_order = IntegerField() active = BooleanField(default=True) class LineupMemberModel(pydantic.BaseModel): game_id: int team_id: int player_id: int position: str batting_order: int active: Optional[bool] = True def post_lineup_member(member_dict: dict): lineup_model = LineupMemberModel.parse_obj(member_dict) new_lineup = LineupMember( game=Game.get_by_id(lineup_model.game_id), team_id=lineup_model.team_id, player_id=lineup_model.player_id, position=lineup_model.position, batting_order=lineup_model.batting_order, active=lineup_model.active, ) new_lineup.save() return_lineup = model_to_dict(new_lineup) db.close() return return_lineup def get_lineup_members(game_id=None, team_id=None, player_id=None, position=None, batting_order=None, active=None): this_member = LineupMember.select() if game_id is not None: this_game = Game.get_by_id(game_id) this_member = this_member.where(LineupMember.game == this_game) if team_id is not None: this_member = this_member.where(LineupMember.team_id == team_id) if player_id is not None: this_member = this_member.where(LineupMember.player_id == player_id) if position is not None: this_member = this_member.where(LineupMember.position == position) if batting_order is not None: this_member = this_member.where(LineupMember.batting_order == batting_order) if active is not None: this_member = this_member.where(LineupMember.active == active) return_data = {'members': []} for x in this_member: return_data['members'].append(model_to_dict(x)) db.close() return return_data def get_one_member(game_id=None, team_id=None, player_id=None, position=None, batting_order=None, active=None): all_members = get_lineup_members(game_id, team_id, player_id, position, batting_order, active) if len(all_members['members']) == 0: raise KeyError('Lineup member not found') else: return all_members['members'][len(all_members['members']) - 1] def patch_lineup_member( member_id, game_id=None, team_id=None, player_id=None, position=None, batting_order=None, active=None): this_member = LineupMember.get_by_id(member_id) if game_id is not None: this_member.game_id = game_id if team_id is not None: this_member.team_id = team_id if player_id is not None: this_member.player_id = player_id if position is not None: this_member.position = position if batting_order is not None: this_member.batting_order = batting_order if active is not None: this_member.active = active this_member.save() return_member = model_to_dict(this_member) db.close() return return_member class GameState(BaseModel): game = ForeignKeyField(Game) top_half = BooleanField(default=True) inning = IntegerField(default=1) outs = IntegerField(default=0) away_batter_up = IntegerField(default=1) home_batter_up = IntegerField(default=1) away_score = IntegerField(default=0) home_score = IntegerField(default=0) on_first = ForeignKeyField(LineupMember, null=True) on_second = ForeignKeyField(LineupMember, null=True) on_third = ForeignKeyField(LineupMember, null=True) final = BooleanField(default=False) class GameStateModel(pydantic.BaseModel): game_id: int top_half: Optional[bool] = True inning: Optional[int] = 1 outs: Optional[int] = 0 away_batter_up: Optional[int] = 1 home_batter_up: Optional[int] = 1 away_score: Optional[int] = 0 home_score: Optional[int] = 0 on_first_id: Optional[int] = None on_second_id: Optional[int] = None on_third_id: Optional[int] = None final: Optional[bool] = False def post_game_state(state_dict: dict): state_model = GameStateModel.parse_obj(state_dict) if state_model.on_first_id: on_first = LineupMember.get_by_id(state_model.on_first_id) else: on_first = None if state_model.on_second_id: on_second = LineupMember.get_by_id(state_model.on_second_id) else: on_second = None if state_model.on_second_id: on_third = LineupMember.get_by_id(state_model.on_third_id) else: on_third = None new_state = GameState( game=Game.get_by_id(state_model.game_id), top_half=state_model.top_half, inning=state_model.inning, outs=state_model.outs, away_batter_up=state_model.away_batter_up, home_batter_up=state_model.home_batter_up, away_score=state_model.away_score, home_score=state_model.home_score, on_first=on_first, on_second=on_second, on_third=on_third ) new_state.save() return_state = new_state db.close() return return_state def get_game_state(game_id=None, top_half=None, inning=None, outs=None, away_batter_up=None, home_batter_up=None, away_score=None, home_score=None, final=None): this_state = GameState.select() if game_id is not None: this_game = Game.get_by_id(game_id) this_state = this_state.where(GameState.game == this_game) if top_half is not None: this_state = this_state.where(GameState.top_half == top_half) if inning is not None: this_state = this_state.where(GameState.inning == inning) if outs is not None: this_state = this_state.where(GameState.outs == outs) if away_batter_up is not None: this_state = this_state.where(GameState.away_batter_up == away_batter_up) if home_batter_up is not None: this_state = this_state.where(GameState.home_batter_up == home_batter_up) if away_score is not None: this_state = this_state.where(GameState.away_score == away_score) if home_score is not None: this_state = this_state.where(GameState.home_score == home_score) if final is not None: this_state = this_state.where(GameState.final == final) return_data = {'states': []} for x in this_state: return_data['states'].append(model_to_dict(x)) db.close() return return_data def get_one_game_state(game_id=None, top_half=None, inning=None, outs=None, away_batter_up=None, home_batter_up=None, away_score=None, home_score=None, final=None): all_states = get_game_state( game_id, top_half, inning, outs, away_batter_up, home_batter_up, away_score, home_score, final ) if len(all_states['states']) == 0: raise KeyError('Game State not found') else: return all_states['states'][(len(all_states['states'])) - 1] def patch_game_state( state_id, top_half=None, inning=None, outs=None, away_batter_up=None, home_batter_up=None, away_score=None, home_score=None, final=None, on_first_id=None, on_second_id=None, on_third_id=None): this_state = GameState.get_by_id(state_id) if top_half is not None: this_state.top_half = top_half if inning is not None: this_state.inning = inning if outs is not None: this_state.outs = outs if away_batter_up is not None: this_state.away_batter_up = away_batter_up if home_batter_up is not None: this_state.home_batter_up = home_batter_up if away_score is not None: this_state.away_score = away_score if home_score is not None: this_state.home_score = home_score if final is not None: this_state.final = final if on_first_id is not None: if on_first_id: on_first = LineupMember.get_by_id(on_first_id) else: on_first = None this_state.on_first = on_first if on_second_id is not None: if on_second_id: on_second = LineupMember.get_by_id(on_second_id) else: on_second = None this_state.on_second = on_second if on_third_id is not None: if on_third_id: on_third = LineupMember.get_by_id(on_third_id) else: on_third = None this_state.on_third = on_third this_state.save() return_state = model_to_dict(this_state) db.close() return return_state db.create_tables([Game, AtBat, Running, Defense, LineupMember, GameState, Chaos, Defense, Running]) db.close()