Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

More robust battle tracking #666

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions src/poke_env/environment/abstract_battle.py
Original file line number Diff line number Diff line change
@@ -449,8 +449,8 @@ def parse_message(self, split_message: List[str]):
if event[-1].startswith("[anim]"):
event = event[:-1]

if event[-1].startswith("[from]move: "):
override_move = event.pop()[12:]
if event[-1].startswith("[from] move: "):
override_move = event.pop()[13:]

if override_move == "Sleep Talk":
# Sleep talk was used, but also reveals another move
@@ -461,7 +461,7 @@ def parse_message(self, split_message: List[str]):
override_move = None
elif self.logger is not None:
self.logger.warning(
"Unmanaged [from]move message received - move %s in cleaned up "
"Unmanaged [from] move message received - move %s in cleaned up "
"message %s in battle %s turn %d",
override_move,
event,
@@ -472,8 +472,8 @@ def parse_message(self, split_message: List[str]):
if event[-1] == "null":
event = event[:-1]

if event[-1].startswith("[from]ability: "):
revealed_ability = event.pop()[15:]
if event[-1].startswith("[from] ability: "):
revealed_ability = event.pop()[16:]
pokemon = event[2]
self.get_pokemon(pokemon).ability = revealed_ability

@@ -483,7 +483,7 @@ def parse_message(self, split_message: List[str]):
return
elif self.logger is not None:
self.logger.warning(
"Unmanaged [from]ability: message received - ability %s in "
"Unmanaged [from] ability: message received - ability %s in "
"cleaned up message %s in battle %s turn %d",
revealed_ability,
event,
10 changes: 10 additions & 0 deletions src/poke_env/environment/battle.py
Original file line number Diff line number Diff line change
@@ -94,6 +94,16 @@
self._teampreview = False
self._update_team_from_request(request["side"])

if self.active_pokemon is not None:
active_mon = self.get_pokemon(
request["side"]["pokemon"][0]["ident"],
force_self_team=True,
details=request["side"]["pokemon"][0]["details"],
)
if active_mon != self.active_pokemon:
self.active_pokemon.switch_out()
active_mon.switch_in()

Check warning on line 105 in src/poke_env/environment/battle.py

Codecov / codecov/patch

src/poke_env/environment/battle.py#L104-L105

Added lines #L104 - L105 were not covered by tests

if "active" in request:
active_request = request["active"][0]

42 changes: 7 additions & 35 deletions src/poke_env/environment/double_battle.py
Original file line number Diff line number Diff line change
@@ -130,6 +130,13 @@ def parse_request(self, request: Dict[str, Any]) -> None:
if side["pokemon"]:
self._player_role = side["pokemon"][0]["ident"][:2]
self._update_team_from_request(side)
if self.player_role is not None:
self._active_pokemon[f"{self.player_role}a"] = self.team[
request["side"]["pokemon"][0]["ident"]
]
self._active_pokemon[f"{self.player_role}b"] = self.team[
request["side"]["pokemon"][1]["ident"]
]

if "active" in request:
for active_pokemon_number, active_request in enumerate(request["active"]):
@@ -139,41 +146,6 @@ def parse_request(self, request: Dict[str, Any]) -> None:
force_self_team=True,
details=pokemon_dict["details"],
)
if self.player_role is not None:
if (
active_pokemon_number == 0
and f"{self.player_role}a" not in self._active_pokemon
):
self._active_pokemon[f"{self.player_role}a"] = active_pokemon
elif f"{self.player_role}b" not in self._active_pokemon:
self._active_pokemon[f"{self.player_role}b"] = active_pokemon
elif (
active_pokemon_number == 0
and self._active_pokemon[f"{self.player_role}a"].fainted
and self._active_pokemon[f"{self.player_role}b"]
== active_pokemon
):
(
self._active_pokemon[f"{self.player_role}a"],
self._active_pokemon[f"{self.player_role}b"],
) = (
self._active_pokemon[f"{self.player_role}b"],
self._active_pokemon[f"{self.player_role}a"],
)
elif (
active_pokemon_number == 1
and self._active_pokemon[f"{self.player_role}b"].fainted
and not active_pokemon.fainted
and self._active_pokemon[f"{self.player_role}a"]
== active_pokemon
):
(
self._active_pokemon[f"{self.player_role}a"],
self._active_pokemon[f"{self.player_role}b"],
) = (
self._active_pokemon[f"{self.player_role}b"],
self._active_pokemon[f"{self.player_role}a"],
)

if active_pokemon.fainted:
continue
1 change: 1 addition & 0 deletions src/poke_env/environment/pokemon.py
Original file line number Diff line number Diff line change
@@ -378,6 +378,7 @@ def set_hp_status(self, hp_status: str, store=False):
self.end_effect("yawn")
else:
hp = hp_status
self._status = None

current_hp, max_hp = "".join([c for c in hp if c in "0123456789/"]).split("/")
self._current_hp = int(current_hp)
245 changes: 140 additions & 105 deletions src/poke_env/player/player.py
Original file line number Diff line number Diff line change
@@ -9,7 +9,7 @@
from asyncio import Condition, Event, Queue, Semaphore
from logging import Logger
from time import perf_counter
from typing import Any, Awaitable, Dict, List, Optional, Union
from typing import Any, Awaitable, Dict, List, Optional, Tuple, Union

import orjson

@@ -146,6 +146,7 @@

self._battles: Dict[str, AbstractBattle] = {}
self._battle_semaphore: Semaphore = create_in_poke_loop(Semaphore, 0)
self._reqs: Dict[str, List[List[str]]] = {}

self._battle_start_condition: Condition = create_in_poke_loop(Condition)
self._battle_count_queue: Queue[Any] = create_in_poke_loop(
@@ -264,119 +265,153 @@
:type split_message: str
"""
# Battle messages can be multiline
battle_tag = split_messages[0][0]
if (
len(split_messages) > 1
and len(split_messages[1]) > 1
and split_messages[1][1] == "init"
):
battle_info = split_messages[0][0].split("-")
battle_info = battle_tag.split("-")
battle = await self._create_battle(battle_info)
else:
battle = await self._get_battle(split_messages[0][0])

for split_message in split_messages[1:]:
if len(split_message) <= 1:
continue
elif split_message[1] == "":
battle.parse_message(split_message)
elif split_message[1] in self.MESSAGES_TO_IGNORE:
pass
elif split_message[1] == "request":
if split_message[2]:
request = orjson.loads(split_message[2])
battle.parse_request(request)
if battle.move_on_next_request:
await self._handle_battle_request(battle)
battle.move_on_next_request = False
elif split_message[1] == "win" or split_message[1] == "tie":
if split_message[1] == "win":
battle.won_by(split_message[2])
else:
battle.tied()
await self._battle_count_queue.get()
self._battle_count_queue.task_done()
self._battle_finished_callback(battle)
async with self._battle_end_condition:
self._battle_end_condition.notify_all()
elif split_message[1] == "error":
self.logger.log(
25, "Error message received: %s", "|".join(split_message)
battle = await self._get_battle(battle_tag)
request = self._reqs.pop(battle_tag, None)
if split_messages[1][1] == "request":
protocol = None
self._reqs[battle_tag] = split_messages

Check warning on line 281 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L280-L281

Added lines #L280 - L281 were not covered by tests
else:
protocol = split_messages
if protocol is not None or request is not None:
split_messages = protocol or [[f">{battle_tag}"]]
if request is not None:
split_messages += [request[1]]

Check warning on line 287 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L287

Added line #L287 was not covered by tests
results = [
await self._process_split_message(m, battle) for m in split_messages
]
should_process_request = any([r[0] for r in results])
is_from_teampreview = any([r[1] for r in results])
should_maybe_default = any([r[2] for r in results])
if should_process_request:
await self._handle_battle_request(

Check warning on line 295 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L295

Added line #L295 was not covered by tests
battle,
from_teampreview_request=is_from_teampreview,
maybe_default_order=should_maybe_default,
)
if split_message[2].startswith(
"[Invalid choice] Sorry, too late to make a different move"
):
if battle.trapped:
await self._handle_battle_request(battle)
elif split_message[2].startswith(
"[Unavailable choice] Can't switch: The active Pokémon is "
"trapped"
) or split_message[2].startswith(
"[Invalid choice] Can't switch: The active Pokémon is trapped"
):
battle.trapped = True
await self._handle_battle_request(battle)
elif split_message[2].startswith(
"[Invalid choice] Can't switch: You can't switch to an active "
"Pokémon"
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Can't switch: You can't switch to a fainted "
"Pokémon"
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Can't move: Invalid target for"
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Can't move: You can't choose a target for"
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Can't move: "
) and split_message[2].endswith("needs a target"):
await self._handle_battle_request(battle, maybe_default_order=True)
elif (
split_message[2].startswith("[Invalid choice] Can't move: Your")
and " doesn't have a move matching " in split_message[2]
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Incomplete choice: "
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Unavailable choice]"
) and split_message[2].endswith("is disabled"):
battle.move_on_next_request = True
elif split_message[2].startswith("[Invalid choice]") and split_message[
2
].endswith("is disabled"):
battle.move_on_next_request = True
elif split_message[2].startswith(
"[Invalid choice] Can't move: You sent more choices than unfainted"
" Pokémon."
):
await self._handle_battle_request(battle, maybe_default_order=True)
elif split_message[2].startswith(
"[Invalid choice] Can't move: You can only Terastallize once per battle."
):
await self._handle_battle_request(battle, maybe_default_order=True)
else:
self.logger.critical("Unexpected error message: %s", split_message)
elif split_message[1] == "turn":
battle.parse_message(split_message)
await self._handle_battle_request(battle)
elif split_message[1] == "teampreview":
battle.parse_message(split_message)
await self._handle_battle_request(battle, from_teampreview_request=True)
elif split_message[1] == "bigerror":
self.logger.warning("Received 'bigerror' message: %s", split_message)
elif split_message[1] == "uhtml" and split_message[2] == "otsrequest":
await self._handle_ots_request(battle.battle_tag)

async def _process_split_message(
self, split_message: List[str], battle: AbstractBattle
) -> Tuple[bool, bool, bool]:
should_process_request = False
is_from_teampreview = False
should_maybe_default = False
if len(split_message) <= 1:
pass
elif split_message[1] == "":
battle.parse_message(split_message)

Check warning on line 310 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L310

Added line #L310 was not covered by tests
elif split_message[1] in self.MESSAGES_TO_IGNORE:
pass

Check warning on line 312 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L312

Added line #L312 was not covered by tests
elif split_message[1] == "request":
if split_message[2]:
request = orjson.loads(split_message[2])
battle.parse_request(request)
if battle.move_on_next_request:
should_process_request = True
battle.move_on_next_request = False

Check warning on line 319 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L314-L319

Added lines #L314 - L319 were not covered by tests
elif split_message[1] == "win" or split_message[1] == "tie":
if split_message[1] == "win":
battle.won_by(split_message[2])
else:
battle.parse_message(split_message)
battle.tied()

Check warning on line 324 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L324

Added line #L324 was not covered by tests
await self._battle_count_queue.get()
self._battle_count_queue.task_done()
self._battle_finished_callback(battle)
async with self._battle_end_condition:
self._battle_end_condition.notify_all()
elif split_message[1] == "error":
self.logger.log(25, "Error message received: %s", "|".join(split_message))
if split_message[2].startswith(

Check warning on line 332 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L331-L332

Added lines #L331 - L332 were not covered by tests
"[Invalid choice] Sorry, too late to make a different move"
):
if battle.trapped:
should_process_request = True
elif split_message[2].startswith(

Check warning on line 337 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L335-L337

Added lines #L335 - L337 were not covered by tests
"[Unavailable choice] Can't switch: The active Pokémon is " "trapped"
) or split_message[2].startswith(
"[Invalid choice] Can't switch: The active Pokémon is trapped"
):
battle.trapped = True
should_process_request = True
elif split_message[2].startswith(

Check warning on line 344 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L342-L344

Added lines #L342 - L344 were not covered by tests
"[Invalid choice] Can't switch: You can't switch to an active "
"Pokémon"
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith(

Check warning on line 350 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L348-L350

Added lines #L348 - L350 were not covered by tests
"[Invalid choice] Can't switch: You can't switch to a fainted "
"Pokémon"
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith(

Check warning on line 356 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L354-L356

Added lines #L354 - L356 were not covered by tests
"[Invalid choice] Can't move: Invalid target for"
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith(

Check warning on line 361 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L359-L361

Added lines #L359 - L361 were not covered by tests
"[Invalid choice] Can't move: You can't choose a target for"
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith(

Check warning on line 366 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L364-L366

Added lines #L364 - L366 were not covered by tests
"[Invalid choice] Can't move: "
) and split_message[2].endswith("needs a target"):
should_process_request = True
should_maybe_default = True
elif (

Check warning on line 371 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L369-L371

Added lines #L369 - L371 were not covered by tests
split_message[2].startswith("[Invalid choice] Can't move: Your")
and " doesn't have a move matching " in split_message[2]
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith("[Invalid choice] Incomplete choice: "):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith("[Unavailable choice]") and split_message[

Check warning on line 380 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L375-L380

Added lines #L375 - L380 were not covered by tests
2
].endswith("is disabled"):
battle.move_on_next_request = True
elif split_message[2].startswith("[Invalid choice]") and split_message[

Check warning on line 384 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L383-L384

Added lines #L383 - L384 were not covered by tests
2
].endswith("is disabled"):
battle.move_on_next_request = True
elif split_message[2].startswith(

Check warning on line 388 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L387-L388

Added lines #L387 - L388 were not covered by tests
"[Invalid choice] Can't move: You sent more choices than unfainted"
" Pokémon."
):
should_process_request = True
should_maybe_default = True
elif split_message[2].startswith(

Check warning on line 394 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L392-L394

Added lines #L392 - L394 were not covered by tests
"[Invalid choice] Can't move: You can only Terastallize once per battle."
):
should_process_request = True
should_maybe_default = True

Check warning on line 398 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L397-L398

Added lines #L397 - L398 were not covered by tests
else:
self.logger.critical("Unexpected error message: %s", split_message)

Check warning on line 400 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L400

Added line #L400 was not covered by tests
elif split_message[1] == "turn":
battle.parse_message(split_message)
should_process_request = True

Check warning on line 403 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L402-L403

Added lines #L402 - L403 were not covered by tests
elif split_message[1] == "teampreview":
battle.parse_message(split_message)
should_process_request = True
is_from_teampreview = True

Check warning on line 407 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L405-L407

Added lines #L405 - L407 were not covered by tests
elif split_message[1] == "bigerror":
self.logger.warning("Received 'bigerror' message: %s", split_message)

Check warning on line 409 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L409

Added line #L409 was not covered by tests
elif split_message[1] == "uhtml" and split_message[2] == "otsrequest":
await self._handle_ots_request(battle.battle_tag)

Check warning on line 411 in src/poke_env/player/player.py

Codecov / codecov/patch

src/poke_env/player/player.py#L411

Added line #L411 was not covered by tests
else:
battle.parse_message(split_message)
return should_process_request, is_from_teampreview, should_maybe_default

async def _handle_battle_request(
self,
Loading