1050 lines
36 KiB
Python
1050 lines
36 KiB
Python
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()
|