import asyncio import logging import os from typing import Literal import discord from discord import app_commands from discord import SelectOption from discord.app_commands import Choice from discord.ext import commands, tasks import pygsheets import sqlalchemy from sqlmodel import func, or_ from api_calls import db_get from command_logic.logic_gameplay import ( bunts, chaos, complete_game, defender_dropdown_view, doubles, flyballs, frame_checks, get_full_roster_from_sheets, checks_log_interaction, complete_play, get_scorebug_embed, groundballs, hit_by_pitch, homeruns, is_game_over, lineouts, manual_end_game, new_game_checks, new_game_conflicts, popouts, read_lineup, relief_pitcher_dropdown_view, select_ai_reliever, show_defense_cards, singles, starting_pitcher_dropdown_view, steals, strikeouts, sub_batter_dropdown_view, substitute_player, triples, undo_play, update_game_settings, walks, xchecks, activate_last_play, ) from command_logic.play_context import locked_play from dice import ab_roll from exceptions import * import gauntlets from helpers import ( CARDSETS, DEFENSE_LITERAL, DEFENSE_NO_PITCHER_LITERAL, PD_PLAYERS_ROLE_NAME, SELECT_CARDSET_OPTIONS, Dropdown, get_channel, send_to_channel, team_role, user_has_role, random_gif, random_from_list, ) # from in_game import ai_manager from in_game.ai_manager import get_starting_pitcher, get_starting_lineup from in_game.game_helpers import PUBLIC_FIELDS_CATEGORY_NAME, legal_check from in_game.gameplay_models import ( GameCardsetLink, Lineup, Play, Session, engine, player_description, select, Game, ) from in_game.gameplay_queries import ( get_all_positions, get_cardset_or_none, get_one_lineup, get_plays_by_pitcher, get_position, get_channel_game_or_none, get_active_games_by_team, get_game_lineups, get_team_or_none, ) from utilities.buttons import Confirm, ScorebugButtons, ask_confirm, ask_with_buttons from utilities.dropdown import DropdownView logger = logging.getLogger("discord_app") CLASSIC_EMBED = True CARDSETS class Gameplay(commands.Cog): def __init__(self, bot): self.bot = bot self.sheets = None self.game_states = {} # game_id: {play: , ack: } self.get_sheets.start() self.live_scorecard.start() @tasks.loop(count=1) async def get_sheets(self): logger.info(f"Getting sheets") self.sheets = pygsheets.authorize( service_file="storage/paper-dynasty-service-creds.json", retries=1 ) @tasks.loop(minutes=1) async def live_scorecard(self): try: logger.info(f"Checking live scorecard loop") guild_id = os.environ.get("GUILD_ID") if not guild_id: logger.error("GUILD_ID env var is not set") return guild = self.bot.get_guild(int(guild_id)) score_channel = discord.utils.get( guild.text_channels, name="live-pd-scores" ) if score_channel is None: logger.error(f"Could not find live-pd-channel") return if len(self.game_states) == 0: logger.info(f"No active game_states") return player_role = discord.utils.get(guild.roles, name=PD_PLAYERS_ROLE_NAME) all_embeds = [] logger.info(f"player role: {player_role}") with Session(engine) as session: for key in self.game_states: if not self.game_states[key]["ack"]: this_game = session.get(Game, key) if this_game is None: log_exception( GameNotFoundException, f"Could not pull game #{key} for live scorecard", ) if not this_game.active: logger.info( f"Game {this_game.id} is complete, removing from game_states" ) del self.game_states[key] else: try: logger.info( f"Appending scorebug for Game {this_game.id}" ) this_channel = discord.utils.get( guild.text_channels, id=this_game.channel_id ) logger.info(f"this_channel: {this_channel}") this_embed = await get_scorebug_embed( session, this_game, full_length=False, live_scorecard=True, ) this_embed.set_image(url=None) this_embed.insert_field_at( index=0, name="Ballpark", value=f"{this_channel.mention}", ) all_embeds.append(this_embed) self.game_states[key]["ack"] = True except Exception as e: logger.error(f"Unable to add to game_states: {e}") logger.error(f"Game: {this_game.id}") if len(all_embeds) == 0: logger.info(f"No active game embeds, returning") await score_channel.set_permissions(player_role, read_messages=False) return async for message in score_channel.history(limit=25): await message.delete() await score_channel.set_permissions(player_role, read_messages=True) await score_channel.send(content=None, embeds=all_embeds) except Exception as e: logger.error(f"Failed running live scorecard: {e}") # try: # await send_to_channel(self.bot, 'commissioners-office', f'PD Live Scorecard just failed: {e}') # except Exception as e: # logger.error(f'Couldn\'t even send the error to the private channel :/') @live_scorecard.before_loop async def before_live_scoreboard(self): await self.bot.wait_until_ready() @get_sheets.before_loop async def before_get_sheets(self): logger.info(f"Waiting to get sheets") await self.bot.wait_until_ready() async def cog_command_error(self, ctx, error): logger.error(msg=error, stack_info=True) await ctx.send( f"{error}\n\nRun !help to see the command requirements" ) async def slash_error(self, ctx, error): logger.error(msg=error, stack_info=True) await ctx.send(f"{error[:1600]}") async def post_play( self, session: Session, interaction: discord.Interaction, this_play: Play, buffer_message: str = None, full_length: bool = False, ): logger.info(f"post_play - Posting new play: {this_play}") if this_play is None: logger.info( f"this_play is None, searching for game in channel {interaction.channel.id}" ) this_game = get_channel_game_or_none(session, interaction.channel.id) try: this_play = activate_last_play(session, this_game) except Exception as e: this_play = this_game.initialize_play(session) finally: if this_play is None: log_exception( PlayNotFoundException, f"Attempting to display gamestate, but cannot find current play", ) if is_game_over(this_play): logger.info(f"Game {this_play.game.id} seems to be over") await interaction.edit_original_response( content=f"Looks like this one is over!" ) submit_game = await ask_confirm( interaction=interaction, question=f"Final score: {this_play.game.away_team.abbrev} {this_play.away_score} - {this_play.home_score} {this_play.game.home_team.abbrev}\n{this_play.scorebug_ascii}\nShould I go ahead and submit this game or roll it back a play?", custom_confirm_label="Submit", custom_cancel_label="Roll Back", ) if submit_game: logger.info( f"post_play - is_game_over - {interaction.user.display_name} rejected game completion" ) await complete_game(session, interaction, this_play, self.bot) return else: logger.warning( f"post_play - is_game_over - {interaction.user.display_name} rejected game completion in Game {this_play.game.id}" ) cal_channel = get_channel(interaction, "commissioners-office") await cal_channel.send( content=f"{interaction.user.display_name} just rejected game completion down in {interaction.channel.mention}" ) this_play = undo_play(session, this_play) await self.post_play(session, interaction, this_play) await interaction.channel.send( content=f"I let Cal know his bot is stupid" ) if this_play.pitcher.is_fatigued and not this_play.ai_is_batting: if this_play.managerai.replace_pitcher(session, this_play.game): logger.info(f"Running a pitcher sub") await interaction.edit_original_response( content="The AI is making a pitching change..." ) new_pitcher_card = await select_ai_reliever( session, this_play.pitcher.team, this_play ) new_pitcher_lineup = substitute_player( session, this_play, this_play.pitcher, new_pitcher_card, "P" ) logger.info(f"Sub complete") scorebug_buttons, this_ab_roll = None, None scorebug_embed = await get_scorebug_embed( session, this_play.game, full_length=full_length, classic=CLASSIC_EMBED ) if this_play.game.roll_buttons and interaction.user.id in [ this_play.game.away_team.gmid, this_play.game.home_team.gmid, ]: logger.info(f"Including scorebug buttons") scorebug_buttons = ScorebugButtons(this_play, scorebug_embed, timeout=8) if ( this_play.on_base_code == 0 and this_play.game.auto_roll and not this_play.batter.team.is_ai and not this_play.is_new_inning ): logger.info(f"Rolling ab") this_ab_roll = ab_roll( this_play.batter.team, this_play.game, allow_chaos=False ) scorebug_buttons = None if this_ab_roll is not None and this_ab_roll.d_six_one > 3: logger.info(f"Setting embed image to pitcher") scorebug_embed.set_image(url=this_play.pitcher.player.pitcher_card_url) if buffer_message is not None: logger.info(f"Posting buffered message") await interaction.edit_original_response(content=buffer_message) sb_message = await interaction.channel.send( content=None, embed=scorebug_embed, view=scorebug_buttons ) else: logger.info(f"Posting unbuffered message") sb_message = await interaction.edit_original_response( content=None, embed=scorebug_embed, view=scorebug_buttons ) if this_ab_roll is not None: logger.info(f"Posting ab roll") await interaction.channel.send(content=None, embeds=this_ab_roll.embeds) if scorebug_buttons is not None: logger.info(f"Posting scorebug buttons roll") await scorebug_buttons.wait() if not scorebug_buttons.value: await sb_message.edit(view=None) async def complete_and_post_play( self, session: Session, interaction: discord.Interaction, this_play: Play, buffer_message: str = None, ): next_play = complete_play(session, this_play) logger.info(f"Completed play {this_play.id}") logger.info(f"Updating self.game_states") self.game_states[this_play.game.id] = {"play": this_play, "ack": False} logger.info(f"New state: {self.game_states}") await self.post_play(session, interaction, next_play, buffer_message) def kickstart_live_scorecard(self): try: self.live_scorecard.start() logger.info(f"Kick started the live scorecard") except RuntimeError as e: logger.info(f"Live scorecard is already running") @commands.command(name="test-write", help="Test concurrent db writes", hidden=True) @commands.is_owner() async def test_write_command(self, ctx): await ctx.send(f"I am going to open a connection, delay, then try to write") with Session(engine) as session: ncb_team = await get_team_or_none(session, team_id=31) await ctx.send( f"The {ncb_team.lname} has_guide value is: {ncb_team.has_guide}. Now to delay for 10 seconds..." ) ncb_team.has_guide = not ncb_team.has_guide session.add(ncb_team) await asyncio.sleep(10) await ctx.send(f"Now to attempt committing the NCB change...") session.commit() await ctx.send(f"Am I alive? Did it work?") group_new_game = app_commands.Group( name="new-game", description="Start a new baseball game" ) @group_new_game.command( name="mlb-campaign", description="Start a new MLB campaign game against an AI" ) @app_commands.choices( league=[ Choice(value="minor-league", name="Minor League"), Choice(value="flashback", name="Flashback"), Choice(value="major-league", name="Major League"), Choice(value="hall-of-fame", name="Hall of Fame"), ], roster=[ Choice(value="1", name="Primary"), Choice(value="2", name="Secondary"), Choice(value="3", name="Ranked"), ], ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def new_game_mlb_campaign_command( self, interaction: discord.Interaction, league: Choice[str], away_team_abbrev: str, home_team_abbrev: str, roster: Choice[str], ): await interaction.response.defer() self.kickstart_live_scorecard() with Session(engine) as session: teams = await new_game_checks( session, interaction, away_team_abbrev, home_team_abbrev ) if teams is None: logger.error(f"Received None from new_game_checks, cancelling new game") return away_team = teams["away_team"] home_team = teams["home_team"] ai_team = away_team if away_team.is_ai else home_team human_team = away_team if home_team.is_ai else home_team conflict_games = get_active_games_by_team(session, team=human_team) if len(conflict_games) > 0: await interaction.edit_original_response( content=f"Ope. The {human_team.sname} are already playing over in {interaction.guild.get_channel(conflict_games[0].channel_id).mention}" ) return conflict_games = get_active_games_by_team(session, team=human_team) if len(conflict_games) > 0: await interaction.edit_original_response( content=f"Ope. The {human_team.sname} are already playing over in {interaction.guild.get_channel(conflict_games[0].channel_id).mention}" ) return current = await db_get("current") week_num = current["week"] logger.info( f'gameplay - new_game_mlb_campaign - Season: {current["season"]} / Week: {week_num} / Away Team: {away_team.description} / Home Team: {home_team.description}' ) def role_error(required_role: str, league_name: str, lower_league: str): return f"Ope. Looks like you haven't received the **{required_role}** role, yet!\n\nTo play **{league_name}** games, you need to defeat all 30 MLB teams in the {lower_league} campaign. You can see your progress with `/record`.\n\nIf you have completed the {lower_league} campaign, go ping Cal to get your new role!" if league.value == "flashback": if not user_has_role(interaction.user, "PD - Major League"): await interaction.edit_original_response( content=role_error( "PD - Major League", league_name="Flashback", lower_league="Minor League", ) ) return elif league.value == "major-league": if not user_has_role(interaction.user, "PD - Major League"): await interaction.edit_original_response( content=role_error( "PD - Major League", league_name="Major League", lower_league="Minor League", ) ) return elif league.value == "hall-of-fame": if not user_has_role(interaction.user, "PD - Hall of Fame"): await interaction.edit_original_response( content=role_error( "PD - Hall of Fame", league_name="Hall of Fame", lower_league="Major League", ) ) return this_game = Game( away_team_id=away_team.id, home_team_id=home_team.id, away_roster_id=69 if away_team.is_ai else int(roster.value), home_roster_id=69 if home_team.is_ai else int(roster.value), channel_id=interaction.channel_id, season=current["season"], week=week_num, first_message=( None if interaction.message is None else interaction.message.channel.id ), ai_team="away" if away_team.is_ai else "home", game_type=league.value, ) game_info_log = f"{league.name} game between {away_team.description} and {home_team.description} / first message: {this_game.first_message}" logger.info(game_info_log) # Get AI SP await interaction.edit_original_response( content=f"{ai_team.gmname} is looking for a Starting Pitcher..." ) ai_sp_lineup = await get_starting_pitcher( session, ai_team, this_game, True if home_team.is_ai else False, league.value, ) logger.info( f"Chosen SP in Game {this_game.id}: {ai_sp_lineup.player.name_with_desc}" ) await interaction.edit_original_response( content=f"The {ai_team.sname} are starting **{ai_sp_lineup.player.name_with_desc}**:\n\n{ai_sp_lineup.player.pitcher_card_url}" ) # Get AI Lineup final_message = await interaction.channel.send( content=f"{ai_team.gmname} is filling out the {ai_team.sname} lineup card..." ) logger.info(f"Pulling lineup...") batter_lineups = await get_starting_lineup( session, team=ai_team, game=this_game, league_name=this_game.league_name, sp_name=ai_sp_lineup.player.name, ) # Check for last game settings logger.info(f"Checking human team's automation preferences...") g_query = session.exec( select(Game) .where(or_(Game.home_team == human_team, Game.away_team == human_team)) .order_by(Game.id.desc()) .limit(1) ).all() if len(g_query) > 0: last_game = g_query[0] this_game.auto_roll = last_game.auto_roll this_game.roll_buttons = last_game.roll_buttons logger.info( f"Setting auto_roll to {last_game.auto_roll} and roll_buttons to {last_game.roll_buttons}" ) # Commit game and lineups session.add(this_game) session.commit() await final_message.edit( content=f"The {ai_team.sname} lineup is in, pulling in scouting data..." ) for batter in batter_lineups: pos_count = await get_all_positions( session=session, this_card=batter.card ) if pos_count != 0: logger.info( f"logged position ratings for {batter.player.name_with_desc}" ) else: logger.warning( f"received no positions for {batter.player.name_with_desc}" ) if batter.position not in ["P", "DH"]: log_exception( PositionNotFoundException, f"{batter.player.name_with_desc} is listed at {batter.position} but no ratings were found.", ) logger.info(f"Pulling team roles") away_role = await team_role(interaction, this_game.away_team) home_role = await team_role(interaction, this_game.home_team) logger.info(f"Building scorebug embed") embed = await get_scorebug_embed(session, this_game) embed.clear_fields() embed.add_field( name=f"{ai_team.abbrev} Lineup", value=this_game.team_lineup(session, ai_team), ) logger.info(f"Pulling and caching full {human_team.abbrev} roster") done = await get_full_roster_from_sheets( session, interaction, self.sheets, this_game, human_team, int(roster.value), ) roster_choice = await ask_with_buttons( interaction, ["vs Left", "vs Right"], "Which lineup will you be using?", delete_question=False, confirmation_message="Got it!", ) sp_view = starting_pitcher_dropdown_view( session, this_game, human_team, this_game.league_name, [interaction.user], ) await interaction.channel.send( content=f"### {human_team.lname} Starting Pitcher", view=sp_view ) try: await asyncio.sleep(5) this_play = await read_lineup( session, interaction, this_game=this_game, lineup_team=human_team, sheets_auth=self.sheets, lineup_num=1 if roster_choice == "vs Right" else 2, league_name=this_game.game_type, ) except PositionNotFoundException as e: logger.error(f"Position validation failed during lineup load: {e}") this_game.active = False session.add(this_game) session.commit() await interaction.channel.send(content=str(e)) return except LineupsMissingException as e: logger.error(f"Attempting to start game, pausing for 5 seconds: {e}") await asyncio.sleep(5) try: this_play = this_game.current_play_or_none(session) await self.post_play( session, interaction, this_play, buffer_message="Game on!" ) except LineupsMissingException as e: await interaction.channel.send( content=f"Run `/gamestate` once you have selected a Starting Pitcher to get going!" ) @group_new_game.command( name="gauntlet", description="Start a new Gauntlet game against an AI" ) @app_commands.choices( roster=[ Choice(value="1", name="Primary"), Choice(value="2", name="Secondary"), Choice(value="3", name="Ranked"), ] ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def new_game_gauntlet_command( self, interaction: discord.Interaction, roster: Choice[str] ): await interaction.response.defer() self.kickstart_live_scorecard() with Session(engine) as session: try: await new_game_conflicts(session, interaction) except GameException as e: return main_team = await get_team_or_none( session, gm_id=interaction.user.id, main_team=True ) human_team = await get_team_or_none( session, gm_id=interaction.user.id, gauntlet_team=True ) if not main_team: await interaction.edit_original_response( content=f"I don't see a team for you, yet. You can sign up with the `/newteam` command!" ) return if not human_team: await interaction.edit_original_response( content=f"I don't see an active run for you. You can get started with the `/gauntlets start` command!" ) return e_query = await db_get("events", params=[("active", True)]) if e_query["count"] == 0: await interaction.edit_original_response( content=f"Hm. It looks like there aren't any active gauntlets. What do we even pay Cal for?" ) return elif e_query["count"] == 1: this_event = e_query["events"][0] r_query = await db_get( "gauntletruns", params=[ ("team_id", human_team.id), ("gauntlet_id", this_event["id"]), ("is_active", True), ], ) if r_query["count"] == 0: await interaction.edit_original_response( content=f"I don't see an active run for you. If you would like to start a new one, run " f'`/gauntlets start {this_event["name"]}` and we can get you started in no time!' ) return this_run = r_query["runs"][0] else: r_query = await db_get( "gauntletruns", params=[("team_id", human_team.id), ("is_active", True)], ) if r_query["count"] == 0: await interaction.edit_original_response( content=f"I don't see an active run for you. If you would like to start a new one, run " f'`/gauntlets start {e_query["events"][0]["name"]}` and we can get you started in no time!' ) return else: this_run = r_query["runs"][0] this_event = r_query["runs"][0]["gauntlet"] # If not new or after draft, create new AI game is_home = gauntlets.is_home_team(human_team, this_event, this_run) ai_team = await gauntlets.get_opponent( session, human_team, this_event, this_run ) if ai_team is None: await interaction.edit_original_response( content=f"Yike. I'm not sure who your next opponent is. Plz ping the shit out of Cal!" ) return else: logger.info(f"opponent: {ai_team}") ai_role = await team_role(interaction, ai_team) human_role = await team_role(interaction, main_team) away_role = ai_role if is_home else human_role home_role = human_role if is_home else ai_role conflict_games = get_active_games_by_team(session, team=human_team) if len(conflict_games) > 0: await interaction.edit_original_response( content=f"Ope. The {human_team.sname} are already playing over in {interaction.guild.get_channel(conflict_games[0].channel_id).mention}" ) return current = await db_get("current") game_code = gauntlets.get_game_code(human_team, this_event, this_run) this_game = Game( away_team_id=ai_team.id if is_home else human_team.id, home_team_id=human_team.id if is_home else ai_team.id, channel_id=interaction.channel_id, season=current["season"], week=current["week"], first_message=( None if interaction.message is None else interaction.message.id ), ai_team="away" if is_home else "home", away_roster_id=69 if is_home else int(roster.value), home_roster_id=int(roster.value) if is_home else 69, game_type=game_code, ) logger.info( f"Game between {human_team.abbrev} and {ai_team.abbrev} is initializing!" ) # Get AI SP await interaction.edit_original_response( content=f"{ai_team.gmname} is looking for a Starting Pitcher..." ) ai_sp_lineup = await gauntlets.get_starting_pitcher( session, ai_team, this_game, this_event, this_run ) logger.info( f"Chosen SP in Game {this_game.id}: {ai_sp_lineup.player.name_with_desc}" ) await interaction.edit_original_response( content=f"The {ai_team.sname} are starting **{ai_sp_lineup.player.name_with_desc}**:\n\n{ai_sp_lineup.player.pitcher_card_url}" ) # Get AI Lineup final_message = await interaction.channel.send( content=f"{ai_team.gmname} is filling out the {ai_team.sname} lineup card..." ) logger.info(f"Pulling lineup in Game {this_game.id}") batter_lineups = await get_starting_lineup( session, team=ai_team, game=this_game, league_name=f'gauntlet-{this_event["id"]}', sp_name=ai_sp_lineup.player.name, ) # Check for last game settings logger.info(f"Checking human team's automation preferences...") g_query = session.exec( select(Game) .where(or_(Game.home_team == human_team, Game.away_team == human_team)) .order_by(Game.id.desc()) .limit(1) ).all() if len(g_query) > 0: last_game = g_query[0] this_game.auto_roll = last_game.auto_roll this_game.roll_buttons = last_game.roll_buttons logger.info( f"Setting auto_roll to {last_game.auto_roll} and roll_buttons to {last_game.roll_buttons}" ) # Commit game and lineups session.add(this_game) session.commit() await final_message.edit( content=f"The {ai_team.sname} lineup is in, pulling in scouting data..." ) for batter in batter_lineups: await get_all_positions(session=session, this_card=batter.card) embed = await get_scorebug_embed(session, this_game) embed.clear_fields() embed.add_field( name=f"{ai_team.abbrev} Lineup", value=this_game.team_lineup(session, ai_team), ) # Get pitchers from rosterlinks done = await get_full_roster_from_sheets( session, interaction, self.sheets, this_game, human_team, 1 ) # if done: # sp_view = starting_pitcher_dropdown_view(session, this_game, human_team, game_type=this_game.league_name, responders=[interaction.user]) # sp_message = await interaction.channel.send(content=f'### {human_team.lname} Starting Pitcher', view=sp_view) # await final_message.edit( # content=f'{away_role.mention} @ {home_role.mention} is set!', # embed=embed # ) roster_choice = await ask_with_buttons( interaction, ["vs Left", "vs Right"], "Which lineup will you be using?", delete_question=False, confirmation_message="Got it!", ) # Read the 9 field players from sheets (this will fail to initialize play without SP) try: await read_lineup( session, interaction, this_game=this_game, lineup_team=human_team, sheets_auth=self.sheets, lineup_num=1 if roster_choice == "vs Right" else 2, league_name=this_game.game_type, ) except PositionNotFoundException as e: logger.error(f"Position validation failed during lineup load: {e}") this_game.active = False session.add(this_game) session.commit() await interaction.channel.send(content=str(e)) return except LineupsMissingException as e: # Expected - can't initialize play without SP yet logger.info( f"Field player lineup read from sheets, waiting for SP selection: {e}" ) sp_view = starting_pitcher_dropdown_view( session, this_game, human_team, this_game.league_name, [interaction.user], ) await interaction.channel.send( content=f"### {human_team.lname} Starting Pitcher", view=sp_view ) # Don't try to initialize play immediately - wait for user to select SP # The play will be initialized when they run /gamestate await interaction.channel.send( content=f"Once you've selected your Starting Pitcher, run `/gamestate` to get the game started!" ) @group_new_game.command( name="exhibition", description="Start a new custom game against an AI" ) @app_commands.choices( roster=[ Choice(value="1", name="Primary"), Choice(value="2", name="Secondary"), Choice(value="3", name="Ranked"), ] ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def new_game_exhibition_command( self, interaction: discord.Interaction, away_team_abbrev: str, home_team_abbrev: str, roster: Choice[str], cardsets: Literal[ "Minor League", "Major League", "Hall of Fame", "Flashback", "Custom" ] = "Custom", ): await interaction.response.defer() self.kickstart_live_scorecard() with Session(engine) as session: teams = await new_game_checks( session, interaction, away_team_abbrev, home_team_abbrev ) if teams is None: logger.error(f"Received None from new_game_checks, cancelling new game") return away_team = teams["away_team"] home_team = teams["home_team"] if not away_team.is_ai ^ home_team.is_ai: await interaction.edit_original_response( content=f"I don't see an AI team in this Exhibition game. Run `/new-game exhibition` again with an AI for a custom game or `/new-game ` for a PvP game." ) return current = await db_get("current") week_num = current["week"] logger.info( f'gameplay - new_game_mlb_campaign - Season: {current["season"]} / Week: {week_num} / Away Team: {away_team.description} / Home Team: {home_team.description}' ) ai_team = away_team if away_team.is_ai else home_team human_team = away_team if home_team.is_ai else home_team this_game = Game( away_team_id=away_team.id, home_team_id=home_team.id, away_roster_id=69 if away_team.is_ai else int(roster.value), home_roster_id=69 if home_team.is_ai else int(roster.value), channel_id=interaction.channel_id, season=current["season"], week=week_num, first_message=( None if interaction.message is None else interaction.message.id ), ai_team="away" if away_team.is_ai else "home", game_type="exhibition", ) async def new_game_setup(): # Get AI SP await interaction.edit_original_response( content=f"{ai_team.gmname} is looking for a Starting Pitcher..." ) ai_sp_lineup = await get_starting_pitcher( session, ai_team, this_game, True if home_team.is_ai else False, "exhibition", ) logger.info(f"Chosen SP: {ai_sp_lineup.player.name_with_desc}") session.add(ai_sp_lineup) await interaction.edit_original_response( content=f"The {ai_team.sname} are starting **{ai_sp_lineup.player.name_with_desc}**:\n\n{ai_sp_lineup.player.pitcher_card_url}" ) # Get AI Lineup final_message = await interaction.channel.send( content=f"{ai_team.gmname} is filling out the {ai_team.sname} lineup card..." ) logger.info(f"Pulling lineup...") batter_lineups = await get_starting_lineup( session, team=ai_team, game=this_game, league_name=this_game.league_name, sp_name=ai_sp_lineup.player.name, ) for x in batter_lineups: session.add(x) # Check for last game settings logger.info(f"Checking human team's automation preferences...") g_query = session.exec( select(Game) .where( or_(Game.home_team == human_team, Game.away_team == human_team) ) .order_by(Game.id.desc()) .limit(1) ).all() if len(g_query) > 0: last_game = g_query[0] this_game.auto_roll = last_game.auto_roll this_game.roll_buttons = last_game.roll_buttons logger.info( f"Setting auto_roll to {last_game.auto_roll} and roll_buttons to {last_game.roll_buttons}" ) # Commit game and lineups session.add(this_game) session.commit() await final_message.edit( content=f"The {ai_team.sname} lineup is in, pulling in scouting data..." ) for batter in batter_lineups: await get_all_positions(session=session, this_card=batter.card) logger.info(f"Pulling team roles") away_role = await team_role(interaction, this_game.away_team) home_role = await team_role(interaction, this_game.home_team) logger.info(f"Building scorebug embed") embed = await get_scorebug_embed(session, this_game) embed.clear_fields() embed.add_field( name=f"{ai_team.abbrev} Lineup", value=this_game.team_lineup(session, ai_team), ) logger.info(f"Pulling and caching full {human_team.abbrev} roster") done = await get_full_roster_from_sheets( session, interaction, self.sheets, this_game, human_team, int(roster.value), ) if done: sp_view = starting_pitcher_dropdown_view( session, this_game, human_team, this_game.league_name, [interaction.user], ) await interaction.channel.send( content=f"### {human_team.lname} Starting Pitcher", view=sp_view ) await final_message.edit( content=f"{away_role.mention} @ {home_role.mention} is set!\n\n" f"Go ahead and set your lineup with the `/set lineup` command!", embed=embed, ) if cardsets != "Custom": c_list = CARDSETS[cardsets] for row in c_list["primary"]: this_cardset = await get_cardset_or_none(session, cardset_id=row) if this_cardset is not None: this_link = GameCardsetLink( game=this_game, cardset=this_cardset, priority=1 ) session.add(this_link) for row in c_list["secondary"]: this_cardset = await get_cardset_or_none(session, cardset_id=row) this_link = GameCardsetLink( game=this_game, cardset=this_cardset, priority=2 ) await new_game_setup() else: async def my_callback(interaction: discord.Interaction, values): logger.info(f"Setting custom cardsets inside callback") await interaction.response.defer(thinking=True) logger.info(f"values: {values}") for cardset_id in values: logger.info(f"Getting cardset: {cardset_id}") this_cardset = await get_cardset_or_none(session, cardset_id) logger.info(f"this_cardset: {this_cardset}") this_link = GameCardsetLink( game=this_game, cardset=this_cardset, priority=1 ) session.add(this_link) logger.info(f"Done processing links") session.commit() await interaction.edit_original_response(content="Got it...") await new_game_setup() my_dropdown = Dropdown( option_list=SELECT_CARDSET_OPTIONS, placeholder="Select up to 8 cardsets to include", callback=my_callback, max_values=len(SELECT_CARDSET_OPTIONS), ) view = DropdownView([my_dropdown]) await interaction.edit_original_response(content=None, view=view) # TODO: add new-game ranked @group_new_game.command( name="unlimited", description="Start a new Unlimited game against another human" ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def new_game_unlimited_command( self, interaction: discord.Interaction, away_team_abbrev: str, home_team_abbrev: str, ): await interaction.response.defer() self.kickstart_live_scorecard() with Session(engine) as session: teams = await new_game_checks( session, interaction, away_team_abbrev, home_team_abbrev ) if teams is None: logger.error(f"Received None from new_game_checks, cancelling new game") return away_team = teams["away_team"] home_team = teams["home_team"] if away_team.is_ai or home_team.is_ai: await interaction.edit_original_response( content=f"Unlimited games are for two human-run teams. To play against the AI, you can play `mlb-campaign`, `gauntlet`, or `exhibition` game modes." ) return current = await db_get("current") week_num = current["week"] logger.info( f'gameplay - new_game_unlimited - Season: {current["season"]} / Week: {week_num} / Away Team: {away_team.description} / Home Team: {home_team.description}' ) this_game = Game( away_team_id=away_team.id, home_team_id=home_team.id, away_roster_id=None, home_roster_id=None, channel_id=interaction.channel_id, season=current["season"], week=week_num, first_message=( None if interaction.message is None else interaction.message.id ), game_type="exhibition", ) await interaction.edit_original_response( content=f"Let's get set up for **{away_team.lname}** @ **{home_team.lname}**!" ) away_role = await team_role(interaction, away_team) home_role = await team_role(interaction, home_team) away_roster_id = await ask_with_buttons( interaction=interaction, button_options=["Primary", "Secondary", "Ranked"], question=f"{away_role.mention}\nWhich roster should I pull for you?", delete_question=False, confirmation_message=f"Got it! As soon as the {home_team.sname} select their roster, I will pull them both in at once.", ) home_roster_id = await ask_with_buttons( interaction=interaction, button_options=["Primary", "Secondary", "Ranked"], question=f"{home_role.mention}\nWhich roster should I pull for you?", delete_question=False, confirmation_message=f"Got it! Off to Sheets I go for the {away_team.abbrev} roster...", ) if away_roster_id and home_roster_id: if away_roster_id == "Primary": away_roster_id = 1 elif away_roster_id == "Secondary": away_roster_id = 2 else: away_roster_id = 3 if home_roster_id == "Primary": home_roster_id = 1 elif home_roster_id == "Secondary": home_roster_id = 2 else: home_roster_id = 3 logger.info( f"Setting roster IDs - away: {away_roster_id} / home: {home_roster_id}" ) this_game.away_roster_id = away_roster_id this_game.home_roster_id = home_roster_id session.add(this_game) session.commit() logger.info(f"Pulling away team's roster") away_roster = await get_full_roster_from_sheets( session, interaction, self.sheets, this_game, away_team, away_roster_id ) # if away_roster: logger.info(f"Pulling home team's roster") await interaction.channel.send( content=f"And now for the {home_team.abbrev} sheet..." ) home_roster = await get_full_roster_from_sheets( session, interaction, self.sheets, this_game, home_team, home_roster_id ) # if home_roster: await interaction.channel.send( content=f"{away_role.mention} @ {home_role.mention}\n\nThe game is set, both of you may run `/set ` to start!" ) @commands.command( name="force-endgame", help="Mod: Force a game to end without stats" ) async def force_end_game_command(self, ctx: commands.Context): with Session(engine) as session: this_game = get_channel_game_or_none(session, ctx.channel.id) if this_game is None: await ctx.send( f"I do not see a game here - are you in the right place?" ) return try: await ctx.send( content=None, embed=await get_scorebug_embed( session, this_game, full_length=True ), ) except Exception as e: logger.error( f"Unable to display scorebug while forcing game to end: {e}" ) await ctx.send( content="This game is so boned that I can't display the scorebug." ) nuke_game = await ask_confirm( ctx, question=f"Is this the game I should nuke?", label_type="yes", timeout=15, ) # if view.value: if nuke_game: this_game.active = False session.add(this_game) session.commit() await ctx.channel.send( content=random_gif( random_from_list( ["i killed it", "deed is done", "gone forever"] ) ) ) else: await ctx.send(f"It stays. For now.") group_set_rosters = app_commands.Group(name="set", description="Set SP and lineup") @group_set_rosters.command( name="lineup", description="Import a saved lineup for this channel's PD game." ) @app_commands.describe(lineup="Which handedness lineup are you using?") @app_commands.choices( lineup=[Choice(value="1", name="v Right"), Choice(value="2", name="v Left")] ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def read_lineup_command( self, interaction: discord.Interaction, lineup: Choice[str] ): await interaction.response.defer() with Session(engine) as session: this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content=f"Hm. I don't see a game going on in this channel. Am I drunk?" ) return if this_game.away_team.gmid == interaction.user.id: this_team = this_game.away_team elif this_game.home_team.gmid == interaction.user.id: this_team = this_game.home_team else: logger.info( f"{interaction.user.name} tried to run a command in Game {this_game.id} when they aren't a GM in the game." ) await interaction.edit_original_response( content="Bruh. Only GMs of the active teams can pull lineups." ) return all_lineups = get_game_lineups(session, this_game, this_team) if len(all_lineups) > 1: play_count = session.exec( select(func.count(Play.id)).where( Play.game == this_game, Play.complete == True ) ).one() if play_count > 0: await interaction.edit_original_response( content=f'Since {play_count} play{"s" if play_count != 1 else ""} ha{"ve" if play_count != 1 else "s"} been logged, you will have to run `/substitution batter` to replace any of your batters.' ) return logger.info( f"lineup: {lineup} / value: {lineup.value} / name: {lineup.name}" ) try: this_play = await read_lineup( session, interaction, this_game=this_game, lineup_team=this_team, sheets_auth=self.sheets, lineup_num=int(lineup.value), league_name=this_game.game_type, ) except PositionNotFoundException as e: logger.error(f"Position validation failed during lineup load: {e}") await interaction.edit_original_response(content=str(e)) return except LineupsMissingException as e: await interaction.edit_original_response( content="Run `/set starting-pitcher` to select your SP" ) return if this_play is not None: await self.post_play(session, interaction, this_play) @group_set_rosters.command(name="starting-pitcher") @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def set_starting_pitcher(self, interaction: discord.Interaction): await interaction.response.defer() with Session(engine) as session: this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content=f"Hm. I don't see a game going on in this channel. Am I drunk?" ) return if this_game.away_team.gmid == interaction.user.id: this_team = this_game.away_team elif this_game.home_team.gmid == interaction.user.id: this_team = this_game.home_team else: logger.info( f"{interaction.user.name} tried to run a command in Game {this_game.id} when they aren't a GM in the game." ) await interaction.edit_original_response( content="Bruh. Only GMs of the active teams can pull lineups." ) return try: check_sp = get_one_lineup(session, this_game, this_team, position="P") play_count = session.exec( select(func.count(Play.id)).where( Play.game == this_game, Play.complete == True, Play.pitcher == check_sp, ) ).one() if play_count > 0: await interaction.edit_original_response( content=f'Since {play_count} play{"s" if play_count != 1 else ""} ha{"ve" if play_count != 1 else "s"} been logged, you will have to run `/substitution pitcher` to replace {check_sp.player.name}.' ) return except sqlalchemy.exc.NoResultFound as e: # if 'NoResultFound' not in str(e): # logger.error(f'Error checking for existing sp: {e}') # log_exception(e, 'Unable to check your lineup for an existing SP') # else: logger.info(f"No pitcher in game, good to go") check_sp = None if check_sp is not None: logger.info( f"Already an SP in Game {this_game.id}, asking if we should swap" ) swap_sp = await ask_confirm( interaction, question=f"{check_sp.player.name} is already scheduled to start this game - would you like to switch?", label_type="yes", ) if not swap_sp: logger.info(f"No swap being made") await interaction.edit_original_response( content=f"We will leave {check_sp.player.name} on the lineup card." ) return session.delete(check_sp) session.commit() sp_view = starting_pitcher_dropdown_view( session, this_game, this_team, game_type=this_game.league_name, responders=[interaction.user], ) await interaction.edit_original_response( content=f"### {this_team.lname} Starting Pitcher", view=sp_view ) @app_commands.command(name="gamestate", description="Post the current game state") async def gamestate_command( self, interaction: discord.Interaction, include_lineups: bool = False ): await interaction.response.defer(ephemeral=True, thinking=True) with Session(engine) as session: this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content=f"Hm. I don't see a game going on in this channel. Am I drunk?" ) return this_play = this_game.current_play_or_none(session) try: await self.post_play( session, interaction, this_play, full_length=include_lineups, buffer_message=( None if this_game.human_team.gmid != interaction.user.id else "Posting current play" ), ) except LineupsMissingException as e: logger.info(f"Could not post full scorebug embed, posting lineups") ai_team = ( this_game.away_team if this_game.ai_team == "away" else this_game.home_team ) embed = await get_scorebug_embed(session, this_game) embed.clear_fields() embed.add_field( name=f"{ai_team.abbrev} Lineup", value=this_game.team_lineup(session, ai_team), ) @app_commands.command(name="settings-ingame", description="Change in-game settings") @app_commands.describe( roll_buttons='Display the "Roll AB" and "Check Jump" buttons along with the scorebug', auto_roll="When there are no baserunners, automatically roll the next AB", ) async def game_settings_command( self, interaction: discord.Interaction, roll_buttons: bool = None, auto_roll: bool = None, ): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction( session, interaction, command_name="settings-ingame", lock_play=False ) await interaction.edit_original_response( content=None, embed=await update_game_settings( session, interaction, this_game, roll_buttons=roll_buttons, auto_roll=auto_roll, ), ) @app_commands.command( name="end-game", description="End the current game in this channel" ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def end_game_command(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "end-game") as ( this_game, owner_team, this_play, ): await manual_end_game( session, interaction, this_game, current_play=this_play ) group_substitution = app_commands.Group( name="substitute", description="Make a substitution in active game" ) @group_substitution.command(name="batter", description="Make a batter substitution") async def sub_batter_command( self, interaction: discord.Interaction, batting_order: Literal[ "this-spot", "1", "2", "3", "4", "5", "6", "7", "8", "9" ] = "this-spot", ): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction( session, interaction, command_name="substitute batter", lock_play=False ) if batting_order == "this-spot": if this_play.batter.team != owner_team: logger.info( f"Batting order not included while on defense; returning" ) await interaction.edit_original_response( content=f"When you make a defensive substitution, please include the batting order where they should enter." ) return this_order = this_play.batting_order else: this_order = int(batting_order) logger.info(f"sub batter - this_play: {this_play}") bat_view = sub_batter_dropdown_view( session, this_game, owner_team, this_order, [interaction.user] ) await interaction.edit_original_response( content=f"### {owner_team.lname} Substitution", view=bat_view ) @group_substitution.command( name="pitcher", description="Make a pitching substitution" ) async def sub_pitcher_command( self, interaction: discord.Interaction, batting_order: Literal[ "dh-spot", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" ] = "10", ): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction( session, interaction, command_name="substitute pitcher", lock_play=False ) if owner_team != this_play.pitcher.team: logger.warning( f"User {interaction.user.name} ({owner_team.abbrev}) tried to run a sub for the {this_play.pitcher.team.lname}" ) await interaction.edit_original_response( content=f"Please run pitcher subs when your team is on defense. If you are pinch-hitting for a pitcher already in the lineup, use `/substitute batter`" ) return if batting_order != "10" and this_play.pitcher.batting_order == 10: forfeit_dh = await ask_confirm( interaction, f"Are you sure you want to forfeit the DH?" ) if not forfeit_dh: await interaction.edit_original_response( content=f"Fine, be that way." ) return if not this_play.is_new_inning: pitcher_plays = get_plays_by_pitcher( session, this_game, this_play.pitcher ) batters_faced = sum(1 for x in pitcher_plays if x.pa == 1) if batters_faced < 3: await interaction.edit_original_response( content=f"Looks like **{this_play.pitcher.player.name}** has only faced {batters_faced} of the 3-batter minimum." ) return rp_view = relief_pitcher_dropdown_view( session, this_game, this_play.pitcher.team, batting_order, responders=[interaction.user], ) rp_message = await interaction.edit_original_response( content=f"### {this_play.pitcher.team.lname} Relief Pitcher", view=rp_view, ) @group_substitution.command( name="defense", description="Make a defensive substitution or move defenders between positions", ) async def sub_defense_command( self, interaction: discord.Interaction, new_position: DEFENSE_NO_PITCHER_LITERAL ): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction( session, interaction, command_name="substitute defense", lock_play=False ) defense_view = await defender_dropdown_view( session=session, this_game=this_game, human_team=owner_team, new_position=new_position, responders=[interaction.user], ) defense_message = await interaction.edit_original_response( content=f"### {owner_team.lname} {new_position} Change", view=defense_view, ) group_log = app_commands.Group( name="log", description="Log a play in this channel's game" ) @group_log.command(name="flyball", description="Flyballs: a, b, ballpark, bq, c") async def log_flyball( self, interaction: discord.Interaction, flyball_type: Literal["a", "b", "ballpark", "b?", "c"], ): with Session(engine) as session: async with locked_play(session, interaction, "log flyball") as ( this_game, owner_team, this_play, ): logger.info(f"log flyball {flyball_type} - this_play: {this_play}") this_play = await flyballs( session, interaction, this_play, flyball_type ) await self.complete_and_post_play( session, interaction, this_play, buffer_message=( "Flyball logged" if this_play.starting_outs + this_play.outs < 3 and ( (this_play.on_second and flyball_type in ["b", "ballpark"]) or (this_play.on_third and flyball_type == "b?") ) else None ), ) @group_log.command( name="frame-pitch", description=f"Walk/strikeout split; determined by home plate umpire", ) async def log_frame_check(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log frame-check") as ( this_game, owner_team, this_play, ): logger.info(f"log frame-check - this_play: {this_play}") this_play = await frame_checks(session, interaction, this_play) await self.complete_and_post_play( session, interaction, this_play, buffer_message="Frame check logged" ) @group_log.command( name="lineout", description="Lineouts: one out, ballpark, max outs" ) async def log_lineout( self, interaction: discord.Interaction, lineout_type: Literal["one-out", "ballpark", "max-outs"], ): with Session(engine) as session: async with locked_play(session, interaction, "log lineout") as ( this_game, owner_team, this_play, ): logger.info(f"log lineout - this_play: {this_play}") this_play = await lineouts( session, interaction, this_play, lineout_type ) await self.complete_and_post_play( session, interaction, this_play, buffer_message=( "Lineout logged" if this_play.on_base_code > 3 else None ), ) @group_log.command(name="single", description="Singles: *, **, ballpark, uncapped") async def log_single( self, interaction: discord.Interaction, single_type: Literal["*", "**", "ballpark", "uncapped"], ): with Session(engine) as session: async with locked_play(session, interaction, "log single") as ( this_game, owner_team, this_play, ): logger.info(f"log single {single_type} - this_play: {this_play}") this_play = await singles(session, interaction, this_play, single_type) await self.complete_and_post_play( session, interaction, this_play, buffer_message=( "Single logged" if ( (this_play.on_first or this_play.on_second) and single_type == "uncapped" ) else None ), ) @group_log.command(name="double", description="Doubles: **, ***, uncapped") async def log_double( self, interaction: discord.Interaction, double_type: Literal["**", "***", "uncapped"], ): with Session(engine) as session: async with locked_play(session, interaction, "log double") as ( this_game, owner_team, this_play, ): logger.info(f"log double {double_type} - this_play: {this_play}") this_play = await doubles(session, interaction, this_play, double_type) await self.complete_and_post_play( session, interaction, this_play, buffer_message=( "Double logged" if (this_play.on_first and double_type == "uncapped") else None ), ) @group_log.command(name="triple", description="Triples: no sub-types") async def log_triple(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log triple") as ( this_game, owner_team, this_play, ): logger.info(f"log triple - this_play: {this_play}") this_play = await triples(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name="homerun", description="Home Runs: ballpark, no-doubt") async def log_homerun( self, interaction: discord.Interaction, homerun_type: Literal["ballpark", "no-doubt"], ): with Session(engine) as session: async with locked_play(session, interaction, "log homerun") as ( this_game, owner_team, this_play, ): logger.info(f"log homerun {homerun_type} - this_play: {this_play}") this_play = await homeruns( session, interaction, this_play, homerun_type ) await self.complete_and_post_play(session, interaction, this_play) @group_log.command( name="walk", description="Walks: unintentional (default), intentional" ) async def log_walk( self, interaction: discord.Interaction, walk_type: Literal["unintentional", "intentional"] = "unintentional", ): with Session(engine) as session: async with locked_play(session, interaction, "log walk") as ( this_game, owner_team, this_play, ): logger.info(f"log walk {walk_type} - this_play: {this_play}") this_play = await walks(session, interaction, this_play, walk_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name="strikeout", description="Strikeout") async def log_strikeout(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log strikeout") as ( this_game, owner_team, this_play, ): logger.info(f"log strikeout - this_play: {this_play}") this_play = await strikeouts(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name="popout", description="Popout") async def log_popout(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log popout") as ( this_game, owner_team, this_play, ): logger.info(f"log popout - this_play: {this_play}") this_play = await popouts(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name="groundball", description="Groundballs: a, b, c") async def log_groundball( self, interaction: discord.Interaction, groundball_type: Literal["a", "b", "c"] ): with Session(engine) as session: async with locked_play( session, interaction, f"log groundball {groundball_type}" ) as (this_game, owner_team, this_play): logger.info( f"log groundball {groundball_type} - this_play: {this_play}" ) this_play = await groundballs( session, interaction, this_play, groundball_type ) await self.complete_and_post_play(session, interaction, this_play) @group_log.command( name="hit-by-pitch", description="Hit by pitch: batter to first; runners advance if forced", ) async def log_hit_by_pitch(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log hit-by-pitch") as ( this_game, owner_team, this_play, ): logger.info(f"log hit-by-pitch - this_play: {this_play}") this_play = await hit_by_pitch(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command( name="chaos", description="Chaos: wild-pitch, passed-ball, balk, pickoff" ) async def log_chaos( self, interaction: discord.Interaction, chaos_type: Literal["wild-pitch", "passed-ball", "balk", "pickoff"], ): with Session(engine) as session: async with locked_play(session, interaction, "log chaos") as ( this_game, owner_team, this_play, ): if this_play.on_base_code == 0: await interaction.edit_original_response( content=f"There cannot be chaos when the bases are empty." ) return logger.info(f"log chaos - this_play: {this_play}") this_play = await chaos(session, interaction, this_play, chaos_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command( name="bunt", description="Bunts: sacrifice, bad, popout, double-play, defense" ) async def log_sac_bunt( self, interaction: discord.Interaction, bunt_type: Literal["sacrifice", "bad", "popout", "double-play", "defense"], ): with Session(engine) as session: async with locked_play(session, interaction, "log bunt") as ( this_game, owner_team, this_play, ): if this_play.on_base_code == 0: await interaction.edit_original_response( content=f"You cannot bunt when the bases are empty." ) return elif this_play.starting_outs == 2: await interaction.edit_original_response( content=f"You cannot bunt with two outs." ) return logger.info(f"log bunt - this_play: {this_play}") this_play = await bunts(session, interaction, this_play, bunt_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command( name="stealing", description="Running: stolen-base, caught-stealing" ) @app_commands.describe( to_base="Base the runner is advancing to; 2 for 2nd, 3 for 3rd, 4 for Home" ) async def log_stealing( self, interaction: discord.Interaction, running_type: Literal["stolen-base", "caught-stealing", "steal-plus-overthrow"], to_base: Literal[2, 3, 4], ): with Session(engine) as session: async with locked_play(session, interaction, "log stealing") as ( this_game, owner_team, this_play, ): if ( (to_base == 2 and this_play.on_first is None) or (to_base == 3 and this_play.on_second is None) or (to_base == 4 and this_play.on_third is None) ): logger.info(f"Illegal steal attempt") await interaction.edit_original_response( content=f"I don't see a runner there." ) return if (to_base == 3 and this_play.on_third is not None) or ( to_base == 2 and this_play.on_second is not None ): logger.info(f"Stealing runner is blocked") if to_base == 3: content = f"{this_play.on_second.player.name} is blocked by {this_play.on_third.player.name}" else: content = f"{this_play.on_first.player.name} is blocked by {this_play.on_second.player.name}" await interaction.edit_original_response(content=content) return logger.info(f"log stealing - this_play: {this_play}") this_play = await steals( session, interaction, this_play, running_type, to_base ) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name="xcheck", description="Defender makes an x-check") @app_commands.choices( position=[ Choice(name="Pitcher", value="P"), Choice(name="Catcher", value="C"), Choice(name="First Base", value="1B"), Choice(name="Second Base", value="2B"), Choice(name="Third Base", value="3B"), Choice(name="Shortstop", value="SS"), Choice(name="Left Field", value="LF"), Choice(name="Center Field", value="CF"), Choice(name="Right Field", value="RF"), ] ) async def log_xcheck_command( self, interaction: discord.Interaction, position: Choice[str] ): with Session(engine) as session: async with locked_play(session, interaction, "log xcheck") as ( this_game, owner_team, this_play, ): logger.info(f"log xcheck - this_play: {this_play}") this_play = await xchecks( session, interaction, this_play, position.value ) await self.complete_and_post_play( session, interaction, this_play, buffer_message="X-Check logged" ) @group_log.command( name="undo-play", description="Roll back most recent play from the log" ) async def log_undo_play_command(self, interaction: discord.Interaction): with Session(engine) as session: async with locked_play(session, interaction, "log undo-play") as ( this_game, owner_team, this_play, ): logger.info(f"log undo-play - this_play: {this_play}") original_play = this_play this_play = undo_play(session, this_play) original_play.locked = ( False # prevent finally from committing deleted row ) await self.post_play(session, interaction, this_play) group_show = app_commands.Group( name="show-card", description="Display the player card for an active player" ) @group_show.command(name="defense", description="Display a defender's player card") async def show_defense_command( self, interaction: discord.Interaction, position: DEFENSE_LITERAL ): with Session(engine) as session: # Read-only command - no play locking needed await interaction.response.defer(thinking=True) this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content="I don't see an active game in this channel." ) return this_play = this_game.current_play_or_none(session) if this_play is None: await interaction.edit_original_response( content="No active play found." ) return logger.info(f"show-card defense - position: {position}") await show_defense_cards(session, interaction, this_play, position) async def setup(bot): await bot.add_cog(Gameplay(bot))