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