Pokemon Playing AI training tools built with a modern, async, type-hinting stack
from .battle import Battle
from .choices import (
    AnyChoice,
    BattleChoice,
    DefaultChoice,
    ForceSwitchChoice,
    ItemChoice,
    MoveChoice,
    MoveDecisionChoice,
    PassChoice,
    QuitChoice,
    ResignChoice,
    SwitchChoice,
    TeamChoice,
    TeamOrderChoice,
)
from .pokemon import BattleAbility, BattleItem, BattleMove, BattlePokemon, BoostBlock, StatBlock
from .state import BattleState

pokesage/battle/__init__.py
trevorWieland/pokesage
from typing import List, Literal, Optional

from poketypes.dex import DexGen
from pydantic import BaseModel, Field

from .choices import AnyChoice, BattleChoice
from .state import BattleState


class Battle(BaseModel):
    """
    Representing a battle in its entirety.
    One Battle contains multiple BattleStates in an ordered list.
    Each BattleState should represent the battlefield itself at the point of a decision.

    This class should (other than the BattleState list) contain no state-specific details.

    Things like weather, field conditions, teams, etc should be stored *only* in the BattleState.
    Things like player names, victory information, administrative data should be stored here.

    Built to support singles, doubles, or triples gametypes
    """

    battle_states: List[BattleState] = Field(
        [],
        description="The ordered list of battle states as they were before a decision by the player",
    )
    battle_actions: Optional[AnyChoice] = Field(
        None,
        description="The list of decisions the player made at each corresponding BattleState in battle_states",
    )

    player_name: Optional[str] = Field(None, description="The player's name")
    player_rating: Optional[int] = Field(None, description="The rating of the player")
    player_id: Optional[str] = Field(None, description="A unique identifier for the player")

    opponent_name: Optional[str] = Field(None, description="The opponent's name")
    opponent_rating: Optional[int] = Field(None, description="The rating of the opponent")
    opponent_id: Optional[str] = Field(None, description="A unique identifier for the opponent")

    rated: bool = Field(False, description="Whether this match is rated or not")

    gametype: Optional[Literal["singles", "doubles", "triples"]] = Field(
        None, description="The gametype of this battle"
    )
    format: Optional[str] = Field(None, description="The format of this match")
    gen: Optional[DexGen.ValueType] = Field(None, description="The generation this battle format corresponds to")
    turn: int = Field(None, description="The current / total number of turns in this battle")

    player_victory: Optional[bool] = Field(None, description="Whether this resulted in the player's victory")

    player_team_size: Optional[int] = Field(None, description="The integer team size of the player")
    opponent_team_size: Optional[int] = Field(None, description="The integer team size of the opponent")

    def slot_length(self) -> int:
        """
        Helper function to quickly tell how many slots there are for this gametype
        """

        if self.gametype is None:
            raise RuntimeError("slot_length was called but no gametype was set!")

        if self.gametype == "singles":
            return 1
        if self.gametype == "doubles":
            return 2
        if self.gametype == "triples":
            return 3

        raise RuntimeError(f"Unknown gametype: {self.gametype}")

pokesage/battle/battle.py
trevorWieland/pokesage
from beartype.typing import List, Optional, Union

from pydantic import BaseModel, Field


class TeamChoice(BaseModel):
    """
    Data Model representing a team order decision

    Contains information about move, target, and any extra move actions
    """

    team_order: List[int] = Field(
        ...,
        description="A list of integer pokemon slots in the order you want them.",
    )

    def to_showdown(self) -> str:
        """
        Converts the choice to a showdown-formatted decision
        """

        if len(self.team_order) >= 10:
            return "team " + ",".join([str(s) for s in self.team_order])
        else:
            return "team " + "".join([str(s) for s in self.team_order])


class MoveChoice(BaseModel):
    """
    Data Model representing a single move option for a pokemon.

    Contains information about move, target, and any extra move actions
    """

    move_number: int = Field(..., description="The 1-indexed position of the move")

    target_number: Optional[int] = Field(None, description="The 1-indexed target of the move, if needed.")

    tera: bool = Field(False, description="Whether this choice involves teratyping first")
    mega: bool = Field(False, description="Whether this choice involves mega-evolving first")
    dyna: bool = Field(False, description="Whether this choice involves dynamaxing first")
    zmove: bool = Field(False, description="Whether this choice is using the zmove form of the move")

    def to_showdown(self) -> str:
        """
        Converts the choice to a showdown-formatted decision
        """

        choice = f"move {self.move_number}"

        if self.target_number is not None:
            choice += f" {self.target_number}"
        if self.tera:
            choice += " terastallize"
        if self.mega:
            choice += " mega"
        if self.dyna:
            choice += " dynamax"
        if self.zmove:
            choice += " zmove"

        return choice.strip()


class SwitchChoice(BaseModel):
    """
    Data Model representing a switch-out option for a pokemon.

    Contains information about target switch slot number
    """

    slot: int = Field(..., description="The slot to switch to")

    def to_showdown(self) -> str:
        """
        Converts the choice to a showdown-formatted decision
        """

        choice = f"switch {self.slot}"

        return choice.strip()


class ItemChoice(BaseModel):
    """
    Data Model representing a item-use option for a pokemon.

    Contains information about target and which item to use.

    Unused for showdown. Reserved for future emulator interaction
    """

    def to_showdown(self) -> str:
        """
        Converts the choice to a showdown-formatted decision
        """

        raise NotImplementedError


class PassChoice(BaseModel):
    """
    Data Model representing that this slot doesn't need to do anything, and thus passes.

    Contains no extra information, and while we could design our system to not need this,
     it makes action shapes consistent to require it
    """


class ResignChoice(BaseModel):
    """
    Data Model representing the option to resign.

    Used for type-hinting consistency, and for connectors to gracefully terminate when a sage-class fails for just one battle
    """


class QuitChoice(BaseModel):
    """
    Data Model representing the option to resign all games and close connection.

    Used for type-hinting consistency, and for connectors to gracefully terminate when a sage-class fails irrecoverably
    """


class DefaultChoice(BaseModel):
    """
    Data Model representing a single move option for a pokemon.

    Contains information about move, target, and any extra move actions
    """

    def to_showdown(self) -> str:
        """
        Converts the choice to a showdown-formatted decision
        """

        return "default"


AnyChoice = Union[
    List[Union[MoveChoice, SwitchChoice, ItemChoice, PassChoice]],
    TeamChoice,
    ResignChoice,
    DefaultChoice,
    QuitChoice,
    None,
]
BattleChoice = Union[
    List[
        Union[
            List[Union[MoveChoice, SwitchChoice, ItemChoice],],
            PassChoice,
        ]
    ],
    TeamChoice,
]

TeamOrderChoice = Union[TeamChoice, ResignChoice, DefaultChoice, QuitChoice]
MoveDecisionChoice = Union[
    List[Union[MoveChoice, SwitchChoice, ItemChoice, PassChoice]],
    ResignChoice,
    DefaultChoice,
    QuitChoice,
]
ForceSwitchChoice = Union[List[Union[SwitchChoice, PassChoice]], ResignChoice, DefaultChoice, QuitChoice]

pokesage/battle/choices.py
trevorWieland/pokesage
from typing import Dict, List, Literal, Optional, Union

from poketypes.dex import (
    DexAbility,
    DexCondition,
    DexGen,
    DexItem,
    DexMove,
    DexMoveCategory,
    DexNature,
    DexPokemon,
    DexStat,
    DexStatus,
    DexType,
    DexWeather,
)
from pydantic import BaseModel, Field


class StatBlock(BaseModel):
    """"""

    min_attack: Optional[int] = None
    min_defence: Optional[int] = None
    min_spattack: Optional[int] = None
    min_spdefence: Optional[int] = None
    min_speed: Optional[int] = None
    min_hp: Optional[int] = None

    max_attack: Optional[int] = None
    max_defence: Optional[int] = None
    max_spattack: Optional[int] = None
    max_spdefence: Optional[int] = None
    max_speed: Optional[int] = None
    max_hp: Optional[int] = None


class BoostBlock(BaseModel):
    """"""

    attack: int = 0
    defence: int = 0
    spattack: int = 0
    spdefence: int = 0
    speed: int = 0

    accuracy: int = 0
    evasion: int = 0


class BattleMove(BaseModel):
    """"""

    name: DexMove.ValueType = Field(..., description="The move")
    probability: float = Field(1.0, description="The probability that this pokemon has this move.")

    use_count: int = Field(..., description="The number of times this move has been seen")


class BattleAbility(BaseModel):
    """"""

    name: DexAbility.ValueType = Field(..., description="The ability")
    probability: float = Field(1.0, description="The probability that this pokemon has this ability.")


class BattleItem(BaseModel):
    """"""

    name: DexItem.ValueType = Field(..., description="The item")
    probability: float = Field(1.0, description="The probability that this pokemon is holding this item.")


class BattlePokemon(BaseModel):
    """ """

    player_id: str = Field(..., description="A unique identifier for the player that controls this pokemon")

    species: DexPokemon.ValueType = Field(..., description="The species of the pokemon")
    base_species: DexPokemon.ValueType = Field(..., description="The base species of the pokemon")
    nickname: Optional[str] = Field(
        None,
        description="The nickname of this pokemon. If we don't know the nickname yet, put None",
    )

    level: int = Field(100, description="The level of the pokemon")
    gender: Optional[Literal["M", "F"]] = Field(None, description="The gender of this pokemon")

    hp_type: Literal["exact", "fraction"] = Field(
        "fraction",
        description="Whether the max_hp and cur_hp fields are fractional out of 100 or exact.",
    )
    max_hp: Optional[int] = Field(
        None,
        description="The maximum health of the pokemon. If hp_type set to fractional, will always be 100",
    )
    cur_hp: Optional[int] = Field(
        None,
        description="The current health of the pokemon. If hp_type set to fractional, will be percentage",
    )

    team_pos: Optional[int] = Field(None, description="Which team-position the pokemon is currently in")
    slot: Optional[int] = Field(None, description="Which slot (if any) the pokemon is currently in", ge=1, le=3)

    active: bool = Field(False, description="Whether the current pokemon is active or not")

    stats: StatBlock = Field(
        StatBlock(),
        description="The min-max stat block of the pokemon. If you know the exact stats, min=max",
    )
    boosts: BoostBlock = Field(StatBlock(), description="The current stat boosts of the pokemon.")

    has_item: Optional[bool] = Field(
        None,
        description="Whether the pokemon is holding an item or not. None if unknown",
    )
    possible_items: List[BattleItem] = Field([], description="The list of possible items that the pokemon is holding")

    possible_abilities: List[BattleAbility] = Field([], description="The list of possible abilities of this pokemon")
    overwritten_ability: Optional[DexAbility.ValueType] = Field(
        None,
        description="If this pokemon's ability has been overwritten, which ability does it now have?",
    )

    moveset: List[BattleMove] = Field(
        [],
        description="The list of moves this pokemon has. If doing things probablistically, this can exceed 4!",
    )

    status: Optional[DexStatus.ValueType] = Field(None, description="The current status this pokemon is dealing with")
    conditions: List[DexCondition.ValueType] = Field(
        [], description="The list of current conditions this pokemon is dealing with"
    )

    tera_type: Optional[DexType.ValueType] = Field(None, description="The teratype of this pokemon")

    is_tera: bool = Field(False, description="Whether the pokemon is currently teratyped")
    is_mega: bool = Field(False, description="Whether the pokemon is currently mega-evolved")
    is_dynamax: bool = Field(False, description="Whether the pokemon is currently dynamaxed")

    is_reviving: bool = Field(False, description="Revival Blessing mechanic support")

    def to_id(self) -> str:
        """
        Provides a (hopefully) unique id for this pokemon based on currently known information

        When a player has multiple mostly-identical pokemon (basespecies, forme, level, gender, nickname) this breaks
        """
        id_str = f"{self.player_id}_{self.species}_{self.level}_{self.gender}_{self.nickname}"

        return id_str

    def to_base_id(self) -> str:
        """
        Provides a simplified id based on information known from `poke` messages in showdown

        When a player has multiple partially-identical pokemon (basespecies, level, gender) this breaks
        """
        id_str = f"{self.player_id}_{self.base_species}_{self.level}_{self.gender}_None"

        return id_str

pokesage/battle/pokemon.py
trevorWieland/pokesage
from beartype.typing import Dict, Optional

from poketypes.dex import DexWeather
from pydantic import BaseModel, Field

from .choices import BattleChoice
from .pokemon import BattlePokemon


class BattleState(BaseModel):
    """
    Representing the current state of the field in a given pokemon battle.

    Built to support singles, doubles, or triples gametypes
    """

    turn: int = Field(
        ...,
        description="The turn number of this battle state. Note: We might have multiple states per turn due to force-switches!",
    )

    player_team: Dict[str, BattlePokemon] = Field(
        {},
        description="A list containing Pokemon on the player's team. The keys must be unique to this pokemon",
    )
    opponent_team: Dict[str, BattlePokemon] = Field(
        {},
        description="A list containing Pokemon on the opponent's team.",
    )

    player_slots: Dict[int, Optional[str]] = Field(
        {}, description="Maps the player'ss battle slots to either None or the string id of the pokemon in that slot"
    )
    opponent_slots: Dict[int, Optional[str]] = Field(
        {}, description="Maps the opponent's battle slots to either None or the string id of the pokemon in that slot"
    )

    weather: Optional[DexWeather.ValueType] = Field(None, description="The current weather in the field")

    battle_choice: Optional[BattleChoice] = Field(
        None, description="The choices the player can make in response to this battle state"
    )

pokesage/battle/state.py
trevorWieland/pokesage
from .abstractconnector import ConnectionTermination, ConnectionTerminationCode, Connector
from .websocketconnector import WebsocketConnector

pokesage/connectors/__init__.py
trevorWieland/pokesage
from abc import ABC, abstractmethod
from enum import Enum, unique
from beartype.typing import AsyncGenerator, Optional, Tuple, Union

from aiohttp import ClientSession
from pydantic import BaseModel, Field

from ..battle.choices import AnyChoice
from ..battle.state import BattleState
from ..processors import ProgressState


@unique
class ConnectionTerminationCode(Enum):
    OBJECTIVE_COMPLETE = 0
    INVALID_CHOICE = 1
    SETUP_ERROR = 2
    CONNECTION_ERROR = 3
    PROCESSOR_ERROR = 4
    OTHER_ERROR = 5


class ConnectionTermination(BaseModel):
    """
    A returned object by the Connector's launch_connection function.
    """

    code: ConnectionTerminationCode = Field(..., description="The status code for this termination")
    message: Optional[str] = Field(None, description="An optional extra string message about this termination")


class Connector(ABC):
    """
    An abstract class defining function headers mandatory for connector objects

    Since the entrypoint for the user is the sage class, the sage class should create the aiohttp session,
    so that all downstream connections use the same async connection pool
    """

    @abstractmethod
    async def launch_connection(
        self, session: ClientSession
    ) -> AsyncGenerator[Tuple[ProgressState, Union[BattleState, ConnectionTermination, None]], AnyChoice]:
        """
        This function should be a generator that:
        - yields BattleState
        - accepts Choice as send (See Choice type alias for options)

        This function should be instantiated by the connection handling subclass, not the sage
        """

pokesage/connectors/abstractconnector.py
trevorWieland/pokesage
from .abstractconnector import Connector

pokesage/connectors/simulatorconnector.py
trevorWieland/pokesage
import json
import os
from beartype.typing import AsyncGenerator, Dict, List, Literal, Optional, Tuple, Type, Union
from urllib.parse import quote

from aiohttp import ClientSession, ClientWebSocketResponse, WSMsgType
from poketypes.showdown.showdownmessage import (
    Message,
    Message_challstr,
    MType,
    Message_updateuser,
    Message_updatesearch,
)
from yarl import URL
from beartype.door import is_bearable

from ..battle.battle import Battle
from ..battle.choices import AnyChoice, ForceSwitchChoice, MoveDecisionChoice, QuitChoice, ResignChoice, TeamOrderChoice
from ..battle.state import BattleState
from ..processors import ShowdownProcessor
from .abstractconnector import ConnectionTermination, ConnectionTerminationCode, Connector, ProgressState


class WebsocketConnector(Connector):
    """
    An abstract class defining function headers mandatory for connector objects

    Since the entrypoint for the user is the sage class, the sage class should create the aiohttp session,
    so that all downstream connections use the same async connection pool
    """

    def __init__(
        self,
        showdown_username: str,
        showdown_password: str,
        target_format: str,
        gametype: Literal["singles", "doubles", "triples"],
        objective: Literal["accept", "challenge", "ladder"],
        whitelist_users: List[str] = [],
        max_concurrent_battles: int = 1,
        total_battles: int = 1,
        processor_class: Type[ShowdownProcessor] = ShowdownProcessor,
        showdown_uri: str = "ws://localhost:8000/showdown/websocket",
        save_logs: bool = False,
        save_json: bool = False,
    ) -> None:
        """
        Save connection parameters so that the connection can be initialized later
        """

        self.showdown_username = showdown_username
        self.showdown_password = showdown_password
        self.target_format = target_format
        self.gametype = gametype
        self.objective = objective

        if self.objective == "challenge" and len(whitelist_users) == 0:
            raise RuntimeError(
                "Objective was set to challenge but no users were in the whitelist! Please add at least one user to the list!"
            )

        self.whitelist_users = whitelist_users
        self.max_concurrent_battles = max_concurrent_battles
        self.total_battles = total_battles
        self.processor_class = processor_class
        self.showdown_uri = showdown_uri
        self.save_logs = save_logs
        self.save_json = save_json

        # Parameter initialization
        self.logged_in: bool = False
        self.last_search: Optional[Message_updatesearch] = None
        self.valid_formats: List[str] = []

        self.pending_challenges: Dict[str, str] = {}
        self.battle_processors: Dict[str, ShowdownProcessor] = {}
        self.completed_battles: Dict[str, Battle] = {}

        self.pending_battle_amt: int = 0
        self.active_battle_amt: int = 0

    async def launch_connection(
        self, session: ClientSession
    ) -> AsyncGenerator[Tuple[ProgressState, Union[BattleState, ConnectionTermination, None]], AnyChoice]:
        """
        This function should be an async generator that:
        - yields BattleState
        - accepts Choice as send (See Choice type alias for options)

        This function should be instantiated by the connection handling subclass, not the sage
        """

        conn_term = None

        async with session.ws_connect(self.showdown_uri) as ws:
            async for msg in ws:
                if msg.type == WSMsgType.TEXT:
                    message: str = msg.data

                    if message[0] == ">" and "battle" in message:
                        # In this case this is a battle message block
                        battle_id = message.strip().split("\n")[0][1:]

                        for m in message.strip().split("\n")[1:]:
                            progress_state, data = await self.handle_battle_message(session, ws, battle_id, m)

                            if data is None:
                                continue

                            action = yield progress_state, data
                            conn_term = await self.process_action(ws, progress_state, action, battle_id)
                            if conn_term is not None:
                                break
                    elif message[0] == "|":
                        # In this case this is a standard message block
                        for m in message.strip().split("\n"):
                            (
                                progress_state,
                                conn_term,
                            ) = await self.handle_general_message(session, ws, m)
                            if conn_term is not None:
                                break

                            action = yield progress_state, None
                            conn_term = await self.process_action(ws, progress_state, action)
                            if conn_term is not None:
                                break
                    elif message[0] != ">":
                        # In this case this is a global message block
                        progress_state, conn_term = await self.handle_general_message(session, ws, f"||{message}")
                        if conn_term is not None:
                            break

                        action = yield progress_state, None
                        conn_term = await self.process_action(ws, progress_state, action)
                        if conn_term is not None:
                            break
                    else:
                        continue

                    if conn_term is not None:
                        yield ProgressState.FULL_END, conn_term
                        break
                else:
                    # We should never see a websocket error from showdown
                    # If we do see one, we should end and return the error information
                    yield ProgressState.FULL_END, ConnectionTermination(
                        code=ConnectionTerminationCode.CONNECTION_ERROR,
                        message=msg.data,
                    )
                    break

    async def handle_general_message(
        self, session: ClientSession, ws: ClientWebSocketResponse, message: str
    ) -> Tuple[ProgressState, Optional[ConnectionTermination]]:
        """
        Handles general messages given by the showdown server.

        This ranges from login strings to battle requests. This is where most of our administrative tasks should be done.
        Battle-specific tasks (other than original creation) should be handled as part of `handle_battle_message`
        """

        msg = Message.from_message(message)

        if msg.MTYPE == MType.updateuser:
            self.logged_in = msg.NAMED

            if self.logged_in and self.last_search is not None:
                conn_term = await self.handle_updatesearch(session, ws, self.last_search)
                if conn_term is not None:
                    return ProgressState.FULL_END, conn_term
        elif msg.MTYPE == MType.updatesearch:
            if self.objective == "ladder":
                self.pending_battle_amt = len(msg.SEARCHING)
            self.active_battle_amt = 0 if msg.GAMES is None else len(msg.GAMES)

            if self.logged_in:
                conn_term = await self.handle_updatesearch(session, ws, msg)
                if conn_term is not None:
                    return ProgressState.FULL_END, conn_term
            else:
                self.last_search = msg
        elif msg.MTYPE == MType.formats:
            self.valid_formats = [f.lower().replace("[", "").replace("]", "").replace(" ", "") for f in msg.FORMATS]
        elif msg.MTYPE == MType.challstr:
            try:
                await self.send_login(session, ws, msg)
            except Exception as ex:
                return ProgressState.FULL_END, ConnectionTermination(
                    code=ConnectionTerminationCode.SETUP_ERROR, message=str(ex)
                )
        elif msg.MTYPE == MType.pm:
            # For whatever reason showdown seems to use pms for challenge requests, so we
            # need to check this for any pending challenges we've sent/received
            pass
        else:
            # This is unlikely to be a bug since we only care about battle-related things in this class
            # However, for debugging purposes it might be nice to log unexpected messages, or to expand the
            # elif chain to include more types and manually pass so that it is clear they are unneeded
            pass

        return ProgressState.NO_ACTION, None

    async def handle_updatesearch(
        self, session: ClientSession, ws: ClientWebSocketResponse, message: Message_updatesearch
    ) -> Optional[ConnectionTermination]:
        """
        Handles the updatesearch message. This is one of the main objective-processing points, the other being `pm`

        Returns either None, or a ConnectionTermination if the objective is completed
        """

        if len(self.completed_battles.keys()) >= self.total_battles and message.GAMES is None:
            # In this case we've completed the needed amount and all games are over
            return ConnectionTermination(
                code=ConnectionTerminationCode.OBJECTIVE_COMPLETE, message="REQUIRED NUMBER OF GAMES COMPLETED"
            )
        elif len(self.completed_battles.keys()) >= self.total_battles:
            # In this case we've somehow completed the necessary amt of battles but not all games are over?
            return None
        elif self.pending_battle_amt + self.active_battle_amt >= self.max_concurrent_battles:
            # In this case we already have too many concurrent games, so we skip creating any new ones
            return None

        if self.objective == "accept":
            # if we're just trying to accept challenges, we don't want to initiate anything ourselves
            pass
        elif self.objective == "challenge":
            # TODO: Add support for multiple formats
            for u in self.whitelist_users:
                if u not in self.pending_challenges.keys():
                    await ws.send_str(f"|/challenge {u}, {self.target_format}")
                    self.pending_challenges[u] = self.target_format
                    self.pending_battle_amt += 1
                    break
        elif self.objective == "ladder":
            # We can ladder as long as we don't already have an ongoing or searching match for this format
            # We can also ladder once per format simulatneously, though for now we stick to just one format
            # TODO: Add support for multi-format searching instead of only one format

            if self.target_format not in message.SEARCHING and (
                message.GAMES is None or all([self.target_format not in g for g in message.GAMES])
            ):
                # in this case we aren't already searching for a match in this format, so we can submit a ladder request
                await ws.send_str(f"|/search, {self.target_format}")
                self.pending_battle_amt += 1

        return None

    async def send_login(
        self,
        session: ClientSession,
        ws: ClientWebSocketResponse,
        message: Message_challstr,
    ) -> None:
        """
        Handles the challengestr login flow.

        Sends a request to showdown's official login site with the user/pass/challstr to get a login token first
        Then sends that login token to the target showdown websocket to validate username selection

        Returns None, but raises a runtime error if the login fails
        """

        # TODO: Do some better especping for the user/pass
        params = {
            "name": quote(self.showdown_username).replace("-", "%2D"),
            "pass": quote(self.showdown_password),
            "challstr": message.CHALLSTR.strip(),
        }

        url = URL(
            f"https://play.pokemonshowdown.com/api/login?name={params['name']}&pass={params['pass']}&challstr={params['challstr']}",
            encoded=True,
        )

        async with session.post(url) as resp:
            r = await resp.text()
            r = json.loads(r[1:])

            if "assertion" not in r.keys():
                raise RuntimeError(f"Failed to log in: {r}")

            token = r["assertion"]

        await ws.send_str(f"|/trn {self.showdown_username},0,{token.strip()}")

    async def handle_battle_message(
        self,
        session: ClientSession,
        ws: ClientWebSocketResponse,
        battle_id: str,
        message: str,
    ) -> Tuple[ProgressState, Optional[BattleState]]:
        """
        Handles battle-specific messages. Mostly serves to pass messages to the corresponding processor for the battle.

        If the battle has already been completed, we will skip all messages afterwards by returning None for the battle state
        """

        if battle_id in self.completed_battles.keys():
            return ProgressState.NO_ACTION, None

        bp = self.battle_processors.get(
            battle_id,
            self.processor_class(session, self.showdown_username, self.showdown_username),
        )

        progress_state = await bp.process_message(message_str=message)

        self.battle_processors[battle_id] = bp

        return progress_state, bp.battle.battle_states[-1]

    async def process_action(
        self,
        ws: ClientWebSocketResponse,
        progress_state: ProgressState,
        action: AnyChoice,
        battle_id: Optional[str] = None,
    ) -> Optional[ConnectionTermination]:
        """
        This function will process the action based on the current progress state, closing any connections as needed
        All action verification is handled later in `submit_action`, this just looks at progress_state and determines what to do
        """

        if progress_state == ProgressState.FULL_END:
            # Process any final logging / closing operations needed if any (e.g. resign from all ongoing battles)
            # TODO

            return ConnectionTermination(code=ConnectionTerminationCode.OBJECTIVE_COMPLETE)
        elif progress_state == ProgressState.GAME_END:
            # Process game end movement
            if self.save_logs:
                os.makedirs(f"logs/{self.target_format}/", exist_ok=True)
                with open(f"logs/{self.target_format}/{battle_id}.log", "w", encoding="utf8") as f:
                    f.writelines([f"{l}\n" for l in self.battle_processors[battle_id].log])

            if self.save_json:
                os.makedirs(f"logs/{self.target_format}/", exist_ok=True)
                with open(f"logs/{self.target_format}/{battle_id}.json", "w", encoding="utf8") as f:
                    json.dump(self.battle_processors[battle_id].battle.model_dump(mode="json"), f)

            self.completed_battles[battle_id] = self.battle_processors.pop(battle_id).battle
            return None
        else:
            # Verify/Submit the action to pokemon showdown
            try:
                return await self.submit_action(ws, progress_state, action, battle_id=battle_id)
            except AssertionError as ex:
                return ConnectionTermination(code=ConnectionTerminationCode.INVALID_CHOICE, message=f"{type(ex)}: {ex}")
            except NotImplementedError as ex:
                return ConnectionTermination(code=ConnectionTerminationCode.INVALID_CHOICE, message=f"{type(ex)}: {ex}")
            except Exception as ex:
                return ConnectionTermination(code=ConnectionTerminationCode.OTHER_ERROR, message=f"{type(ex)}: {ex}")

    async def submit_action(
        self,
        ws: ClientWebSocketResponse,
        progress_state: ProgressState,
        action: AnyChoice,
        battle_id: Optional[str] = None,
    ) -> Optional[ConnectionTermination]:
        """
        This function will:
        - Verify that the action(s) returned the by the user is valid to submit at this time
        - Submit it to showdown

        It will not check if a given move is a legal move, because that step should be done by the processor
        at choice-build time
        """

        # Handle quit and resign options first since they are the simple cases
        if is_bearable(action, QuitChoice):
            # Process all-resignation decision
            # TODO: Resign from all active games / close anything else as needed

            return ConnectionTermination(
                code=ConnectionTerminationCode.OTHER_ERROR,
                message="USER REQUESTED SHUTDOWN",
            )
        if is_bearable(action, ResignChoice):
            # Process resignation decision
            # TODO: Resign from just this game, don't return ConnectionTermination
            assert (
                battle_id is not None
            ), "battle_id was None but you sent a resignation! When not in a battle, send QuitChoice instead!"
            return

        assert is_bearable(action, AnyChoice)

        action_str = f"{battle_id}|/choose "

        if progress_state == ProgressState.TEAM_ORDER:
            # Check valid formatting for team order submission:
            assert is_bearable(action, TeamOrderChoice), f"action was not a valid TeamOrderChoice!\n{action}"
            assert (
                battle_id is not None
            ), "battle_id was None! This likely means a bug in websocketconnector, not your code!"

            # TeamOrderChoice can either be a TeamChoice or a DefaultChoice, easy
            action_str += action.to_showdown()
        elif progress_state == ProgressState.SWITCH:
            # Check valid formatting for force-switch submission:
            assert is_bearable(action, ForceSwitchChoice), f"action was not a valid ForceSwitchChoice!\n{action}"
            assert (
                battle_id is not None
            ), "battle_id was None! This likely means a bug in websocketconnector, not your code!"

            # ForceSwitchChoice can either be a List of either Switch/Pass choices, or a single DefaultChoice
            if is_bearable(action, list):
                assert (
                    (self.gametype == "singles" and len(action) == 1)
                    or (self.gametype == "doubles" and len(action) == 2)
                    or (self.gametype == "triples" and len(action) == 3)
                ), f"gametype is {self.gametype}, but you sent {len(action)} commands!"
                action_str += ",".join([sub_action.to_showdown() for sub_action in action])
            else:
                action_str += action.to_showdown()
        elif progress_state == ProgressState.MOVE:
            # Check valid formatting for move submission:
            assert is_bearable(action, MoveDecisionChoice), f"action was not a valid MoveDecisionChoice!\n{action}"
            assert (
                battle_id is not None
            ), "battle_id was None! This likely means a bug in websocketconnector, not your code!"

            # ForceSwitchChoice can either be a List of Move/Switch/Pass choices, or a single DefaultChoice
            if is_bearable(action, list):
                assert (
                    (self.gametype == "singles" and len(action) == 1)
                    or (self.gametype == "doubles" and len(action) == 2)
                    or (self.gametype == "triples" and len(action) == 3)
                ), f"gametype is {self.gametype}, but you sent {len(action)} commands!"
                action_str += ",".join([sub_action.to_showdown() for sub_action in action])
            else:
                action_str += action.to_showdown()
        else:
            assert action is None
            return

        await self.battle_processors[battle_id].process_action(action, action_str)
        await ws.send_str(action_str)

pokesage/connectors/websocketconnector.py
trevorWieland/pokesage
from .abstractprocessor import Processor, ProgressState
from .showdownprocessor import ShowdownProcessor

pokesage/processors/__init__.py
trevorWieland/pokesage
from abc import ABC, abstractmethod
from typing import Type
from enum import Enum, unique
from aiohttp import ClientSession

from ..battle.battle import Battle
from ..battle.state import BattleState
from ..battle.choices import AnyChoice


@unique
class ProgressState(Enum):
    NO_ACTION = 0
    TEAM_ORDER = 1
    MOVE = 2
    SWITCH = 3
    GAME_END = 4
    FULL_END = 5


class Processor(ABC):
    """
    Processes incoming `messages` and updates the BattleState accordingly.

    A Processor should be implemented to handle a specific format of messages (Showdown, GBA emulator, etc.)
    The Battle/BattleState that gets updated SHOULD NOT contain anything directly tied to the formatting.
    """

    BATTLE_CLASS: Type[Battle] = Battle
    BATTLE_STATE_CLASS: Type[BattleState] = BattleState

    def __init__(self, session: ClientSession, battle_id: str, player_name: str) -> None:
        """
        Initialize the Processor with a battle_id, as well as the session to use for any http requests
        """

        self.battle_id = battle_id
        self.session = session

        self.battle: Battle = self.BATTLE_CLASS(battle_actions=[], battle_states=[], player_name=player_name)
        self.log = []

    @abstractmethod
    async def process_message(self, message_str: str) -> ProgressState:
        """
        This function should process the given message and return a ProgressState accordingly.

        Remember, everything that you want to keep should be stored in self.battle / a state inside self.battle.battle_states!
        self.log should only be used for troubleshooting measures, or other logging purposes
        """

    @abstractmethod
    async def process_action(self, action: AnyChoice, action_str: str) -> None:
        """
        This function should take an action given by the sage and process it.

        This shouldn't *submit* the action to the game, but instead log the action as part of the state
        """

pokesage/processors/abstractprocessor.py
trevorWieland/pokesage
from beartype.typing import Optional
from copy import deepcopy
from poketypes.dex import clean_forme
from poketypes.showdown.battlemessage import *

from ..battle import Battle, BattleAbility, BattleItem, BattleMove, BattlePokemon, BattleState, BoostBlock, StatBlock
from ..battle.choices import MoveChoice, PassChoice, SwitchChoice, TeamChoice, AnyChoice
from .abstractprocessor import Processor, ProgressState


class ShowdownProcessor(Processor):
    """
    Processor class for showdown-style messages. Built to work for showdown-style battle messages exclusively

    This class on its own only provides the framework to process the battle. You should then extend this class
    to actually implement battle state processing. As is, it will provide the minimal request-handling to provide
    valid actions. Almost no other details will be available in the battle state if you use this!
    """

    async def process_action(self, action: AnyChoice, action_str: str) -> None:
        """
        Takes the action given by the sage and adds it to the log

        Also duplicates the battle_state so we save a copy of the state previous to the decision
        """

        self.log.append(action_str)
        self.battle.battle_actions.append(action)
        self.battle.battle_states.append(deepcopy(self.battle.battle_states[-1]))

    async def process_message(self, message_str: str) -> ProgressState:
        """
        This function shouild process the given message and return a ProgressState accordingly.

        Remember, everything that you want to keep should be stored in self.battle / a state inside self.battle.battle_states!
        self.log should only be used for troubleshooting measures, or other logging purposes
        """

        bm = BattleMessage.from_message(message_str)

        try:
            progress_state = await self.process_bm(bm)
        except Exception as e:
            print("Log so far:")
            for m in self.log:
                print(m)
            print()
            print("Current Battle State:")
            print(self.battle.battle_states[-1].model_dump_json(indent=4))
            print()
            print("Message that caused error:")
            print(bm.BATTLE_MESSAGE)
            print()
            raise e

        return progress_state

    async def preprocess_bm(self, bm: BattleMessage) -> Optional[ProgressState]:
        """
        A function to hook into the battle message processing flow.
        This will run *before* the battle message is processed by its specific method

        If the function returns a ProgressState, we will *SKIP* the remaining processing
        """

        self.log.append(bm.BATTLE_MESSAGE)

    async def postprocess_bm(self, bm: BattleMessage) -> Optional[ProgressState]:
        """
        A function to hook into the battle message processing flow.
        This will run *after* the battle message is processed by its specific method

        If the function returns a ProgressState, it will overwrite the original ProgressState
        """

    async def process_bm(self, bm: BattleMessage) -> ProgressState:
        """
        The main processing flow for Battle Messages.

        Flow:
        - preprocess_bm -> If not None, break and return result
        - processbm_{BMTYPE} runs for the specific battle message type
        - postprocess_bm -> If not None, result overrides previous progress state result

        In subclasses, you shouldn't need to override this function, instead you would override individual
        process_bm_{BMTYPE} functions for specific feature extraction, or the pre/post process functions

        The returned ProgressState will be acted on, so make sure that if the server is expecting an action,
        you return the correct corresponding ProgressState for the expected action.
        """

        progress_state = await self.preprocess_bm(bm)

        if progress_state is not None:
            return progress_state

        if bm.BMTYPE == BMType.player:
            await self.processbm_player(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.teamsize:
            await self.processbm_teamsize(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.gametype:
            await self.processbm_gametype(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.gen:
            await self.processbm_gen(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.tier:
            await self.processbm_tier(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.rated:
            await self.processbm_rated(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.rule:
            await self.processbm_rule(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.clearpoke:
            await self.processbm_clearpoke(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.poke:
            await self.processbm_poke(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.start:
            await self.processbm_start(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.teampreview:
            await self.processbm_teampreview(bm)
            progress_state = ProgressState.TEAM_ORDER
        elif bm.BMTYPE == BMType.empty:
            await self.processbm_empty(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.request:
            await self.processbm_request(bm)

            bm: BattleMessage_request = bm

            if bm.REQUEST_TYPE == "TEAMPREVIEW":
                progress_state = ProgressState.NO_ACTION
            elif bm.REQUEST_TYPE == "ACTIVE":
                progress_state = ProgressState.NO_ACTION
            elif bm.REQUEST_TYPE == "FORCESWITCH":
                progress_state = ProgressState.SWITCH
            elif bm.REQUEST_TYPE == "WAIT":
                progress_state = ProgressState.NO_ACTION
            else:
                progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.inactive:
            await self.processbm_inactive(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.inactiveoff:
            await self.processbm_inactiveoff(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.upkeep:
            await self.processbm_upkeep(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.turn:
            await self.processbm_turn(bm)
            progress_state = ProgressState.MOVE
        elif bm.BMTYPE == BMType.win:
            await self.processbm_win(bm)
            progress_state = ProgressState.GAME_END
        elif bm.BMTYPE == BMType.tie:
            await self.processbm_tie(bm)
            progress_state = ProgressState.GAME_END
        elif bm.BMTYPE == BMType.expire:
            await self.processbm_expire(bm)
            progress_state = ProgressState.GAME_END
        elif bm.BMTYPE == BMType.t:
            await self.processbm_t(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.move:
            await self.processbm_move(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.switch:
            await self.processbm_switch(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.drag:
            await self.processbm_drag(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.detailschange:
            await self.processbm_detailschange(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.replace:
            await self.processbm_replace(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.swap:
            await self.processbm_swap(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.cant:
            await self.processbm_cant(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.faint:
            await self.processbm_faint(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.fail:
            await self.processbm_fail(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.block:
            await self.processbm_block(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.notarget:
            await self.processbm_notarget(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.miss:
            await self.processbm_miss(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.damage:
            await self.processbm_damage(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.heal:
            await self.processbm_heal(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.sethp:
            await self.processbm_sethp(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.status:
            await self.processbm_status(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.curestatus:
            await self.processbm_curestatus(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.cureteam:
            await self.processbm_cureteam(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.boost:
            await self.processbm_boost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.unboost:
            await self.processbm_unboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.setboost:
            await self.processbm_setboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.swapboost:
            await self.processbm_swapboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.invertboost:
            await self.processbm_invertboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.clearboost:
            await self.processbm_clearboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.clearallboost:
            await self.processbm_clearallboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.clearpositiveboost:
            await self.processbm_clearpositiveboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.clearnegativeboost:
            await self.processbm_clearnegativeboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.copyboost:
            await self.processbm_copyboost(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.weather:
            await self.processbm_weather(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.fieldstart:
            await self.processbm_fieldstart(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.fieldend:
            await self.processbm_fieldend(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.terastallize:
            await self.processbm_terastallize(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.fieldactivate:
            await self.processbm_fieldactivate(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.sidestart:
            await self.processbm_sidestart(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.sideend:
            await self.processbm_sideend(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.swapsideconditions:
            await self.processbm_swapsideconditions(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.volstart:
            await self.processbm_volstart(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.volend:
            await self.processbm_volend(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.crit:
            await self.processbm_crit(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.supereffective:
            await self.processbm_supereffective(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.resisted:
            await self.processbm_resisted(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.immune:
            await self.processbm_immune(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.item:
            await self.processbm_item(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.enditem:
            await self.processbm_enditem(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.ability:
            await self.processbm_ability(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.endability:
            await self.processbm_endability(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.transform:
            await self.processbm_transform(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.mega:
            await self.processbm_mega(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.primal:
            await self.processbm_primal(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.burst:
            await self.processbm_burst(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.zpower:
            await self.processbm_zpower(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.zbroken:
            await self.processbm_zbroken(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.activate:
            await self.processbm_activate(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.hint:
            await self.processbm_hint(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.center:
            await self.processbm_center(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.message:
            await self.processbm_message(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.combine:
            await self.processbm_combine(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.waiting:
            await self.processbm_waiting(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.prepare:
            await self.processbm_prepare(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.mustrecharge:
            await self.processbm_mustrecharge(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.nothing:
            await self.processbm_nothing(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.hitcount:
            await self.processbm_hitcount(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.singlemove:
            await self.processbm_singlemove(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.singleturn:
            await self.processbm_singleturn(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.formechange:
            await self.processbm_formechange(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.error:
            await self.processbm_error(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.bigerror:
            await self.processbm_bigerror(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.unknown:
            await self.processbm_unknown(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.init:
            await self.processbm_init(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.title:
            await self.processbm_title(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.join:
            await self.processbm_join(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.leave:
            await self.processbm_leave(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.raw:
            await self.processbm_raw(bm)
            progress_state = ProgressState.NO_ACTION
        elif bm.BMTYPE == BMType.anim:
            await self.processbm_anim(bm)
            progress_state = ProgressState.NO_ACTION
        else:
            print(f"BattleStateProcessor does not have a handler for BMType: {bm.BMTYPE}!")
            progress_state = ProgressState.NO_ACTION

        other_state = await self.postprocess_bm(bm)

        if other_state is not None:
            progress_state = other_state

        return progress_state

    async def processbm_player(self, bm: BattleMessage_player) -> None:
        """
        Processes this message to update the current battle state
        """

        if self.battle.player_name == bm.USERNAME:
            self.battle.player_id = bm.PLAYER
            self.battle.player_rating = bm.RATING
        else:
            self.battle.opponent_name = bm.USERNAME
            self.battle.opponent_id = bm.PLAYER
            self.battle.opponent_rating = bm.RATING

    async def processbm_teamsize(self, bm: BattleMessage_teamsize) -> None:
        """
        Processes this message to update the current battle state
        """

        if self.battle.player_id == bm.PLAYER:
            self.battle.player_team_size = bm.NUMBER
        else:
            self.battle.opponent_team_size = bm.NUMBER

    async def processbm_gametype(self, bm: BattleMessage_gametype) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.gametype = bm.GAMETYPE

        if bm.GAMETYPE == "singles":
            self.battle.battle_states[-1].player_slots = {1: None}
            self.battle.battle_states[-1].opponent_slots = {1: None}
        elif bm.GAMETYPE == "doubles":
            self.battle.battle_states[-1].player_slots = {1: None, 2: None}
            self.battle.battle_states[-1].opponent_slots = {1: None, 2: None}
        elif bm.GAMETYPE == "triples":
            self.battle.battle_states[-1].player_slots = {1: None, 2: None, 3: None}
            self.battle.battle_states[-1].opponent_slots = {1: None, 2: None, 3: None}

    async def processbm_gen(self, bm: BattleMessage_gen) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.gen = bm.GENNUM

    async def processbm_tier(self, bm: BattleMessage_tier) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.format = bm.FORMATNAME.lower().replace("[", "").replace("]", "").replace(" ", "")

    async def processbm_rated(self, bm: BattleMessage_rated) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.rated = True

    async def processbm_rule(self, bm: BattleMessage_rule) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_clearpoke(self, bm: BattleMessage_clearpoke) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_poke(self, bm: BattleMessage_poke) -> None:
        """
        Processes this message to update the current battle state
        """

        if bm.PLAYER == self.battle.player_id:
            # We already get better information from the request object, so we don't care about this for us
            return

        # In showdown we first see opponent pokemon base-species details, without knowing what the exact forme is
        # So later, when checking, check for a match on species first, then on base-species.
        # Once base-species finds a match, set species accordingly
        poke = BattlePokemon(
            player_id=bm.PLAYER,
            species=bm.SPECIES,
            base_species=clean_forme(bm.SPECIES),
            nickname=None,
            level=bm.LEVEL,
            gender=bm.GENDER,
            tera_type=bm.TERA,
            has_item=bm.HAS_ITEM,
        )

        temp_id = poke.to_base_id()

        self.battle.battle_states[-1].opponent_team[temp_id] = poke

    async def processbm_start(self, bm: BattleMessage_start) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_teampreview(self, bm: BattleMessage_teampreview) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_empty(self, bm: BattleMessage_empty) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_request(self, bm: BattleMessage_request) -> None:
        """
        Processes this message to update the current battle state
        """

        assert self.battle.player_name == bm.USERNAME

        assert (self.battle.player_id == bm.PLAYER) or (self.battle.player_id is None)

        if self.battle.player_id is None:
            self.battle.player_id = bm.PLAYER

        # Set our choices back to empty so we're ready to add details
        current_state = self.battle.battle_states[-1]
        current_state.battle_choice = None

        # Process all pokemon in the `side` data
        for e, bm_poke in enumerate(bm.POKEMON):
            # TODO: Process Reviving/Commanding mechanics properly (Don't know what they're for yet)

            if len(current_state.player_team) == e:
                # This means we have never initialized this pokemon before, so we need to do so
                s_block = StatBlock(
                    min_attack=bm_poke.STATS["atk"],
                    max_attack=bm_poke.STATS["atk"],
                    min_defence=bm_poke.STATS["def"],
                    max_defence=bm_poke.STATS["def"],
                    min_spattack=bm_poke.STATS["spa"],
                    max_apattack=bm_poke.STATS["spa"],
                    min_spdefence=bm_poke.STATS["spd"],
                    max_spdefence=bm_poke.STATS["spd"],
                    min_speed=bm_poke.STATS["spe"],
                    max_speed=bm_poke.STATS["spe"],
                    min_hp=bm_poke.MAX_HP,
                    max_hp=bm_poke.MAX_HP,
                )
                poke = BattlePokemon(
                    player_id=bm.PLAYER,
                    species=bm_poke.SPECIES,
                    base_species=clean_forme(bm_poke.SPECIES),
                    nickname=bm_poke.IDENT.IDENTITY,
                    level=bm_poke.LEVEL,
                    gender=bm_poke.GENDER,
                    hp_type="exact",
                    max_hp=bm_poke.MAX_HP,
                    cur_hp=bm_poke.CUR_HP,
                    team_pos=e + 1,
                    stats=s_block,
                    has_item=bm_poke.ITEM is not None,
                    possible_items=[] if bm_poke.ITEM is None else [BattleItem(name=bm_poke.ITEM, probability=1.0)],
                    possible_abilities=[BattleAbility(name=bm_poke.BASE_ABILITY, probability=1.0)],
                    overwritten_ability=None if (bm_poke.ABILITY == bm_poke.BASE_ABILITY) else bm_poke.ABILITY,
                    moveset=[BattleMove(name=m, probability=1.0, use_count=0) for m in bm_poke.MOVES],
                    status=bm_poke.STATUS,
                    tera_type=bm_poke.TERATYPE,
                    is_tera=bm_poke.TERASTALLIZED is not None,
                    active=bm_poke.ACTIVE,
                )

                current_state.player_team[poke.to_id()] = poke
            else:
                poke_id = f"{bm.PLAYER}_{bm_poke.SPECIES}_{bm_poke.LEVEL}_{bm_poke.GENDER}_{bm_poke.IDENT.IDENTITY}"
                poke = current_state.player_team[poke_id]

                poke.max_hp = bm_poke.MAX_HP if bm_poke.MAX_HP is not None else poke.max_hp
                poke.cur_hp = bm_poke.CUR_HP

                poke.has_item = bm_poke.ITEM is not None
                poke.possible_items = [] if bm_poke.ITEM is None else [BattleItem(name=bm_poke.ITEM, probability=1.0)]

                poke.possible_abilities = (
                    [BattleAbility(name=bm_poke.BASE_ABILITY, probability=1.0)]
                    if len(poke.possible_abilities) == 0
                    else poke.possible_abilities
                )
                poke.overwritten_ability = (
                    None
                    if (bm_poke.ABILITY == bm_poke.BASE_ABILITY or bm_poke.ABILITY == poke.possible_abilities[0].name)
                    else bm_poke.ABILITY
                )

                poke.moveset = (
                    [BattleMove(name=m, probability=1.0, use_count=0) for m in bm_poke.MOVES]
                    if len(poke.moveset) == 0
                    else poke.moveset
                )

                poke.status = bm_poke.STATUS

                poke.tera_type = bm_poke.TERATYPE
                poke.is_tera = bm_poke.TERASTALLIZED is not None

                poke.team_pos = e + 1

                poke.active = bm_poke.ACTIVE

                poke.is_reviving = bm_poke.REVIVING

                current_state.player_team[poke_id] = poke

        if bm.REQUEST_TYPE == "WAIT":
            # In this case there is no action to be made, so we just set the state and move on
            self.battle.battle_states[-1] = current_state
            return

        if bm.REQUEST_TYPE == "TEAMPREVIEW":
            # In this case we have a teamchoice decision to make, though unlike other choices where we calculate every
            # option ahead of time for the player, here we just send a sorted list of team slots.
            # No point in returning a list of every possible teamchoice permutation
            current_state.battle_choice = TeamChoice(team_order=list(range(1, len(current_state.player_team) + 1)))

            self.battle.battle_states[-1] = current_state
            return

        # Process the active pokemon
        switch_options = [
            SwitchChoice(slot=p.team_pos)
            for p in current_state.player_team.values()
            if (p.status != DexStatus.STATUS_FNT and not p.active)
        ]
        revive_options = [
            SwitchChoice(slot=p.team_pos)
            for p in current_state.player_team.values()
            if (p.status == DexStatus.STATUS_FNT and not p.active)
        ]

        if bm.REQUEST_TYPE == "FORCESWITCH":
            # If the request type is a forceswitch, then we have already identified every option the player has

            choices = []
            for i in range(self.battle.slot_length()):
                if bm.FORCESWITCH_SLOTS[i]:
                    choices.append(switch_options)
                else:
                    choices.append(PassChoice())

            current_state.battle_choice = choices

            self.battle.battle_states[-1] = current_state
            return

        assert bm.REQUEST_TYPE == "ACTIVE"

        choices = []
        for i in range(self.battle.slot_length()):
            if len(bm.ACTIVE_OPTIONS) <= i:
                # This implies we have less alive pokemon than slots, so we fill the remaining parts with pass choices
                choices.append(PassChoice())
                continue

            # This has extra details about moves and options for our pokemon in the active slots
            # We use this information to build current_state.move_choices / switch_choices
            ao = bm.ACTIVE_OPTIONS[i]

            # Process all the moves that are open to us
            move_options = []
            for move_num, m_data in enumerate(ao.MOVES):
                if m_data.DISABLED or m_data.CUR_PP == 0:
                    continue

                # Since the request comes before we see any switches, we have to process valid targets later in `turn`
                mo = MoveChoice(move_number=move_num + 1)
                move_options.append(mo)

                if ao.CAN_TERA:
                    mo = MoveChoice(move_number=move_num + 1, tera=True)
                    move_options.append(mo)

                if ao.CAN_MEGA:
                    mo = MoveChoice(move_number=move_num + 1, mega=True)
                    move_options.append(mo)

                if ao.CAN_DYNA:
                    mo = MoveChoice(move_number=move_num + 1, dyna=True)
                    move_options.append(mo)

                if ao.CAN_ZMOVE:
                    mo = MoveChoice(move_number=move_num + 1, zmove=True)
                    move_options.append(mo)

            if not ao.TRAPPED:
                slot_choices = move_options + switch_options
            else:
                slot_choices = move_options

            choices.append(slot_choices)

        current_state.battle_choice = choices
        self.battle.battle_states[-1] = current_state

    async def processbm_inactive(self, bm: BattleMessage_inactive) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_inactiveoff(self, bm: BattleMessage_inactiveoff) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_upkeep(self, bm: BattleMessage_upkeep) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_turn(self, bm: BattleMessage_turn) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.turn = bm.NUMBER

        # We need to process valid targets here, as this will happen `after` any switches are made
        # TODO: process move target options

    async def processbm_win(self, bm: BattleMessage_win) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.player_victory = bm.USERNAME == self.battle.player_name

    async def processbm_tie(self, bm: BattleMessage_tie) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.player_victory = False

    async def processbm_expire(self, bm: BattleMessage_expire) -> None:
        """
        Processes this message to update the current battle state
        """

        self.battle.player_victory = False

    async def processbm_t(self, bm: BattleMessage_t) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_move(self, bm: BattleMessage_move) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_switch(self, bm: BattleMessage_switch) -> None:
        """
        Processes this message to update the current battle state
        """

        # TODO: Figure out a way of supporting duplicate base species w/ different nicknames
        ## Maybe a numbering system based on appearance?
        ## Current setup breaks when a pokemon changes forme back to a basespecies
        ## Current setup also relies on perfect accounting of slot information

        full_ident = f"{bm.POKEMON.PLAYER}_{bm.SPECIES}_{bm.LEVEL}_{bm.GENDER}_{bm.POKEMON.IDENTITY}"
        base_ident = f"{bm.POKEMON.PLAYER}_{clean_forme(bm.SPECIES)}_{bm.LEVEL}_{bm.GENDER}_None"

        slot = bm.POKEMON.SLOT
        assert slot is not None

        slot = "abc".find(slot) + 1

        current_state = self.battle.battle_states[-1]

        if full_ident in current_state.player_team.keys():
            # Process this as a player slot switch
            if current_state.player_slots[slot] is not None:
                old_poke_id = current_state.player_slots[slot]
                current_state.player_team[old_poke_id].slot = None
                current_state.player_team[old_poke_id].active = False

            current_state.player_team[full_ident].slot = slot
            current_state.player_team[full_ident].active = True
            current_state.player_slots[slot] = full_ident
        elif full_ident in current_state.opponent_team.keys():
            # Process this as an opponent slot switch
            # This also means this is a pokemon we have seen before from the opponent
            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident
        elif base_ident in current_state.opponent_team.keys():
            # Process this as an opponent slot switch
            # This also means this is a pokemon we have not yet seen before from the opponent
            poke = current_state.opponent_team.pop(base_ident)
            poke.species = bm.SPECIES
            poke.nickname = bm.POKEMON.IDENTITY
            current_state.opponent_team[full_ident] = poke

            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident
        else:
            # In this case, we must be playing without teampreview, since we haven't seen this before
            # TODO: assert that teampreview is part of this format for bugtracking reasons

            assert (
                bm.POKEMON.PLAYER == self.battle.opponent_id
            ), f"Previously unindentified player pokemon: {full_ident}"

            poke = BattlePokemon(
                player_id=bm.POKEMON.PLAYER,
                species=bm.SPECIES,
                base_species=clean_forme(bm.SPECIES),
                nickname=bm.POKEMON.IDENTITY,
                level=bm.LEVEL,
                gender=bm.GENDER,
                tera_type=bm.TERA,
            )
            current_state.opponent_team[full_ident] = poke

            # Process this as an opponent slot switch
            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident

        self.battle.battle_states[-1] = current_state

    async def processbm_drag(self, bm: BattleMessage_drag) -> None:
        """
        Processes this message to update the current battle state
        """

        full_ident = f"{bm.POKEMON.PLAYER}_{bm.SPECIES}_{bm.LEVEL}_{bm.GENDER}_{bm.POKEMON.IDENTITY}"
        base_ident = f"{bm.POKEMON.PLAYER}_{clean_forme(bm.SPECIES)}_{bm.LEVEL}_{bm.GENDER}_None"

        slot = bm.POKEMON.SLOT
        assert slot is not None

        slot = "abc".find(slot) + 1

        current_state = self.battle.battle_states[-1]

        if full_ident in current_state.player_team.keys():
            # Process this as a player slot switch
            if current_state.player_slots[slot] is not None:
                old_poke_id = current_state.player_slots[slot]
                current_state.player_team[old_poke_id].slot = None
                current_state.player_team[old_poke_id].active = False

            current_state.player_team[full_ident].slot = slot
            current_state.player_team[full_ident].active = True
            current_state.player_slots[slot] = full_ident
        elif full_ident in current_state.opponent_team.keys():
            # Process this as an opponent slot switch
            # This also means this is a pokemon we have seen before from the opponent
            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident
        elif base_ident in current_state.opponent_team.keys():
            # Process this as an opponent slot switch
            # This also means this is a pokemon we have not yet seen before from the opponent
            poke = current_state.opponent_team.pop(base_ident)
            poke.species = bm.SPECIES
            poke.nickname = bm.POKEMON.IDENTITY
            current_state.opponent_team[full_ident] = poke

            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident
        else:
            # In this case, we must be playing without teampreview, since we haven't seen this before
            # TODO: assert that teampreview is part of this format for bugtracking reasons

            assert (
                bm.POKEMON.PLAYER == self.battle.opponent_id
            ), f"Previously unindentified player pokemon: {full_ident}"

            poke = BattlePokemon(
                player_id=bm.POKEMON.PLAYER,
                species=bm.SPECIES,
                base_species=clean_forme(bm.SPECIES),
                nickname=bm.POKEMON.IDENTITY,
                level=bm.LEVEL,
                gender=bm.GENDER,
                tera_type=bm.TERA,
            )
            current_state.opponent_team[full_ident] = poke

            # Process this as an opponent slot switch
            if current_state.opponent_slots[slot] is not None:
                old_poke_id = current_state.opponent_slots[slot]
                current_state.opponent_team[old_poke_id].slot = None
                current_state.opponent_team[old_poke_id].active = False

            current_state.opponent_team[full_ident].slot = slot
            current_state.opponent_team[full_ident].active = True
            current_state.opponent_slots[slot] = full_ident

        self.battle.battle_states[-1] = current_state

    async def processbm_detailschange(self, bm: BattleMessage_detailschange) -> None:
        """
        Processes this message to update the current battle state
        """

        full_ident = f"{bm.POKEMON.PLAYER}_{bm.SPECIES}_{bm.LEVEL}_{bm.GENDER}_{bm.POKEMON.IDENTITY}"

        slot = bm.POKEMON.SLOT
        assert slot is not None

        slot = "abc".find(slot) + 1

        current_state = self.battle.battle_states[-1]

        if bm.POKEMON.PLAYER == self.battle.player_id:
            # Process this as a player update
            assert current_state.player_slots[slot] is not None, "Failed to setup slots before change occured"

            old_poke_id = current_state.player_slots[slot]
            poke = current_state.player_team.pop(old_poke_id)

            poke.species = bm.SPECIES
            poke.base_species = clean_forme(bm.SPECIES)

            current_state.player_team[full_ident] = poke
            current_state.player_slots[slot] = full_ident
        else:
            # Process this as a opponent update
            assert current_state.opponent_slots[slot] is not None, "Failed to setup slots before change occured"

            old_poke_id = current_state.opponent_slots[slot]
            poke = current_state.opponent_team.pop(old_poke_id)

            poke.species = bm.SPECIES
            poke.base_species = clean_forme(bm.SPECIES)

            current_state.opponent_team[full_ident] = poke
            current_state.opponent_slots[slot] = full_ident

        self.battle.battle_states[-1] = current_state

    async def processbm_replace(self, bm: BattleMessage_replace) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_swap(self, bm: BattleMessage_swap) -> None:
        """
        Processes this message to update the current battle state
        """

        player = bm.POKEMON.PLAYER

        original_slot = bm.POKEMON.SLOT
        assert original_slot is not None

        original_slot = "abc".find(original_slot) + 1
        new_slot = bm.POSITION + 1

        current_state = self.battle.battle_states[-1]

        if player == self.battle.player_id:
            # Process this swap as a player-side swap
            assert current_state.player_slots[original_slot] is not None, "Failed to setup slots before swap occured"
            assert current_state.player_slots[new_slot] is not None, "Failed to setup slots before swap occured"

            orig_id = current_state.player_slots[original_slot]
            new_id = current_state.player_slots[new_slot]

            current_state.player_team[orig_id].slot = new_slot
            current_state.player_team[new_id].slot = original_slot

            current_state.player_slots[original_slot] = new_id
            current_state.player_slots[new_slot] = orig_id
        else:
            # Process this swap as an opponent-side swap
            assert current_state.opponent_slots[original_slot] is not None, "Failed to setup slots before swap occured"
            assert current_state.opponent_slots[new_slot] is not None, "Failed to setup slots before swap occured"

            orig_id = current_state.opponent_slots[original_slot]
            new_id = current_state.opponent_slots[new_slot]

            current_state.opponent_team[orig_id].slot = new_slot
            current_state.opponent_team[new_id].slot = original_slot

            current_state.opponent_slots[original_slot] = new_id
            current_state.opponent_slots[new_slot] = orig_id

        self.battle.battle_states[-1] = current_state

    async def processbm_cant(self, bm: BattleMessage_cant) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_faint(self, bm: BattleMessage_faint) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_fail(self, bm: BattleMessage_fail) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_block(self, bm: BattleMessage_block) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_notarget(self, bm: BattleMessage_notarget) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_miss(self, bm: BattleMessage_miss) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_damage(self, bm: BattleMessage_damage) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_heal(self, bm: BattleMessage_heal) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_sethp(self, bm: BattleMessage_sethp) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_status(self, bm: BattleMessage_status) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_curestatus(self, bm: BattleMessage_curestatus) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_cureteam(self, bm: BattleMessage_cureteam) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_boost(self, bm: BattleMessage_boost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_unboost(self, bm: BattleMessage_unboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_setboost(self, bm: BattleMessage_setboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_swapboost(self, bm: BattleMessage_swapboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_invertboost(self, bm: BattleMessage_invertboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_clearboost(self, bm: BattleMessage_clearboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_clearallboost(self, bm: BattleMessage_clearallboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_clearpositiveboost(self, bm: BattleMessage_clearpositiveboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_clearnegativeboost(self, bm: BattleMessage_clearnegativeboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_copyboost(self, bm: BattleMessage_copyboost) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_weather(self, bm: BattleMessage_weather) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_fieldstart(self, bm: BattleMessage_fieldstart) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_terastallize(self, bm: BattleMessage_terastallize) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_fieldend(self, bm: BattleMessage_fieldend) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_fieldactivate(self, bm: BattleMessage_fieldactivate) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_sidestart(self, bm: BattleMessage_sidestart) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_sideend(self, bm: BattleMessage_sideend) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_swapsideconditions(self, bm: BattleMessage_swapsideconditions) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_volstart(self, bm: BattleMessage_volstart) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_volend(self, bm: BattleMessage_volend) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_crit(self, bm: BattleMessage_crit) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_supereffective(self, bm: BattleMessage_supereffective) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_resisted(self, bm: BattleMessage_resisted) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_immune(self, bm: BattleMessage_immune) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_item(self, bm: BattleMessage_item) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_enditem(self, bm: BattleMessage_enditem) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_ability(self, bm: BattleMessage_ability) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_endability(self, bm: BattleMessage_endability) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_transform(self, bm: BattleMessage_transform) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_mega(self, bm: BattleMessage_mega) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_primal(self, bm: BattleMessage_primal) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_burst(self, bm: BattleMessage_burst) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_zpower(self, bm: BattleMessage_zpower) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_zbroken(self, bm: BattleMessage_zbroken) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_activate(self, bm: BattleMessage_activate) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_hint(self, bm: BattleMessage_hint) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_center(self, bm: BattleMessage_center) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_message(self, bm: BattleMessage_message) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_combine(self, bm: BattleMessage_combine) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_waiting(self, bm: BattleMessage_waiting) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_prepare(self, bm: BattleMessage_prepare) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_mustrecharge(self, bm: BattleMessage_mustrecharge) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_nothing(self, bm: BattleMessage_nothing) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_hitcount(self, bm: BattleMessage_hitcount) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_singlemove(self, bm: BattleMessage_singlemove) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_singleturn(self, bm: BattleMessage_singleturn) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_formechange(self, bm: BattleMessage_formechange) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_error(self, bm: BattleMessage_error) -> None:
        """
        Processes this message to update the current battle state
        """

        raise RuntimeError(f"Reached error state: {bm.MESSAGE}")

    async def processbm_bigerror(self, bm: BattleMessage_bigerror) -> None:
        """
        Processes this message to update the current battle state
        """

        raise RuntimeError(f"Reached big error state: {bm.MESSAGE}")

    async def processbm_unknown(self, bm: BattleMessage) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_init(self, bm: BattleMessage_init) -> None:
        """
        Processes this message to update the current battle state
        """

        # This signifies that the game has started, so we need to initialize our first state
        bs = BattleState(turn=1)
        self.battle.battle_states.append(bs)

    async def processbm_title(self, bm: BattleMessage_title) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_join(self, bm: BattleMessage_join) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_leave(self, bm: BattleMessage_leave) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_raw(self, bm: BattleMessage_raw) -> None:
        """
        Processes this message to update the current battle state
        """

    async def processbm_anim(self, bm: BattleMessage_anim) -> None:
        """
        Processes this message to update the current battle state
        """

pokesage/processors/showdownprocessor.py
trevorWieland/pokesage
from .abstractsage import AbstractSage
from .defaultsage import DefaultSage, Gorm
from .randomsage import Bronius, RandomSage

pokesage/sages/__init__.py
trevorWieland/pokesage
from abc import ABC, abstractmethod
from typing import Optional
from tqdm.asyncio import tqdm
from aiohttp import ClientSession

from ..battle.choices import ForceSwitchChoice, MoveDecisionChoice, TeamOrderChoice
from ..battle.state import BattleState
from ..connectors import ConnectionTermination, ConnectionTerminationCode, Connector
from ..processors import ProgressState


class AbstractSage(ABC):
    """
    Abstract Sage class for other player classes to implement with their decision functions.

    Each subclass should implement 3 main functions:
    - team_choice:
        - This function will be called when the player is asked which team order they should use.
        - For example, a teampreview message from showdown might trigger this function
        - In formats where less than a full team is used, this should submit the correct sample
    - move_choice:
        - This function will be called when the player is asked for a standard decision in the match
        - For example, every turn the player will recieve a call to this function
        - The player could choose to use moves, items, switch pokemon, or of course, resign
    - forceswitch_choice:
        - This function will be called when the player is forcibly asked to switch one or more pokemon
        - For example, when a pokemon faints they must be switched out before the next move_choice will be triggered
        - The player will have to choose which switches to make, or of course, resign
    """

    def __init__(
        self, name: str, connector: Connector, session: Optional[ClientSession] = None, use_tqdm: bool = False
    ) -> None:
        """ """

        self.name = name
        self.connector = connector
        self.session = session
        self.use_tqdm = use_tqdm

    async def launch(self) -> None:
        """
        Launches the associated connector object, as well as the aiohttp client session if needed
        """

        if self.use_tqdm:
            pbar = tqdm(total=self.connector.total_battles)
        else:
            pbar = None

        if self.session is not None:
            await self.play(session=self.session, pbar=pbar)
        else:
            async with ClientSession() as session:
                await self.play(session=session, pbar=pbar)

        if self.use_tqdm:
            pbar.close()

    async def play(self, session: ClientSession, pbar: Optional[tqdm]) -> None:
        """
        Communicates with the game using the connector.launch_connection generator
        """

        connection = self.connector.launch_connection(session)
        action = None

        while True:
            progress_state, data = await connection.asend(action)

            if progress_state == ProgressState.TEAM_ORDER:
                battle_state: BattleState = data
                action = await self.team_choice(session=session, battle_state=battle_state)
            elif progress_state == ProgressState.MOVE:
                battle_state: BattleState = data
                action = await self.move_choice(session=session, battle_state=battle_state)
            elif progress_state == ProgressState.SWITCH:
                battle_state: BattleState = data
                action = await self.forceswitch_choice(session=session, battle_state=battle_state)
            elif progress_state == ProgressState.GAME_END:
                # Not particularly helpful in this use case, but if built as a gymnasium env, this would provide `terminated`
                # Note: This means that the individual game has ended, *NOT* that the connection is closed.
                action = None

                if pbar is not None:
                    pbar.update(1)
            elif progress_state == ProgressState.FULL_END:
                # This tells us that the connection itself has been ended
                action = None

                print(data.model_dump_json(indent=2))

                break
            else:
                # This means a NO_ACTION state was returned.
                # since no action is needed, we can just continue
                action = None

            # print(progress_state)
            if action is not None:
                # print(f"Sending: {type(action)}")
                # print(action.model_dump_json(indent=2))
                pass

    @abstractmethod
    async def team_choice(self, session: ClientSession, battle_state: BattleState) -> TeamOrderChoice:
        """
        This function should return a teamorder decision based on the battle_processor given.

        The function may optionally use the session object given to make any http requests as needed.

        This function should be instantiated by the sage class, not the connection handler
        """

    @abstractmethod
    async def move_choice(self, session: ClientSession, battle_state: BattleState) -> MoveDecisionChoice:
        """
        This function should return a move decision based on the battle_processor given.

        The function may optionally use the session object given to make any http requests as needed.

        This function should be instantiated by the sage class, not the connection handler
        """

    @abstractmethod
    async def forceswitch_choice(self, session: ClientSession, battle_state: BattleState) -> ForceSwitchChoice:
        """
        This function should return a forced-switch decision based on the battle_processor given.

        The function may optionally use the session object given to make any http requests as needed.

        This function should be instantiated by the sage class, not the connection handler
        """

pokesage/sages/abstractsage.py
trevorWieland/pokesage
from aiohttp import ClientSession

from ..battle.choices import DefaultChoice, ForceSwitchChoice, MoveDecisionChoice, TeamOrderChoice
from ..battle.state import BattleState
from .abstractsage import AbstractSage


class DefaultSage(AbstractSage):
    """
    Picks the first legal option every time.

    Best used for testing connections, as this is even less sophisticated than random choices.
    """

    async def team_choice(self, session: ClientSession, battle_state: BattleState) -> TeamOrderChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        return DefaultChoice()

    async def move_choice(self, session: ClientSession, battle_state: BattleState) -> MoveDecisionChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        return DefaultChoice()

    async def forceswitch_choice(self, session: ClientSession, battle_state: BattleState) -> ForceSwitchChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        return DefaultChoice()


class Gorm(DefaultSage):
    """
    Sage-named version of the DefaultSage class.

    Cool kids use this instead of DefaultSage
    """

pokesage/sages/defaultsage.py
trevorWieland/pokesage
from aiohttp import ClientSession

from ..battle.choices import (
    DefaultChoice,
    ForceSwitchChoice,
    MoveDecisionChoice,
    TeamOrderChoice,
    TeamChoice,
    PassChoice,
    SwitchChoice,
)
from ..battle.state import BattleState
from .abstractsage import AbstractSage

from beartype.door import is_bearable
import random


class RandomSage(AbstractSage):
    """
    Picks the first legal option every time.

    Best used for testing connections, as this is even less sophisticated than random choices.
    """

    async def team_choice(self, session: ClientSession, battle_state: BattleState) -> TeamOrderChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        # TODO: Implement random sub-team ordering (ex: of the 6 options, pick 4)

        assert is_bearable(
            battle_state.battle_choice, TeamChoice
        ), f"Expected to receive a TeamChoice to decide from but got: {type(battle_state.battle_choice)} instead!"

        team_order = battle_state.battle_choice.team_order

        random.shuffle(team_order)

        return TeamChoice(team_order=team_order)

    async def move_choice(self, session: ClientSession, battle_state: BattleState) -> MoveDecisionChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        assert is_bearable(battle_state.battle_choice, list), "The given choices should be a list!"

        selected_choices = []
        for slot_choices in battle_state.battle_choice:
            if is_bearable(slot_choices, PassChoice):
                selected_choices.append(slot_choices)
            else:
                valid_choices = [c for c in slot_choices if (not is_bearable(c, SwitchChoice) or c not in slot_choices)]
                selected_choices.append(valid_choices[random.randint(0, len(valid_choices) - 1)])

        return selected_choices

    async def forceswitch_choice(self, session: ClientSession, battle_state: BattleState) -> ForceSwitchChoice:
        """
        Simply returns default choice, which will signify picking the first legal option.

        This is the equivalent to picking /choose default on showdown or running out of time in VGC
        """

        assert is_bearable(battle_state.battle_choice, list), "The given choices should be a list!"

        selected_choices = []
        for slot_choices in battle_state.battle_choice:
            if is_bearable(slot_choices, PassChoice):
                selected_choices.append(slot_choices)
            else:
                valid_choices = [c for c in slot_choices if (c not in selected_choices)]

                if len(valid_choices) == 0:
                    print(slot_choices)
                    print(selected_choices)
                    print(valid_choices)

                selected_choices.append(valid_choices[random.randint(0, len(valid_choices) - 1)])

        return selected_choices


class Bronius(RandomSage):
    """
    Sage-named version of the RandomSage class.

    Cool kids use this instead of RandomSage
    """

pokesage/sages/randomsage.py
trevorWieland/pokesage