1c24161e76
- Fix TypeError in check_steal_opportunity by properly mocking catcher defense - Correct tag_from_third test calculation to account for all adjustment conditions - Fix pitcher replacement test by setting appropriate allowed runners threshold - Add comprehensive test coverage for AI service business logic - Implement VS Code testing panel configuration with pytest integration - Create pytest.ini for consistent test execution and warning management - Add test isolation guidelines and factory pattern implementation - Establish 102 passing tests with zero failures 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
675 lines
27 KiB
Python
675 lines
27 KiB
Python
"""
|
|
AIService - AI decision-making business logic.
|
|
|
|
Extracted from Discord app ManagerAi model methods.
|
|
Handles all AI decision-making for gameplay mechanics.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Literal
|
|
from sqlmodel import Session, select, func, or_
|
|
from datetime import datetime
|
|
|
|
from .base_service import BaseService
|
|
from ..models.manager_ai import ManagerAi
|
|
from ..models.position_rating import PositionRating
|
|
from ..models.ai_responses import (
|
|
JumpResponse,
|
|
TagResponse,
|
|
ThrowResponse,
|
|
UncappedRunResponse,
|
|
DefenseResponse,
|
|
RunResponse,
|
|
)
|
|
|
|
|
|
class AIService(BaseService):
|
|
"""Service for AI decision-making in gameplay."""
|
|
|
|
def __init__(self, session: Session):
|
|
super().__init__(session)
|
|
|
|
def check_steal_opportunity(
|
|
self,
|
|
manager_ai: ManagerAi,
|
|
game: "Game",
|
|
to_base: Literal[2, 3, 4]
|
|
) -> JumpResponse:
|
|
"""
|
|
Check if AI should attempt a steal to the specified base.
|
|
|
|
Migrated from ManagerAi.check_jump() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
to_base: Target base (2, 3, or 4)
|
|
|
|
Returns:
|
|
JumpResponse with steal decision details
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
CardNotFoundException: If no runner found on required base
|
|
"""
|
|
self._log_operation(f"check_steal_opportunity", f"to base {to_base} in game {game.id}")
|
|
|
|
this_resp = JumpResponse(min_safe=20)
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError(f"No game found while checking for steal")
|
|
|
|
num_outs = this_play.starting_outs
|
|
run_diff = this_play.away_score - this_play.home_score
|
|
if game.ai_team == 'home':
|
|
run_diff = run_diff * -1
|
|
|
|
pitcher_hold = this_play.pitcher.card.pitcherscouting.pitchingcard.hold
|
|
catcher_defense = self.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
|
|
|
|
self.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:
|
|
raise ValueError(f"Attempted to check a steal to 2nd base, but no runner found on first.")
|
|
|
|
self.logger.info(f"Checking steal numbers for {runner.player.name} in Game {game.id}")
|
|
|
|
match manager_ai.steal:
|
|
case 10:
|
|
this_resp.min_safe = 12 + num_outs
|
|
case steal if steal > 8 and run_diff <= 5:
|
|
this_resp.min_safe = 13 + num_outs
|
|
case steal if steal > 6 and run_diff <= 5:
|
|
this_resp.min_safe = 14 + num_outs
|
|
case steal if steal > 4 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.min_safe = 15 + num_outs
|
|
case steal if 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 manager_ai.steal > 7 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.run_if_auto_jump = True
|
|
elif manager_ai.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:
|
|
self.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
|
|
self.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:
|
|
raise ValueError(f"Attempted to check a steal to 3rd base, but no runner found on second.")
|
|
|
|
match manager_ai.steal:
|
|
case 10:
|
|
this_resp.min_safe = 12 + num_outs
|
|
case steal if steal > 6 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.min_safe = 15 + num_outs
|
|
case _:
|
|
this_resp.min_safe = None
|
|
|
|
if manager_ai.steal == 10 and num_outs < 2 and run_diff <= 5:
|
|
this_resp.run_if_auto_jump = True
|
|
elif manager_ai.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:
|
|
self.logger.info("No jump ai note")
|
|
else:
|
|
jump_safe_range = runner_card.steal_low + battery_hold
|
|
self.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:
|
|
raise ValueError(f"Attempted to check a steal to home, but no runner found on third.")
|
|
|
|
if manager_ai.steal == 10:
|
|
this_resp.min_safe = 5
|
|
elif this_play.inning_num > 7 and manager_ai.steal >= 5:
|
|
this_resp.min_safe = 6
|
|
elif manager_ai.steal > 5:
|
|
this_resp.min_safe = 7
|
|
elif manager_ai.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!"
|
|
|
|
self.logger.info(f"Returning steal response for game {game.id}: {this_resp}")
|
|
return this_resp
|
|
|
|
def check_tag_from_second(self, manager_ai: ManagerAi, game: "Game") -> TagResponse:
|
|
"""
|
|
Check if runner on second should tag up on a fly ball.
|
|
|
|
Migrated from ManagerAi.tag_from_second() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
TagResponse with tag decision details
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("check_tag_from_second", f"game {game.id}")
|
|
|
|
this_resp = TagResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking tag_from_second")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression_mod = abs(manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.behind_aggression - 5)
|
|
adjusted_running = manager_ai.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
|
|
|
|
self.logger.info(f"tag_from_second response: {this_resp}")
|
|
return this_resp
|
|
|
|
def check_tag_from_third(self, manager_ai: ManagerAi, game: "Game") -> TagResponse:
|
|
"""
|
|
Check if runner on third should tag up on a fly ball.
|
|
|
|
Migrated from ManagerAi.tag_from_third() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
TagResponse with tag decision details
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("check_tag_from_third", f"game {game.id}")
|
|
|
|
this_resp = TagResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking tag_from_third")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression_mod = abs(manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.behind_aggression - 5)
|
|
adjusted_running = manager_ai.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
|
|
|
|
self.logger.info(f"tag_from_third response: {this_resp}")
|
|
return this_resp
|
|
|
|
def decide_throw_target(self, manager_ai: ManagerAi, game: "Game") -> ThrowResponse:
|
|
"""
|
|
Decide where to throw on uncapped advances.
|
|
|
|
Migrated from ManagerAi.throw_at_uncapped() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
ThrowResponse with throw target decision
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("decide_throw_target", f"game {game.id}")
|
|
|
|
this_resp = ThrowResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking throw_at_uncapped")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = manager_ai.ahead_aggression if ai_rd > 0 else manager_ai.behind_aggression
|
|
current_outs = this_play.starting_outs + this_play.outs
|
|
|
|
if ai_rd > 5:
|
|
if manager_ai.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 manager_ai.ahead_aggression > 8:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4 + current_outs
|
|
elif ai_rd > 0:
|
|
if manager_ai.ahead_aggression > 8:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -6 + current_outs
|
|
elif ai_rd > -3:
|
|
if manager_ai.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -6 + current_outs
|
|
elif ai_rd > -6:
|
|
if manager_ai.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4 + current_outs
|
|
else:
|
|
if manager_ai.behind_aggression < 5:
|
|
this_resp.at_trail_runner = True
|
|
this_resp.trail_max_safe_delta = -4
|
|
|
|
self.logger.info(f"throw_at_uncapped response: {this_resp}")
|
|
return this_resp
|
|
|
|
def decide_runner_advance(
|
|
self,
|
|
manager_ai: ManagerAi,
|
|
game: "Game",
|
|
lead_base: int,
|
|
trail_base: int
|
|
) -> UncappedRunResponse:
|
|
"""
|
|
Decide if runners should advance on uncapped situations.
|
|
|
|
Migrated from ManagerAi.uncapped_advance() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
lead_base: Base number for lead runner
|
|
trail_base: Base number for trail runner
|
|
|
|
Returns:
|
|
UncappedRunResponse with advance decisions
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("decide_runner_advance", f"game {game.id}, lead_base {lead_base}, trail_base {trail_base}")
|
|
|
|
this_resp = UncappedRunResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking uncapped_advance")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.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
|
|
|
|
# Bounds checking
|
|
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.trail_min_safe = 20
|
|
if this_resp.trail_min_safe < 1:
|
|
this_resp.trail_min_safe = 1
|
|
|
|
self.logger.info(f"uncapped advance response: {this_resp}")
|
|
return this_resp
|
|
|
|
def set_defensive_alignment(self, manager_ai: ManagerAi, game: "Game") -> DefenseResponse:
|
|
"""
|
|
Determine defensive alignment and holds.
|
|
|
|
Migrated from ManagerAi.defense_alignment() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
DefenseResponse with defensive decisions
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("set_defensive_alignment", f"game {game.id}")
|
|
|
|
this_resp = DefenseResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking defense_alignment")
|
|
|
|
self.logger.info(f"defense_alignment - this_play: {this_play}")
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.behind_aggression - 5
|
|
pitcher_hold = this_play.pitcher.card.pitcherscouting.pitchingcard.hold
|
|
|
|
catcher_defense = self.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
|
|
|
|
# Hold decisions
|
|
if this_play.starting_outs == 2 and this_play.on_base_code > 0:
|
|
self.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]:
|
|
self.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]:
|
|
self.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"
|
|
|
|
self.logger.info(f"Defense alignment response: {this_resp}")
|
|
return this_resp
|
|
|
|
def decide_groundball_running(self, manager_ai: ManagerAi, game: "Game") -> RunResponse:
|
|
"""
|
|
Decide if AI should run on groundball.
|
|
|
|
Migrated from ManagerAi.gb_decide_run() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
RunResponse with running decision
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("decide_groundball_running", f"game {game.id}")
|
|
|
|
this_resp = RunResponse()
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking gb_decide_run")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.behind_aggression - 5
|
|
|
|
this_resp.min_safe = 15 - aggression # TODO: write this algorithm
|
|
self.logger.info(f"gb_decide_run response: {this_resp}")
|
|
return this_resp
|
|
|
|
def decide_groundball_throw(
|
|
self,
|
|
manager_ai: ManagerAi,
|
|
game: "Game",
|
|
runner_speed: int,
|
|
defender_range: int
|
|
) -> ThrowResponse:
|
|
"""
|
|
Decide where to throw on groundball with runner.
|
|
|
|
Migrated from ManagerAi.gb_decide_throw() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
runner_speed: Speed of the runner
|
|
defender_range: Range of the fielding defender
|
|
|
|
Returns:
|
|
ThrowResponse with throw decision
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("decide_groundball_throw", f"game {game.id}")
|
|
|
|
this_resp = ThrowResponse(at_lead_runner=True)
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking gb_decide_throw")
|
|
|
|
ai_rd = this_play.ai_run_diff
|
|
aggression = manager_ai.ahead_aggression - 5 if ai_rd > 0 else manager_ai.behind_aggression - 5
|
|
|
|
if (runner_speed - 4 + defender_range) <= (10 + aggression):
|
|
this_resp.at_lead_runner = True
|
|
|
|
self.logger.info(f"gb_decide_throw response: {this_resp}")
|
|
return this_resp
|
|
|
|
def should_replace_pitcher(self, manager_ai: ManagerAi, game: "Game") -> bool:
|
|
"""
|
|
Determine if fatigued pitcher should be replaced.
|
|
|
|
Migrated from ManagerAi.replace_pitcher() method.
|
|
|
|
Args:
|
|
manager_ai: ManagerAi configuration
|
|
game: Current game
|
|
|
|
Returns:
|
|
bool: True if pitcher should be replaced
|
|
|
|
Raises:
|
|
GameException: If no current play found
|
|
"""
|
|
self._log_operation("should_replace_pitcher", f"game {game.id}")
|
|
|
|
this_play = game.current_play_or_none(self.session)
|
|
if this_play is None:
|
|
raise ValueError("No game found while checking replace_pitcher")
|
|
|
|
this_pitcher = this_play.pitcher
|
|
outs = self.session.exec(
|
|
select(func.sum("Play.outs")).where(
|
|
"Play.game" == game,
|
|
"Play.pitcher" == this_pitcher,
|
|
"Play.complete" == True
|
|
)
|
|
).one()
|
|
self.logger.info(f"Pitcher: {this_pitcher.card.player.name_with_desc} / Outs: {outs}")
|
|
|
|
allowed_runners = self.session.exec(
|
|
select(func.count("Play.id")).where(
|
|
"Play.game" == game,
|
|
"Play.pitcher" == this_pitcher,
|
|
or_("Play.hit" == 1, "Play.bb" == 1)
|
|
)
|
|
).one()
|
|
run_diff = this_play.ai_run_diff
|
|
|
|
self.logger.info(f"run diff: {run_diff} / allowed runners: {allowed_runners} / behind aggro: {manager_ai.behind_aggression} / ahead aggro: {manager_ai.ahead_aggression}")
|
|
self.logger.info(f"this play: {this_play}")
|
|
|
|
if this_pitcher.replacing_id is None:
|
|
# Starter logic
|
|
pitcher_pow = this_pitcher.card.pitcherscouting.pitchingcard.starter_rating
|
|
self.logger.info(f"Starter POW: {pitcher_pow}")
|
|
|
|
if outs >= pitcher_pow * 3 + 6:
|
|
self.logger.info("Starter has thrown POW + 3 - being pulled")
|
|
return True
|
|
|
|
elif allowed_runners < 5:
|
|
self.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:
|
|
self.logger.info("Starter is fatigued")
|
|
return True
|
|
|
|
elif (run_diff > 5 or (run_diff > 2 and manager_ai.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 3):
|
|
self.logger.info(f"AI team has big lead of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif (run_diff > 2 or (run_diff >= 0 and manager_ai.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 1):
|
|
self.logger.info(f"AI team has lead of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif (run_diff >= 0 or (run_diff >= -2 and manager_ai.behind_aggression > 5)) and (allowed_runners < 5 and this_play.on_base_code <= run_diff):
|
|
self.logger.info(f"AI team in close game with run diff of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif run_diff >= -3 and manager_ai.behind_aggression > 5 and allowed_runners < 5 and this_play.on_base_code <= 1:
|
|
self.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:
|
|
self.logger.info("AI team is way behind and starter is going to wear it - staying in")
|
|
return False
|
|
|
|
else:
|
|
self.logger.info("AI team found no exceptions - pull starter")
|
|
return True
|
|
|
|
else:
|
|
# Reliever logic
|
|
pitcher_pow = this_pitcher.card.pitcherscouting.pitchingcard.relief_rating
|
|
self.logger.info(f"Reliever POW: {pitcher_pow}")
|
|
|
|
if outs >= pitcher_pow * 3 + 3:
|
|
self.logger.info("Only allow POW + 1 IP - pull reliever")
|
|
return True
|
|
|
|
elif this_pitcher.is_fatigued and this_play.is_new_inning:
|
|
self.logger.info("Reliever is fatigued to start the inning - pull reliever")
|
|
return True
|
|
|
|
elif (run_diff > 5 or (run_diff > 2 and manager_ai.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):
|
|
self.logger.info(f"AI team has big lead of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif (run_diff > 2 or (run_diff >= 0 and manager_ai.ahead_aggression > 5)) and (allowed_runners < run_diff or this_play.on_base_code <= 1 or this_play.starting_outs == 2):
|
|
self.logger.info(f"AI team has lead of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif (run_diff >= 0 or (run_diff >= -2 and manager_ai.behind_aggression > 5)) and (allowed_runners < 5 or this_play.on_base_code <= run_diff or this_play.starting_outs == 2):
|
|
self.logger.info(f"AI team in close game with run diff of {run_diff} - staying in")
|
|
return False
|
|
|
|
elif run_diff >= -3 and manager_ai.behind_aggression > 5 and allowed_runners < 5 and this_play.on_base_code <= 1:
|
|
self.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:
|
|
self.logger.info("AI team is way behind and reliever is going to wear it - staying in")
|
|
return False
|
|
|
|
else:
|
|
self.logger.info("AI team found no exceptions - pull reliever")
|
|
return True |