diff --git a/botbowl/core/forward_model.py b/botbowl/core/forward_model.py index dcc12379..faca59f7 100644 --- a/botbowl/core/forward_model.py +++ b/botbowl/core/forward_model.py @@ -7,12 +7,13 @@ """ from copy import deepcopy, copy from enum import Enum -from pytest import set_trace class Reversible: - def __init__(self, ignored_keys=[]): + def __init__(self, ignored_keys=None): + if ignored_keys is None: + ignored_keys = [] super().__setattr__("_trajectory", None) super().__setattr__("_ignored_keys", set(ignored_keys)) @@ -89,7 +90,6 @@ def next_step(self): self.current_step += 1 - class Step: def undo(self): raise NotImplementedError("Method to be overwritten by subclass") @@ -332,6 +332,12 @@ def is_immutable(obj): immutable_types = {int, float, str, tuple, bool, range, type(None)} +def treat_as_immutable(cls): + """Used as decorator for classes that should never be tracked by forward model""" + immutable_types.add(cls) + return cls + + def add_reversibility(value, trajectory): if is_immutable(value): return value diff --git a/botbowl/core/game.py b/botbowl/core/game.py index efc21786..4f8e686e 100755 --- a/botbowl/core/game.py +++ b/botbowl/core/game.py @@ -5,17 +5,44 @@ ========================== This module contains the Game class, which is the main class and interface used to interact with a game in botbowl. """ - from botbowl.core.load import * from botbowl.core.procedure import * from botbowl.core.forward_model import Trajectory, MovementStep +from copy import deepcopy +from typing import Optional, Tuple, List, Union, Any + + class InvalidActionError(Exception): pass -class Game: - def __init__(self, game_id, home_team, away_team, home_agent, away_agent, config=None, arena=None, ruleset=None, state=None, seed=None, record=False): +class Game: + replay: Optional[Replay] + game_id: str + home_agent: Agent + away_agent: Agent + arena: TwoPlayerArena + config: Configuration + ruleset: RuleSet + state: GameState + rnd: np.random.RandomState + ff_map: Any #?? + start_time: Optional[float] + end_time: Optional[float] + last_request_time: Optional[float] + last_action_time: Optional[float] + action: Optional[Action] + trajectory: Trajectory + square_shortcut: List[List[Square]] + + def __init__(self, game_id, home_team: Team, away_team: Team, home_agent: Agent, away_agent: Agent, + config: Optional[Configuration] = None, + arena: Optional[TwoPlayerArena] = None, + ruleset: Optional[RuleSet] = None, + state: Optional[GameState] = None, + seed=None, + record: bool = False): assert config is not None or arena is not None assert config is not None or ruleset is not None assert home_team.team_id != away_team.team_id @@ -33,12 +60,11 @@ def __init__(self, game_id, home_team, away_team, home_agent, away_agent, config self.end_time = None self.last_request_time = None self.last_action_time = None - self.forced_action = None self.action = None self.trajectory = Trajectory() self.square_shortcut = self.state.pitch.squares - def to_json(self, ignore_reports=False): + def to_json(self, ignore_reports: bool = False): return { 'game_id': self.game_id, 'start_time': self.start_time, @@ -58,7 +84,7 @@ def to_json(self, ignore_reports=False): 'rounds': self.config.rounds, } - def enable_forward_model(self): + def enable_forward_model(self) -> None: """ Enables the forward model. Should not be called before Game.init(). Can only be called once. Can't be undone """ @@ -68,14 +94,14 @@ def enable_forward_model(self): self.trajectory.enabled = True self.state.set_trajectory(self.trajectory) - def get_step(self): + def get_step(self) -> int: """ Returns an int that is the forward model step counter. The step counter can be used to revert the game state to this state with Game.revert() """ return self.trajectory.current_step - def revert(self, to_step): + def revert(self, to_step: int) -> None: """ Reverts the game state to how a the step to_step """ @@ -83,13 +109,13 @@ def revert(self, to_step): self.trajectory.revert(to_step) @property - def actor(self): + def actor(self) -> Optional[Agent]: if len(self.state.available_actions) > 0: return self.get_team_agent(self.state.available_actions[0].team) else: return None - def init(self): + def init(self) -> None: """ Initialized the Game. The START_GAME action must still be called after this if humans are in the game. """ @@ -111,7 +137,7 @@ def init(self): self.replay.record_action(start_action) self.step(start_action) - def step(self, action=None): + def step(self, action=None) -> None: """ Runs until an action from a human is required. If game requires an action to continue one must be given. :param action: Action to perform, can be None if game does not require any. @@ -122,7 +148,6 @@ def step(self, action=None): if action is not None: action.player = self.get_player(action.player.player_id) if action.player is not None else None - # Set action as a property so other methods can access it self.action = action @@ -139,7 +164,7 @@ def step(self, action=None): if self.state.stack.is_empty(): print("Somethings wrong") - + # if procedure is ready for input if done: @@ -150,7 +175,7 @@ def step(self, action=None): # Query agent for action self.last_request_time = time.time() self.action = self._safe_act() - + # Check if time limit was violated self.last_action_time = time.time() @@ -167,13 +192,13 @@ def step(self, action=None): # If not in fast mode - wait for input before continuing if not self.config.fast_mode: break - + # Else continue procedure with no action self.action = None self.trajectory.next_step() - def refresh(self): + def refresh(self) -> None: """ Checks clocks and runs forced actions. Useful in called in human games. """ @@ -183,7 +208,7 @@ def refresh(self): if self.action is not None: self.step(self.action) - def _check_clocks(self): + def _check_clocks(self) -> None: """ Checks if clocks are done. """ @@ -201,7 +226,7 @@ def _check_clocks(self): actor = self.actor clock = self.get_agent_clock(actor) while clock in self.state.clocks: - + # Request timout action if done: action = self._forced_action() @@ -222,10 +247,10 @@ def _check_clocks(self): if not clock.is_running(): clock.resume() - def _end_game(self): - ''' + def _end_game(self) -> None: + """ End the game - ''' + """ # Game ended when the last action was received - to avoid timout during finishing procedures self.end_time = self.last_action_time @@ -240,7 +265,7 @@ def _end_game(self): self.replay.record_step(self) self.replay.dump(self) - def _is_action_allowed(self, action): + def _is_action_allowed(self, action: Action) -> bool: """ Checks whether the specified action is allowed by comparing to actions in self.state.available_actions. :param action: @@ -283,7 +308,7 @@ def _is_action_allowed(self, action): return True return False - def _safe_act(self): + def _safe_act(self) -> Optional[Action]: """ Gets action from agent and sets correct player reference. """ @@ -296,15 +321,19 @@ def _safe_act(self): print(f"Unknown player id {action.player.player_id}") action.player = None else: - action.player = self.state.player_by_id[action.player.player_id] + action.player = self.state.player_by_id[action.player.player_id] return action - def _forced_action(self): - ''' + def _forced_action(self) -> Action: + """ Return action that prioritize to end the player's turn. - ''' + """ # Take first negative action - for action_type in [ActionType.END_TURN, ActionType.END_SETUP, ActionType.END_PLAYER_TURN, ActionType.SELECT_NONE, ActionType.HEADS, ActionType.KICK, ActionType.SELECT_DEFENDER_DOWN, ActionType.SELECT_DEFENDER_STUMBLES, ActionType.SELECT_ATTACKER_DOWN, ActionType.SELECT_PUSH, ActionType.SELECT_BOTH_DOWN, ActionType.DONT_USE_REROLL, ActionType.DONT_USE_APOTHECARY]: + for action_type in [ActionType.END_TURN, ActionType.END_SETUP, ActionType.END_PLAYER_TURN, + ActionType.SELECT_NONE, ActionType.HEADS, ActionType.KICK, ActionType.SELECT_DEFENDER_DOWN, + ActionType.SELECT_DEFENDER_STUMBLES, ActionType.SELECT_ATTACKER_DOWN, + ActionType.SELECT_PUSH, ActionType.SELECT_BOTH_DOWN, ActionType.DONT_USE_REROLL, + ActionType.DONT_USE_APOTHECARY]: for action in self.state.available_actions: if action_type == ActionType.END_SETUP and not self.is_setup_legal(self.get_agent_team(self.actor)): continue @@ -315,8 +344,8 @@ def _forced_action(self): position = self.rnd.choice(action_choice.positions) if len(action_choice.positions) > 0 else None player = self.rnd.choice(action_choice.players) if len(action_choice.players) > 0 else None return Action(action_choice.action_type, position=position, player=player) - - def _squares_moved(self): + + def _squares_moved(self) -> list: """ :return: The squares moved by the active player in json - used by the web app. """ @@ -331,9 +360,8 @@ def _squares_moved(self): return [] out = [sq.to_json() for sq in self.state.active_player.state.squares_moved] return out - #return [] - def _one_step(self, action): + def _one_step(self, action: Action) -> bool: """ Executes one step in the game if it is allowed. :param action: Action from agent. Can be None if no action is required. @@ -362,11 +390,11 @@ def _one_step(self, action): if not self._is_action_allowed(action): if type(action) is Action: - raise InvalidActionError(f"Action not allowed {action.to_json() if action is not None else 'None'}") + raise InvalidActionError( + f"Action not allowed {action.to_json() if action is not None else 'None'}") else: raise InvalidActionError(f"Action not allowed {action}") - # Run proc if self.config.debug_mode: print("Proc={}".format(proc)) @@ -385,11 +413,13 @@ def _one_step(self, action): for y in range(len(self.state.pitch.board)): for x in range(len(self.state.pitch.board)): assert self.state.pitch.board[y][x] is None or \ - (self.state.pitch.board[y][x].position.x == x and self.state.pitch.board[y][x].position.y == y) + (self.state.pitch.board[y][x].position.x == x and self.state.pitch.board[y][ + x].position.y == y) for team in self.state.teams: for player in team.players: - if not (player.position is None or self.state.pitch.board[player.position.y][player.position.x] == player): + if not (player.position is None or + self.state.pitch.board[player.position.y][player.position.x] == player): raise Exception("Player position violation") # Remove all finished procs @@ -429,89 +459,90 @@ def _one_step(self, action): return False # Can continue without user input # End player turn if only action available - if len(self.state.available_actions) == 1 and self.state.available_actions[0].action_type == ActionType.END_PLAYER_TURN: + if len(self.state.available_actions) == 1 and \ + self.state.available_actions[0].action_type == ActionType.END_PLAYER_TURN: return self._one_step(Action(ActionType.END_PLAYER_TURN)) return True # Game needs user input - - def remove_clocks(self): - ''' + + def remove_clocks(self) -> None: + """ Remove all clocks. - ''' + """ self.state.clocks.clear() - def remove_secondary_clocks(self): - ''' + def remove_secondary_clocks(self) -> None: + """ Remove all secondary clocks and resume the primary clock - if any. - ''' + """ self.state.clocks = [clock for clock in self.state.clocks if clock.is_primary] for clock in self.state.clocks: if not clock.is_running(): clock.resume() - def get_clock(self, team): - ''' + def get_clock(self, team: Team) -> Optional[Clock]: + """ Returns the clock belonging to the given team. - ''' + """ for clock in self.state.clocks: if clock.team == team: return clock return None - def get_agent_clock(self, agent): - ''' + def get_agent_clock(self, agent: Agent) -> Optional[Clock]: + """ Returns the clock belonging to the given agent's team. - ''' + """ for clock in self.state.clocks: if clock.team == self.get_agent_team(agent): return clock return None - def has_clock(self, team): - ''' + def has_clock(self, team: Team) -> bool: + """ Returns true if the given team has a clock. - ''' + """ for clock in self.state.clocks: if clock.team == team: return True return False - def has_agent_clock(self, agent): - ''' + def has_agent_clock(self, agent: Agent) -> bool: + """ Returns true if the given agent's team has a clock. - ''' + """ for clock in self.state.clocks: if clock.team == self.get_agent_team(agent): return True return False - def pause_clocks(self): - ''' + def pause_clocks(self) -> None: + """ Pauses all clocks. - ''' + """ for clock in self.state.clocks: if clock.is_running(): clock.pause() - def add_secondary_clock(self, team): - ''' + def add_secondary_clock(self, team: Team) -> None: + """ Adds a secondary clock for quick decisions. - ''' + """ self.pause_clocks() assert team is not None and type(team) == Team clock = Clock(team, self.config.time_limits.secondary) self.state.clocks.append(clock) - def add_primary_clock(self, team): - ''' + def add_primary_clock(self, team: Team) -> None: + """ Adds a primary clock that will be paused if secondary clocks are added. - ''' + """ self.state.clocks.clear() assert team is not None and type(team) == Team clock = Clock(team, self.config.time_limits.turn, is_primary=True) self.state.clocks.append(clock) - def get_seconds_left(self, team=None): + def get_seconds_left(self, team: Team = None) -> Optional[int]: ''' Returns the number of seconds left on the clock for the given team and None if the given team has no clock. ''' @@ -526,13 +557,14 @@ def get_seconds_left(self, team=None): return clock.get_seconds_left() return None - def is_started(self): - """ - Returns true if the game is started else false. - """ - return (not self.state.game_over) and len(self.state.stack.items) > 0 + # redefined below + #def is_started(self): + # """ + # Returns true if the game is started else false. + # """ + # return (not self.state.game_over) and len(self.state.stack.items) > 0 - def get_team_agent(self, team): + def get_team_agent(self, team: Team) -> Optional[Agent]: """ :param team: :return: The agent who's controlling the specified team. @@ -543,7 +575,7 @@ def get_team_agent(self, team): return self.home_agent return self.away_agent - def get_agent_team(self, agent): + def get_agent_team(self, agent: Agent) -> Optional[Team]: """ :param agent: The agent controlling the team :return: The team controlled by the specified agent. @@ -554,30 +586,30 @@ def get_agent_team(self, agent): return self.state.home_team return self.state.away_team - def set_seed(self, seed): + def set_seed(self, seed: int) -> None: ''' Sets the random seed of the game. ''' self.seed = seed self.rnd = np.random.RandomState(self.seed) - def set_available_actions(self): + def set_available_actions(self) -> None: """ Calls the current procedure's available_actions() method and sets the game's available actions to the returned list. """ self.state.available_actions = self.state.stack.peek().available_actions() - - def report(self, outcome): + + def report(self, outcome) -> None: """ Adds the outcome to the game's reports. """ self.state.reports.append(outcome) - def is_started(self): + def is_started(self) -> bool: return self.start_time is not None - def is_team_side(self, position, team): + def is_team_side(self, position: Square, team: Team) -> bool: """ :param position: :param team: @@ -587,7 +619,7 @@ def is_team_side(self, position, team): return self.arena.board[position.y][position.x] in TwoPlayerArena.home_tiles return self.arena.board[position.y][position.x] in TwoPlayerArena.away_tiles - def get_team_side(self, team): + def get_team_side(self, team: Team) -> List[Square]: """ :param team: :return: a list of squares on team's side of the arena. @@ -595,18 +627,19 @@ def get_team_side(self, team): tiles = [] for y in range(len(self.arena.board)): for x in range(len(self.arena.board[y])): - if self.arena.board[y][x] in (TwoPlayerArena.home_tiles if team == self.state.home_team else TwoPlayerArena.away_tiles): + if self.arena.board[y][x] in \ + (TwoPlayerArena.home_tiles if team == self.state.home_team else TwoPlayerArena.away_tiles): tiles.append(self.get_square(x, y)) return tiles - def is_scrimmage(self, position): + def is_scrimmage(self, position: Square) -> bool: """ :param position: :return: Returns True if pos is on the scrimmage line. """ return self.arena.board[position.y][position.x] in TwoPlayerArena.scrimmage_tiles - def is_wing(self, position, right): + def is_wing(self, position: Square, right) -> bool: """ :param position: :param right: Whether to check on the right side of the arena. If False, it will check on the left side. @@ -616,34 +649,34 @@ def is_wing(self, position, right): return self.arena.board[position.y][position.x] in TwoPlayerArena.wing_right_tiles return self.arena.board[position.y][position.x] in TwoPlayerArena.wing_left_tiles - def remove_balls(self): + def remove_balls(self) -> None: """ Removes all balls from the arena. """ self.state.pitch.balls.clear() - def is_last_turn(self): + def is_last_turn(self) -> bool: """ :return: True if this turn is the last turn of the game. """ return self.get_next_team().state.turn == self.config.rounds and self.state.half == 2 - def is_last_round(self): + def is_last_round(self) -> bool: """ :return: True if this round is the las round of the game. """ return self.state.round == self.config.rounds - def get_next_team(self): + def get_next_team(self) -> Team: """ :return: The team who's turn it is next. """ idx = self.state.turn_order.index(self.state.current_team) - if idx+1 == len(self.state.turn_order): + if idx + 1 == len(self.state.turn_order): return self.state.turn_order[0] - return self.state.turn_order[idx+1] + return self.state.turn_order[idx + 1] - def add_or_skip_turn(self, turns): + def add_or_skip_turn(self, turns: None) -> None: """ Adds or removes a number of turns from the current half. This method will raise an assertion error if the turn counter goes to a negative number. @@ -653,21 +686,21 @@ def add_or_skip_turn(self, turns): team.state.turn += turns assert team.state.turn >= 0 - def get_player(self, player_id): + def get_player(self, player_id: str) -> Optional[Player]: """ :param player_id: :return: Returns the player with player_id """ return self.state.player_by_id[player_id] - def get_player_at(self, position): + def get_player_at(self, position: Square) -> Optional[Player]: """ :param position: :return: Returns the player at pos else None. """ return self.state.pitch.board[position.y][position.x] - def set_turn_order_from(self, first_team): + def set_turn_order_from(self, first_team: Team) -> None: """ Sets the turn order starting from first_team. :param first_team: The first team to start. @@ -686,7 +719,7 @@ def set_turn_order_from(self, first_team): after.append(team) self.state.turn_order = after + before - def set_turn_order_after(self, last_team): + def set_turn_order_after(self, last_team: Team) -> None: """ Sets the turn order starting after last_team. :param last_team: The last team to start. @@ -705,57 +738,57 @@ def set_turn_order_after(self, last_team): added = True self.state.turn_order = after + before - def get_turn_order(self): + def get_turn_order(self) -> List[Team]: """ :return: The list of teams sorted by turn order. """ return self.state.turn_order - def is_home_team(self, team): + def is_home_team(self, team: Team) -> bool: """ :return: True if team is the home team. """ return team == self.state.home_team - def get_opp_team(self, team): + def get_opp_team(self, team: Team) -> Team: """ :param team: :return: The opponent team of team. """ return self.state.home_team if self.state.away_team == team else self.state.away_team - def get_dugout(self, team): + def get_dugout(self, team: Team) -> Dugout: return self.state.dugouts[team.team_id] - def get_reserves(self, team): + def get_reserves(self, team: Team) -> List[Player]: """ :param team: :return: The reserves in the dugout of this team. """ return self.get_dugout(team).reserves - def get_knocked_out(self, team): + def get_knocked_out(self, team: Team) -> List[Player]: """ :param team: :return: The knocked out players in the dugout of this team. """ return self.get_dugout(team).kod - def get_casualties(self, team): + def get_casualties(self, team: Team) -> List[Player]: """ :param team: :return: The badly hurt, injured, and dead players in th dugout of this team. """ return self.get_dugout(team).casualties - def get_dungeon(self, team): + def get_dungeon(self, team: Team) -> List[Player]: """ :param team: :return: The ejected players of this team, who's locked to a cell in the dungeon. """ return self.get_dugout(team).dungeon - def current_turn(self): + def current_turn(self) -> Optional[Turn]: """ :return: The top-most Turn procedure in the stack. """ @@ -765,7 +798,7 @@ def current_turn(self): return proc return None - def can_use_reroll(self, team): + def can_use_reroll(self, team: Team) -> bool: """ :param team: :return: True if the team can use reroll right now (i.e. this turn). @@ -776,7 +809,7 @@ def can_use_reroll(self, team): return not current_turn.quick_snap return False - def get_kicking_team(self, half=None): + def get_kicking_team(self, half: Optional[int] = None): """ :param half: Set this to None if you want the team who's kicking this drive. :return: The team who's kicking in the specified half. If half is None, the team who's kicking this drive. @@ -785,7 +818,7 @@ def get_kicking_team(self, half=None): return self.state.kicking_this_drive return self.state.kicking_first_half if half == 1 else self.state.receiving_first_half - def get_receiving_team(self, half=None): + def get_receiving_team(self, half: Optional[int] = None) -> Team: """ :param half: Set this to None if you want the team who's receiving this drive. :return: The team who's receiving in the specified half. If half is None, the team who's receiving this drive. @@ -794,29 +827,30 @@ def get_receiving_team(self, half=None): return self.state.receiving_this_drive return self.state.receiving_first_half if half == 1 else self.state.kicking_first_half - def has_ball(self, player): + def has_ball(self, player: Player) -> bool: """ :param player: :return: True if player has the ball. """ ball = self.get_ball_at(player.position) - return True if ball is not None and ball.is_carried else False + return ball is not None and ball.is_carried - def get_ball(self): + def get_ball(self) -> Optional[Ball]: """ :return: A ball on the pitch or None. """ for ball in self.state.pitch.balls: return ball - def is_touchdown(self, player): + def is_touchdown(self, player: Player) -> bool: """ :param player: :return: True if player is in the opponent's endzone with the ball. """ - return self.arena.is_in_opp_endzone(player.position, player.team == self.state.home_team) and self.has_ball(player) + return self.arena.is_in_opp_endzone(player.position, player.team == self.state.home_team) and \ + self.has_ball(player) - def is_blitz_available(self): + def is_blitz_available(self) -> bool: """ :return: True if the current team can make a blitz this turn. """ @@ -824,7 +858,7 @@ def is_blitz_available(self): if turn is not None: return turn.blitz_available - def use_blitz_action(self): + def use_blitz_action(self) -> None: """ Uses this turn's blitz action. """ @@ -832,7 +866,7 @@ def use_blitz_action(self): if turn is not None: turn.blitz_available = False - def unuse_blitz_action(self): + def unuse_blitz_action(self) -> None: """ Unuses this turn's blitz action. """ @@ -840,7 +874,7 @@ def unuse_blitz_action(self): if turn is not None: turn.blitz_available = True - def is_pass_available(self): + def is_pass_available(self) -> bool: """ :return: True if the current team can make a pass this turn. """ @@ -848,7 +882,7 @@ def is_pass_available(self): if turn is not None: return turn.pass_available - def use_pass_action(self): + def use_pass_action(self) -> None: """ Use this turn's pass action. """ @@ -856,7 +890,7 @@ def use_pass_action(self): if turn is not None: turn.pass_available = False - def unuse_pass_action(self): + def unuse_pass_action(self) -> None: """ Use this turn's pass action. """ @@ -864,7 +898,7 @@ def unuse_pass_action(self): if turn is not None: turn.pass_available = True - def is_handoff_available(self): + def is_handoff_available(self) -> bool: """ :return: True if the current team can make a handoff this turn. """ @@ -872,7 +906,7 @@ def is_handoff_available(self): if turn is not None: return turn.handoff_available - def use_handoff_action(self): + def use_handoff_action(self) -> None: """ Uses this turn's handoff action. """ @@ -880,7 +914,7 @@ def use_handoff_action(self): if turn is not None: turn.handoff_available = False - def unuse_handoff_action(self): + def unuse_handoff_action(self) -> None: """ Uses this turn's handoff action. """ @@ -888,7 +922,7 @@ def unuse_handoff_action(self): if turn is not None: turn.handoff_available = True - def is_foul_available(self): + def is_foul_available(self) -> bool: """ :return: True if the current team can make a foul this turn. """ @@ -896,7 +930,7 @@ def is_foul_available(self): if turn is not None: return turn.foul_available - def use_foul_action(self): + def use_foul_action(self) -> None: """ Uses this turn's foul action. """ @@ -904,7 +938,7 @@ def use_foul_action(self): if turn is not None: turn.foul_available = False - def unuse_foul_action(self): + def unuse_foul_action(self) -> None: """ Uses this turn's foul action. """ @@ -912,7 +946,7 @@ def unuse_foul_action(self): if turn is not None: turn.foul_available = True - def is_blitz(self): + def is_blitz(self) -> bool: """ :return: True if the current turn is a Blitz! """ @@ -920,7 +954,7 @@ def is_blitz(self): if turn is not None: return turn.blitz - def is_quick_snap(self): + def is_quick_snap(self) -> bool: """ :return: True if the current turn is a Quick Snap! """ @@ -928,7 +962,7 @@ def is_quick_snap(self): if turn is not None: return turn.quick_snap - def get_players_on_pitch(self, team=None, used=None, up=None): + def get_players_on_pitch(self, team: Team = None, used=None, up=None) -> List[Player]: """ :param team: The team of the players. :param used: If specified, filter by ther players used state. @@ -939,11 +973,12 @@ def get_players_on_pitch(self, team=None, used=None, up=None): for y in range(len(self.state.pitch.board)): for x in range(len(self.state.pitch.board[y])): player = self.state.pitch.board[y][x] - if player is not None and (team is None or player.team == team) and (used is None or used == player.state.used) and (up is None or up == player.state.up): + if player is not None and (team is None or player.team == team) and ( + used is None or used == player.state.used) and (up is None or up == player.state.up): players.append(player) return players - def pitch_to_reserves(self, player): + def pitch_to_reserves(self, player: Player) -> None: """ Moves player from the pitch to the reserves section in the dugout. :param player: @@ -953,10 +988,11 @@ def pitch_to_reserves(self, player): player.state.used = False player.state.up = True - def reserves_to_pitch(self, player, position): + def reserves_to_pitch(self, player: Player, position: Square) -> None: """ Moves player from the reserves section in the dugout to the pitch. :param player: + :param position: position on pitch to put player """ self.get_reserves(player.team).remove(player) player_at = self.get_player_at(position) @@ -965,7 +1001,7 @@ def reserves_to_pitch(self, player, position): self.put(player, position) player.state.up = True - def pitch_to_kod(self, player): + def pitch_to_kod(self, player: Player) -> None: """ Moves player from the pitch to the KO section in the dugout. :param player: @@ -975,7 +1011,7 @@ def pitch_to_kod(self, player): player.state.knocked_out = True player.state.up = True - def kod_to_reserves(self, player): + def kod_to_reserves(self, player: Player) -> None: """ Moves player from the KO section in the dugout to the pitch. This also resets the players knocked_out state. :param player: @@ -985,7 +1021,7 @@ def kod_to_reserves(self, player): player.state.knocked_out = False player.state.up = True - def pitch_to_casualties(self, player): + def pitch_to_casualties(self, player: Player) -> None: """ Moves player from the pitch to the CAS section in the dugout and applies the casualty and effect to the player. :param player: @@ -995,7 +1031,7 @@ def pitch_to_casualties(self, player): player.state.stunned = False self.get_casualties(player.team).append(player) - def pitch_to_dungeon(self, player): + def pitch_to_dungeon(self, player: Player) -> None: """ Moves player from the pitch to the dungeon and ejects the player from the game. :param player: @@ -1005,7 +1041,7 @@ def pitch_to_dungeon(self, player): player.state.ejected = True player.state.up = True - def lift(self, player): + def lift(self, player: Player) -> None: """ Lifts player from the board. Call put_down(player) to set player down again. """ @@ -1013,7 +1049,7 @@ def lift(self, player): player.state.in_air = True self.state.pitch.board[player.position.y][player.position.x] = None - def put_down(self, player): + def put_down(self, player: Player) -> None: """ Puts a player down on the board on the square it was hovering. """ @@ -1021,7 +1057,7 @@ def put_down(self, player): player.state.in_air = False self.put(player, player.position) - def put(self, piece, position): + def put(self, piece: Union[Catchable, Player], position: Square) -> None: """ Put a piece on or above a square. :param piece: Ball or player @@ -1036,13 +1072,14 @@ def put(self, piece, position): put=True) self.trajectory.log_state_change(log_entry) elif type(piece) is Ball: + piece: Ball self.state.pitch.balls.append(piece) elif type(piece) is Bomb: self.state.pitch.bomb = piece else: raise Exception("Unknown piece type") - def remove(self, piece): + def remove(self, piece: Union[Catchable, Player]) -> None: """ Remove a piece from the board. :param piece: @@ -1052,16 +1089,18 @@ def remove(self, piece): if not piece.state.in_air: self.state.pitch.board[piece.position.y][piece.position.x] = None - log_entry = MovementStep(self.state.pitch.board if not piece.state.in_air else None, piece, piece.position, put=False) + log_entry = MovementStep(self.state.pitch.board if not piece.state.in_air else None, piece, piece.position, + put=False) self.trajectory.log_state_change(log_entry) piece.position = None elif type(piece) is Ball: + piece: Ball self.state.pitch.balls.remove(piece) elif type(piece) is Bomb: self.state.pitch.bomb = None - def move(self, piece, position): + def move(self, piece: Union[Catchable, Player], position: Square) -> None: """ Move a piece already on the board. If the piece is a ball carrier, the ball is moved as well. :param piece: @@ -1078,7 +1117,7 @@ def move(self, piece, position): elif piece.is_catchable(): piece.move_to(position) - def shove(self, piece, x, y): + def shove(self, piece: Union[Catchable, Player], x: int, y: int) -> None: """ Shove a push x number of step in the horizontal direction and y number of steps in the vertical direction. :param piece @@ -1087,7 +1126,7 @@ def shove(self, piece, x, y): """ self.move(piece, self.get_square(piece.position.x + x, piece.position.y + y)) - def swap(self, piece_a, piece_b): + def swap(self, piece_a: Union[Catchable, Player], piece_b: Union[Catchable, Player]) -> None: """ Swap two pieces on the board. :param piece_a: @@ -1115,12 +1154,13 @@ def swap(self, piece_a, piece_b): elif type(piece_b) is Catchable: piece_a.move_to(pos_b) - def get_catch_modifiers(self, catcher, accurate=False, interception=False, handoff=False): + def get_catch_modifiers(self, catcher: Player, accurate: bool = False, interception: bool = False, + handoff: bool = False) -> int: """ :param catcher: :param accurate: whether it is an accurate pass. :param interception: whether it is an interception catch. - :param interception: whether it is a handoff catch. + :param handoff: whether it is a handoff catch. :return: the modifier to be added to the pass roll. """ modifiers = 1 if accurate or handoff else 0 @@ -1142,7 +1182,7 @@ def get_catch_modifiers(self, catcher, accurate=False, interception=False, hando modifiers -= 1 return modifiers - def get_pass_modifiers(self, passer, pass_distance, ttm=False): + def get_pass_modifiers(self, passer: Player, pass_distance: PassDistance, ttm: bool = False) -> int: """ :param passer: :param pass_distance: the PassDistance to the target. @@ -1180,17 +1220,18 @@ def get_pass_modifiers(self, passer, pass_distance, ttm=False): return modifiers - def get_leap_modifiers(self, player): + def get_leap_modifiers(self, player: Player) -> int: """ :param player: :return: the modifier to be added to the leap roll. """ return 1 if player.has_skill(Skill.VERY_LONG_LEGS) else 0 - def get_dodge_modifiers(self, player, position, include_diving_tackle=False): + def get_dodge_modifiers(self, player: Player, position: Square, include_diving_tackle: bool = False) -> int: """ :param player: :param position: The position the player is dodging to + :param include_diving_tackle: :return: the modifier to be added to the dodge roll. """ modifiers = 1 @@ -1219,7 +1260,7 @@ def get_dodge_modifiers(self, player, position, include_diving_tackle=False): return modifiers - def get_pickup_modifiers(self, player, position): + def get_pickup_modifiers(self, player: Player, position: Square) -> int: """ :param player: :param position: the square of the ball. @@ -1242,14 +1283,14 @@ def get_pickup_modifiers(self, player, position): return modifiers - def num_tackle_zones_in(self, player): + def num_tackle_zones_in(self, player: Player) -> int: """ :param player: :return: Number of opponent tackle zones the player is in. """ return self.num_tackle_zones_at(player, player.position) - def num_tackle_zones_at(self, player, position): + def num_tackle_zones_at(self, player: Player, position: Square) -> int: """ :param player: :param position: @@ -1261,7 +1302,7 @@ def num_tackle_zones_at(self, player, position): tackle_zones += 1 return tackle_zones - def get_catcher(self, position): + def get_catcher(self, position: Square) -> Optional[Player]: """ :param position: A square on the board :return: A player if the ball can be catched by one at the given square, otherwise None. @@ -1275,13 +1316,13 @@ def get_catcher(self, position): else: return None - def is_setup_legal(self, team): + def is_setup_legal(self, team: Team) -> bool: """ :param team: :return: Whether the team has set up legally. """ if not self.is_setup_legal_count(team, max_players=self.config.pitch_max, - min_players=self.config.pitch_min): + min_players=self.config.pitch_min): return False elif not self.is_setup_legal_scrimmage(team, min_players=self.config.scrimmage_min): return False @@ -1289,7 +1330,7 @@ def is_setup_legal(self, team): return False return True - def is_setup_legal_count(self, team, tile=None, max_players=11, min_players=3): + def is_setup_legal_count(self, team: Team, tile=None, max_players=11, min_players=3) -> bool: """ :param team: :param tile: The tile area to check. @@ -1311,7 +1352,7 @@ def is_setup_legal_count(self, team, tile=None, max_players=11, min_players=3): return False return True - def num_casualties(self, team=None): + def num_casualties(self, team: Team = None) -> int: """ :param team: If None, return the sum of both teams casualties. :return: The number of casualties suffered by team. @@ -1321,7 +1362,7 @@ def num_casualties(self, team=None): else: return len(self.get_casualties(self.state.home_team)) + len(self.get_casualties(self.state.away_team)) - def get_winning_team(self): + def get_winning_team(self) -> Optional[Team]: """ :return: The team with most touchdowns, otherwise None. """ @@ -1331,7 +1372,7 @@ def get_winning_team(self): return self.state.away_team return None - def is_setup_legal_scrimmage(self, team, min_players=3): + def is_setup_legal_scrimmage(self, team: Team, min_players=3) -> bool: """ :param team: :param min_players: @@ -1341,7 +1382,7 @@ def is_setup_legal_scrimmage(self, team, min_players=3): return self.is_setup_legal_count(team, tile=Tile.HOME_SCRIMMAGE, min_players=min_players) return self.is_setup_legal_count(team, tile=Tile.AWAY_SCRIMMAGE, min_players=min_players) - def is_setup_legal_wings(self, team, min_players=0, max_players=2): + def is_setup_legal_wings(self, team: Team, min_players=0, max_players=2) -> bool: """ :param team: :param min_players: @@ -1354,7 +1395,7 @@ def is_setup_legal_wings(self, team, min_players=0, max_players=2): return self.is_setup_legal_count(team, tile=Tile.AWAY_WING_LEFT, max_players=max_players, min_players=min_players) and \ self.is_setup_legal_count(team, tile=Tile.AWAY_WING_RIGHT, max_players=max_players, min_players=min_players) - def get_procedure_names(self): + def get_procedure_names(self) -> List[str]: """ :return: a list of procedure names in the stack. """ @@ -1368,14 +1409,13 @@ def get_procedure_names(self): procs.append(proc.__class__.__name__) return procs - def get_player_action_type(self): + def get_player_action_type(self) -> Optional[PlayerActionType]: """ - :param player: :return: the player PlayerActionType if there is any on the stack. """ return self.state.player_action_type - def remove_recursive_refs(self): + def remove_recursive_refs(self) -> None: """ Removes recursive references. Must be called before serializing. """ @@ -1383,7 +1423,7 @@ def remove_recursive_refs(self): for player in team.players: player.team = None - def add_recursive_refs(self): + def add_recursive_refs(self) -> None: """ Adds recursive references. Can be called after serializing. """ @@ -1391,15 +1431,15 @@ def add_recursive_refs(self): for player in team.players: player.team = team - def get_termination_time(self): - """ - The time at which the current turn must be terminated - or the opponent's action choice (like selecting block die). - """ - if self.state.termination_opp is not None: - return self.state.termination_opp - return self.state.termination_turn + #def get_termination_time(self): + # """ + # The time at which the current turn must be terminated - or the opponent's action choice (like selecting block die). + # """ + # if self.state.termination_opp is not None: + # return self.state.termination_opp + # return self.state.termination_turn - def get_team_by_id(self, team_id): + def get_team_by_id(self, team_id) -> Optional[Team]: """ :param team_id: :return: returns the team with the id or None @@ -1410,7 +1450,7 @@ def get_team_by_id(self, team_id): return self.state.away_team return None - def get_winner(self): + def get_winner(self) -> Optional[Agent]: """ returns the winning agent of the game. None if it's a draw. If the game timed out the current player loses. @@ -1427,10 +1467,10 @@ def get_winner(self): # If the game is over the player with most TDs wins if self.state.game_over: return self.get_team_agent(self.get_winning_team()) - + return None - def get_other_agent(self, agent): + def get_other_agent(self, agent: Agent) -> Optional[Agent]: """ Returns the other agent in the game. """ @@ -1440,7 +1480,7 @@ def get_other_agent(self, agent): return self.away_agent return self.home_agent - def get_other_active_player_id(self): + def get_other_active_player_id(self) -> Optional[str]: """ Returns the player id of the other player involved in current procedures - if any. """ @@ -1462,21 +1502,21 @@ def get_other_active_player_id(self): return proc.picked_up_teammate.player_id return None - def replace_home_agent(self, agent): + def replace_home_agent(self, agent: Agent) -> None: """ Replaces the home agent safely. :param agent: """ self.home_agent = agent - def replace_away_agent(self, agent): + def replace_away_agent(self, agent: Agent) -> None: """ Replaces the away agent safely. :param agent: """ self.away_agent = agent - def has_report_of_type(self, outcome_type, last=None): + def has_report_of_type(self, outcome_type, last=None) -> bool: """ :param outcome_type: :return: True if the the game has reported an outcome of the given type. If last is specified, only the recent number of reports are checked. @@ -1489,7 +1529,7 @@ def has_report_of_type(self, outcome_type, last=None): return True return False - def get_balls_at(self, position, in_air=False): + def get_balls_at(self, position: Square, in_air=False) -> List[Ball]: """ Assumes there is only one ball on the square :param position: @@ -1502,7 +1542,7 @@ def get_balls_at(self, position, in_air=False): balls.append(ball) return balls - def get_ball_at(self, position, in_air=False): + def get_ball_at(self, position: Square, in_air=False) -> Optional[Ball]: """ Assumes there is only one ball on the square. :param position: @@ -1512,34 +1552,34 @@ def get_ball_at(self, position, in_air=False): balls_at = self.get_balls_at(position, in_air) return balls_at[0] if balls_at else None - def get_bomb(self): + def get_bomb(self) -> Optional[Bomb]: """ Returns a bomb or None. :return: Bomb or None """ return self.state.pitch.bomb - def remove_bomb(self): + def remove_bomb(self) -> None: """ Removes the bombe from the pitch. """ self.state.pitch.bomb = None - def put_bomb(self, bomb): + def put_bomb(self, bomb) -> None: """ Adds a bomb to the pitch. """ assert self.state.pitch.bomb is None self.state.pitch.bomb = bomb - def get_ball_positions(self): + def get_ball_positions(self) -> List[Square]: """ :return: The position of the ball. If no balls are in the arena None is returned. If multiple balls are in the arena, the position of the first ball is return. """ return [ball.position for ball in self.state.pitch.balls] - def get_ball_position(self): + def get_ball_position(self) -> Optional[Square]: """ Assumes there is only one ball on the square :return: Ball or None @@ -1548,7 +1588,7 @@ def get_ball_position(self): return ball.position return None - def get_ball_carrier(self): + def get_ball_carrier(self) -> Optional[Player]: """ :return: the ball carrier if any - otherwise None. """ @@ -1558,14 +1598,15 @@ def get_ball_carrier(self): else: return self.get_player_at(ball_position) - def is_out_of_bounds(self, position): + def is_out_of_bounds(self, position: Square) -> bool: """ :param position: :return: True if pos is out of bounds. """ - return position.x < 1 or position.x >= self.state.pitch.width-1 or position.y < 1 or position.y >= self.state.pitch.height-1 + return position.x < 1 or position.x >= self.state.pitch.width - 1 or \ + position.y < 1 or position.y >= self.state.pitch.height - 1 - def get_push_squares(self, from_position, to_position): + def get_push_squares(self, from_position: Square, to_position: Square) -> List[Square]: """ :param from_position: The position of the attacker. :param to_position: The position of the defender. @@ -1600,7 +1641,7 @@ def get_push_squares(self, from_position, to_position): assert len(squares) > 0 return squares - def get_square(self, x, y): + def get_square(self, x, y) -> Square: """ Returns an existing square object for the given position to avoid a new instantiation. If the square object is out of bounds it may be instantiated. @@ -1614,7 +1655,8 @@ def get_square(self, x, y): return Square(x, y) return self.square_shortcut[y][x] - def get_adjacent_squares(self, position, diagonal=True, out=False, occupied=True, distance=1): + def get_adjacent_squares(self, position: Square, diagonal=True, out=False, occupied=True, distance=1) \ + -> List[Square]: """ Returns a list of adjacent squares from the position. :param position: @@ -1625,12 +1667,12 @@ def get_adjacent_squares(self, position, diagonal=True, out=False, occupied=True :return: """ squares = [] - r = range(-distance, distance+1) + r = range(-distance, distance + 1) for yy in r: for xx in r: if yy == 0 and xx == 0: continue - sq = self.get_square(position.x+xx, position.y+yy) + sq = self.get_square(position.x + xx, position.y + yy) if not out and self.is_out_of_bounds(sq): continue if not occupied and self.get_player_at(sq) is not None: @@ -1641,7 +1683,7 @@ def get_adjacent_squares(self, position, diagonal=True, out=False, occupied=True squares.append(sq) return squares - def get_adjacent_opponents(self, player, diagonal=True, down=True, standing=True, stunned=True, skill=None): + def get_adjacent_opponents(self, player: Player, diagonal=True, down=True, standing=True, stunned=True, skill=None) -> List[Player]: """ Returns a list of adjacent opponents to the player it its current position. :param player: @@ -1652,9 +1694,10 @@ def get_adjacent_opponents(self, player, diagonal=True, down=True, standing=True :param skill: Only include players with this skill. :return: """ - return self.get_adjacent_players(player.position, self.get_opp_team(player.team), diagonal, down, standing, stunned, skill=skill) + return self.get_adjacent_players(player.position, self.get_opp_team(player.team), diagonal, down, standing, + stunned, skill=skill) - def get_adjacent_teammates(self, player, diagonal=True, down=True, standing=True, stunned=True, skill=None): + def get_adjacent_teammates(self, player: Player, diagonal=True, down=True, standing=True, stunned=True, skill=None) -> List[Player]: """ Returns a list of adjacent teammates to the player it its current position. :param player: @@ -1667,7 +1710,8 @@ def get_adjacent_teammates(self, player, diagonal=True, down=True, standing=True """ return self.get_adjacent_players(player.position, player.team, diagonal, down, standing, stunned, skill=skill) - def get_adjacent_players(self, position, team=None, diagonal=True, down=True, standing=True, stunned=True, skill=None): + def get_adjacent_players(self, position: Square, team: Team=None, diagonal=True, down=True, standing=True, stunned=True, + skill=None) -> List[Player]: """ Returns a list of adjacent player to the position. :param position: @@ -1697,7 +1741,7 @@ def get_adjacent_players(self, position, team=None, diagonal=True, down=True, st players.append(player_at) return players - def get_assisting_players(self, player, opp_player, foul=False): + def get_assisting_players(self, player: Player, opp_player: Player, foul=False) -> List[Player]: """ :param player: :param opp_player: @@ -1709,7 +1753,7 @@ def get_assisting_players(self, player, opp_player, foul=False): for xx in range(-1, 2, 1): if yy == 0 and xx == 0: continue - p = self.get_square(opp_player.position.x+xx, opp_player.position.y+yy) + p = self.get_square(opp_player.position.x + xx, opp_player.position.y + yy) if not self.is_out_of_bounds(p) and player.position != p: player_at = self.get_player_at(p) if player_at is not None: @@ -1717,25 +1761,24 @@ def get_assisting_players(self, player, opp_player, foul=False): if not player_at.can_assist(): continue if (not foul and player_at.has_skill(Skill.GUARD)) or \ - self.num_tackle_zones_in(player_at) <= 1: + self.num_tackle_zones_in(player_at) <= 1: assists.append(player_at) return assists - def can_assist(self, player, foul=False): + def can_assist(self, player: Player, foul: bool = False) -> bool: """ - :param assister: The player which potentially can assist - :param opp_player: The opponent player to assist against + :param player: The player which potentially can assist :param foul: :return: """ if not player.can_assist(): return False if (not foul and player.has_skill(Skill.GUARD)) or \ - self.num_tackle_zones_in(player) <= 1: + self.num_tackle_zones_in(player) <= 1: return True return False - def get_assisting_players_at(self, player, opp_player, foul=False): + def get_assisting_players_at(self, player: Player, opp_player: Player, foul: bool=False) -> List[Player]: """ :param player: :param opp_player: @@ -1747,15 +1790,15 @@ def get_assisting_players_at(self, player, opp_player, foul=False): for xx in range(-1, 2, 1): if yy == 0 and xx == 0: continue - p = self.get_square(opp_player.position.x+xx, opp_player.position.y+yy) + p = self.get_square(opp_player.position.x + xx, opp_player.position.y + yy) if not self.is_out_of_bounds(p) and player.position != p: player_at = self.get_player_at(p) if player_at is not None: - if player_at.team == player.team and self.can_assist(player_at): + if player_at.team == player.team and self.can_assist(player_at, foul): assists.append(player_at) return assists - def get_block_strengths(self, attacker, defender, blitz=False): + def get_block_strengths(self, attacker: Player, defender: Player, blitz: bool = False) -> Tuple[int, int]: """ :param attacker: :param defender: @@ -1770,7 +1813,7 @@ def get_block_strengths(self, attacker, defender, blitz=False): defender_strength += len(self.get_assisting_players(defender, attacker)) return attacker_strength, defender_strength - def num_block_dice(self, attacker, defender, blitz=False, dauntless_success=False): + def num_block_dice(self, attacker: Player, defender: Player, blitz: bool = False, dauntless_success: bool = False) -> int: """ :param attacker: :param defender: @@ -1780,7 +1823,7 @@ def num_block_dice(self, attacker, defender, blitz=False, dauntless_success=Fals """ return self.num_block_dice_at(attacker, defender, attacker.position, blitz, dauntless_success) - def get_block_probs(self, attacker, defender): + def get_block_probs(self, attacker: Player, defender: Player) -> Tuple[float, float, float, float]: """ :param attacker: :param defender: @@ -1817,7 +1860,7 @@ def get_block_probs(self, attacker, defender): p_fumble_self = p_self return p_self, p_opp, p_fumble_self, p_fumble_opp - def get_blitz_probs(self, attacker, attack_position, defender): + def get_blitz_probs(self, attacker: Player, attack_position: Square, defender: Player) -> Tuple[float, float, float, float]: """ :param attacker: :param attack_position: @@ -1833,7 +1876,7 @@ def get_blitz_probs(self, attacker, attack_position, defender): self.move(attacker, orig_position) return p_self, p_opp, p_fumble_self, p_fumble_opp - def get_dodge_prob(self, player, position, allow_dodge_reroll=True, allow_team_reroll=False): + def get_dodge_prob(self, player: Player, position: Square, allow_dodge_reroll: bool=True, allow_team_reroll: bool=False) -> float: """ :param player: :param position: @@ -1845,15 +1888,16 @@ def get_dodge_prob(self, player, position, allow_dodge_reroll=True, allow_team_r return 1.0 ag_roll = Rules.agility_table[player.get_ag()] - self.get_dodge_modifiers(player, position) ag_roll = max(2, min(6, ag_roll)) - successful_outcomes = 6-(ag_roll-1) + successful_outcomes = 6 - (ag_roll - 1) p = successful_outcomes / 6.0 if allow_dodge_reroll and player.has_skill(Skill.DODGE) and not self.get_adjacent_opponents(player, down=False, skill=Skill.TACKLE): p += (1.0-p)*p elif allow_team_reroll and self.can_use_reroll(player.team): - p += (1.0-p)*p + p += (1.0 - p) * p return p - def get_catch_prob(self, player, accurate=False, interception=False, handoff=False, allow_catch_reroll=True, allow_team_reroll=False): + def get_catch_prob(self, player: Player, accurate: bool=False, interception: bool=False, handoff: bool=False, allow_catch_reroll: bool=True, + allow_team_reroll: bool=False) -> float: """ :param player: :param accurate: whether it is an accurate pass @@ -1865,18 +1909,20 @@ def get_catch_prob(self, player, accurate=False, interception=False, handoff=Fal """ ag_roll = Rules.agility_table[player.get_ag()] - self.get_catch_modifiers(player, accurate=accurate, interception=interception, handoff=handoff) ag_roll = max(2, min(6, ag_roll)) - successful_outcomes = 6-(ag_roll-1) + successful_outcomes = 6 - (ag_roll - 1) p = successful_outcomes / 6.0 if allow_catch_reroll and player.has_skill(Skill.CATCH): - p += (1.0-p)*p + p += (1.0 - p) * p elif allow_team_reroll and self.can_use_reroll(player.team): - p += (1.0-p)*p + p += (1.0 - p) * p return p - def get_dodge_prob_from(self, player, from_position, to_position, allow_dodge_reroll=False, allow_team_reroll=False): + def get_dodge_prob_from(self, player: Player, from_position: Square, to_position: Square, + allow_dodge_reroll: bool=False, allow_team_reroll: bool=False) -> float: """ :param player: - :param position: + :param from_position: + :param to_position :param allow_dodge_reroll: :param allow_team_reroll: :return: the probability of a successful dodge for player from from_position to to_position. @@ -1887,7 +1933,8 @@ def get_dodge_prob_from(self, player, from_position, to_position, allow_dodge_re self.move(player, orig_position) return p - def get_pickup_prob(self, player, position, allow_pickup_reroll=True, allow_team_reroll=False): + def get_pickup_prob(self, player: Player, position: Square, allow_pickup_reroll: bool=True, + allow_team_reroll: bool=False) -> float: """ :param player: :param position: the position of the ball @@ -1897,40 +1944,49 @@ def get_pickup_prob(self, player, position, allow_pickup_reroll=True, allow_team """ ag_roll = Rules.agility_table[player.get_ag()] - self.get_pickup_modifiers(player, position=position) ag_roll = max(2, min(6, ag_roll)) - successful_outcomes = 6-(ag_roll-1) + successful_outcomes = 6 - (ag_roll - 1) p = successful_outcomes / 6.0 if allow_pickup_reroll and player.has_skill(Skill.SURE_HANDS): - p += (1.0-p)*p + p += (1.0 - p) * p elif allow_team_reroll and self.can_use_reroll(player.team): - p += (1.0-p)*p + p += (1.0 - p) * p return p - def get_pass_prob(self, player, piece, position, allow_pickup_reroll=True, allow_team_reroll=False): + def get_pass_prob(self, player: Player, piece: Piece, position: Square, + allow_pass_reroll: bool = True, allow_team_reroll: bool = False) -> float: """ :param player: passer :param piece: piece to pass :param position: the position of the ball - :param allow_pickup_reroll: + :param allow_pass_reroll: :param allow_team_reroll: :return: the probability of a successful catch for player. """ - ag_roll = Rules.agility_table[player.get_ag()] - self.get_pass_modifiers(player, piece, position) + distance = self.get_pass_distance(from_position=player.position, to_position=position) + ttm = type(piece) != Ball + if ttm: + assert distance in {PassDistance.SHORT_PASS, PassDistance.QUICK_PASS}, "Throw team mate distance is too far" + + modifiers = self.get_pass_modifiers(player, pass_distance=distance, ttm=ttm) + ag_roll = Rules.agility_table[player.get_ag()] - modifiers ag_roll = max(2, min(6, ag_roll)) - successful_outcomes = 6-(ag_roll-1) + successful_outcomes = 6 - (ag_roll - 1) p = successful_outcomes / 6.0 - if allow_pickup_reroll and player.has_skill(Skill.Pass): - p += (1.0-p)*p + if allow_pass_reroll and player.has_skill(Skill.Pass): + p += (1.0 - p) * p elif allow_team_reroll and self.can_use_reroll(player.team): - p += (1.0-p)*p + p += (1.0 - p) * p return p - def num_block_dice_at(self, attacker, defender, position, blitz=False, dauntless_success=False): + def num_block_dice_at(self, attacker, defender, position: Square, blitz: bool=False, dauntless_success: bool=False): """ :param attacker: :param defender: + :param position: attackers position :param blitz: if it is a blitz :param dauntless_success: If a dauntless rolls was successful. - :return: The number of block dice used in a block between the attacker and defender if the attacker block at the given position. + :return: The number of block dice used in a block between the attacker and defender if the attacker block at + the given position. """ # Determine dice and favor @@ -1958,23 +2014,21 @@ def num_block_dice_at(self, attacker, defender, position, blitz=False, dauntless return 2 elif st_for == st_against: return 1 - elif st_for*2 < st_against: + elif st_for * 2 < st_against: return -3 elif st_for < st_against: return -2 - def num_assists_at(self, attacker, defender, position, foul: bool = False): + def num_assists_at(self, attacker: Player, defender: Player, position: Square, foul: bool = False) \ + -> Tuple[int, int]: ''' Return net assists for a block of player on opp_player when player has moved to position first. Required for calculating assists after moving in a Blitz action. - :param attacker: Player - :param defender: Player - :param position: Square - :param ignore_guard: bool :return: int - Net # of assists ''' - # Note that because blitzing/fouling player may have moved, calculating assists for is slightly different to against. + # Note that because blitzing/fouling player may have moved, + # calculating assists for is slightly different to against. # Assists against opp_assisters = self.get_adjacent_players(position, team=self.get_opp_team(attacker.team), down=False) n_assist_against: int = 0 @@ -1991,7 +2045,8 @@ def num_assists_at(self, attacker, defender, position, foul: bool = False): adjacent_to_assisters = self.get_adjacent_opponents(assister, down=False) found_adjacent = False for adjacent_to_assister in adjacent_to_assisters: - # Need to make sure we take into account the blocking/blitzing player may be in a different square than currently represented on the board. + # Need to make sure we take into account the blocking/blitzing player may be in a different square + # than currently represented on the board. if adjacent_to_assister.position == position or adjacent_to_assister.position == attacker.position or not adjacent_to_assister.can_assist(): continue else: @@ -2024,19 +2079,21 @@ def num_assists_at(self, attacker, defender, position, foul: bool = False): n_assists_for += 1 return n_assists_for, n_assist_against - def get_pass_distances(self, passer, piece, dump_off=False): + def get_pass_distances(self, passer: Player, piece: Piece, dump_off: bool = False) -> Tuple[List[Square], List[PassDistance]]: """ - :param passer: - :param weather: :return: two lists (squares, distances) indicating the PassDistance to each square that the passer can pass to. """ return self.get_pass_distances_at(passer, piece, passer.position, dump_off=dump_off) - def get_pass_distances_at(self, passer, piece, position, dump_off=False): + def get_pass_distances_at(self, passer: Player, piece: Piece, position: Square, dump_off: bool = False) \ + -> Tuple[List[Square], List[PassDistance]]: """ :param passer: - :param weather: - :return: two lists (squares, distances) indicating the PassDistance to each square that the passer can pass to if at the given position. + :param piece: + :param position: + :param dump_off: + :return: two lists (squares, distances) indicating the PassDistance to each square that the passer can pass to + if at the given position. """ squares = [] distances = [] @@ -2062,7 +2119,7 @@ def get_pass_distances_at(self, passer, piece, position, dump_off=False): distances.append(distance) return squares, distances - def get_pass_distance(self, from_position, to_position): + def get_pass_distance(self, from_position: Square, to_position: Square) -> PassDistance: """ :param from_position: :param to_position: @@ -2075,7 +2132,7 @@ def get_pass_distance(self, from_position, to_position): distance = Rules.pass_matrix[distance_y][distance_x] return PassDistance(distance) - def get_distance_to_endzone(self, player): + def get_distance_to_endzone(self, player: Player) -> int: """ :param player: :return: direct distance to the nearest opponent endzone tile. @@ -2084,9 +2141,9 @@ def get_distance_to_endzone(self, player): x = self.get_opp_endzone_x(player.team) return abs(x - player.position.x) - def get_opp_endzone_x(self, team): + def get_opp_endzone_x(self, team: Team) -> int: """ - :param player: + :param team: :return: the x-coordinate of the opponents endzone """ if team == self.state.home_team: @@ -2094,7 +2151,7 @@ def get_opp_endzone_x(self, team): else: return self.arena.width - 2 - def get_interceptors(self, position_from, position_to, team): + def get_interceptors(self, position_from: Square, position_to: Square, team: Team) -> List[Player]: """ Finds interceptors using the following rules: 1) Find line x from a to b @@ -2105,6 +2162,7 @@ def get_interceptors(self, position_from, position_to, team): 6) Determine players on squares :param position_from where the passer is :param position_to where the ball is passed to + :param team: team that can attempt interception """ # 1) Find line x from a to b @@ -2158,13 +2216,13 @@ def get_interceptors(self, position_from, position_to, team): return players - def get_available_actions(self): + def get_available_actions(self) -> List[ActionChoice]: """ :return: a list of available action choices in the current state. """ return self.state.available_actions - def clear_board(self): + def clear_board(self) -> None: """ Moves all players from the board to their respective reserves box. """ @@ -2173,25 +2231,26 @@ def clear_board(self): for player in self.get_players_on_pitch(self.state.away_team): self.pitch_to_reserves(player) - def get_active_player(self): + def get_active_player(self) -> Optional[Player]: """ :return: the current player to make a move if any, else None. """ return self.state.active_player - def get_procedure(self): + def get_procedure(self) -> Procedure: """ :return: The current procedure on the top of the stack. """ return self.state.stack.peek() - def get_weather(self): + def get_weather(self) -> WeatherType: """ :return: The current weather. """ return self.state.weather - def apply_casualty(self, player, inflictor, casualty, effect, roll): + def apply_casualty(self, player: Player, inflictor: Player, casualty, effect: CasualtyEffect, roll: DiceRoll) \ + -> None: """ Applies a casualty to a player and moves it to the dugout. :param player: the player to apply the casualty to. @@ -2219,7 +2278,7 @@ def apply_casualty(self, player, inflictor, casualty, effect, roll): if effect is not CasualtyEffect.MNG and effect is not CasualtyEffect.NONE: player.state.injuries_gained.append(effect) - def get_current_turn_proc(self): + def get_current_turn_proc(self) -> Optional[Procedure]: """ :return: the Turn procedure that is highest on the stack. """ @@ -2229,46 +2288,46 @@ def get_current_turn_proc(self): return self.state.stack.items[idx] return None - def get_tile(self, position): + def get_tile(self, position: Square) -> Tile: """ :param position: a Square on the board. :return: the tile type at the given position. """ return self.arena.board[position.y][position.x] - def get_adjacent_blood_lust_victims(self, player): + def get_adjacent_blood_lust_victims(self, player: Player) -> List[Square]: """ :param player: a player on the board. :return: return positions of adjecent players that can be bitten because of a failed blood lust roll. """ - return [p.position for p in self.get_adjacent_teammates(player) if not p.has_skill(Skill.BLOOD_LUST) ] + return [p.position for p in self.get_adjacent_teammates(player) if not p.has_skill(Skill.BLOOD_LUST)] - def get_hypno_targets(self, player): + def get_hypno_targets(self, player: Player) -> List[Square]: """ :param player: player on the board. :return: available targets for given player to hypnotize if player has Hypnotic Gaze skill """ - + if not player.has_skill(Skill.HYPNOTIC_GAZE): return [] return [o.position for o in self.get_adjacent_opponents(player, down=False) if o.has_tackle_zone()] - def get_hypno_modifier(self, player): + def get_hypno_modifier(self, player: Player) -> int: """ :param player: player on the board with hypnotic gaze skill. :return: modifier for player to hypnotize target. """ - return 1 - self.num_tackle_zones_in(player) + return 1 - self.num_tackle_zones_in(player) - def get_landing_modifiers(self, player): + def get_landing_modifiers(self, player: Player) -> int: """ :param player: Player attempting to land. """ return self.num_tackle_zones_in(player) - def get_handoff_actions(self, player): + def get_handoff_actions(self, player: Player) -> List[ActionChoice]: """ :param player: Hand-offing player :return: Available hand-off actions for the player. @@ -2279,7 +2338,7 @@ def get_handoff_actions(self, player): for player_to in self.get_adjacent_teammates(player): if player_to.can_catch(): hand_off_positions.append(player_to.position) - modifiers = self.get_catch_modifiers(player, player_to.position) + modifiers = self.get_catch_modifiers(player_to, handoff=True) target = Rules.agility_table[player.get_ag()] rolls.append([min(6, max(2, target - modifiers))]) @@ -2288,19 +2347,19 @@ def get_handoff_actions(self, player): positions=hand_off_positions, rolls=rolls)) return actions - def get_stand_up_actions(self, player): + def get_stand_up_actions(self, player: Player) -> List[ActionChoice]: rolls = [] if not player.state.up: moves = 0 if player.has_skill(Skill.JUMP_UP) else 3 if player.get_ma() < moves: - stand_up_roll = max(2, min(6, 4-self.get_stand_up_modifier(player))) + stand_up_roll = max(2, min(6, 4 - self.get_stand_up_modifier(player))) rolls.append([stand_up_roll]) else: rolls.append([]) return [ActionChoice(ActionType.STAND_UP, team=player.team, rolls=rolls)] return [] - def get_adjacent_move_actions(self, player): + def get_adjacent_move_actions(self, player: Player) -> List[ActionChoice]: quick_snap = self.is_quick_snap() actions = [] move_positions = [] @@ -2310,8 +2369,8 @@ def get_adjacent_move_actions(self, player): sprints = 3 if player.has_skill(Skill.SPRINT) else 2 gfi_roll = 3 if self.state.weather == WeatherType.BLIZZARD else 2 if (not quick_snap - and not player.state.taken_root - and player.state.moves + move_needed <= player.get_ma() + sprints) \ + and not player.state.taken_root + and player.state.moves + move_needed <= player.get_ma() + sprints) \ or (quick_snap and player.state.moves == 0): # Regular movement for square in self.get_adjacent_squares(player.position, occupied=False): @@ -2336,7 +2395,7 @@ def get_adjacent_move_actions(self, player): return actions - def get_leap_actions(self, player): + def get_leap_actions(self, player: Player) -> List[ActionChoice]: actions = [] if player.can_use_skill(Skill.LEAP) and not self.is_quick_snap(): sprints = 3 if player.has_skill(Skill.SPRINT) else 2 @@ -2365,7 +2424,7 @@ def get_leap_actions(self, player): positions=leap_positions, rolls=leap_rolls)) return actions - def get_foul_actions(self, player): + def get_foul_actions(self, player: Player) -> List[ActionChoice]: """ :param player: Fouling player :return: Available foul actions for the player. @@ -2385,7 +2444,7 @@ def get_foul_actions(self, player): positions=foul_positions, rolls=foul_rolls)) return actions - def get_pickup_teammate_actions(self, player): + def get_pickup_teammate_actions(self, player: Player) -> List[ActionChoice]: actions = [] teammates = self.get_adjacent_teammates(player, down=False, skill=Skill.RIGHT_STUFF) if teammates: @@ -2396,7 +2455,7 @@ def get_pickup_teammate_actions(self, player): positions=positions, rolls=d6_rolls)) return actions - def get_block_actions(self, player, blitz=False): + def get_block_actions(self, player: Player, blitz = False) -> List[ActionChoice]: if player.state.has_blocked: return [] @@ -2450,7 +2509,7 @@ def get_block_actions(self, player, blitz=False): return actions - def get_pass_actions(self, player, piece, dump_off=False): + def get_pass_actions(self, player: Player, piece, dump_off = False) -> List[ActionChoice]: actions = [] if piece: ttm = type(piece) == Player @@ -2485,7 +2544,7 @@ def get_pass_actions(self, player, piece, dump_off=False): actions.append(ActionChoice(ActionType.DONT_USE_SKILL, team=player.team, skill=Skill.DUMP_OFF)) return actions - def get_hypnotic_gaze_actions(self, player): + def get_hypnotic_gaze_actions(self, player: Player) -> List[ActionChoice]: actions = [] if player.has_skill(Skill.HYPNOTIC_GAZE) and player.state.up: @@ -2502,7 +2561,7 @@ def get_hypnotic_gaze_actions(self, player): rolls=rolls)) return actions - def purge_stack_until(self, proc_class, inclusive=False): + def purge_stack_until(self, proc_class, inclusive = False) -> None: assert proc_class in [proc.__class__ for proc in self.state.stack.items] while not isinstance(self.state.stack.peek(), proc_class): self.state.stack.pop() @@ -2510,13 +2569,13 @@ def purge_stack_until(self, proc_class, inclusive=False): self.state.stack.pop() assert not self.state.stack.is_empty() - def get_proc(self, proc_type): + def get_proc(self, proc_type) -> Optional[Procedure]: for proc in self.state.stack.items: if type(proc) == proc_type: return proc return None - def get_stand_up_modifier(self, player): + def get_stand_up_modifier(self, player: Player) -> int: """ :param player: player on the board with MA < 3. :return: modifier for player to stand up. diff --git a/botbowl/core/model.py b/botbowl/core/model.py index e6e8ff06..7ac46265 100755 --- a/botbowl/core/model.py +++ b/botbowl/core/model.py @@ -5,15 +5,18 @@ ========================== This module contains most of the model classes. """ +from abc import ABC, abstractmethod +from copy import copy +from typing import List, Optional, Set, Dict -from copy import copy, deepcopy import numpy as np import uuid import time -import json import pickle from math import sqrt -from botbowl.core.util import * +import os + +from botbowl.core.util import get_data_path, Stack, compare_iterable from botbowl.core.table import * from botbowl.core.forward_model import Immutable, Reversible, CallableStep @@ -59,7 +62,8 @@ def record_action(self, action): def dump(self, game): replay_id = game.game_id self.reports = game.state.reports - name = self.steps[0].game['home_agent']['name'] + "_VS_" + self.steps[0].game['away_agent']['name'] + "_" + str(replay_id) + name = self.steps[0].game['home_agent']['name'] + "_VS_" + self.steps[0].game['away_agent']['name'] + "_" + str( + replay_id) directory = get_data_path('replays') if not os.path.exists(directory): os.mkdir(directory) @@ -73,7 +77,7 @@ def next(self): return None self.idx += 1 while self.idx not in self.steps: - #print(self.actions[self.idx]) + # print(self.actions[self.idx]) self.idx += 1 return self.steps[self.idx] @@ -82,7 +86,7 @@ def prev(self): return None self.idx -= 1 while self.idx not in self.steps: - #print(self.actions[self.idx]) + # print(self.actions[self.idx]) self.idx -= 1 return self.steps[self.idx] @@ -129,6 +133,25 @@ def to_json(self): class Configuration: + name: str + arena: Optional['TwoPlayerArena'] + ruleset: Optional['RuleSet'] + roster_size: int + pitch_max: int + pitch_min: int + scrimmage_min: int + wing_max: int + rounds: int + kick_off_table: bool + fast_mode: bool + debug_mode: bool + competition_mode: bool + kick_scatter_distance: str + offensive_formations: List['Formation'] + defensive_formations: List['Formation'] + time_limits: Optional[TimeLimits] + pathfinding_enabled: bool + pathfinding_directly_to_adjacent: bool def __init__(self): self.name = "Default" @@ -153,6 +176,26 @@ def __init__(self): class PlayerState(Reversible): + up: bool + in_air: bool + used: bool + spp_earned: int + moves: int + stunned: bool + bone_headed: bool + hypnotized: bool + really_stupid: bool + heated: bool + knocked_out: bool + ejected: bool + injuries_gained: List + wild_animal: bool + taken_root: bool + blood_lust: bool + picked_up: bool + used_skills: Set[Skill] + squares_moved: List['Square'] + has_blocked: bool def __init__(self): super().__init__() @@ -171,7 +214,8 @@ def __init__(self): self.injuries_gained = [] self.wild_animal = False self.taken_root = False - self.blood_lust = False + self.blood_lust = False + self.picked_up = False self.used_skills = set() self.squares_moved = [] self.has_blocked = False @@ -193,7 +237,7 @@ def to_json(self): 'injuries_gained': [injury.name for injury in self.injuries_gained], 'squares_moved': [square.to_json() for square in self.squares_moved], 'wild_animal': self.wild_animal, - 'taken_root': self.taken_root, + 'taken_root': self.taken_root, 'blood_lust': self.blood_lust, 'has_blocked': self.has_blocked } @@ -220,8 +264,20 @@ def reset_turn(self): self.used_skills.clear() self.squares_moved.clear() + always_show_attr = ['up'] + show_if_true_attr = ['used', 'stunned', 'bone_headed', 'hypnotized', 'really_stupid', 'heated', 'knocked_out', + 'ejected', 'wild_animal', 'taken_root', 'blood_lust', 'picked_up', 'has_blocked'] + + def __repr__(self): + states_to_show = [f"{attr}={getattr(self, attr)}" for attr in PlayerState.always_show_attr] + \ + [f"{attr}=True" for attr in PlayerState.show_if_true_attr if getattr(self, attr)] + return f'PlayerState({", ".join(states_to_show)})' + class Agent: + name: str + human: bool + agent_id: str def __init__(self, name, human=False, agent_id=None): if agent_id is not None: @@ -257,6 +313,20 @@ def end_game(self, game): class TeamState(Reversible): + bribes: int + babes: int + apothecaries: int + wizard_available: bool + masterchef: bool + score: int + turn: int + rerolls_start: int + rerolls: int + ass_coaches: int + cheerleaders: int + fame: int + reroll_used: bool + time_violation: int def __init__(self, team): super().__init__() @@ -301,6 +371,12 @@ def use_reroll(self): class Clock: + seconds: int + started_at: float + paused_at: float + paused_seconds: int + is_primary: bool + team: 'Team' def __init__(self, team, seconds, is_primary=False): self.seconds = seconds @@ -310,20 +386,20 @@ def __init__(self, team, seconds, is_primary=False): self.is_primary = is_primary self.team = team - def is_running(self): + def is_running(self) -> bool: return self.paused_at is None - def pause(self): + def pause(self) -> None: assert self.paused_at is None self.paused_at = time.time() - def resume(self): + def resume(self) -> None : assert self.paused_at is not None now = time.time() self.paused_seconds += now - self.paused_at self.paused_at = None - def get_running_time(self): + def get_running_time(self) -> float: now = time.time() if self.is_running(): return now - self.started_at - self.paused_seconds @@ -355,6 +431,35 @@ def to_json(self): class GameState(Reversible): + stack: Stack + reports: List['Outcome'] + half: int + round: int + coin_toss_winner: Optional['Team'] + kicking_first_half: Optional['Team'] + receiving_first_half: Optional['Team'] + kicking_this_drive: Optional['Team'] + receiving_this_drive: Optional['Team'] + current_team: Optional['Team'] + teams: List['Team'] + home_team: 'Team' + away_team: 'Team' + team_by_id: Dict[str, 'Team'] + player_by_id: Dict[str, 'Player'] + team_by_player_id: Dict[str, 'Team'] + + pitch: 'Pitch' + dugouts: Dict[str, 'Dugout'] + weather: WeatherType + gentle_gust: bool + turn_order: List['Team'] + spectators: int + active_player: Optional['Player'] + game_over: bool + available_actions: List['ActionChoice'] + clocks: List[Clock] + rerolled_procs: Set['Procedure'] + player_action_type: Optional[ActionType] def __init__(self, game, home_team, away_team): super().__init__(ignored_keys=["clocks"]) @@ -431,22 +536,19 @@ def to_json(self, ignore_reports=False, ignore_clocks=False): class Pitch(Reversible): - - range = [-1, 0, 1] + balls: List['Ball'] + bomb: Optional['Bomb'] + board: List[List[Optional['Player']]] + squares: List[List['Square']] + height: int + width: int def __init__(self, width, height): super().__init__(ignored_keys=["board", "squares"]) self.balls = [] self.bomb = None - #self.board = [] - self.board = np.full((height, width), None) - self.squares = [] - for y in range(height): - # self.board.append([]) - self.squares.append([]) - for x in range(width): - # self.board[y].append(None) - self.squares[y].append(Square(x, y)) + self.board = [[None for x in range(width)] for y in range(height)] + self.squares = [[Square(x, y) for x in range(width)] for y in range(height)] self.height = len(self.board) self.width = len(self.board[0]) @@ -465,8 +567,18 @@ def to_json(self): class ActionChoice(Immutable): - - def __init__(self, action_type, team, positions=None, players=None, rolls=None, block_dice=None, skill=None, paths=None, disabled=False): + action_type: ActionType + positions: List['Square'] + players: List['Player'] + team: 'Team' + rolls: List[int] + block_dice: List + disabled: bool + skill: Optional[Skill] + paths: List['Path'] + + def __init__(self, action_type, team, positions=None, players=None, rolls=None, block_dice=None, skill=None, + paths=None, disabled=False): self.action_type = action_type self.positions = [] if positions is None else positions self.players = [] if players is None else players @@ -477,6 +589,10 @@ def __init__(self, action_type, team, positions=None, players=None, rolls=None, self.skill = skill self.paths = [] if paths is None else paths + def __repr__(self): + return f"ActionChoice({self.action_type}, len(positions)={len(self.positions)}, " \ + f"len(players)={len(self.players)}, len(paths)={len(self.paths)})" + def to_json(self): return { 'action_type': self.action_type.name, @@ -501,6 +617,9 @@ def to_json(self): class Action(Reversible): + action_type: ActionType + position: Optional['Square'] + player: Optional['Player'] def __init__(self, action_type, position=None, player=None): super().__init__() @@ -508,6 +627,11 @@ def __init__(self, action_type, position=None, player=None): self.position = position self.player = player + def __repr__(self): + pos_str = f", position={self.position}" if self.position is not None else "" + player_str = f", player={self.player}" if self.player is not None else "" + return f"Action({self.action_type}{pos_str}{player_str})" + def to_json(self): return { 'action_type': self.action_type.name, @@ -517,7 +641,6 @@ def to_json(self): class TwoPlayerArena: - home_tiles = [Tile.HOME, Tile.HOME_TOUCHDOWN, Tile.HOME_WING_LEFT, Tile.HOME_WING_RIGHT, Tile.HOME_SCRIMMAGE] away_tiles = [Tile.AWAY, Tile.AWAY_TOUCHDOWN, Tile.AWAY_WING_LEFT, Tile.AWAY_WING_RIGHT, Tile.AWAY_SCRIMMAGE] scrimmage_tiles = [Tile.HOME_SCRIMMAGE, Tile.AWAY_SCRIMMAGE] @@ -554,15 +677,31 @@ def to_json(self): return self.json -class Die: +class Die(ABC): + value: str + @abstractmethod def get_value(self): - Exception("Method not implemented") + pass + @abstractmethod + def to_json(self): + pass -class DiceRoll(Reversible): - def __init__(self, dice, modifiers=0, target=None, d68=False, roll_type=RollType.AGILITY_ROLL, target_higher=True, target_lower=False, highest_succeed=True, lowest_fail=True): +class DiceRoll(Reversible): + dice: List[Die] + modifiers: int + target: Optional[int] + d68: bool + roll_type: RollType + target_higher: int + target_lower: int + highest_succeed: bool + lowest_fail: bool + + def __init__(self, dice, modifiers=0, target=None, d68=False, roll_type=RollType.AGILITY_ROLL, target_higher=True, + target_lower=False, highest_succeed=True, lowest_fail=True): super().__init__() self.dice = dice self.sum = 0 @@ -582,6 +721,9 @@ def __init__(self, dice, modifiers=0, target=None, d68=False, roll_type=RollType else: self.sum += d.get_value() + def __repr__(self): + return f"DiceRoll(dice={self.dice})" + def to_json(self): dice = [] for die in self.dice: @@ -602,7 +744,7 @@ def to_json(self): def modified_target(self): if self.target is not None: - return max(1*len(self.dice), min(6*len(self.dice), self.target - self.modifiers)) + return max(1 * len(self.dice), min(6 * len(self.dice), self.target - self.modifiers)) return None def contains(self, value): @@ -621,10 +763,10 @@ def get_result(self): return self.sum + self.modifiers def is_d6_success(self): - if self.lowest_fail and self.sum == 1*len(self.dice): + if self.lowest_fail and self.sum == 1 * len(self.dice): return False - if self.highest_succeed and self.sum == 6*len(self.dice): + if self.highest_succeed and self.sum == 6 * len(self.dice): return True if self.target_higher: @@ -645,7 +787,6 @@ def same(self): class D3(Die): - FixedRolls = [] @staticmethod @@ -661,6 +802,9 @@ def __init__(self, rnd): else: self.value = rnd.randint(1, 4) + def __repr__(self): + return f"D3({self.value})" + def get_value(self): return self.value @@ -672,21 +816,20 @@ def to_json(self): class D6(Die, Immutable): - FixedRolls = [] TWO_PROBS = { - 2: (1/6 * 1/6), - 3: 2 * (1/6 * 1/6), - 4: 3 * (1/6 * 1/6), - 5: 4 * (1/6 * 1/6), - 6: 5 * (1/6 * 1/6), - 7: 6 * (1/6 * 1/6), - 8: 5 * (1/6 * 1/6), - 9: 4 * (1/6 * 1/6), - 10: 3 * (1/6 * 1/6), - 11: 2 * (1/6 * 1/6), - 12: 1 * (1/6 * 1/6) + 2: (1 / 6 * 1 / 6), + 3: 2 * (1 / 6 * 1 / 6), + 4: 3 * (1 / 6 * 1 / 6), + 5: 4 * (1 / 6 * 1 / 6), + 6: 5 * (1 / 6 * 1 / 6), + 7: 6 * (1 / 6 * 1 / 6), + 8: 5 * (1 / 6 * 1 / 6), + 9: 4 * (1 / 6 * 1 / 6), + 10: 3 * (1 / 6 * 1 / 6), + 11: 2 * (1 / 6 * 1 / 6), + 12: 1 * (1 / 6 * 1 / 6) } @staticmethod @@ -702,6 +845,9 @@ def __init__(self, rnd): else: self.value = rnd.randint(1, 7) + def __repr__(self): + return f"D6({self.value})" + def get_value(self): return self.value @@ -713,7 +859,6 @@ def to_json(self): class D8(Die, Immutable): - FixedRolls = [] @staticmethod @@ -729,6 +874,9 @@ def __init__(self, rnd): else: self.value = rnd.randint(1, 9) + def __repr__(self): + return f"D8({self.value})" + def get_value(self): return self.value @@ -740,7 +888,7 @@ def to_json(self): class BBDie(Die, Immutable): - + value: BBDieResult FixedRolls = [] @staticmethod @@ -763,7 +911,10 @@ def __init__(self, rnd): r = 3 self.value = BBDieResult(r) - def get_value(self): + def __repr__(self): + return f"BBDie({self.value})" + + def get_value(self) -> BBDieResult: return self.value def to_json(self): @@ -812,6 +963,7 @@ def __init__(self, name, races, ma, st, ag, av, skills, cost, feeder, n_skill_se class Piece: + position: 'Square' def __init__(self, position=None): self.position = position @@ -831,8 +983,8 @@ def move(self, x, y): # This is unfortunately way slower than below, but Square is Immutable self.position = Square(self.position.x + x, self.position.y + y) - #self.position.x += x - #self.position.y += y + # self.position.x += x + # self.position.y += y def move_to(self, position): self.position = position @@ -869,6 +1021,12 @@ def move_to(self, position): super().move_to(position) + def __repr__(self): + return f"Ball(position={self.position if self.position is not None else 'None'}, " \ + f"on_ground={self.on_ground}, " \ + f"is_carried={self.is_carried})" + + class Bomb(Catchable): def __init__(self, position, on_ground=True, is_carried=False): @@ -876,9 +1034,18 @@ def __init__(self, position, on_ground=True, is_carried=False): class Player(Piece, Reversible): - - def __init__(self, player_id, role, name, nr, team, extra_skills=None, extra_ma=0, extra_st=0, extra_ag=0, extra_av=0, - niggling_injuries=0, mng=False, spp=0, injuries=None, position=None): + player_id: str + role: Role + nr: int + team: 'Team' + extra_ma: int + extra_st: int + extra_ag: int + extra_av: int + injuries: List + + def __init__(self, player_id, role, name, nr, team, extra_skills=None, extra_ma=0, extra_st=0, extra_ag=0, + extra_av=0, niggling_injuries=0, mng=False, spp=0, injuries=None, position=None): Reversible.__init__(self, ignored_keys=["position", "role"]) super().__init__(position) self.player_id = player_id @@ -919,14 +1086,16 @@ def to_json(self): } def get_ag(self): - ag = self.role.ag + self.extra_ag - self.injuries.count(CasualtyEffect.AG) - self.state.injuries_gained.count(CasualtyEffect.AG) + ag = self.role.ag + self.extra_ag - self.injuries.count(CasualtyEffect.AG) - self.state.injuries_gained.count( + CasualtyEffect.AG) ag = max(self.role.ag - 2, ag) ag = max(1, ag) ag = min(10, ag) return ag def get_st(self): - st = self.role.st + self.extra_st - self.injuries.count(CasualtyEffect.ST) - self.state.injuries_gained.count(CasualtyEffect.ST) + st = self.role.st + self.extra_st - self.injuries.count(CasualtyEffect.ST) - self.state.injuries_gained.count( + CasualtyEffect.ST) st = max(self.role.st - 2, st) st = max(1, st) st = min(10, st) @@ -936,14 +1105,16 @@ def get_ma(self): if self.state.taken_root: return 0 else: - ma = self.role.ma + self.extra_ma - self.injuries.count(CasualtyEffect.MA) - self.state.injuries_gained.count(CasualtyEffect.MA) + ma = self.role.ma + self.extra_ma - self.injuries.count( + CasualtyEffect.MA) - self.state.injuries_gained.count(CasualtyEffect.MA) ma = max(self.role.ma - 2, ma) ma = max(1, ma) ma = min(10, ma) return ma def get_av(self): - av = self.role.av + self.extra_av - - self.injuries.count(CasualtyEffect.AV) - self.state.injuries_gained.count(CasualtyEffect.AV) + av = self.role.av + self.extra_av - - self.injuries.count(CasualtyEffect.AV) - self.state.injuries_gained.count( + CasualtyEffect.AV) av = max(self.role.av - 2, av) av = max(1, av) av = min(10, av) @@ -1006,6 +1177,9 @@ def place_prone(self): self.state.up = False self.state.taken_root = False + def __repr__(self): + return f"Player(position={self.position if self.position is not None else 'None'}, {self.role.name}, state={self.state})" + class Square(Immutable): @@ -1029,7 +1203,6 @@ def y(self): def y(self, y): raise AttributeError("Squares is immutable, how dare you?!") - def to_json(self): return { 'x': self.x, @@ -1048,13 +1221,16 @@ def distance(self, other, manhattan=False, flight=False): if manhattan: return abs(other.x - self.x) + abs(other.y - self.y) elif flight: - return sqrt((other.x - self.x)**2 + (other.y - self.y)**2) + return sqrt((other.x - self.x) ** 2 + (other.y - self.y) ** 2) else: return max(abs(other.x - self.x), abs(other.y - self.y)) def is_adjacent(self, other, manhattan=False): return self.distance(other, manhattan) == 1 + def __repr__(self): + return f"Square({self.x}, {self.y})" + class Race: @@ -1111,8 +1287,17 @@ def __hash__(self): class Outcome(Immutable): - - def __init__(self, outcome_type, position=None, player=None, opp_player=None, rolls=None, team=None, n=0, skill=None): + outcome_type: OutcomeType + position: Optional[Square] + player: Optional[Player] + opp_player: Optional[Player] + rolls: List[DiceRoll] + team: Optional[Team] + n: int + skill: Optional[Skill] + + def __init__(self, outcome_type, position=None, player=None, opp_player=None, rolls=None, team=None, n=0, + skill=None): self.outcome_type = outcome_type self.position = position self.player = player @@ -1122,6 +1307,10 @@ def __init__(self, outcome_type, position=None, player=None, opp_player=None, ro self.n = n self.skill = skill + def __repr__(self): + pos_str = "" if self.position is None else f", position={self.position}" + return f"Outcome({self.outcome_type}{pos_str}, rolls={self.rolls})" + def to_json(self): rolls = [] for roll in self.rolls: @@ -1149,7 +1338,8 @@ def __init__(self, name, cost, max_num, reduced=0): class RuleSet: - def __init__(self, name, races=[], star_players=[], inducements=[], spp_actions={}, spp_levels={}, improvements={}, se_start=0, se_interval=0, se_pace=0): + def __init__(self, name, races=[], star_players=[], inducements=[], spp_actions={}, spp_levels={}, improvements={}, + se_start=0, se_interval=0, se_pace=0): self.name = name self.races = races self.star_players = star_players @@ -1179,7 +1369,8 @@ def __init__(self, name, formation): def _get_player(self, players, t): if t == 'S': - idx = np.argmax([player.get_st() + (0.5 if player.has_skill(Skill.BLOCK) else 0) - (0.5 if player.has_skill(Skill.SURE_HANDS) else 0) for player in players]) + idx = np.argmax([player.get_st() + (0.5 if player.has_skill(Skill.BLOCK) else 0) - ( + 0.5 if player.has_skill(Skill.SURE_HANDS) else 0) for player in players]) return players[idx] if t == 'm': idx = np.argmax([player.get_ma() for player in players]) @@ -1209,7 +1400,8 @@ def _get_player(self, players, t): idx = np.argmin([len(player.get_skills()) for player in players]) return players[idx] if t == 'x': - idx = np.argmax([1 if player.has_skill(Skill.BLOCK) else (0 if player.has_skill(Skill.PASS) or player.has_skill(Skill.CATCH) else 0.5) for player in players]) + idx = np.argmax([1 if player.has_skill(Skill.BLOCK) else ( + 0 if player.has_skill(Skill.PASS) or player.has_skill(Skill.CATCH) else 0.5) for player in players]) return players[idx] return players[0] @@ -1233,7 +1425,7 @@ def actions(self, game, team): for y in range(len(self.formation)): if len(players) == 0: return actions - x = len(self.formation[0])-1 + x = len(self.formation[0]) - 1 tp = self.formation[y][x] if tp == '-' or tp != t: continue @@ -1274,9 +1466,9 @@ def compare(self, other, path): if self.name != other.name: diff.append(f"{path}.name: {self.name} _NotEqual_ {other.name}") - formations_equal = all([(self_a == other_a).all() for self_a, other_a in zip(self.formation, other.formation)]) + formations_equal = all((self_a == other_a).all() for self_a, other_a in zip(self.formation, other.formation)) if not formations_equal: diff.append(f"{path}.formation: _NotEqual_ ") - return diff \ No newline at end of file + return diff diff --git a/botbowl/core/pathfinding/cython_pathfinding.pyx b/botbowl/core/pathfinding/cython_pathfinding.pyx index 527cc65b..4d07913f 100644 --- a/botbowl/core/pathfinding/cython_pathfinding.pyx +++ b/botbowl/core/pathfinding/cython_pathfinding.pyx @@ -154,7 +154,7 @@ cdef class Pathfinder: def __init__(self, game, player, trr=False, directly_to_adjacent=False, can_block=False, can_handoff=False, can_foul=False): - self.players_on_pitch = game.state.pitch.board.flatten() + self.players_on_pitch = game.state.pitch.board self.game = game self.pitch_width = self.game.arena.width - 1 @@ -312,7 +312,7 @@ cdef class Pathfinder: to_pos = node.get().position + direction if not (1 <= to_pos.x < self.pitch_width and 1 <= to_pos.y < self.pitch_height): return NodePtr() - player_at = self.players_on_pitch[to_pos.y * 28 + to_pos.x] + player_at = self.players_on_pitch[to_pos.y][to_pos.x] if player_at is not None: if node.get().can_handoff and player_at.team == self.player.team and player_at.can_catch(): @@ -382,7 +382,7 @@ cdef class Pathfinder: object player_at best_node = self.nodes[to_pos.y][to_pos.x] best_before = self.locked_nodes[to_pos.y][to_pos.x] - player_at = self.players_on_pitch[to_pos.y * 28 + to_pos.x] + player_at = self.players_on_pitch[to_pos.y][to_pos.x] next_node = NodePtr( new Node(node, to_pos, 0, 0, node.get().euclidean_distance)) target = self._get_handoff_target(player_at) diff --git a/botbowl/core/pathfinding/python_pathfinding.py b/botbowl/core/pathfinding/python_pathfinding.py index ff37d465..4e933ec1 100644 --- a/botbowl/core/pathfinding/python_pathfinding.py +++ b/botbowl/core/pathfinding/python_pathfinding.py @@ -8,15 +8,15 @@ from botbowl.core.table import Rules from botbowl.core.model import Square -from botbowl.core.forward_model import Reversible -from botbowl.core.util import compare_object +from botbowl.core.forward_model import treat_as_immutable from botbowl.core.table import Skill, WeatherType import copy import numpy as np from queue import PriorityQueue -class Path(Reversible): +@treat_as_immutable +class Path: def __init__(self, node: 'Node'): super().__init__() @@ -49,9 +49,6 @@ def get_last_step(self) -> 'Square': def is_empty(self) -> bool: return len(self) == 0 - def compare(self, other, path=""): - return compare_object(self, other, path) - def collect_path(self): steps = [] rolls = [] @@ -64,6 +61,14 @@ def collect_path(self): self._steps = list(reversed(steps)) self._rolls = list(reversed(rolls)) + def __eq__(self, other): + return self.prob == other.prob and \ + self.steps == other.steps and \ + self.rolls == other.rolls and \ + self.block_dice == other.block_dice and \ + self.handoff_roll == other.handoff_roll and \ + self.foul_roll == other.foul_roll + class Node: diff --git a/botbowl/core/procedure.py b/botbowl/core/procedure.py index 68d902d6..8ff92af0 100755 --- a/botbowl/core/procedure.py +++ b/botbowl/core/procedure.py @@ -7,18 +7,22 @@ responsible for an isolated part of the game. Procedures are added to a stack, and the top-most procedure must finish before other procedures are run. Procedures can add other procedures to the stack simply by instantiating procedures. """ -from abc import abstractmethod, ABCMeta -from botbowl.core.pathfinding import Pathfinder +from botbowl.core.pathfinding import Pathfinder +from botbowl.core.util import compare_object from botbowl.core.model import * from botbowl.core.table import * + import time +from abc import abstractmethod, ABCMeta class Procedure(Reversible): - def __init__(self, game, context=None, ignored_keys=[]): + def __init__(self, game, context=None, ignored_keys=None): + ignored_keys = [] if ignored_keys is None else ignored_keys + super().__init__(ignored_keys=["game"]+ignored_keys) self.game = game self.context = context @@ -55,6 +59,9 @@ def available_actions(self): def compare(self, other, path=""): return compare_object(self, other, path, ignored_keys={"game"}, ignored_types={Procedure}) + def __repr__(self): + return f"{self.__class__.__name__}(done={self.done}, started={self.started})" + class Regeneration(Procedure): diff --git a/botbowl/core/util.py b/botbowl/core/util.py index eeac7aae..e2ea53d1 100755 --- a/botbowl/core/util.py +++ b/botbowl/core/util.py @@ -9,6 +9,8 @@ import os from collections.abc import Iterable from copy import copy +from typing import Sized + import botbowl from botbowl.core.forward_model import Reversible from botbowl.core.model import * @@ -121,7 +123,7 @@ def compare_iterable(s1, s2, path=""): elif hasattr(s1, "compare"): diff.extend(s1.compare(s2, f"{path}")) - elif isinstance(s1, Iterable) and len(s1) != len(s2): + elif isinstance(s1, Sized) and len(s1) != len(s2): diff.append(f"{path}: __len__: '{len(s1)}' _notEqual_ '{len(s2)}'") elif isinstance(s1, dict):