paper-dynasty-discord/in_game/gameplay_models.py
Cal Corum 04d6cf3b5e fix: replace star imports with explicit named imports across codebase
Convert all `from x import *` to explicit imports in 12 files, resolving
925 F403/F405 ruff violations. Each name traced to its canonical source
module. Also fixes: duplicate Session import (players.py), missing
sample_team_data fixture param, duplicate method name paperdex_cardset_slash,
unused commands import in package __init__ files, and Play type hint in
play_lock.py.

3 pre-existing code bugs remain (F811 duplicate test names, F821 undefined
`question` variable) — these need investigation, not mechanical fixes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-09 12:48:07 -05:00

1697 lines
62 KiB
Python

import datetime
import logging
from typing import Literal
import os
import discord
from pydantic import field_validator
from sqlmodel import (
Session,
SQLModel,
UniqueConstraint,
create_engine,
select,
or_,
Field,
Relationship,
BigInteger,
)
from sqlalchemy import Column, func
from exceptions import (
CardNotFoundException,
GameException,
LineupsMissingException,
MultipleHumanTeamsException,
NoHumanTeamsException,
PlayInitException,
log_exception,
)
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, pool_size=10, max_overflow=30)
CACHE_LIMIT = 259200 # 1209600 # in seconds
SBA_COLOR = "a6ce39"
SBA_LOGO = "https://paper-dynasty.s3.us-east-1.amazonaws.com/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) # noqa: E712
.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 = "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()
logger.info(f"all_lineups: {all_lineups}")
lineup_val = ""
for line in all_lineups:
logger.info(f"line in all_lineups: {line}")
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":
if line.card.pitcherscouting:
this_hand = line.card.pitcherscouting.pitchingcard.hand
elif line.card.batterscouting:
# Fallback to batting hand if pitcherscouting is missing
this_hand = line.card.batterscouting.battingcard.hand
else:
this_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("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("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,
"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("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,
"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("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,
"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("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("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("No game found while checking throw_at_uncapped")
ai_rd = this_play.ai_run_diff
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("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:
logger.info(f"checking defensive alignment in game {this_game.id}")
this_resp = DefenseResponse()
this_play = this_game.current_play_or_none(session)
if this_play is None:
raise GameException("No game found while checking defense_alignment")
logger.info(f"defense_alignment - this_play: {this_play}")
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("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("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("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 += "- 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 += "- 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 += "- play the whole infield in\n"
else:
this_resp.corners_in = True
this_resp.ai_note += "- play the corners in\n"
if len(this_resp.ai_note) == 0 and this_play.on_base_code > 0:
this_resp.ai_note += "- 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("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("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("Checking if fatigued pitcher should be replaced")
this_play = this_game.current_play_or_none(session)
if this_play is None:
raise GameException("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, # noqa: E712
)
).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("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("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(
"AI team is way behind and starter is going to wear it - staying in"
)
return False
else:
logger.info("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("Only allow POW + 1 IP - pull reliever")
return True
elif this_pitcher.is_fatigued and this_play.is_new_inning:
logger.info("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(
"AI team is way behind and reliever is going to wear it - staying in"
)
return False
else:
logger.info("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 self.image and "batting" in self.image:
return self.image
elif self.image2 and "batting" in self.image2:
return self.image2
else:
return None
@property
def pitcher_card_url(self):
if self.image and "pitching" in self.image:
return self.image
elif self.image2 and "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)
variant: int = Field(default=0, index=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, # noqa: E712
)
).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) # noqa: E712
).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()