1352 lines
58 KiB
Python
1352 lines
58 KiB
Python
import datetime
|
|
import logging
|
|
import math
|
|
from typing import Literal
|
|
import os
|
|
|
|
import discord
|
|
import pydantic
|
|
|
|
from pydantic import field_validator
|
|
from sqlmodel import Session, SQLModel, UniqueConstraint, create_engine, select, or_, Field, Relationship, text, BigInteger
|
|
from sqlalchemy import Column, func, desc
|
|
|
|
from exceptions import *
|
|
from in_game.managerai_responses import DefenseResponse, JumpResponse, RunResponse, TagResponse, ThrowResponse, UncappedRunResponse
|
|
|
|
|
|
logger = logging.getLogger('discord_app')
|
|
# sqlite_url = 'sqlite:///storage/gameplay.db'
|
|
# connect_args = {"check_same_thread": False}
|
|
# engine = create_engine(sqlite_url, echo=False, connect_args=connect_args)
|
|
postgres_url = f'postgresql://{os.getenv('DB_USERNAME')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_URL')}/{os.getenv('DB_NAME')}'
|
|
engine = create_engine(postgres_url)
|
|
CACHE_LIMIT = 1209600 # in seconds
|
|
SBA_COLOR = 'a6ce39'
|
|
SBA_LOGO = 'https://sombaseball.ddns.net/static/images/sba-logo.png'
|
|
|
|
|
|
class ManagerAiBase(SQLModel):
|
|
id: int | None = Field(sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
name: str = Field(index=True)
|
|
steal: int | None = Field(default=5)
|
|
running: int | None = Field(default=5)
|
|
hold: int | None = Field(default=5)
|
|
catcher_throw: int | None = Field(default=5)
|
|
uncapped_home: int | None = Field(default=5)
|
|
uncapped_third: int | None = Field(default=5)
|
|
uncapped_trail: int | None = Field(default=5)
|
|
bullpen_matchup: int | None = Field(default=5)
|
|
behind_aggression: int | None = Field(default=5)
|
|
ahead_aggression: int | None = Field(default=5)
|
|
decide_throw: int | None = Field(default=5)
|
|
|
|
|
|
class GameCardsetLink(SQLModel, table=True):
|
|
game_id: int | None = Field(default=None, foreign_key='game.id', primary_key=True)
|
|
cardset_id: int | None = Field(default=None, foreign_key='cardset.id', primary_key=True)
|
|
priority: int | None = Field(default=1, index=True)
|
|
|
|
game: 'Game' = Relationship(back_populates='cardset_links')
|
|
cardset: 'Cardset' = Relationship(back_populates='game_links')
|
|
|
|
|
|
class RosterLink(SQLModel, table=True):
|
|
game_id: int | None = Field(default=None, foreign_key='game.id', primary_key=True)
|
|
card_id: int | None = Field(default=None, foreign_key='card.id', primary_key=True)
|
|
team_id: int = Field(index=True, foreign_key='team.id')
|
|
|
|
game: 'Game' = Relationship(back_populates='roster_links')
|
|
card: 'Card' = Relationship()
|
|
team: 'Team' = Relationship()
|
|
|
|
|
|
class TeamBase(SQLModel):
|
|
id: int = Field(sa_column=Column(BigInteger(), primary_key=True, autoincrement=False, unique=True))
|
|
abbrev: str = Field(index=True)
|
|
sname: str
|
|
lname: str
|
|
gmid: int = Field(sa_column=Column(BigInteger(), autoincrement=False, index=True))
|
|
gmname: str
|
|
gsheet: str
|
|
wallet: int
|
|
team_value: int
|
|
collection_value: int
|
|
logo: str | None = Field(default=None)
|
|
color: str
|
|
season: int
|
|
career: int
|
|
ranking: int
|
|
has_guide: bool
|
|
is_ai: bool
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return f'{self.id}. {self.abbrev} {self.lname}, {"is_ai" if self.is_ai else "human"}'
|
|
|
|
|
|
class Team(TeamBase, table=True):
|
|
cards: list['Card'] = Relationship(back_populates='team', cascade_delete=True)
|
|
lineups: list['Lineup'] = Relationship(back_populates='team', cascade_delete=True)
|
|
# away_games: list['Game'] = Relationship(back_populates='away_team')
|
|
# home_games: list['Game'] = Relationship(back_populates='home_team')
|
|
|
|
@property
|
|
def embed(self) -> discord.Embed:
|
|
embed = discord.Embed(
|
|
title=f'{self.lname}',
|
|
color=int(self.color, 16) if self.color else int(SBA_COLOR, 16)
|
|
)
|
|
embed.set_footer(text=f'Paper Dynasty Season {self.season}', icon_url=SBA_LOGO)
|
|
embed.set_thumbnail(url=self.logo if self.logo else SBA_LOGO)
|
|
return embed
|
|
|
|
|
|
class Game(SQLModel, table=True):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
away_team_id: int = Field(foreign_key='team.id')
|
|
home_team_id: int = Field(foreign_key='team.id')
|
|
channel_id: int = Field(sa_column=(Column(BigInteger(), index=True)))
|
|
season: int
|
|
active: bool | None = Field(default=True)
|
|
is_pd: bool | None = Field(default=True)
|
|
ranked: bool | None = Field(default=False)
|
|
short_game: bool | None = Field(default=False)
|
|
week: int | None = Field(default=None)
|
|
game_num: int | None = Field(default=None)
|
|
away_roster_id: int | None = Field(default=None)
|
|
home_roster_id: int | None = Field(default=None)
|
|
first_message: str | None = Field(default=None)
|
|
ai_team: str | None = Field(default=None)
|
|
game_type: str
|
|
roll_buttons: bool | None = Field(default=True)
|
|
auto_roll: bool | None = Field(default=False)
|
|
|
|
cardset_links: list[GameCardsetLink] = Relationship(back_populates='game', cascade_delete=True)
|
|
roster_links: list[RosterLink] = Relationship(back_populates='game', cascade_delete=True)
|
|
away_team: Team = Relationship(
|
|
# back_populates='away_games',
|
|
# sa_relationship_kwargs={
|
|
# 'primaryjoin': 'Game.away_team_id==Team.id',
|
|
# 'foreign_keys': '[Game.away_team.id]',
|
|
# 'lazy': 'joined'
|
|
# }
|
|
sa_relationship_kwargs=dict(foreign_keys="[Game.away_team_id]")
|
|
)
|
|
home_team: Team = Relationship(
|
|
# back_populates='home_games',
|
|
# sa_relationship_kwargs={
|
|
# 'primaryjoin': 'Game.home_team_id==Team.id',
|
|
# 'foreign_keys': '[Game.home_team.id]',
|
|
# 'lazy': 'joined'
|
|
# }
|
|
sa_relationship_kwargs=dict(foreign_keys="[Game.home_team_id]")
|
|
)
|
|
lineups: list['Lineup'] = Relationship(back_populates='game', cascade_delete=True)
|
|
plays: list['Play'] = Relationship(back_populates='game', cascade_delete=True)
|
|
|
|
@field_validator('ai_team', 'game_type')
|
|
def lowercase_strings(cls, value: str) -> str:
|
|
return value.lower()
|
|
|
|
@property
|
|
def cardset_param_string(self) -> str:
|
|
pri_cardsets = ''
|
|
back_cardsets = ''
|
|
for link in self.cardset_links:
|
|
if link.priority == 1:
|
|
pri_cardsets += f'&cardset_id={link.cardset_id}'
|
|
else:
|
|
back_cardsets += f'&backup_cardset_id={link.cardset_id}'
|
|
return f'{pri_cardsets}{back_cardsets}'
|
|
|
|
def current_play_or_none(self, session: Session):
|
|
this_play = session.exec(select(Play).where(Play.game == self, Play.complete == False).order_by(Play.play_num.desc()).limit(1)).all()
|
|
if len(this_play) == 1:
|
|
return this_play[0]
|
|
else:
|
|
return None
|
|
|
|
def initialize_play(self, session: Session):
|
|
"""
|
|
Commits new_play
|
|
"""
|
|
existing_play = self.current_play_or_none(session)
|
|
if existing_play is not None:
|
|
return existing_play
|
|
|
|
all_plays = session.exec(select(func.count(Play.id)).where(Play.game == self)).one()
|
|
if all_plays > 0:
|
|
raise PlayInitException(f'{all_plays} plays for game {self.id} already exist, but all are complete.')
|
|
|
|
leadoff_batter, home_pitcher, home_catcher = None, None, None
|
|
home_positions, away_positions = [], []
|
|
for line in [x for x in self.lineups if x.active]:
|
|
if line.team == self.away_team:
|
|
if line.position not in away_positions:
|
|
away_positions.append(line.position)
|
|
if line.batting_order == 1:
|
|
leadoff_batter = line
|
|
else:
|
|
if line.position not in home_positions:
|
|
home_positions.append(line.position)
|
|
if line.position == 'P':
|
|
home_pitcher = line
|
|
elif line.position == 'C':
|
|
home_catcher = line
|
|
|
|
if len(home_positions) != 10:
|
|
e_msg = f'Only {len(home_positions)} players found on home team'
|
|
log_exception(LineupsMissingException, e_msg)
|
|
if len(away_positions) != 10:
|
|
e_msg = f'Only {len(away_positions)} players found on away team'
|
|
log_exception(LineupsMissingException, e_msg)
|
|
if None in [leadoff_batter, home_pitcher, home_catcher]:
|
|
e_msg = f'Could not set the initial pitcher, catcher, and batter'
|
|
log_exception(LineupsMissingException, e_msg)
|
|
|
|
manager_ai_id = ((datetime.datetime.now().day * (self.away_team_id if self.ai_team == 'away' else self.home_team_id)) % 3) + 1
|
|
if manager_ai_id > 3 or manager_ai_id < 1:
|
|
manager_ai_id = 1
|
|
|
|
new_play = Play(
|
|
game=self,
|
|
play_num=1,
|
|
batter=leadoff_batter,
|
|
pitcher=home_pitcher,
|
|
batter_pos=leadoff_batter.position,
|
|
catcher=home_catcher,
|
|
is_tied=True,
|
|
is_new_inning=True,
|
|
managerai_id=manager_ai_id
|
|
)
|
|
session.add(new_play)
|
|
session.commit()
|
|
session.refresh(new_play)
|
|
|
|
new_play.init_ai(session)
|
|
|
|
return new_play
|
|
|
|
def team_lineup(self, session: Session, team: Team, with_links: bool = True) -> str:
|
|
all_lineups = session.exec(select(Lineup).where(Lineup.team == team, Lineup.game == self, Lineup.active).order_by(Lineup.batting_order)).all()
|
|
|
|
lineup_val = ''
|
|
for line in all_lineups:
|
|
if with_links:
|
|
name_string = {line.player.name_card_link("batting" if line.position != "P" else "pitching")}
|
|
else:
|
|
name_string = f'{line.player.name_with_desc}'
|
|
|
|
if line.position == 'P':
|
|
this_hand = line.card.pitcherscouting.pitchingcard.hand
|
|
else:
|
|
this_hand = line.card.batterscouting.battingcard.hand
|
|
|
|
lineup_val += f'{line.batting_order}. {this_hand.upper()} | {name_string}, {line.position}\n'
|
|
|
|
return lineup_val
|
|
|
|
@property
|
|
def human_team(self):
|
|
if self.home_team.is_ai and not self.away_team.is_ai:
|
|
return self.away_team
|
|
elif self.away_team.is_ai and not self.home_team.is_ai:
|
|
return self.home_team
|
|
elif self.home_team.is_ai and self.away_team.is_ai:
|
|
raise NoHumanTeamsException
|
|
else:
|
|
raise MultipleHumanTeamsException
|
|
|
|
@property
|
|
def league_name(self):
|
|
if 'gauntlet' in self.game_type:
|
|
parts = self.game_type.split('-')
|
|
return f'{parts[0]}-{parts[1]}'
|
|
else:
|
|
return self.game_type
|
|
|
|
|
|
class ManagerAi(ManagerAiBase, table=True):
|
|
plays: list['Play'] = Relationship(back_populates='managerai')
|
|
def create_ai(session: Session = None):
|
|
def get_new_ai(this_session: Session):
|
|
all_ai = this_session.exec(select(ManagerAi.id)).all()
|
|
if len(all_ai) == 0:
|
|
logger.info(f'Creating ManagerAI records')
|
|
new_ai = [
|
|
ManagerAi(
|
|
name='Balanced'
|
|
),
|
|
ManagerAi(
|
|
name='Yolo',
|
|
steal=10,
|
|
running=10,
|
|
hold=5,
|
|
catcher_throw=10,
|
|
uncapped_home=10,
|
|
uncapped_third=10,
|
|
uncapped_trail=10,
|
|
bullpen_matchup=3,
|
|
behind_aggression=10,
|
|
ahead_aggression=10,
|
|
decide_throw=10
|
|
),
|
|
ManagerAi(
|
|
name='Safe',
|
|
steal=3,
|
|
running=3,
|
|
hold=8,
|
|
catcher_throw=5,
|
|
uncapped_home=5,
|
|
uncapped_third=3,
|
|
uncapped_trail=5,
|
|
bullpen_matchup=8,
|
|
behind_aggression=5,
|
|
ahead_aggression=1,
|
|
decide_throw=1
|
|
)
|
|
]
|
|
for x in new_ai:
|
|
session.add(x)
|
|
session.commit()
|
|
|
|
if session is None:
|
|
with Session(engine) as session:
|
|
get_new_ai(session)
|
|
else:
|
|
get_new_ai(session)
|
|
|
|
return True
|
|
|
|
def check_jump(self, session: Session, this_game: Game, to_base: Literal[2, 3, 4]) -> JumpResponse:
|
|
logger.info(f'Checking jump to {to_base} in Game {this_game.id}')
|
|
this_resp = JumpResponse(min_safe=20)
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking for jump')
|
|
|
|
num_outs = this_play.starting_outs
|
|
run_diff = this_play.away_score - this_play.home_score
|
|
if this_game.ai_team == 'home':
|
|
run_diff = run_diff * -1
|
|
|
|
pitcher_hold = this_play.pitcher.card.pitcherscouting.pitchingcard.hold
|
|
catcher_defense = session.exec(select(PositionRating).where(PositionRating.player_id == this_play.catcher.player_id, PositionRating.position == 'C', PositionRating.variant == this_play.catcher.card.variant)).one()
|
|
catcher_hold = catcher_defense.arm
|
|
battery_hold = pitcher_hold + catcher_hold
|
|
logger.info(f'game state: {num_outs} outs, {run_diff} run diff, battery_hold: {battery_hold}')
|
|
|
|
if to_base == 2:
|
|
runner = this_play.on_first
|
|
if runner is None:
|
|
log_exception(CardNotFoundException, f'Attempted to check a jump to 2nd base, but no runner found on first.')
|
|
logger.info(f'Checking steal numbers for {runner.player.name} in Game {this_game.id}')
|
|
|
|
match self.steal:
|
|
case 10:
|
|
this_resp.min_safe = 12 + num_outs
|
|
case self.steal if self.steal > 8 and run_diff <= 5:
|
|
this_resp.min_safe = 13 + num_outs
|
|
case self.steal if self.steal > 6 and run_diff <= 5:
|
|
this_resp.min_safe = 14 + num_outs
|
|
case self.steal if self.steal > 4 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.min_safe = 15 + num_outs
|
|
case self.steal if self.steal > 2 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.min_safe = 16 + num_outs
|
|
case _:
|
|
this_resp.min_safe = 17 + num_outs
|
|
|
|
if self.steal > 7 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.run_if_auto_jump = True
|
|
elif self.steal < 5:
|
|
this_resp.must_auto_jump = True
|
|
|
|
runner_card = runner.card.batterscouting.battingcard
|
|
if this_resp.run_if_auto_jump and runner_card.steal_auto:
|
|
this_resp.ai_note = f'- WILL SEND **{runner.player.name}** to second!'
|
|
|
|
elif this_resp.must_auto_jump and not runner_card.steal_auto:
|
|
logger.info(f'No jump ai note')
|
|
|
|
else:
|
|
jump_safe_range = runner_card.steal_high + battery_hold
|
|
nojump_safe_range = runner_card.steal_low + battery_hold
|
|
logger.info(f'jump_safe_range: {jump_safe_range} / nojump_safe_range: {nojump_safe_range} / min_safe: {this_resp.min_safe}')
|
|
|
|
if this_resp.min_safe <= nojump_safe_range:
|
|
this_resp.ai_note = f'- SEND **{runner.player.name}** to second!'
|
|
|
|
elif this_resp.min_safe <= jump_safe_range:
|
|
this_resp.ai_note = f'- SEND **{runner.player.name}** to second if they get the jump'
|
|
|
|
elif to_base == 3:
|
|
runner = this_play.on_second
|
|
if runner is None:
|
|
log_exception(CardNotFoundException, f'Attempted to check a jump to 3rd base, but no runner found on second.')
|
|
|
|
match self.steal:
|
|
case 10:
|
|
this_resp.min_safe = 12 + num_outs
|
|
case self.steal if self.steal > 6 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.min_safe = 15 + num_outs
|
|
case _:
|
|
this_resp.min_safe = None
|
|
|
|
if self.steal == 10 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.run_if_auto_jump = True
|
|
elif self.steal <= 5:
|
|
this_resp.must_auto_jump = True
|
|
|
|
runner_card = runner.card.batterscouting.battingcard
|
|
if this_resp.run_if_auto_jump and runner_card.steal_auto:
|
|
this_resp.ai_note = f'- SEND **{runner.player.name}** to third!'
|
|
|
|
elif this_resp.must_auto_jump and not runner_card.steal_auto or this_resp.min_safe is None:
|
|
logger.info(f'No jump ai note')
|
|
|
|
else:
|
|
jump_safe_range = runner_card.steal_low + battery_hold
|
|
logger.info(f'jump_safe_range: {jump_safe_range} / min_safe: {this_resp.min_safe}')
|
|
|
|
if this_resp.min_safe <= jump_safe_range:
|
|
this_resp.ai_note = f'- SEND **{runner.player.name}** to third!'
|
|
|
|
elif run_diff in [-1, 0]:
|
|
runner = this_play.on_third
|
|
if runner is None:
|
|
log_exception(CardNotFoundException, f'Attempted to check a jump to home, but no runner found on third.')
|
|
|
|
if self.steal == 10:
|
|
this_resp.min_safe = 5
|
|
elif this_play.inning_num > 7 and self.steal >= 5:
|
|
this_resp.min_safe = 6
|
|
elif self.steal > 5:
|
|
this_resp.min_safe = 7
|
|
elif self.steal > 2:
|
|
this_resp.min_safe = 8
|
|
else:
|
|
this_resp.min_safe = 10
|
|
|
|
runner_card = runner.card.batterscouting.battingcard
|
|
jump_safe_range = runner_card.steal_low - 9
|
|
|
|
if this_resp.min_safe <= jump_safe_range:
|
|
this_resp.ai_note = f'- SEND **{runner.player.name}** to third!'
|
|
|
|
logger.info(f'Returning jump resp to game {this_game.id}: {this_resp}')
|
|
return this_resp
|
|
|
|
def tag_from_second(self, session: Session, this_game: Game) -> TagResponse:
|
|
this_resp = TagResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking tag_from_second')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression_mod = abs(self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5)
|
|
adjusted_running = self.running + aggression_mod
|
|
|
|
if adjusted_running >= 8:
|
|
this_resp.min_safe = 4
|
|
elif adjusted_running >= 5:
|
|
this_resp.min_safe = 7
|
|
else:
|
|
this_resp.min_safe = 10
|
|
|
|
if this_play.starting_outs == 1:
|
|
this_resp.min_safe -= 2
|
|
else:
|
|
this_resp.min_safe += 2
|
|
|
|
logger.info(f'tag_from_second response: {this_resp}')
|
|
return this_resp
|
|
|
|
def tag_from_third(self, session: Session, this_game: Game) -> TagResponse:
|
|
this_resp = TagResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking tag_from_third')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression_mod = abs(self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5)
|
|
adjusted_running = self.running + aggression_mod
|
|
|
|
if adjusted_running >= 8:
|
|
this_resp.min_safe = 7
|
|
elif adjusted_running >= 5:
|
|
this_resp.min_safe = 10
|
|
else:
|
|
this_resp.min_safe = 12
|
|
|
|
if ai_rd in [-1, 0]:
|
|
this_resp.min_safe -= 2
|
|
|
|
if this_play.starting_outs == 1:
|
|
this_resp.min_safe -= 2
|
|
|
|
logger.info(f'tag_from_third response: {this_resp}')
|
|
return this_resp
|
|
|
|
def throw_at_uncapped(self, session: Session, this_game: Game) -> ThrowResponse:
|
|
this_resp = ThrowResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking throw_at_uncapped')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = self.ahead_aggression if ai_rd > 0 else self.behind_aggression
|
|
current_outs = this_play.starting_outs + this_play.outs
|
|
|
|
if ai_rd > 5:
|
|
if self.ahead_aggression > 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4 + current_outs
|
|
else:
|
|
this_resp.cutoff = True
|
|
elif ai_rd > 2:
|
|
if self.ahead_aggression > 8:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4 + current_outs
|
|
elif ai_rd > 0:
|
|
if self.ahead_aggression > 8:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -6 + current_outs
|
|
elif ai_rd > -3:
|
|
if self.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -6 + current_outs
|
|
elif ai_rd > -6:
|
|
if self.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4 + current_outs
|
|
else:
|
|
if self.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4
|
|
|
|
logger.info(f'throw_at_uncapped response: {this_resp}')
|
|
return this_resp
|
|
|
|
def uncapped_advance(self, session: Session, this_game: Game, lead_base: int, trail_base: int) -> UncappedRunResponse:
|
|
this_resp = UncappedRunResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking uncapped_advance_lead')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5
|
|
|
|
if ai_rd > 4:
|
|
if lead_base == 4:
|
|
this_resp.min_safe = 16 - this_play.starting_outs - aggression
|
|
this_resp.send_trail = True
|
|
this_resp.trail_min_safe = 10 - aggression - this_play.starting_outs - this_play.outs
|
|
elif lead_base == 3:
|
|
this_resp.min_safe = 14 + (this_play.starting_outs * 2) - aggression
|
|
if this_play.starting_outs + this_play.outs >= 2:
|
|
this_resp.send_trail = False
|
|
elif ai_rd > 1 or ai_rd < -2:
|
|
if lead_base == 4:
|
|
this_resp.min_safe = 12 - this_play.starting_outs - aggression
|
|
this_resp.send_trail = True
|
|
this_resp.trail_min_safe = 10 - aggression - this_play.starting_outs - this_play.outs
|
|
elif lead_base == 3:
|
|
this_resp.min_safe = 12 + (this_play.starting_outs * 2) - (aggression * 2)
|
|
if this_play.starting_outs + this_play.outs >= 2:
|
|
this_resp.send_trail = False
|
|
else:
|
|
if lead_base == 4:
|
|
this_resp.min_safe = 10 - this_play.starting_outs - aggression
|
|
this_resp.send_trail = True
|
|
this_resp.trail_min_safe = 2
|
|
elif lead_base == 3:
|
|
this_resp.min_safe = 14 + (this_play.starting_outs * 2) - aggression
|
|
if this_play.starting_outs + this_play.outs >= 2:
|
|
this_resp.send_trail = False
|
|
|
|
if this_resp.min_safe > 20:
|
|
this_resp.min_safe = 20
|
|
if this_resp.min_safe < 1:
|
|
this_resp.min_safe = 1
|
|
if this_resp.trail_min_safe > 20:
|
|
this_resp.min_safe = 20
|
|
if this_resp.trail_min_safe < 1:
|
|
this_resp.min_safe = 1
|
|
|
|
logger.info(f'Uncapped advance response: {this_resp}')
|
|
return this_resp
|
|
|
|
def defense_alignment(self, session: Session, this_game: Game) -> DefenseResponse:
|
|
this_resp = DefenseResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking uncapped_advance_lead')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5
|
|
pitcher_hold = this_play.pitcher.card.pitcherscouting.pitchingcard.hold
|
|
|
|
catcher_defense = session.exec(select(PositionRating).where(PositionRating.player_id == this_play.catcher.player_id, PositionRating.position == 'C', PositionRating.variant == this_play.catcher.card.variant)).one()
|
|
catcher_hold = catcher_defense.arm
|
|
battery_hold = pitcher_hold + catcher_hold
|
|
|
|
if this_play.starting_outs == 2 and this_play.on_base_code > 0:
|
|
logger.info(f'Checking for holds with 2 outs')
|
|
if this_play.on_base_code == 1:
|
|
this_resp.hold_first = True
|
|
this_resp.ai_note += f'- hold {this_play.on_first.player.name} on 1st\n'
|
|
elif this_play.on_base_code == 2:
|
|
this_resp.hold_second = True
|
|
this_resp.ai_note += f'- hold {this_play.on_second.player.name} on 2nd\n'
|
|
elif this_play.on_base_code in [4, 7]:
|
|
this_resp.hold_first = True
|
|
this_resp.hold_second = True
|
|
this_resp.ai_note += f'- hold {this_play.on_first.player.name} on 1st\n- hold {this_play.on_second.player.name} on 2nd\n'
|
|
elif this_play.on_base_code == 5:
|
|
this_resp.hold_first = True
|
|
this_resp.ai_note += f'- hold {this_play.on_first.player.name} on first\n'
|
|
elif this_play.on_base_code == 6:
|
|
this_resp.hold_second = True
|
|
this_resp.ai_note += f'- hold {this_play.on_second.player.name} on 2nd\n'
|
|
elif this_play.on_base_code in [1, 5]:
|
|
logger.info(f'Checking for hold with runner on first')
|
|
runner = this_play.on_first.player
|
|
if this_play.on_first.card.batterscouting.battingcard.steal_auto and ((this_play.on_first.card.batterscouting.battingcard.steal_high + battery_hold) >= (12 - aggression)):
|
|
this_resp.hold_first = True
|
|
this_resp.ai_note += f'- hold {runner.name} on 1st\n'
|
|
elif this_play.on_base_code in [2, 4]:
|
|
logger.info(f'Checking for hold with runner on second')
|
|
if (this_play.on_second.card.batterscouting.battingcard.steal_low + max(battery_hold, 5)) >= (14 - aggression):
|
|
this_resp.hold_second = True
|
|
this_resp.ai_note += f'- hold {this_play.on_second.player.name} on 2nd\n'
|
|
|
|
# Defensive Alignment
|
|
if this_play.on_third and this_play.starting_outs < 2:
|
|
if this_play.could_walkoff:
|
|
this_resp.outfield_in = True
|
|
this_resp.infield_in = True
|
|
this_resp.ai_note += f'- play the outfield and infield in'
|
|
elif this_play.on_first and this_play.starting_outs == 1:
|
|
this_resp.corners_in = True
|
|
this_resp.ai_note += f'- play the corners in\n'
|
|
elif abs(this_play.away_score - this_play.home_score) <= 3:
|
|
this_resp.infield_in = True
|
|
this_resp.ai_note += f'- play the whole infield in\n'
|
|
else:
|
|
this_resp.corners_in = True
|
|
this_resp.ai_note += f'- play the corners in\n'
|
|
|
|
if len(this_resp.ai_note) == 0 and this_play.on_base_code > 0:
|
|
this_resp.ai_note += f'- play straight up\n'
|
|
|
|
logger.info(f'Defense alignment response: {this_resp}')
|
|
return this_resp
|
|
|
|
def gb_decide_run(self, session: Session, this_game: Game) -> RunResponse:
|
|
this_resp = RunResponse()
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking gb_decide_run')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5
|
|
|
|
this_resp.min_safe = 15 - aggression # TODO: write this algorithm
|
|
logger.info(f'gb_decide_run response: {this_resp}')
|
|
return this_resp
|
|
|
|
def gb_decide_throw(self, session: Session, this_game: Game, runner_speed: int, defender_range: int) -> ThrowResponse:
|
|
this_resp = ThrowResponse(at_lead_runner=True)
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking gb_decide_throw')
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = self.ahead_aggression - 5 if ai_rd > 0 else self.behind_aggression - 5
|
|
|
|
if (runner_speed - 4 + defender_range) <= (10 + aggression):
|
|
this_resp.at_lead_runner = True
|
|
|
|
logger.info(f'gb_decide_throw response: {this_resp}')
|
|
return this_resp
|
|
|
|
def replace_pitcher(self, session: Session, this_game: Game) -> bool:
|
|
logger.info(f'Checking if fatigued pitcher should be replaced')
|
|
this_play = this_game.current_play_or_none(session)
|
|
if this_play is None:
|
|
raise GameException(f'No game found while checking replace_pitcher')
|
|
|
|
this_pitcher = this_play.pitcher
|
|
outs = session.exec(select(func.sum(Play.outs)).where(
|
|
Play.game == this_game, Play.pitcher == this_pitcher, Play.complete == True
|
|
)).one()
|
|
logger.info(f'Pitcher: {this_pitcher.card.player.name_with_desc} / Outs: {outs}')
|
|
|
|
allowed_runners = session.exec(select(func.count(Play.id)).where(
|
|
Play.game == this_game, Play.pitcher == this_pitcher, or_(Play.hit == 1, Play.bb == 1)
|
|
)).one()
|
|
run_diff = this_play.ai_run_diff
|
|
|
|
logger.info(f'run diff: {run_diff} / allowed runners: {allowed_runners} / behind aggro: {self.behind_aggression} / ahead aggro: {self.ahead_aggression}')
|
|
logger.info(f'this play: {this_play}')
|
|
|
|
if this_pitcher.replacing_id is None:
|
|
pitcher_pow = this_pitcher.card.pitcherscouting.pitchingcard.starter_rating
|
|
logger.info(f'Starter POW: {pitcher_pow}')
|
|
|
|
if outs >= pitcher_pow * 3 + 6:
|
|
logger.info(f'Starter has thrown POW + 3 - being pulled')
|
|
return True
|
|
|
|
elif allowed_runners < 5:
|
|
logger.info(f'Starter is cooking with {allowed_runners} runners allowed - staying in')
|
|
return False
|
|
|
|
elif this_pitcher.is_fatigued and this_play.on_base_code > 1:
|
|
logger.info(f'Starter is fatigued')
|
|
return True
|
|
|
|
elif (run_diff > 5 or (run_diff > 2 and self.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 3):
|
|
logger.info(f'AI team has big lead of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif (run_diff > 2 or (run_diff >= 0 and self.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 1):
|
|
logger.info(f'AI team has lead of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif (run_diff >= 0 or (run_diff >= -2 and self.behind_aggression > 5)) and (allowed_runners < 5 and this_play.on_base_code <= run_diff):
|
|
logger.info(f'AI team in close game with run diff of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif run_diff >= -3 and self.behind_aggression > 5 and allowed_runners < 5 and this_play.on_base_code <= 1:
|
|
logger.info(f'AI team is close behind with run diff of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif run_diff <= -5 and this_play.inning_num <= 3:
|
|
logger.info(f'AI team is way behind and starter is going to wear it - staying in')
|
|
return False
|
|
|
|
else:
|
|
logger.info(f'AI team found no exceptions - pull starter')
|
|
return True
|
|
|
|
else:
|
|
pitcher_pow = this_pitcher.card.pitcherscouting.pitchingcard.relief_rating
|
|
logger.info(f'Reliever POW: {pitcher_pow}')
|
|
|
|
if outs >= pitcher_pow * 3 + 3:
|
|
logger.info(f'Only allow POW + 1 IP - pull reliever')
|
|
return True
|
|
|
|
elif this_pitcher.is_fatigued and this_play.is_new_inning:
|
|
logger.info(f'Reliever is fatigued to start the inning - pull reliever')
|
|
return True
|
|
|
|
elif (run_diff > 5 or (run_diff > 2 and self.ahead_aggression > 5)) and (this_play.starting_outs == 2 or allowed_runners <= run_diff or this_play.on_base_code <= 3 or this_play.starting_outs == 2):
|
|
logger.info(f'AI team has big lead of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif (run_diff > 2 or (run_diff >= 0 and self.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 1 or this_play.starting_outs == 2):
|
|
logger.info(f'AI team has lead of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif (run_diff >= 0 or (run_diff >= -2 and self.behind_aggression > 5)) and (allowed_runners < 5 or this_play.on_base_code <= run_diff or this_play.starting_outs == 2):
|
|
logger.info(f'AI team in close game with run diff of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif run_diff >= -3 and self.behind_aggression > 5 and allowed_runners < 5 and this_play.on_base_code <= 1:
|
|
logger.info(f'AI team is close behind with run diff of {run_diff} - staying in')
|
|
return False
|
|
|
|
elif run_diff <= -5 and this_play.starting_outs != 0:
|
|
logger.info(f'AI team is way behind and reliever is going to wear it - staying in')
|
|
return False
|
|
|
|
else:
|
|
logger.info(f'AI team found no exceptions - pull reliever')
|
|
return True
|
|
|
|
|
|
class CardsetBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
name: str
|
|
ranked_legal: bool | None = Field(default=False)
|
|
|
|
|
|
class Cardset(CardsetBase, table=True):
|
|
game_links: list[GameCardsetLink] = Relationship(back_populates='cardset', cascade_delete=True)
|
|
players: list['Player'] = Relationship(back_populates='cardset')
|
|
|
|
|
|
class PlayerBase(SQLModel):
|
|
id: int | None = Field(sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
name: str
|
|
cost: int
|
|
image: str
|
|
mlbclub: str
|
|
franchise: str
|
|
cardset_id: int | None = Field(default=None, foreign_key='cardset.id')
|
|
set_num: int
|
|
rarity_id: int | None = Field(default=None)
|
|
pos_1: str
|
|
description: str
|
|
quantity: int | None = Field(default=999)
|
|
image2: str | None = Field(default=None)
|
|
pos_2: str | None = Field(default=None)
|
|
pos_3: str | None = Field(default=None)
|
|
pos_4: str | None = Field(default=None)
|
|
pos_5: str | None = Field(default=None)
|
|
pos_6: str | None = Field(default=None)
|
|
pos_7: str | None = Field(default=None)
|
|
pos_8: str | None = Field(default=None)
|
|
headshot: str | None = Field(default=None)
|
|
vanity_card: str | None = Field(default=None)
|
|
strat_code: str | None = Field(default=None)
|
|
bbref_id: str | None = Field(default=None)
|
|
fangr_id: str | None = Field(default=None)
|
|
mlbplayer_id: int | None = Field(default=None)
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
@field_validator('pos_1', 'pos_2', 'pos_3', 'pos_4', 'pos_5', 'pos_6', 'pos_7', 'pos_8')
|
|
def uppercase_strings(cls, value: str) -> str:
|
|
if value is not None:
|
|
return value.upper()
|
|
else:
|
|
return value
|
|
|
|
@property
|
|
def batter_card_url(self):
|
|
if 'batting' in self.image:
|
|
return self.image
|
|
elif 'batting' in self.image2:
|
|
return self.image2
|
|
else:
|
|
return None
|
|
|
|
@property
|
|
def pitcher_card_url(self):
|
|
if 'pitching' in self.image:
|
|
return self.image
|
|
elif 'pitching' in self.image2:
|
|
return self.image2
|
|
else:
|
|
return None
|
|
|
|
def name_card_link(self, which: Literal['pitching', 'batting']):
|
|
if which == 'pitching':
|
|
return f'[{self.name}]({self.pitcher_card_url})'
|
|
else:
|
|
return f'[{self.name}]({self.batter_card_url})'
|
|
|
|
|
|
class Player(PlayerBase, table=True):
|
|
cardset: Cardset = Relationship(back_populates='players')
|
|
cards: list['Card'] = Relationship(back_populates='player', cascade_delete=True)
|
|
lineups: list['Lineup'] = Relationship(back_populates='player', cascade_delete=True)
|
|
positions: list['PositionRating'] = Relationship(back_populates='player', cascade_delete=True)
|
|
|
|
@property
|
|
def name_with_desc(self):
|
|
return f'{self.description} {self.name}'
|
|
|
|
|
|
def player_description(player: Player = None, player_dict: dict = None) -> str:
|
|
if player is None and player_dict is None:
|
|
err = 'One of "player" or "player_dict" must be included to get full description'
|
|
logger.error(f'gameplay_models - player_description - {err}')
|
|
raise TypeError(err)
|
|
|
|
if player is not None:
|
|
return f'{player.description} {player.name}'
|
|
|
|
r_val = f'{player_dict['description']}'
|
|
if 'name' in player_dict:
|
|
r_val += f' {player_dict["name"]}'
|
|
elif 'p_name' in player_dict:
|
|
r_val += f' {player_dict["p_name"]}'
|
|
return r_val
|
|
|
|
|
|
class BattingCardBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
variant: int | None = Field(default=0)
|
|
steal_low: int = Field(default=0, ge=0, le=20)
|
|
steal_high: int = Field(default=0, ge=0, le=20)
|
|
steal_auto: bool = Field(default=False)
|
|
steal_jump: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
bunting: str = Field(default='C')
|
|
hit_and_run: str = Field(default='C')
|
|
running: int = Field(default=10, ge=1, le=20)
|
|
offense_col: int = Field(ge=1, le=3)
|
|
hand: str
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
# created: datetime.datetime | None = Field(sa_column_kwargs={"server_default": text("CURRENT_TIMESTAMP"),})
|
|
|
|
@field_validator('hand')
|
|
def lowercase_hand(cls, value: str) -> str:
|
|
return value.lower()
|
|
|
|
|
|
class BattingCard(BattingCardBase, table=True):
|
|
pass
|
|
|
|
|
|
class BattingRatingsBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
homerun: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
bp_homerun: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
triple: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_three: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_two: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_pull: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_two: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_one: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_center: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
bp_single: float = Field(default=0.0, ge=0.0, le=10.0)
|
|
hbp: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
walk: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
strikeout: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
lineout: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
popout: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_a: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_bq: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_lf_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_rf_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
groundout_a: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
groundout_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
groundout_c: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
avg: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
obp: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
slg: float = Field(default=0.0, ge=0.0, le=4.0)
|
|
pull_rate: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
center_rate: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
slap_rate: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class BattingRatings(BattingRatingsBase, table=True):
|
|
pass
|
|
|
|
|
|
class BatterScoutingBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
battingcard_id: int | None = Field(default=None, foreign_key='battingcard.id', ondelete='CASCADE')
|
|
ratings_vl_id: int | None = Field(default=None, foreign_key='battingratings.id', ondelete='CASCADE')
|
|
ratings_vr_id: int | None = Field(default=None, foreign_key='battingratings.id', ondelete='CASCADE')
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class BatterScouting(BatterScoutingBase, table=True):
|
|
battingcard: BattingCard = Relationship() #back_populates='batterscouting')
|
|
ratings_vl: BattingRatings = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[BatterScouting.ratings_vl_id]",single_parent=True), cascade_delete=True
|
|
)
|
|
ratings_vr: BattingRatings = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[BatterScouting.ratings_vr_id]",single_parent=True), cascade_delete=True
|
|
)
|
|
cards: list['Card'] = Relationship(back_populates='batterscouting', cascade_delete=False)
|
|
|
|
|
|
class PitchingCardBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
variant: int | None = Field(default=0)
|
|
balk: int = Field(default=0, ge=0, le=20)
|
|
wild_pitch: int = Field(default=0, ge=0, le=20)
|
|
hold: int = Field(default=0, ge=-9, le=9)
|
|
starter_rating: int = Field(default=1, ge=1, le=10)
|
|
relief_rating: int = Field(default=1, ge=1, le=10)
|
|
closer_rating: int | None = Field(default=None, ge=0, le=9)
|
|
offense_col: int = Field(ge=1, le=3)
|
|
batting: str = Field(default='#1WR-C')
|
|
hand: str
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
@field_validator('hand')
|
|
def lowercase_hand(cls, value: str) -> str:
|
|
return value.lower()
|
|
|
|
|
|
class PitchingCard(PitchingCardBase, table=True):
|
|
pass
|
|
|
|
|
|
class PitchingRatingsBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
homerun: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
bp_homerun: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
triple: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_three: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_two: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
double_cf: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_two: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_one: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
single_center: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
bp_single: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
hbp: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
walk: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
strikeout: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_lf_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_cf_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
flyout_rf_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
groundout_a: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
groundout_b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_p: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_c: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_1b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_2b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_3b: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_ss: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_lf: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_cf: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
xcheck_rf: float = Field(default=0.0, ge=0.0, le=108.0)
|
|
avg: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
obp: float = Field(default=0.0, ge=0.0, le=1.0)
|
|
slg: float = Field(default=0.0, ge=0.0, le=4.0)
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class PitchingRatings(PitchingRatingsBase, table=True):
|
|
pass
|
|
|
|
|
|
class PitcherScoutingBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
pitchingcard_id: int | None = Field(default=None, foreign_key='pitchingcard.id', ondelete='CASCADE')
|
|
ratings_vl_id: int | None = Field(default=None, foreign_key='pitchingratings.id', ondelete='CASCADE')
|
|
ratings_vr_id: int | None = Field(default=None, foreign_key='pitchingratings.id', ondelete='CASCADE')
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class PitcherScouting(PitcherScoutingBase, table=True):
|
|
pitchingcard: PitchingCard = Relationship()
|
|
ratings_vl: PitchingRatings = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[PitcherScouting.ratings_vl_id]",single_parent=True), cascade_delete=True
|
|
)
|
|
ratings_vr: PitchingRatings = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[PitcherScouting.ratings_vr_id]",single_parent=True), cascade_delete=True
|
|
)
|
|
cards: list['Card'] = Relationship(back_populates='pitcherscouting')
|
|
|
|
|
|
class CardBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=False))
|
|
player_id: int = Field(foreign_key='player.id', index=True, ondelete='CASCADE')
|
|
team_id: int = Field(foreign_key='team.id', index=True, ondelete='CASCADE')
|
|
batterscouting_id: int | None = Field(default=None, foreign_key='batterscouting.id', ondelete='CASCADE')
|
|
pitcherscouting_id: int | None = Field(default=None, foreign_key='pitcherscouting.id', ondelete='CASCADE')
|
|
variant: int | None = Field(default=0)
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class Card(CardBase, table=True):
|
|
player: Player = Relationship(back_populates='cards')
|
|
team: Team = Relationship(back_populates='cards')
|
|
lineups: list['Lineup'] = Relationship(back_populates='card', cascade_delete=True)
|
|
batterscouting: BatterScouting = Relationship(back_populates='cards')
|
|
pitcherscouting: PitcherScouting = Relationship(back_populates='cards')
|
|
|
|
|
|
class PositionRatingBase(SQLModel):
|
|
__table_args__ = (UniqueConstraint("player_id", "variant", "position"),)
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
player_id: int = Field(foreign_key='player.id', index=True, ondelete='CASCADE')
|
|
variant: int = Field(default=0, index=True)
|
|
position: str = Field(index=True, include=['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF'])
|
|
innings: int = Field(default=0)
|
|
range: int = Field(default=5)
|
|
error: int = Field(default=0)
|
|
arm: int | None = Field(default=None)
|
|
pb: int | None = Field(default=None)
|
|
overthrow: int | None = Field(default=None)
|
|
created: datetime.datetime = Field(default_factory=datetime.datetime.now, nullable=True)
|
|
|
|
|
|
class PositionRating(PositionRatingBase, table=True):
|
|
player: Player = Relationship(back_populates='positions')
|
|
|
|
|
|
class Lineup(SQLModel, table=True):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
position: str = Field(index=True)
|
|
batting_order: int = Field(index=True)
|
|
after_play: int | None = Field(default=0)
|
|
replacing_id: int | None = Field(default=None)
|
|
active: bool = Field(default=True, index=True)
|
|
is_fatigued: bool | None = Field(default=None)
|
|
|
|
game_id: int = Field(foreign_key='game.id', index=True, ondelete='CASCADE')
|
|
game: Game = Relationship(back_populates='lineups')
|
|
|
|
team_id: int = Field(foreign_key='team.id', index=True, ondelete='CASCADE')
|
|
team: Team = Relationship(back_populates='lineups')
|
|
|
|
player_id: int = Field(foreign_key='player.id', index=True, ondelete='CASCADE')
|
|
player: Player = Relationship(back_populates='lineups')
|
|
|
|
card_id: int = Field(foreign_key='card.id', index=True, ondelete='CASCADE')
|
|
card: Card = Relationship(back_populates='lineups')
|
|
|
|
@field_validator('position')
|
|
def uppercase_strings(cls, value: str) -> str:
|
|
return value.upper()
|
|
|
|
|
|
class PlayBase(SQLModel):
|
|
id: int | None = Field(default=None, sa_column=Column(BigInteger(), primary_key=True, autoincrement=True))
|
|
game_id: int = Field(foreign_key='game.id')
|
|
play_num: int
|
|
batter_id: int = Field(foreign_key='lineup.id')
|
|
pitcher_id: int = Field(foreign_key='lineup.id')
|
|
on_base_code: int = Field(default=0)
|
|
inning_half: str = Field(default='top')
|
|
inning_num: int = Field(default=1, ge=1)
|
|
batting_order: int = Field(default=1, ge=1, le=9)
|
|
starting_outs: int = Field(default=0, ge=0, le=2)
|
|
away_score: int = Field(default=0, ge=0)
|
|
home_score: int = Field(default=0, ge=0)
|
|
batter_pos: str | None = Field(default=None)
|
|
in_pow: bool = Field(default=False)
|
|
|
|
on_first_id: int | None = Field(default=None, foreign_key='lineup.id')
|
|
on_first_final: int | None = Field(default=None) # None = out, 1-4 = base
|
|
on_second_id: int | None = Field(default=None, foreign_key='lineup.id')
|
|
on_second_final: int | None = Field(default=None) # None = out, 1-4 = base
|
|
on_third_id: int | None = Field(default=None, foreign_key='lineup.id')
|
|
on_third_final: int | None = Field(default=None) # None = out, 1-4 = base
|
|
batter_final: int | None = Field(default=None) # None = out, 1-4 = base
|
|
|
|
pa: int = Field(default=1, ge=0, le=1)
|
|
ab: int = Field(default=1, ge=0, le=1)
|
|
run: int = Field(default=0, ge=0, le=1)
|
|
e_run: int = Field(default=0, ge=0, le=1)
|
|
hit: int = Field(default=0, ge=0, le=1)
|
|
rbi: int = Field(default=0)
|
|
double: int = Field(default=0, ge=0, le=1)
|
|
triple: int = Field(default=0, ge=0, le=1)
|
|
homerun: int = Field(default=0, ge=0, le=1)
|
|
bb: int = Field(default=0, ge=0, le=1)
|
|
so: int = Field(default=0, ge=0, le=1)
|
|
hbp: int = Field(default=0, ge=0, le=1)
|
|
sac: int = Field(default=0, ge=0, le=1)
|
|
ibb: int = Field(default=0, ge=0, le=1)
|
|
gidp: int = Field(default=0, ge=0, le=1)
|
|
bphr: int = Field(default=0, ge=0, le=1)
|
|
bpfo: int = Field(default=0, ge=0, le=1)
|
|
bp1b: int = Field(default=0, ge=0, le=1)
|
|
bplo: int = Field(default=0, ge=0, le=1)
|
|
sb: int = Field(default=0, ge=0, le=1)
|
|
cs: int = Field(default=0, ge=0, le=1)
|
|
outs: int = Field(default=0, ge=0, le=3)
|
|
|
|
wpa: float = Field(default=0)
|
|
re24: float = Field(default=0)
|
|
|
|
catcher_id: int = Field(foreign_key='lineup.id')
|
|
defender_id: int | None = Field(default=None, foreign_key='lineup.id')
|
|
runner_id: int | None = Field(default=None, foreign_key='lineup.id')
|
|
|
|
check_pos: str | None = Field(default=None)
|
|
error: int = Field(default=0)
|
|
wild_pitch: int = Field(default=0, ge=0, le=1)
|
|
passed_ball: int = Field(default=0, ge=0, le=1)
|
|
pick_off: int = Field(default=0, ge=0, le=1)
|
|
balk: int = Field(default=0, ge=0, le=1)
|
|
complete: bool = Field(default=False)
|
|
locked: bool = Field(default=False)
|
|
is_go_ahead: bool = Field(default=False)
|
|
is_tied: bool = Field(default=False)
|
|
is_new_inning: bool = Field(default=False)
|
|
managerai_id: int | None = Field(default=None, foreign_key='managerai.id')
|
|
|
|
@field_validator('inning_half')
|
|
def lowercase_strings(cls, value: str) -> str:
|
|
return value.lower()
|
|
|
|
@field_validator('check_pos', 'batter_pos')
|
|
def uppercase_strings(cls, value: str) -> str:
|
|
return value.upper()
|
|
|
|
@property
|
|
def ai_run_diff(self) -> int:
|
|
if self.game.ai_team == 'away':
|
|
return self.away_score - self.home_score
|
|
else:
|
|
return self.home_score - self.away_score
|
|
|
|
|
|
class Play(PlayBase, table=True):
|
|
game: Game = Relationship(back_populates='plays')
|
|
batter: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.batter_id]")
|
|
)
|
|
pitcher: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.pitcher_id]")
|
|
)
|
|
on_first: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.on_first_id]")
|
|
)
|
|
on_second: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.on_second_id]")
|
|
)
|
|
on_third: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.on_third_id]")
|
|
)
|
|
catcher: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.catcher_id]")
|
|
)
|
|
defender: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.defender_id]")
|
|
)
|
|
runner: Lineup = Relationship(
|
|
sa_relationship_kwargs=dict(foreign_keys="[Play.runner_id]")
|
|
)
|
|
managerai: ManagerAi = Relationship(back_populates='plays')
|
|
|
|
def init_ai(self, session: Session):
|
|
id = ((datetime.datetime.now().day * self.batter.team.id) % 3) + 1
|
|
if id > 3 or id < 1:
|
|
self.managerai_id = 1
|
|
else:
|
|
self.managerai_id = id
|
|
|
|
session.add(self)
|
|
session.commit()
|
|
|
|
@property
|
|
def scorebug_ascii(self):
|
|
occupied = '●'
|
|
unoccupied = '○'
|
|
|
|
first_base = unoccupied if not self.on_first else occupied
|
|
second_base = unoccupied if not self.on_second else occupied
|
|
third_base = unoccupied if not self.on_third else occupied
|
|
half = '▲' if self.inning_half == 'top' else '▼'
|
|
|
|
if self.game.active:
|
|
inning = f'{half} {self.inning_num}'
|
|
outs = f'{self.starting_outs} Out{"s" if self.starting_outs != 1 else ""}'
|
|
else:
|
|
inning = f'F/{self.inning_num if self.inning_half == "bot" else self.inning_num - 1}'
|
|
outs = ''
|
|
|
|
game_string = f'```\n' \
|
|
f'{self.game.away_team.abbrev.replace("Gauntlet-", ""): ^5}{self.away_score: ^3} {second_base}{inning: >10}\n' \
|
|
f'{self.game.home_team.abbrev.replace("Gauntlet-", ""): ^5}{self.home_score: ^3} {third_base} {first_base}{outs: >8}\n```'
|
|
|
|
return game_string
|
|
|
|
@property
|
|
def ai_is_batting(self) -> bool:
|
|
if self.game.ai_team is None:
|
|
return False
|
|
|
|
if (self.game.ai_team == 'away' and self.inning_half == 'top') or (self.game.ai_team == 'home' and self.inning_half == 'bot'):
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
@property
|
|
def could_walkoff(self) -> bool:
|
|
if self.inning_half == 'bot' and self.on_third is not None:
|
|
runs_needed = self.away_score - self.home_score + 1
|
|
|
|
if runs_needed == 2 or (self.home_score - self.away_score == 9):
|
|
return True
|
|
|
|
else:
|
|
return False
|
|
|
|
|
|
"""
|
|
BEGIN DEVELOPMENT HELPERS
|
|
"""
|
|
|
|
|
|
def create_db_and_tables():
|
|
SQLModel.metadata.create_all(engine)
|
|
ManagerAi.create_ai()
|
|
|
|
|
|
def create_test_games():
|
|
with Session(engine) as session:
|
|
game_1 = Game(
|
|
away_team_id=1,
|
|
home_team_id=2,
|
|
channel_id=1234,
|
|
season=9,
|
|
)
|
|
game_2 = Game(
|
|
away_team_id=3,
|
|
home_team_id=4,
|
|
channel_id=5678,
|
|
season=9,
|
|
)
|
|
|
|
cardset_2024 = Cardset(name='2024 Season', ranked_legal=True)
|
|
cardset_2022 = Cardset(name='2022 Season', ranked_legal=False)
|
|
|
|
game_1_cardset_2024_link = GameCardsetLink(game=game_1, cardset=cardset_2024, priority=1)
|
|
game_1_cardset_2022_link = GameCardsetLink(game=game_1, cardset=cardset_2022, priority=2)
|
|
game_2_cardset_2024_link = GameCardsetLink(game=game_2, cardset=cardset_2024, priority=1)
|
|
|
|
for team_id in [1, 2]:
|
|
for (order, pos) in [(1, 'C'), (2, '1B'), (3, '2B'), (4, '3B'), (5, 'SS'), (6, 'LF'), (7, 'CF'), (8, 'RF'), (9, 'DH')]:
|
|
this_lineup = Lineup(team_id=team_id, card_id=order, player_id=68+order, position=pos, batting_order=order, game=game_1)
|
|
|
|
for team_id in [3, 4]:
|
|
for (order, pos) in [(1, 'C'), (2, '1B'), (3, '2B'), (4, '3B'), (5, 'SS'), (6, 'LF'), (7, 'CF'), (8, 'RF'), (9, 'DH')]:
|
|
this_lineup = Lineup(team_id=team_id, card_id=order, player_id=100+order, position=pos, batting_order=order, game=game_2)
|
|
|
|
session.add(game_1)
|
|
session.add(game_2)
|
|
session.commit()
|
|
|
|
|
|
def select_speed_testing():
|
|
with Session(engine) as session:
|
|
game_1 = session.exec(select(Game).where(Game.id == 1)).one()
|
|
ss_search_start = datetime.datetime.now()
|
|
man_ss = [x for x in game_1.lineups if x.position == 'SS' and x.active]
|
|
ss_search_end = datetime.datetime.now()
|
|
|
|
ss_query_start = datetime.datetime.now()
|
|
query_ss = session.exec(select(Lineup).where(Lineup.game == game_1, Lineup.position == 'SS', Lineup.active == True)).all()
|
|
ss_query_end = datetime.datetime.now()
|
|
|
|
manual_time = ss_search_end - ss_search_start
|
|
query_time = ss_query_end - ss_query_start
|
|
|
|
print(f'Manual Shortstops: time: {manual_time.microseconds} ms / {man_ss}')
|
|
print(f'Query Shortstops: time: {query_time.microseconds} ms / {query_ss}')
|
|
print(f'Game: {game_1}')
|
|
|
|
games = session.exec(select(Game).where(Game.active == True)).all()
|
|
print(f'len(games): {len(games)}')
|
|
|
|
|
|
def select_all_testing():
|
|
with Session(engine) as session:
|
|
game_search = session.exec(select(Team)).all()
|
|
for game in game_search:
|
|
print(f'Game: {game}')
|
|
|
|
|
|
# def select_specic_fields():
|
|
# with Session(engine) as session:
|
|
# games = session.exec(select(Game.id, Game.away_team, Game.home_team))
|
|
# print(f'Games: {games}')
|
|
# print(f'.all(): {games.all()}')
|
|
|
|
|
|
def main():
|
|
create_db_and_tables()
|
|
create_test_games()
|
|
# select_speed_testing()
|
|
# select_all_testing()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|