e54cf885ad
Added random_gif() Moved back from exception-handler cog to local error handling Updated !keepers to be season agnostic Added new !sync param to update and clear local guild Added error checking to !player command
1052 lines
36 KiB
Python
1052 lines
36 KiB
Python
import math
|
|
from typing import Optional
|
|
import pydantic
|
|
from peewee import *
|
|
from playhouse.shortcuts import model_to_dict
|
|
import logging
|
|
|
|
logger = logging.getLogger('discord_app')
|
|
|
|
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
|
|
logger.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
|
|
logger.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()
|