cal/paper-dynasty-discord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
eb2a511507
paper-dynasty-discord/in_game/ai_manager.py
Cal Corum 22d15490dd Add pitcher validation with rank retry in get_starting_pitcher 
When API returns a pitcher without pitching data (e.g., Ohtani with
pos_1=DH), explicitly fetch pitcherscouting and validate before use.
If validation fails, retry with different sp_rank values.

Retry strategy: increment rank first, if > 5 then decrement from
original rank, ensuring all 5 ranks are tried before giving up.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-30 14:04:03 -06:00

780 lines
30 KiB
Python
Raw Blame History

import copy
import logging
import math
import random
# import data_cache
from db_calls_gameplay import StratPlay, StratGame, get_one_lineup, get_manager, get_team_lineups, \
get_last_inning_end_play, make_sub, get_player, StratLineup, get_pitching_stats, patch_play, patch_lineup, \
get_one_game
from api_calls import db_get, db_post
from peewee import *
from typing import Optional, Literal
from in_game import data_cache
from in_game.gameplay_models import Play, Session, Game, Team, Lineup
from in_game.gameplay_queries import get_or_create_ai_card, get_player_id_from_dict, get_player_or_none, get_pitcher_scouting_or_none
from exceptions import DatabaseError
db = SqliteDatabase(
'storage/ai-database.db',
pragmas={
'journal_mode': 'wal',
'cache_size': -1 * 64000,
'synchronous': 0
}
)
logger = logging.getLogger('discord_app')
# 2018, 2024, Mario,
GAUNTLET1_PARAMS = [
('cardset_id', 13), ('cardset_id', 14), ('cardset_id', 17), ('cardset_id', 18), ('cardset_id', 8)
]
# 2008, 2012, 2013, 2016
GAUNTLET2_PARAMS = [
('cardset_id', 8), ('cardset_id', 12), ('cardset_id', 6), ('cardset_id', 7), ('cardset_id', 11)
]
class BaseModel(Model):
class Meta:
database = db
# class Lineup(BaseModel):
# team_id = IntegerField()
# game_level = CharField() # 'minor', 'major', 'hof'
# vs_hand = CharField() # 'left', 'right'
# bat_one_card = IntegerField()
# bat_one_pos = CharField()
# bat_two_card = IntegerField()
# bat_two_pos = CharField()
# bat_three_card = IntegerField()
# bat_three_pos = CharField()
# bat_four_card = IntegerField()
# bat_four_pos = CharField()
# bat_five_card = IntegerField()
# bat_five_pos = CharField()
# bat_six_card = IntegerField()
# bat_six_pos = CharField()
# bat_seven_card = IntegerField()
# bat_seven_pos = CharField()
# bat_eight_card = IntegerField()
# bat_eight_pos = CharField()
# bat_nine_card = IntegerField()
# bat_nine_pos = CharField()
class Rotation(BaseModel):
team_id = IntegerField()
game_level = CharField()
sp_one_card = IntegerField()
sp_two_card = IntegerField()
sp_three_card = IntegerField()
sp_four_card = IntegerField()
sp_five_card = IntegerField()
class Bullpen(BaseModel):
team_id = IntegerField()
game_level = CharField() # 'minor', 'major', 'hof'
vs_hand = CharField() # 'left', 'right'
# def grade_player()
def batter_grading(vs_hand, rg_data):
raw_bat = (rg_data[f'contact-{vs_hand}'] + rg_data[f'power-{vs_hand}'] +
min(rg_data[f'contact-{vs_hand}'], rg_data[f'power-{vs_hand}'])) / 3
other_metrics = [
rg_data['vision'], rg_data['speed'], rg_data['stealing'], rg_data['reaction'], rg_data['arm'],
rg_data['fielding']
]
blended_rating = sum(sorted(other_metrics, reverse=True)[:4], raw_bat * 4) / 8
return {
'overall': rg_data['overall'],
'raw-bat': raw_bat,
'blended': blended_rating
}
def get_cardset_string(this_game: StratGame):
cardsets = ''
bcardsets = ''
if this_game.cardset_ids is not None:
for x in this_game.cardset_ids.split(','):
cardsets += f'&cardset_id={x}'
if this_game.backup_cardset_ids is not None:
for x in this_game.backup_cardset_ids.split(','):
bcardsets += f'&backup_cardset_id={x}'
return f'{cardsets}{bcardsets}'
async def get_or_create_card(player: dict, team: dict) -> int:
# get player card; create one if none found
z = 0
card_id = None
while z < 2 and card_id is None:
z += 1
c_query = await db_get('cards', params=[('team_id', team['id']), ('player_id', player['player_id'])])
if c_query['count'] > 0:
card_id = c_query['cards'][0]['id']
else:
await db_post(
'cards',
payload={'cards': [
{'player_id': player['player_id'], 'team_id': team['id'], 'pack_id': 1}
]}
)
if card_id is None:
logger.error(f'Could not create card for {player["p_name"]} on the {team["lname"]}')
raise DatabaseError(f'Could not create card for {player["p_name"]} on the {team["lname"]}')
return card_id
# First attempt - pulls ratings guide info
async def build_lineup_graded(team_object: dict, vs_hand: str, league_name: str, num_innings: int, batter_rg):
in_lineup = [] # player ids for quick checking
players = {
'c': [],
'1b': [],
'2b': [],
'3b': [],
'ss': [],
'lf': [],
'cf': [],
'rf': [],
'dh': []
}
# Get all eligible players
try:
p_query = await db_get(
endpoint='players',
params=[('mlbclub', team_object['lname']), ('pos_exclude', 'RP'), ('inc_dex', False)],
timeout=10
)
all_players = p_query['players']
except ConnectionError as e:
raise ConnectionError(f'Error pulling players for the {team_object["lname"]}. Cal help plz.')
# Grade players and add to position lists
for x in all_players:
if x['pos_1'] not in ['SP', 'RP']:
this_guy = copy.deepcopy(x)
rg_data = next(x for x in batter_rg if x['player_id'] == this_guy['player_id'])
grading = batter_grading(vs_hand, rg_data)
logger.info(f'player: {this_guy} / grading: {grading}')
this_guy['overall'] = grading['overall']
this_guy['raw-bat'] = grading['raw-bat']
this_guy['blended'] = grading['blended']
players['dh'].append(this_guy)
if 'C' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['c'].append(this_guy)
if '1B' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['1b'].append(this_guy)
if '2B' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['2b'].append(this_guy)
if '3B' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['3b'].append(this_guy)
if 'SS' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['ss'].append(this_guy)
if 'LF' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['lf'].append(this_guy)
if 'CF' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['cf'].append(this_guy)
if 'RF' in [this_guy['pos_1'], this_guy['pos_2'], this_guy['pos_3'], this_guy['pos_4'], this_guy['pos_5'],
this_guy['pos_6'], this_guy['pos_7'], this_guy['pos_8']]:
players['rf'].append(this_guy)
# Select players for lineup
while len(players) > 0:
# Sort players dict by position with fewest eligible players
this_pass_data = sorted(players.items(), key=lambda item: len(item[1]))
logger.info(f'this_pass_data: {this_pass_data}')
# Pull one tuple ('<position>', [<player objects>])
this_pos_data = this_pass_data[0]
logger.info(f'this_pos_data: {this_pos_data}')
# Sort players at this position by blended rating (raw-bat for DH)
if this_pos_data[0] == 'dh':
this_pos = sorted(this_pos_data[1], key=lambda item: item['raw-bat'], reverse=True)
else:
this_pos = sorted(this_pos_data[1], key=lambda item: item['blended'], reverse=True)
logger.info(f'this_pos: {this_pos}')
# Add top player to the lineup
in_lineup.append({'position': copy.deepcopy(this_pos_data[0].upper()), 'player': copy.deepcopy(this_pos[0])})
logger.info(f'adding player: {this_pos[0]}')
logger.info(f'deleting position: {this_pos_data[0]}')
# Remove this position from consideration
del players[this_pos_data[0]]
for key in players:
for x in players[key]:
# Remove duplicate players (even across cardsets) once in lineup
if x['strat_code'] == this_pos[0]['strat_code']:
players[key].remove(x)
# Set batting order as list of lists: [ ['<pos>', <player_id>], ... ]
batting_order = []
return batting_order
async def build_lineup(team_object: dict, game_id: int, league_name: str, sp_name: str, vs_hand: str = 'r') -> list:
build_type = 'fun'
this_game = get_one_game(game_id=game_id)
l_query = await db_get(
f'teams/{team_object["id"]}/lineup/{league_name}?pitcher_name={sp_name}&build_type={build_type}'
f'{get_cardset_string(this_game)}',
timeout=6
)
sorted_players = l_query['array']
logger.info(f'sorted_players: {sorted_players}')
grp_1 = sorted_players[:3]
grp_2 = sorted_players[3:6]
grp_3 = sorted_players[6:]
random.shuffle(grp_1)
random.shuffle(grp_2)
random.shuffle(grp_3)
lineups = []
i = 1
for x in [grp_1, grp_2, grp_3]:
logger.debug(f'group: {x}')
for y in x:
logger.debug(f'y: {y}')
card_id = await get_or_create_card(y[1]['player'], team_object)
lineups.append({
'game_id': game_id,
'team_id': team_object['id'],
'player_id': y[1]['player']['player_id'],
'card_id': card_id,
'position': y[0],
'batting_order': i,
'after_play': 0
})
i += 1
logger.info(f'build_lineup - final lineup: {lineups}')
return lineups
async def get_starting_lineup(session: Session, team: Team, game: Game, league_name: str, sp_name: str, vs_hand: str = 'r') -> list[Lineup]:
build_type = 'fun'
l_query = await db_get(
f'teams/{team.id}/lineup/{league_name}?pitcher_name={sp_name}&build_type={build_type}{game.cardset_param_string}',
timeout=6
)
sorted_players = l_query['array']
logger.debug(f'ai_manager - get_starting_lineup - sorted_players: {sorted_players}')
grp_1 = sorted_players[:3]
grp_2 = sorted_players[3:6]
grp_3 = sorted_players[6:]
random.shuffle(grp_1)
random.shuffle(grp_2)
random.shuffle(grp_3)
lineups = []
i = 1
for x in [grp_1, grp_2, grp_3]:
logger.debug(f'ai_manager - get_starting_lineup - group: {x}')
for y in x:
logger.debug(f'ai_manager - get_starting_lineup - y: {y}')
this_player = await get_player_or_none(session, get_player_id_from_dict(y[1]['player']))
this_card = await get_or_create_ai_card(session, player=this_player, team=team)
lineups.append(Lineup(
position=y[0],
batting_order=i,
game=game,
team=team,
player=this_player,
card=this_card
))
i += 1
logger.debug(f'ai_manager - get_starting_lineup - final lineup: {lineups}')
return lineups
async def get_starting_pitcher(
session: Session, this_team: Team, this_game: Game, is_home: bool, league_name: str) -> Lineup:
d_100 = random.randint(1, 100)
logger.info(f'Getting a {league_name} starting pitcher for the {this_team.lname}; d100: {d_100}')
if is_home:
if d_100 <= 30:
sp_rank = 1
elif d_100 <= 55:
sp_rank = 2
elif d_100 <= 75:
sp_rank = 3
elif d_100 <= 90:
sp_rank = 4
else:
sp_rank = 5
else:
if d_100 <= 50:
sp_rank = 1
elif d_100 <= 75:
sp_rank = 2
elif d_100 <= 85:
sp_rank = 3
elif d_100 <= 95:
sp_rank = 4
else:
sp_rank = 5
logger.info(f'chosen rank: {sp_rank}')
# Try to get a pitcher with valid pitching data, retrying with different ranks if needed
original_rank = sp_rank
tried_ranks = set()
direction = 1 # 1 = incrementing, -1 = decrementing
while len(tried_ranks) < 5:
tried_ranks.add(sp_rank)
logger.info(f'Trying sp_rank: {sp_rank}')
sp_query = await db_get(
f'teams/{this_team.id}/sp/{league_name}?sp_rank={sp_rank}{this_game.cardset_param_string}'
)
this_player = await get_player_or_none(session, get_player_id_from_dict(sp_query))
sp_card = await get_or_create_ai_card(session, this_player, this_team)
# Validate pitcher has pitching data
try:
pitcher_scouting = await get_pitcher_scouting_or_none(session, sp_card)
if pitcher_scouting is not None:
sp_card.pitcherscouting = pitcher_scouting
session.add(sp_card)
session.commit()
session.refresh(sp_card)
logger.info(f'Found valid pitcher at rank {sp_rank}: {this_player.name_with_desc}')
break
else:
logger.warning(f'Pitcher at rank {sp_rank} ({this_player.name_with_desc}) returned None for pitcherscouting')
except DatabaseError:
logger.warning(f'Pitcher at rank {sp_rank} ({this_player.name_with_desc}) lacks pitching data, trying another')
# Adjust rank: increment first, if we hit 6, switch to decrementing from original
sp_rank += direction
if sp_rank > 5:
direction = -1
sp_rank = original_rank - 1
if sp_rank < 1:
# Find any untried rank
untried = [r for r in range(1, 6) if r not in tried_ranks]
if untried:
sp_rank = untried[0]
else:
break
return Lineup(
team=this_team,
player=this_player,
card=sp_card,
position='P',
batting_order=10,
is_fatigued=False,
game=this_game
)
async def get_relief_pitcher(this_play: StratPlay, ai_team: dict, league_name: str = None) -> dict:
used_ids = []
used_players = await get_team_lineups(
game_id=this_play.game.id, team_id=ai_team['id'], inc_inactive=True, as_string=False, pitchers_only=True
)
for x in used_players:
used_ids.append(f'{x.player_id}')
logger.debug(f'used ids: {used_ids}')
id_string = "&used_pitcher_ids=".join(used_ids)
this_game = this_play.game
ai_score = this_play.away_score if this_game.away_team_id == ai_team['id'] else this_play.home_score
human_score = this_play.home_score if this_game.away_team_id == ai_team['id'] else this_play.away_score
logger.debug(f'scores - ai: {ai_score} / human: {human_score}')
if abs(ai_score - human_score) >= 7:
need = 'length'
elif this_play.inning_num >= 9 and abs(ai_score - human_score) <= 3:
need = 'closer'
elif this_play.inning_num in [7, 8] and abs(ai_score - human_score) <= 3:
need = 'setup'
elif abs(ai_score - human_score) <= 3:
need = 'middle'
else:
need = 'length'
logger.debug(f'need: {need}')
rp_query = await db_get(f'teams/{ai_team["id"]}/rp/{league_name.split("-run")[0]}'
f'?need={need}&used_pitcher_ids={id_string}{get_cardset_string(this_game)}')
card_id = await get_or_create_card(rp_query, ai_team)
return {
'game_id': this_play.game.id,
'team_id': ai_team['id'],
'player_id': rp_query['player_id'],
'card_id': card_id,
'position': 'P',
'batting_order': 10,
'after_play': this_play.play_num - 1
}
"""
END NEW GET RP
"""
# used_codes = []
# used_players = await get_team_lineups(
# game_id=this_play.game.id, team_id=ai_team['id'], inc_inactive=True, as_string=False
# )
# for x in used_players:
# c_query = await db_get('cards', object_id=x.card_id)
# used_codes.append(c_query["player"]["strat_code"])
# logger.info(f'get_rp - used_players: {used_codes}')
#
# reliever = None
# attempts = 0
# while reliever is None:
# attempts += 1
# if attempts > 3:
# raise ValueError(f'Could not find a reliever for the {ai_team["sname"]}. Cal plz.')
#
# set_params = [('cardset_id_exclude', 2)]
# if league_name == 'minor-league':
# set_params = copy.deepcopy(MINOR_CARDSET_PARAMS)
# elif league_name == 'major-league':
# set_params = copy.deepcopy(MAJOR_CARDSET_PARAMS)
# elif league_name == 'hall-of-fame':
# set_params = copy.deepcopy(HOF_CARDSET_PARAMS)
# elif 'gauntlet-1' in league_name:
# set_params = copy.deepcopy(MINOR_CARDSET_PARAMS)
# elif 'gauntlet-2' in league_name:
# set_params = copy.deepcopy(GAUNTLET2_PARAMS)
#
# # Pull relievers sorted by current cost
# params = [
# ('mlbclub', ai_team['lname']), ('pos_include', 'RP'), ('inc_dex', False), ('sort_by', 'cost-desc'),
# ('limit', 15)
# ]
# params.extend(set_params)
#
# use_best = False
# if attempts == 1:
# # Try to get long man
# if this_play.inning_num < 6:
# logger.info(f'get_rp - game {this_play.game.id} try for long man')
# params.append(('pos_include', 'SP'))
# # use_best = True
#
# # Try to get closer
# elif this_play.inning_num > 8:
# if (this_play.inning_half == 'top' and this_play.home_score >= this_play.away_score) or \
# (this_play.inning_half == 'bot' and this_play.away_score >= this_play.home_score):
# logger.info(f'get_rp - game {this_play.game.id} try for closer')
# params.append(('pos_include', 'CP'))
# use_best = True
# else:
# params.append(('pos_exclude', 'CP'))
#
# # Try to exclude long men
# elif attempts == 1:
# logger.info(f'get_rp - game {this_play.game.id} try to exclude long men')
# params.append(('pos_exclude', 'SP'))
#
# try:
# pitchers = await db_get(
# endpoint='players',
# params=params,
# timeout=10
# )
# except ConnectionError as e:
# logger.error(f'Could not get pitchers for {ai_team["lname"]}: {e}')
# raise ConnectionError(f'Error pulling starting pitchers for the {ai_team["lname"]}. Cal help plz.')
#
# if pitchers['count'] > 0:
# if use_best or this_play.inning_num > 9 or attempts > 2:
# start = 0
# else:
# start = 9 - this_play.inning_num
#
# for count, guy in enumerate(pitchers['players']):
# if count >= start and guy['strat_code'] not in used_codes:
# card_id = await get_or_create_card(guy, ai_team)
#
# return {
# 'game_id': this_play.game.id,
# 'team_id': ai_team['id'],
# 'player_id': guy['player_id'],
# 'card_id': card_id,
# 'position': 'P',
# 'batting_order': 10,
# 'after_play': this_play.play_num - 1
# }
def get_pitcher(this_game: StratGame, this_play: StratPlay):
p_team_id = this_game.home_team_id
if this_play.inning_half == 'top':
p_team_id = this_game.away_team_id
return get_one_lineup(this_game.id, team_id=p_team_id, position='P', active=True)
async def pitching_ai_note(this_play: StratPlay, this_pitcher: dict):
this_ai = get_manager(this_play.game)
gm_name = f'{this_pitcher["team"]["gmname"]}'
# used_pitchers = await get_team_lineups(
# game_id=this_play.game.id,
# team_id=this_pitcher["team"]['id'],
# inc_inactive=True,
# pitchers_only=True,
# as_string=False
# )
# last_inning_ender = get_last_inning_end_play(
# this_play.game.id,
# this_play.inning_half,
# this_play.inning_num - 1
# )
ai_note = ''
pitcher = this_pitcher
# # Pitcher Substitutions
# new_pitcher = None
# if last_inning_ender and last_inning_ender.pitcher != this_play.pitcher:
# logger.debug(f'{this_pitcher["team"]["sname"]} not making a change.')
#
# ai_note += f'- have {this_pitcher["p_name"]} finish the inning\n'
#
# elif this_play.is_new_inning and this_play.game.short_game and this_play.inning_num != 1:
# logger.debug(f'{this_pitcher["team"]["sname"]} going the to pen.')
#
# if len(used_pitchers) < 8:
# make_sub(await get_relief_pitcher(
# this_play, this_pitcher['team'], this_play.game.game_type
# ))
# pitcher = await get_player(this_play.game, get_pitcher(this_play.game, this_play))
# new_pitcher = pitcher
# else:
# ai_note += f'- continue with auto-fatigued {this_pitcher["p_name"]}\n'
#
# else:
# if len(used_pitchers) == 1:
# ai_note += f'- go to the pen if the pitcher fatigues __and has allowed 5+ baserunners__ ' \
# f'(`/log ai-pitcher-sub`)\n'
# elif len(used_pitchers) < 8:
# ai_note += f'- go to the pen if the pitcher fatigues (`/log ai-pitcher-sub`)\n'
# else:
# ai_note += f' - continue with {this_pitcher["p_name"]}\n'
# Holding Baserunners
if this_play.starting_outs == 2 and this_play.on_base_code > 0:
if this_play.on_base_code in [1, 2]:
ai_note += f'- hold the runner\n'
elif this_play.on_base_code in [4, 7]:
ai_note += f'- hold the runners\n'
elif this_play.on_base_code == 5:
ai_note += f'- hold the runner on first\n'
elif this_play.on_base_code == 6:
ai_note += f'- hold the runner on second\n'
elif this_play.on_base_code in [1, 5]:
ai_note += f'- hold the runner on 1st if they have ***** auto-jump\n'
elif this_play.on_base_code == 2:
ai_note += f'- hold the runner on 2nd if safe range is 14+\n'
# Defensive Alignment
if this_play.on_third and this_play.starting_outs < 2:
if this_play.on_first:
ai_note += f'- play the corners in\n'
elif abs(this_play.away_score - this_play.home_score) <= 3:
ai_note += f'- play the whole infield in\n'
else:
ai_note += f'- play the corners in\n'
return {
'note': ai_note,
'pitcher': pitcher,
'gm_name': gm_name,
'sub': None
}
def batting_ai_note(this_play: StratPlay, this_batter: dict):
this_ai = get_manager(this_play.game)
ai_note = ''
gm_name = f'{this_batter["team"]["gmname"]}'
if this_play.on_first and not this_play.on_second:
ai_note += f'- {this_ai.check_jump(2, this_play.starting_outs)}\n'
elif this_play.on_second and not this_play.on_third:
ai_note += f'- {this_ai.check_jump(3, this_play.starting_outs)}\n'
return {
'note': ai_note,
'batter': this_batter,
'gm_name': gm_name
}
async def check_pitching_sub(this_play: Play, ai_team: Team):
used_pitchers = await get_team_lineups(
game_id=this_play.game.id,
team_id=this_play.pitcher.team_id,
inc_inactive=True,
pitchers_only=True,
as_string=False
)
p_stats = get_pitching_stats(this_play.game.id, lineup_id=this_play.pitcher.id)
if len(p_stats) == 0:
logger.info(f'ai_manager - check_pitching_sub: no stats recorded yet, returning None')
return False
ps = p_stats[0]
logger.info(f'ai_manager - check_pitching_sub: pitcher does have stats')
this_ai = get_manager(this_play.game)
this_pc = await data_cache.get_pd_pitchingcard(this_play.pitcher.player_id, variant=this_play.pitcher.variant)
pof_weakness = this_pc.card.starter_rating if len(used_pitchers) == 1 else this_pc.card.relief_rating
innof_work = math.ceil((ps['pl_outs'] + 1) / 3)
is_starter = True if len(used_pitchers) == 1 else False
gtr = this_ai.go_to_reliever(
this_play,
tot_allowed=ps['pl_hit'] + ps['pl_bb'] + ps['pl_hbp'],
is_starter=is_starter
)
if (this_play.game.short_game or gtr or
(innof_work > pof_weakness + 1 and not is_starter) or
(innof_work > pof_weakness + 3 and is_starter)) and len(used_pitchers) < 8:
rp_lineup = make_sub(await get_relief_pitcher(this_play, ai_team, this_play.game.game_type))
try:
rp_pitcard = await data_cache.get_pd_pitchingcard(rp_lineup.player_id, rp_lineup.variant)
if rp_pitcard.card.relief_rating == 1:
patch_play(this_play.id, in_pow=True)
except Exception as e:
logger.info(f'ai_manager - check_pitching_sub - could not pull card for {rp_lineup.player_id}')
return await get_player(this_play.game, rp_lineup)
return None
async def is_pitcher_fatigued(this_play: StratPlay) -> bool:
# Check Lineup object for 'is_fatigued'
# If yes, return True
# Else, check for fatigue below
# If fatigued, patch Lineup object with 'is_fatigued'
if this_play.pitcher.is_fatigued:
return True
used_pitchers = await get_team_lineups(
game_id=this_play.game.id,
team_id=this_play.pitcher.team_id,
inc_inactive=True,
pitchers_only=True,
as_string=False
)
p_stats = get_pitching_stats(this_play.game.id, lineup_id=this_play.pitcher.id)
if len(p_stats) == 0:
logger.info(f'ai_manager - is_pitcher_fatigued: no stats recorded yet, returning False')
return False
ps = p_stats[0]
if this_play.game.short_game:
pof_weakness = 1
else:
try:
this_pc = await data_cache.get_pd_pitchingcard(this_play.pitcher.player_id, variant=this_play.pitcher.variant)
except:
logger.info(
f'ai_manager - is_pitcher_fatigued: could not pull pitching card for {this_play.pitcher.player_id}, '
f'returning False')
return False
pof_weakness = this_pc.card.starter_rating if len(used_pitchers) == 1 else this_pc.card.relief_rating
# Check starter fatigue
if len(used_pitchers) == 1:
if ps['pl_runs'] >= 7:
logger.info(f'ai_manager - is_pitcher_fatigued: starter allowed 7+, returning True')
return True
elif ps['pl_runs'] >= 6:
logger.info(f'ai_manager - is_pitcher_fatigued: starter allowed 6+, checking for fatigue')
f_query = get_pitching_stats(
this_play.game.id,
lineup_id=this_play.pitcher.id,
in_innings=[this_play.inning_num, this_play.inning_num - 1]
)
if f_query[0]['pl_in_runs'] >= 6:
logger.info(f'ai_manager - is_pitcher_fatigued: starter allowed 6 in 2, returning True')
patch_lineup(this_play.pitcher.id, is_fatigued=True)
return True
elif ps['pl_runs'] >= 5:
logger.info(f'ai_manager - is_pitcher_fatigued: starter allowed 5+, checking for fatigue')
f_query = get_pitching_stats(
this_play.game.id,
lineup_id=this_play.pitcher.id,
in_innings=[this_play.inning_num]
)
if f_query[0]['pl_in_runs'] >= 5:
logger.info(f'ai_manager - is_pitcher_fatigued: starter allowed 5 in 1, returning True')
patch_lineup(this_play.pitcher.id, is_fatigued=True)
return True
innof_work = math.ceil((ps['pl_outs'] + 1) / 3)
if innof_work < pof_weakness:
logger.info(f'ai_manager - is_pitcher_fatigued: not point of weakness, returning False')
return False
elif innof_work == pof_weakness:
patch_play(this_play.id, in_pow=True)
pow_stats = get_pitching_stats(this_play.game.id, lineup_id=this_play.pitcher.id, in_pow=True)
if len(pow_stats) == 0:
logger.info(f'ai_manager - is_pitcher_fatigued: in point of weakness, no stats recorded, returning False')
return False
pows = pow_stats[0]
if pows['pl_hit'] + pows['pl_bb'] + pows['pl_hbp'] < 3:
logger.info(f'ai_manager - is_pitcher_fatigued: in point of weakness, not fatigued, returning False')
return False
elif innof_work > pof_weakness:
patch_play(this_play.id, in_pow=True)
patch_lineup(this_play.pitcher.id, is_fatigued=True)
return True
logger.info(f'ai_manager - is_pitcher_fatigued: beyond point of weakness, fatigued, returning True')
patch_lineup(this_play.pitcher.id, is_fatigued=True)
return True
# async def consider_reliever(
# this_play: StratPlay, this_pitcher: StratLineup, ai_team: dict, run_lead: int, tot_allowed: int,
# used_pitchers: list[StratLineup]):
# this_ai = get_manager(this_play.game)
#
# if (this_play.game.short_game or
# this_ai.go_to_reliever(this_play.starting_outs, this_play.on_base_code, run_lead, tot_allowed)) and \
# len(used_pitchers) < 8:
# make_sub(await get_relief_pitcher(this_play, ai_team, this_play.game.game_type))
# return await get_player(this_play.game, get_pitcher(this_play.game, this_play))
#
# return None
Reference in New Issue View Git Blame Copy Permalink