PokéLLMon 源码解析（一）

本文主要是介绍PokéLLMon 源码解析（一），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

`.\PokeLLMon\poke_env\concurrency.py`

# 导入必要的模块
import asyncio
import atexit
import sys
from logging import CRITICAL, disable
from threading import Thread
from typing import Any, List# 在新线程中运行事件循环
def __run_loop(loop: asyncio.AbstractEventLoop):asyncio.set_event_loop(loop)loop.run_forever()# 停止事件循环
def __stop_loop(loop: asyncio.AbstractEventLoop, thread: Thread):disable(CRITICAL)tasks: List[asyncio.Task[Any]] = []for task in asyncio.all_tasks(loop):task.cancel()tasks.append(task)cancelled = Falseshutdown = asyncio.run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop)shutdown.result()while not cancelled:cancelled = Truefor task in tasks:if not task.done():cancelled = Falseloop.call_soon_threadsafe(loop.stop)thread.join()loop.call_soon_threadsafe(loop.close)# 清理事件循环
def __clear_loop():__stop_loop(POKE_LOOP, _t)# 在事件循环中异步创建对象
async def _create_in_poke_loop_async(cls_: Any, *args: Any, **kwargs: Any) -> Any:return cls_(*args, **kwargs)# 在事件循环中创建对象
def create_in_poke_loop(cls_: Any, *args: Any, **kwargs: Any) -> Any:try:# Python >= 3.7loop = asyncio.get_running_loop()except AttributeError:# Python < 3.7 so get_event_loop won't raise exceptionsloop = asyncio.get_event_loop()except RuntimeError:# asyncio.get_running_loop raised exception so no loop is runningloop = Noneif loop == POKE_LOOP:return cls_(*args, **kwargs)else:return asyncio.run_coroutine_threadsafe(_create_in_poke_loop_async(cls_, *args, **kwargs), POKE_LOOP).result()# 处理线程中的协程
async def handle_threaded_coroutines(coro: Any):task = asyncio.run_coroutine_threadsafe(coro, POKE_LOOP)await asyncio.wrap_future(task)return task.result()# 创建新的事件循环
POKE_LOOP = asyncio.new_event_loop()
py_ver = sys.version_info
_t = Thread(target=__run_loop, args=(POKE_LOOP,), daemon=True)
_t.start()
atexit.register(__clear_loop)

`.\PokeLLMon\poke_env\data\gen_data.py`

# 导入必要的模块和函数
from __future__ import annotations
import os
from functools import lru_cache
from typing import Any, Dict, Optional, Union
import orjson
from poke_env.data.normalize import to_id_str# 定义一个类 GenData
class GenData:# 限制实例的属性，只能包含在 __slots__ 中指定的属性__slots__ = ("gen", "moves", "natures", "pokedex", "type_chart", "learnset")# 定义一个类变量 UNKNOWN_ITEMUNKNOWN_ITEM = "unknown_item"# 定义一个类变量 _gen_data_per_gen，用于存储不同世代的 GenData 实例_gen_data_per_gen: Dict[int, GenData] = {}# 初始化方法，接受一个 gen 参数def __init__(self, gen: int):# 如果该世代的 GenData 已经初始化过，则抛出异常if gen in self._gen_data_per_gen:raise ValueError(f"GenData for gen {gen} already initialized.")# 初始化实例属性self.gen = genself.moves = self.load_moves(gen)self.natures = self.load_natures()self.pokedex = self.load_pokedex(gen)self.type_chart = self.load_type_chart(gen)self.learnset = self.load_learnset()# 定义深拷贝方法，返回当前实例本身def __deepcopy__(self, memodict: Optional[Dict[int, Any]] = None) -> GenData:return self# 加载指定世代的招式数据def load_moves(self, gen: int) -> Dict[str, Any]:with open(os.path.join(self._static_files_root, "moves", f"gen{gen}moves.json")) as f:return orjson.loads(f.read())# 加载自然性格数据def load_natures(self) -> Dict[str, Dict[str, Union[int, float]]]:with open(os.path.join(self._static_files_root, "natures.json")) as f:return orjson.loads(f.read())# 加载学会招式数据def load_learnset(self) -> Dict[str, Dict[str, Union[int, float]]]:with open(os.path.join(self._static_files_root, "learnset.json")) as f:return orjson.loads(f.read())# 加载宝可梦图鉴数据，根据给定的世代号def load_pokedex(self, gen: int) -> Dict[str, Any]:# 打开对应世代号的宝可梦图鉴 JSON 文件with open(os.path.join(self._static_files_root, "pokedex", f"gen{gen}pokedex.json")) as f:# 使用 orjson 库加载 JSON 文件内容dex = orjson.loads(f.read())# 创建一个空字典用于存储其他形态的宝可梦数据other_forms_dex: Dict[str, Any] = {}# 遍历宝可梦图鉴数据for value in dex.values():# 如果存在"cosmeticFormes"字段if "cosmeticFormes" in value:# 遍历所有的其他形态for other_form in value["cosmeticFormes"]:# 将其他形态的数据存入字典中other_forms_dex[to_id_str(other_form)] = value# 处理皮卡丘的特殊形态for name, value in dex.items():# 如果名称以"pikachu"开头且不是"pikachu"或"pikachugmax"if name.startswith("pikachu") and name not in {"pikachu", "pikachugmax"}:# 添加对应的"gmax"形态数据other_forms_dex[name + "gmax"] = dex["pikachugmax"]# 将其他形态数据合并到原始数据中dex.update(other_forms_dex)# 更新宝可梦数据中的"species"字段for name, value in dex.items():# 如果存在"baseSpecies"字段if "baseSpecies" in value:# 将"species"字段设置为"baseSpecies"字段的值value["species"] = value["baseSpecies"]else:# 否则将"baseSpecies"字段设置为名称的标准化形式value["baseSpecies"] = to_id_str(name)# 返回更新后的宝可梦图鉴数据return dex# 加载指定世代的类型相克表def load_type_chart(self, gen: int) -> Dict[str, Dict[str, float]]:# 打开对应世代的类型相克表 JSON 文件with open(os.path.join(self._static_files_root, "typechart", f"gen{gen}typechart.json")) as chart:# 将 JSON 文件内容加载为字典json_chart = orjson.loads(chart.read())# 获取所有类型并转换为大写types = [str(type_).upper() for type_ in json_chart]# 初始化类型相克表字典type_chart = {type_1: {type_2: 1.0 for type_2 in types} for type_1 in types}# 遍历类型相克表数据for type_, data in json_chart.items():type_ = type_.upper()# 遍历每个类型对应的伤害倍数for other_type, damage_taken in data["damageTaken"].items():if other_type.upper() not in types:continue# 确保伤害倍数在合法范围内assert damage_taken in {0, 1, 2, 3}, (data["damageTaken"], type_)# 根据伤害倍数设置相应的伤害值if damage_taken == 0:type_chart[type_][other_type.upper()] = 1elif damage_taken == 1:type_chart[type_][other_type.upper()] = 2elif damage_taken == 2:type_chart[type_][other_type.upper()] = 0.5elif damage_taken == 3:type_chart[type_][other_type.upper()] = 0# 确保所有类型都在类型相克表中assert set(types).issubset(set(type_chart))# 确保类型相克表的长度与类型列表长度相同assert len(type_chart) == len(types)# 确保每个类型的相克效果字典长度与类型列表长度相同for effectiveness in type_chart.values():assert len(effectiveness) == len(types)# 返回类型相克表return type_chart# 返回静态文件根目录路径@propertydef _static_files_root(self) -> str:return os.path.join(os.path.dirname(os.path.realpath(__file__)), "static")# 根据世代创建 GenData 实例@classmethod@lru_cache(None)def from_gen(cls, gen: int) -> GenData:# 创建指定世代的 GenData 实例gen_data = GenData(gen)# 将 GenData 实例存储到类属性中cls._gen_data_per_gen[gen] = gen_datareturn gen_data# 根据格式创建 GenData 实例@classmethod@lru_cache(None)def from_format(cls, format: str) -> GenData:# 解析出世代号gen = int(format[3])  # Update when Gen 10 comes# 根据世代号创建 GenData 实例return cls.from_gen(gen)

`.\PokeLLMon\poke_env\data\normalize.py`

# 导入 functools 模块中的 lru_cache 装饰器
from functools import lru_cache# 使用 lru_cache 装饰器缓存函数的结果，缓存大小为 2 的 13 次方
@lru_cache(2**13)
# 定义函数 to_id_str，将全名转换为对应的 ID 字符串
def to_id_str(name: str) -> str:"""Converts a full-name to its corresponding id string.:param name: The name to convert.:type name: str:return: The corresponding id string.:rtype: str"""# 将输入的名字中的字母和数字提取出来，转换为小写，并拼接成字符串return "".join(char for char in name if char.isalnum()).lower()

`.\PokeLLMon\poke_env\data\replay_template.py`

# 导入 os 模块
import os# 打开 replay_template.html 文件，使用绝对路径拼接得到文件路径
with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), "static", "replay_template.html")
) as f:# 读取文件内容并赋值给 REPLAY_TEMPLATE 变量REPLAY_TEMPLATE = f.read()

`.\PokeLLMon\poke_env\data\static\abilities\construct_ability_json.py`

# 导入 pandas 库，用于处理数据
import pandas as pd
# 导入 json 库，用于处理 JSON 数据# 从 "raw.txt" 文件中读取数据，使用制表符作为分隔符
X = pd.read_csv("raw.txt", "\t")# 从 X 中提取 Name 列的数值，转换为列表
name = list(X.Name.values)
# 从 X 中提取 Description 列的数值，转换为列表
description = list(X.Description.values)
# 将 name 列中的每个元素转换为小写，并去除空格后存储到 name_new 列表中
name_new = list(map(lambda x: x.lower().replace(" ", ""), name))# 创建空字典 ability_dict
ability_dict = {}# 遍历 name 列的长度
for i in range(len(name)):# 将 name_new[i] 作为键，name[i] 和 description[i] 组成的字典作为值，存储到 ability_dict 中ability_dict[name_new[i]] = {"name": name[i], "effect": description[i]}# 打印 "pause"
print("pause")# 打开 "ability_effect.json" 文件，以写入模式打开，文件对象为 f
with open("ability_effect.json", "w") as f:# 将 ability_dict 写入到 f 中，格式化输出，缩进为 4json.dump(ability_dict, f, indent=4)# 打印 "pause"
print("pause")

`.\PokeLLMon\poke_env\data\static\items\construct_item_json.py`

# 导入 pandas 和 json 模块
import pandas as pd
import json# 从 "raw.txt" 文件中读取数据，使用制表符作为分隔符
X = pd.read_csv("raw.txt", "\t")# 获取 Name、Effect 和 Category 列的数值
name = list(X.Name.values)
effect = list(X.Effect.values)
category = list(X.Category.values)# 创建空字典 item_dict
item_dict = {}# 遍历 name 列的长度
for i in range(len(name)):# 将 name 列中的值按 " icon " 分割，取第一个部分作为新的名称new_name = name[i].split(" icon ")[0]# 如果 effect 列中的值不是 NaNif str(effect[i]) != "nan":# 将新名称转换为小写并去除空格，作为字典的键，值为包含名称和效果的字典item_dict[new_name.lower().replace(" ", "")] = {"name":new_name, "effect":effect[i]}# 打印 "pause"
print("pause")# 将 item_dict 写入到 "item_effect.json" 文件中，格式化输出，缩进为 4
with open("item_effect.json", "w") as f:json.dump(item_dict, f, indent=4)# 打印 "pause"
print("pause")

`.\PokeLLMon\poke_env\data\static\moves\extract_gen8_moves.py`

# 导入 json 和 re 模块
import json
import re
import json# 初始化存储招式名称和效果的列表
move_name_list = []
move_effect_list = []# 打开文件 "gen8_raw.txt" 以只读模式
with open("gen8_raw.txt", "r") as f:idx = 0# 循环读取文件中的每一行数据for i in range(2184):data = f.readline()# 每三行数据为一组，分别提取招式名称和效果if idx % 3 == 0:move_name = data.split("    ")[0]move_name_list.append(move_name)elif idx % 3 == 1:effect = data[:-1]move_effect_list.append(effect)idx += 1# 将招式名称和效果列表组合成字典
move2effect = dict(zip(move_name_list, move_effect_list))# 打开文件 "gen8moves.json" 以只读模式
with open("gen8moves.json", "r") as f:# 加载 JSON 文件内容到 gen8moves 字典中gen8moves = json.load(f)# 初始化新的招式名称到效果的字典
move2effect_new = dict()
# 遍历 gen8moves 字典中的每个招式和信息
for move, info in gen8moves.items():try:# 尝试从 move2effect 字典中获取招式对应的效果effect = move2effect[info['name']]# 将招式和效果添加到新的字典中move2effect_new[move] = effectexcept:# 如果出现异常则继续下一个招式continue# 打开文件 "gen8moves_effect.json" 以写入模式
with open("gen8moves_effect.json", "w") as f:# 将新的招式名称到效果的字典以美观的格式写入到文件中json.dump(move2effect_new, f, indent=4)

`.\PokeLLMon\poke_env\data\init.py`

# 从 poke_env.data.gen_data 模块中导入 GenData 类
# 从 poke_env.data.normalize 模块中导入 to_id_str 函数
# 从 poke_env.data.replay_template 模块中导入 REPLAY_TEMPLATE 常量
from poke_env.data.gen_data import GenData
from poke_env.data.normalize import to_id_str
from poke_env.data.replay_template import REPLAY_TEMPLATE# 定义 __all__ 列表，包含需要导出的模块成员
__all__ = ["REPLAY_TEMPLATE","GenData","to_id_str",
]

`.\PokeLLMon\poke_env\environment\abstract_battle.py`

# 导入必要的模块
import os
from abc import ABC, abstractmethod
from logging import Logger
from typing import Any, Dict, List, Optional, Set, Tuple, Union# 导入自定义模块
from poke_env.data import GenData, to_id_str
from poke_env.data.replay_template import REPLAY_TEMPLATE
from poke_env.environment.field import Field
from poke_env.environment.pokemon import Pokemon
from poke_env.environment.side_condition import STACKABLE_CONDITIONS, SideCondition
from poke_env.environment.weather import Weather# 定义一个抽象类 AbstractBattle
class AbstractBattle(ABC):# 定义一个常量集合，包含需要忽略的消息MESSAGES_TO_IGNORE = {"-anim","-burst","-block","-center","-crit","-combine","-fail","-fieldactivate","-hint","-hitcount","-ohko","-miss","-notarget","-nothing","-resisted","-singlemove","-singleturn","-supereffective","-waiting","-zbroken","askreg","debug","chat","c","crit","deinit","gametype","gen","html","init","immune","join","j","J","leave","l","L","name","n","rated","resisted","split","supereffective","teampreview","tier","upkeep","zbroken",}# 定义类的属性，使用 __slots__ 来限制实例的属性，提高内存利用效率__slots__ = ("_anybody_inactive","_available_moves","_available_switches","_battle_tag","_can_dynamax","_can_mega_evolve","_can_tera","_can_z_move","_data","_dynamax_turn","_fields","_finished","_force_switch","_format","in_team_preview","_max_team_size","_maybe_trapped","_move_on_next_request","_opponent_can_dynamax","_opponent_can_mega_evolve","_opponent_can_terrastallize","_opponent_can_z_move","_opponent_dynamax_turn","_opponent_rating","_opponent_side_conditions","_opponent_team","_opponent_username","_player_role","_player_username","_players","_rating","_reconnected","_replay_data","_rqid","rules","_reviving","_save_replays","_side_conditions","_team_size","_team","_teampreview_opponent_team","_teampreview","_trapped","_turn","_wait","_weather","_won","logger",)# 初始化方法，用于创建类的实例def __init__(self,battle_tag: str,  # 战斗标签username: str,  # 用户名logger: Logger,  # 日志记录器save_replays: Union[str, bool],  # 保存重播记录的路径或布尔值gen: int,  # 世代# 加载数据self._data = GenData.from_gen(gen)# 工具属性self._battle_tag: str = battle_tagself._format: Optional[str] = Noneself._max_team_size: Optional[int] = Noneself._opponent_username: Optional[str] = Noneself._player_role: Optional[str] = Noneself._player_username: str = usernameself._players: List[Dict[str, str]] = []self._replay_data: List[List[str]] = []self._save_replays: Union[str, bool] = save_replaysself._team_size: Dict[str, int] = {}self._teampreview: bool = Falseself._teampreview_opponent_team: Set[Pokemon] = set()self._anybody_inactive: bool = Falseself._reconnected: bool = Trueself.logger: Optional[Logger] = logger# 回合选择属性self.in_team_preview: bool = Falseself._move_on_next_request: bool = Falseself._wait: Optional[bool] = None# 战斗状态属性self._dynamax_turn: Optional[int] = Noneself._finished: bool = Falseself._rqid = 0self.rules: List[str] = []self._turn: int = 0self._opponent_can_terrastallize: bool = Trueself._opponent_dynamax_turn: Optional[int] = Noneself._opponent_rating: Optional[int] = Noneself._rating: Optional[int] = Noneself._won: Optional[bool] = None# 游戏中的战斗状态属性self._weather: Dict[Weather, int] = {}self._fields: Dict[Field, int] = {}  # set()self._opponent_side_conditions: Dict[SideCondition, int] = {}  # set()self._side_conditions: Dict[SideCondition, int] = {}  # set()self._reviving: bool = False# Pokemon 属性self._team: Dict[str, Pokemon] = {}self._opponent_team: Dict[str, Pokemon] = {}# 定义一个方法用于获取精灵信息def get_pokemon(self,identifier: str,force_self_team: bool = False,details: str = "",request: Optional[Dict[str, Any]] = None,# 定义一个抽象方法用于清除所有精灵的增益效果@abstractmethoddef clear_all_boosts(self):pass# 检查伤害信息中是否包含关于道具的信息def _check_damage_message_for_item(self, split_message: List[str]):# 捕获对方精灵受到道具伤害的情况# 道具属于未受伤害的一方if (len(split_message) == 6and split_message[4].startswith("[from] item:")and split_message[5].startswith("[of]")):item = split_message[4].split("item:")[-1]pkmn = split_message[5].split("[of]")[-1].strip()self.get_pokemon(pkmn).item = to_id_str(item)# 捕获自身精灵受到道具伤害的情况# 道具属于受伤害的一方elif len(split_message) == 5 and split_message[4].startswith("[from] item:"):item = split_message[4].split("item:")[-1]pkmn = split_message[2]self.get_pokemon(pkmn).item = to_id_str(item)def _check_damage_message_for_ability(self, split_message: List[str]):# 检查是否有对手的能力造成伤害的消息# 物品来自未受伤害的一方# 例如:#   |-damage|p2a: Archeops|88/100|[from] ability: Iron Barbs|[of] p1a: Ferrothornif (len(split_message) == 6and split_message[4].startswith("[from] ability:")and split_message[5].startswith("[of]")):# 从消息中提取能力信息ability = split_message[4].split("ability:")[-1]# 从消息中提取宝可梦信息pkmn = split_message[5].split("[of]")[-1].strip()# 设置宝可梦的能力self.get_pokemon(pkmn).ability = to_id_str(ability)def _check_heal_message_for_item(self, split_message: List[str]):# 检查是否有宝可梦从自己的物品中恢复# 检查物品不为 None 是必要的，因为 PS 模拟器会在消耗掉一颗树果后才显示恢复消息# 例子:#  |-heal|p2a: Quagsire|100/100|[from] item: Leftovers#  |-heal|p2a: Quagsire|100/100|[from] item: Sitrus Berryif len(split_message) == 5 and split_message[4].startswith("[from] item:"):# 从消息中提取宝可梦信息pkmn = split_message[2]# 从消息中提取物品信息item = split_message[4].split("item:")[-1]# 获取宝可梦对象pkmn_object = self.get_pokemon(pkmn)# 如果宝可梦已经有物品，则设置物品if pkmn_object.item is not None:pkmn_object.item = to_id_str(item)# 检查治疗消息中是否包含能力相关信息def _check_heal_message_for_ability(self, split_message: List[str]):# 捕获当一方通过自身能力进行治疗的情况# PS 服务器发送的 "of" 组件有点误导性#   它暗示能力来自对立方# 示例:#   |-heal|p2a: Quagsire|100/100|[from] ability: Water Absorb|[of] p1a: Genesectif len(split_message) == 6 and split_message[4].startswith("[from] ability:"):# 提取能力信息ability = split_message[4].split("ability:")[-1]# 提取宝可梦名称pkmn = split_message[2]# 设置宝可梦的能力self.get_pokemon(pkmn).ability = to_id_str(ability)@abstractmethod# 结束幻象状态的抽象方法def end_illusion(self, pokemon_name: str, details: str):pass# 结束幻象状态的具体实现def _end_illusion_on(self, illusionist: Optional[str], illusioned: Optional[Pokemon], details: str):# 如果没有幻象者，则抛出异常if illusionist is None:raise ValueError("Cannot end illusion without an active pokemon.")# 如果没有被幻象的宝可梦，则抛出异常if illusioned is None:raise ValueError("Cannot end illusion without an illusioned pokemon.")# 获取幻象者的宝可梦对象illusionist_mon = self.get_pokemon(illusionist, details=details)# 如果幻象者和被幻象的宝可梦是同一个，则直接返回幻象者if illusionist_mon is illusioned:return illusionist_mon# 将幻象者切换到战斗状态illusionist_mon.switch_in(details=details)# 设置幻象者的状态illusionist_mon.status = (illusioned.status.name if illusioned.status is not None else None)# 设置幻象者的生命值illusionist_mon.set_hp(f"{illusioned.current_hp}/{illusioned.max_hp}")# 标记被幻象的宝可梦已经解除幻象状态illusioned.was_illusioned()return illusionist_mon# 处理场地结束状态的方法def _field_end(self, field_str: str):# 从 Showdown 消息中创建场地对象field = Field.from_showdown_message(field_str)# 如果场地不是未知状态，则移除该场地if field is not Field.UNKNOWN:self._fields.pop(field)# 定义一个方法，用于处理战场开始的字段信息def field_start(self, field_str: str):# 将传入的字段信息转换为Field对象field = Field.from_showdown_message(field_str)# 如果字段是地形字段if field.is_terrain:# 更新战场上的字段信息，移除之前的地形字段self._fields = {field: turnfor field, turn in self._fields.items()if not field.is_terrain}# 将当前字段信息添加到战场上self._fields[field] = self.turn# 完成战斗def _finish_battle(self):# 如果需要保存战斗回放if self._save_replays:# 根据保存回放的设置确定保存的文件夹if self._save_replays is True:folder = "replays"else:folder = str(self._save_replays)# 如果文件夹不存在，则创建文件夹if not os.path.exists(folder):os.mkdir(folder)# 打开文件，写入格式化后的回放数据with open(os.path.join(folder, f"{self._player_username} - {self.battle_tag}.html"),"w+",encoding="utf-8",) as f:formatted_replay = REPLAY_TEMPLATE# 替换模板中的占位符为实际数据formatted_replay = formatted_replay.replace("{BATTLE_TAG}", f"{self.battle_tag}")formatted_replay = formatted_replay.replace("{PLAYER_USERNAME}", f"{self._player_username}")formatted_replay = formatted_replay.replace("{OPPONENT_USERNAME}", f"{self._opponent_username}")replay_log = f">{self.battle_tag}" + "\n".join(["|".join(split_message) for split_message in self._replay_data])formatted_replay = formatted_replay.replace("{REPLAY_LOG}", replay_log)f.write(formatted_replay)# 标记战斗结束self._finished = True# 抽象方法，用于解析请求@abstractmethoddef parse_request(self, request: Dict[str, Any]):pass# 注册对手的队伍信息def _register_teampreview_pokemon(self, player: str, details: str):# 如果玩家不是当前玩家角色if player != self._player_role:# 创建Pokemon对象，并添加到对手的队伍信息中mon = Pokemon(details=details, gen=self._data.gen)self._teampreview_opponent_team.add(mon)# 根据给定的边（side）和条件字符串（condition_str）来结束边的状态def side_end(self, side: str, condition_str: str):# 如果边的前两个字符与玩家角色相同，则使用边的条件if side[:2] == self._player_role:conditions = self.side_conditionselse:conditions = self.opponent_side_conditions# 从 Showdown 消息中创建边的条件对象condition = SideCondition.from_showdown_message(condition_str)# 如果条件不是未知状态，则从条件中移除if condition is not SideCondition.UNKNOWN:conditions.pop(condition)# 根据给定的边（side）和条件字符串（condition_str）来开始边的状态def _side_start(self, side: str, condition_str: str):# 如果边的前两个字符与玩家角色相同，则使用边的条件if side[:2] == self._player_role:conditions = self.side_conditionselse:conditions = self.opponent_side_conditions# 从 Showdown 消息中创建边的条件对象condition = SideCondition.from_showdown_message(condition_str)# 如果条件可以叠加，则将条件添加到边的条件中if condition in STACKABLE_CONDITIONS:conditions[condition] = conditions.get(condition, 0) + 1# 如果条件不在边的条件中，则将条件添加到边的条件中，并记录回合数elif condition not in conditions:conditions[condition] = self.turn# 交换精灵，暂未实现def _swap(self, pokemon_str: str, slot: str):if self.logger is not None:self.logger.warning("swap method in Battle is not implemented")# 切换精灵的抽象方法@abstractmethoddef switch(self, pokemon_str: str, details: str, hp_status: str):pass# 平局结束战斗def tied(self):self._finish_battle()# 从请求中更新队伍信息def _update_team_from_request(self, side: Dict[str, Any]):for pokemon in side["pokemon"]:# 如果精灵在队伍中，则更新精灵信息，否则创建新的精灵if pokemon["ident"] in self._team:self._team[pokemon["ident"]].update_from_request(pokemon)else:self.get_pokemon(pokemon["ident"], force_self_team=True, request=pokemon)# 根据获胜玩家名字结束战斗def won_by(self, player_name: str):# 如果获胜玩家名字与玩家用户名相同，则设置胜利标志为 True，否则为 Falseif player_name == self._player_username:self._won = Trueelse:self._won = False# 结束战斗self._finish_battle()# 结束回合def end_turn(self, turn: int):# 更新当前回合数self.turn = turn# 对所有活跃的精灵执行结束回合操作for mon in self.all_active_pokemons:if mon:mon.end_turn()# 获取当前活跃的精灵的抽象属性@property@abstractmethoddef active_pokemon(self) -> Any:pass@property@abstractmethoddef all_active_pokemons(self) -> List[Optional[Pokemon]]:pass@property@abstractmethoddef available_moves(self) -> Any:pass@property@abstractmethoddef available_switches(self) -> Any:pass@propertydef battle_tag(self) -> str:""":return: The battle identifier.:rtype: str"""return self._battle_tag@property@abstractmethoddef can_dynamax(self) -> Any:pass@property@abstractmethoddef can_mega_evolve(self) -> Any:pass@property@abstractmethoddef can_z_move(self) -> Any:pass@property@abstractmethoddef can_tera(self) -> Any:pass@propertydef dynamax_turns_left(self) -> Optional[int]:""":return: How many turns of dynamax are left. None if dynamax is not active:rtype: int, optional"""if self._dynamax_turn is not None and any(map(lambda pokemon: pokemon.is_dynamaxed, self._team.values())):return max(3 - (self.turn - self._dynamax_turn), 0)@propertydef fields(self) -> Dict[Field, int]:""":return: A Dict mapping fields to the turn they have been activated.:rtype: Dict[Field, int]"""return self._fields@propertydef finished(self) -> bool:""":return: A boolean indicating whether the battle is finished.:rtype: Optional[bool]"""return self._finished@property@abstractmethoddef force_switch(self) -> Any:pass@propertydef lost(self) -> Optional[bool]:""":return: If the battle is finished, a boolean indicating whether the battle islost. Otherwise None.:rtype: Optional[bool]"""return None if self._won is None else not self._won@property# 返回团队预览中可接受的最大团队大小，如果适用的话def max_team_size(self) -> Optional[int]:return self._max_team_size# 抽象方法，可能被困住的情况@property@abstractmethoddef maybe_trapped(self) -> Any:pass# 抽象方法，对手的当前激活精灵@property@abstractmethoddef opponent_active_pokemon(self) -> Any:pass# 抽象方法，对手是否可以激活极巨化@property@abstractmethoddef opponent_can_dynamax(self) -> Any:pass# 设置对手是否可以激活极巨化@opponent_can_dynamax.setter@abstractmethoddef opponent_can_dynamax(self, value: bool) -> Any:pass# 返回对手的精灵剩余的极巨化回合数@propertydef opponent_dynamax_turns_left(self) -> Optional[int]:if self._opponent_dynamax_turn is not None and any(map(lambda pokemon: pokemon.is_dynamaxed, self._opponent_team.values())):return max(3 - (self.turn - self._opponent_dynamax_turn), 0)# 返回对手的角色在给定的战斗中，p1 或 p2@propertydef opponent_role(self) -> Optional[str]:if self.player_role == "p1":return "p2"if self.player_role == "p2":return "p1"# 返回对手的场地状态@propertydef opponent_side_conditions(self) -> Dict[SideCondition, int]:return self._opponent_side_conditionsdef opponent_team(self) -> Dict[str, Pokemon]:"""During teampreview, keys are not definitive: please rely on values.:return: The opponent's team. Keys are identifiers, values are pokemon objects.:rtype: Dict[str, Pokemon]"""# 如果已经存在对手队伍信息，则直接返回if self._opponent_team:return self._opponent_teamelse:# 否则根据对手队伍预览信息生成对手队伍字典并返回return {mon.species: mon for mon in self._teampreview_opponent_team}@propertydef opponent_username(self) -> Optional[str]:""":return: The opponent's username, or None if unknown.:rtype: str, optional."""# 返回对手的用户名，如果未知则返回 Nonereturn self._opponent_username@opponent_username.setterdef opponent_username(self, value: str):# 设置对手的用户名self._opponent_username = value@propertydef player_role(self) -> Optional[str]:""":return: Player's role in given battle. p1/p2:rtype: str, optional"""# 返回玩家在战斗中的角色，可能是 p1 或 p2return self._player_role@player_role.setterdef player_role(self, value: Optional[str]):# 设置玩家在战斗中的角色self._player_role = value@propertydef player_username(self) -> str:""":return: The player's username.:rtype: str"""# 返回玩家的用户名return self._player_username@player_username.setterdef player_username(self, value: str):# 设置玩家的用户名self._player_username = value@propertydef players(self) -> Tuple[str, str]:""":return: The pair of players' usernames.:rtype: Tuple[str, str]"""# 返回玩家对的用户名组成的元组return self._players[0]["username"], self._players[1]["username"]@players.setterdef players(self, players: Tuple[str, str]):"""Sets the battle player's name::param player_1: First player's username.:type player_1: str:param player_1: Second player's username.:type player_2: str"""# 解包玩家名称元组player_1, player_2 = players# 根据当前玩家用户名设置对手用户名if player_1 != self._player_username:self._opponent_username = player_1else:self._opponent_username = player_2@propertydef rating(self) -> Optional[int]:"""Player's rating after the end of the battle, if it was received.:return: The player's rating after the end of the battle.:rtype: int, optional"""# 返回玩家战斗结束后的评分return self._rating@propertydef opponent_rating(self) -> Optional[int]:"""Opponent's rating after the end of the battle, if it was received.:return: The opponent's rating after the end of the battle.:rtype: int, optional"""# 返回对手战斗结束后的评分return self._opponent_rating@propertydef rqid(self) -> int:"""Should not be used.:return: The last request's rqid.:rtype: Tuple[str, str]"""# 不应该使用，返回最后一个请求的 rqidreturn self._rqid@propertydef side_conditions(self) -> Dict[SideCondition, int]:""":return: The player's side conditions. Keys are SideCondition objects, valuesare:- the number of layers of the side condition if the side condition isstackable- the turn where the SideCondition was setup otherwise:rtype: Dict[SideCondition, int]"""# 返回玩家的边界条件，键为 SideCondition 对象，值为边界条件的层数或设置边界条件的回合数return self._side_conditions@propertydef team(self) -> Dict[str, Pokemon]:""":return: The player's team. Keys are identifiers, values are pokemon objects.:rtype: Dict[str, Pokemon]"""# 返回玩家的队伍，键为标识符，值为 Pokemon 对象return self._team@team.setterdef team(self, value: Dict[str, Pokemon]):# 设置玩家的队伍self._team = value@propertydef team_size(self) -> int:""":return: The number of Pokemon in the player's team.:rtype: int"""# 返回玩家队伍中的 Pokemon 数量if self._player_role is not None:return self._team_size[self._player_role]# 如果没有分配玩家角色，则引发 ValueErrorraise ValueError("Team size cannot be inferred without an assigned player role.")@propertydef teampreview(self) -> bool:""":return: Whether the battle is awaiting a teampreview order.:rtype: bool"""# 返回战斗是否等待 teampreview 命令return self._teampreview@property@abstractmethoddef trapped(self) -> Any:pass@trapped.setter@abstractmethoddef trapped(self, value: Any):pass@propertydef turn(self) -> int:""":return: The current battle turn.:rtype: int"""# 返回当前战斗回合数return self._turn@turn.setterdef turn(self, turn: int):"""Sets the current turn counter to given value.:param turn: Current turn value.:type turn: int"""# 将当前回合计数器设置为给定值self._turn = turn@propertydef weather(self) -> Dict[Weather, int]:""":return: A Dict mapping the battle's weather (if any) to its starting turn:rtype: Dict[Weather, int]"""# 返回将战斗天气（如果有）映射到其起始回合的字典return self._weather@propertydef won(self) -> Optional[bool]:""":return: If the battle is finished, a boolean indicating whether the battle iswon. Otherwise None.:rtype: Optional[bool]"""# 如果战斗结束，返回一个布尔值指示战斗是否获胜，否则返回 Nonereturn self._won@propertydef move_on_next_request(self) -> bool:""":return: Whether the next received request should yield a move order directly.This can happen when a switch is forced, or an error is encountered.:rtype: bool"""# 返回下一个接收到的请求是否应直接产生移动顺序# 当强制切换或遇到错误时会发生这种情况return self._move_on_next_request@move_on_next_request.setter# 设置是否继续处理下一个请求的标志位def move_on_next_request(self, value: bool):# 将传入的布尔值赋给私有属性 _move_on_next_requestself._move_on_next_request = value# 获取是否正在恢复的属性@propertydef reviving(self) -> bool:# 返回私有属性 _reviving 的布尔值return self._reviving

`.\PokeLLMon\poke_env\environment\battle.py`

# 导入所需模块
from logging import Logger
from typing import Any, Dict, List, Optional, Union# 导入自定义模块
from poke_env.environment.abstract_battle import AbstractBattle
from poke_env.environment.move import Move
from poke_env.environment.pokemon import Pokemon
from poke_env.environment.pokemon_type import PokemonType# 定义 Battle 类，继承自 AbstractBattle 类
class Battle(AbstractBattle):# 初始化方法def __init__(self,battle_tag: str,username: str,logger: Logger,gen: int,save_replays: Union[str, bool] = False,):# 调用父类的初始化方法super(Battle, self).__init__(battle_tag, username, logger, save_replays, gen)# 初始化回合选择属性self._available_moves: List[Move] = []self._available_switches: List[Pokemon] = []self._can_dynamax: bool = Falseself._can_mega_evolve: bool = Falseself._can_tera: Optional[PokemonType] = Noneself._can_z_move: bool = Falseself._opponent_can_dynamax = Trueself._opponent_can_mega_evolve = Trueself._opponent_can_z_move = Trueself._opponent_can_tera: bool = Falseself._force_switch: bool = Falseself._maybe_trapped: bool = Falseself._trapped: bool = False# 初始化属性self.battle_msg_history = ""self.pokemon_hp_log_dict = {}self.speed_list = []# 清除所有属性提升def clear_all_boosts(self):if self.active_pokemon is not None:self.active_pokemon.clear_boosts()if self.opponent_active_pokemon is not None:self.opponent_active_pokemon.clear_boosts()# 结束幻象状态def end_illusion(self, pokemon_name: str, details: str):# 根据角色名判断幻象状态的 Pokemonif pokemon_name[:2] == self._player_role:active = self.active_pokemonelse:active = self.opponent_active_pokemon# 如果没有活跃的 Pokemon，则抛出异常if active is None:raise ValueError("Cannot end illusion without an active pokemon.")# 结束幻象状态self._end_illusion_on(illusioned=active, illusionist=pokemon_name, details=details)def switch(self, pokemon_str: str, details: str, hp_status: str):# 从传入的字符串中提取精灵标识符identifier = pokemon_str.split(":")[0][:2]# 如果标识符与玩家角色相同if identifier == self._player_role:# 如果存在活跃的精灵，让其退出战斗if self.active_pokemon:self.active_pokemon.switch_out()else:# 如果对手存在活跃的精灵，让其退出战斗if self.opponent_active_pokemon:self.opponent_active_pokemon.switch_out()# 获取指定的精灵对象pokemon = self.get_pokemon(pokemon_str, details=details)# 让指定的精灵进入战斗，并设置其血量状态pokemon.switch_in(details=details)pokemon.set_hp_status(hp_status)@propertydef active_pokemon(self) -> Optional[Pokemon]:""":return: 活跃的精灵:rtype: Optional[Pokemon]"""# 返回队伍中活跃的精灵for pokemon in self.team.values():if pokemon.active:return pokemon@propertydef all_active_pokemons(self) -> List[Optional[Pokemon]]:""":return: 包含所有活跃精灵和/或 None 的列表:rtype: List[Optional[Pokemon]"""# 返回包含玩家和对手活跃精灵的列表return [self.active_pokemon, self.opponent_active_pokemon]@propertydef available_moves(self) -> List[Move]:""":return: 玩家可以在当前回合使用的招式列表:rtype: List[Move]"""# 返回玩家可以使用的招式列表return self._available_moves@propertydef available_switches(self) -> List[Pokemon]:""":return: 玩家可以在当前回合进行的替换列表:rtype: List[Pokemon]"""# 返回玩家可以进行的替换列表return self._available_switches@propertydef can_dynamax(self) -> bool:""":return: 当前活跃精灵是否可以极巨化:rtype: bool"""# 返回当前活跃精灵是否可以进行极巨化return self._can_dynamax@propertydef can_mega_evolve(self) -> bool:""":return: 当前活跃精灵是否可以超级进化:rtype: bool"""# 返回当前活跃精灵是否可以进行超级进化return self._can_mega_evolve@propertydef can_tera(self) -> Optional[PokemonType]:""":return: None, or the type the active pokemon can terastallize into.:rtype: PokemonType, optional"""# 返回当前活跃宝可梦可以转变成的类型，如果不能则返回 Nonereturn self._can_tera@propertydef can_z_move(self) -> bool:""":return: Whether or not the current active pokemon can z-move.:rtype: bool"""# 返回当前活跃宝可梦是否可以使用 Z 招式return self._can_z_move@propertydef force_switch(self) -> bool:""":return: A boolean indicating whether the active pokemon is forced to switchout.:rtype: Optional[bool]"""# 返回一个布尔值，指示当前活跃宝可梦是否被迫交换出场return self._force_switch@propertydef maybe_trapped(self) -> bool:""":return: A boolean indicating whether the active pokemon is maybe trapped by theopponent.:rtype: bool"""# 返回一个布尔值，指示当前活跃宝可梦是否可能被对手困住return self._maybe_trapped@propertydef opponent_active_pokemon(self) -> Optional[Pokemon]:""":return: The opponent active pokemon:rtype: Pokemon"""# 返回对手当前活跃的宝可梦for pokemon in self.opponent_team.values():if pokemon.active:return pokemonreturn None@propertydef opponent_can_dynamax(self) -> bool:""":return: Whether or not opponent's current active pokemon can dynamax:rtype: bool"""# 返回对手当前活跃的宝可梦是否可以极巨化return self._opponent_can_dynamax@opponent_can_dynamax.setterdef opponent_can_dynamax(self, value: bool):self._opponent_can_dynamax = value@propertydef opponent_can_mega_evolve(self) -> bool:""":return: Whether or not opponent's current active pokemon can mega-evolve:rtype: bool"""# 返回对手当前活跃的宝可梦是否可以超级进化return self._opponent_can_mega_evolve@opponent_can_mega_evolve.setterdef opponent_can_mega_evolve(self, value: bool):self._opponent_can_mega_evolve = valuedef opponent_can_tera(self) -> bool:""":return: Whether or not opponent's current active pokemon can terastallize:rtype: bool"""# 返回对手当前激活的宝可梦是否可以使用 terastallizereturn self._opponent_can_tera@propertydef opponent_can_z_move(self) -> bool:""":return: Whether or not opponent's current active pokemon can z-move:rtype: bool"""# 返回对手当前激活的宝可梦是否可以使用 z-movereturn self._opponent_can_z_move@opponent_can_z_move.setterdef opponent_can_z_move(self, value: bool):# 设置对手当前激活的宝可梦是否可以使用 z-moveself._opponent_can_z_move = value@propertydef trapped(self) -> bool:""":return: A boolean indicating whether the active pokemon is trapped, either bythe opponent or as a side effect of one your moves.:rtype: bool"""# 返回一个布尔值，指示激活的宝可梦是否被困住，无论是被对手困住还是作为你的招式的副作用return self._trapped@trapped.setterdef trapped(self, value: bool):# 设置激活的宝可梦是否被困住self._trapped = value