.PokeLLMonpoke_envconcurrency.py
代码语言：javascript
复制# 导入必要的模块
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 = False
shutdown = asyncio.run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop)
shutdown.result()
while not cancelled:
cancelled = True
for task in tasks:
if not task.done():
cancelled = False
loop.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.7
loop = asyncio.get_running_loop()
except AttributeError:
# Python < 3.7 so get_event_loop won't raise exceptions
loop = asyncio.get_event_loop()
except RuntimeError:
# asyncio.get_running_loop raised exception so no loop is running
loop = None
if 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)
.PokeLLMonpoke_envdatagen_data.py
代码语言：javascript
复制# 导入必要的模块和函数
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_ITEM
UNKNOWN_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 = gen
self.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()] = 1
elif damage_taken == 1:
type_chart[type_][other_type.upper()] = 2
elif damage_taken == 2:
type_chart[type_][other_type.upper()] = 0.5
elif 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
# 返回静态文件根目录路径
@property
def _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_data
return 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)
.PokeLLMonpoke_envdatanormalize.py
代码语言：javascript
复制# 导入 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()
.PokeLLMonpoke_envdatareplay_template.py
代码语言：javascript
复制# 导入 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()
.PokeLLMonpoke_envdatastaticabilitiesconstruct_ability_json.py
代码语言：javascript
复制# 导入 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 中，格式化输出，缩进为 4
json.dump(ability_dict, f, indent=4)
# 打印 "pause"
print("pause")
.PokeLLMonpoke_envdatastaticitemsconstruct_item_json.py
代码语言：javascript
复制# 导入 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 列中的值不是 NaN
if 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")
.PokeLLMonpoke_envdatastaticmovesextract_gen8_moves.py
代码语言：javascript
复制# 导入 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] = effect
except:
# 如果出现异常则继续下一个招式
continue
# 打开文件 "gen8moves_effect.json" 以写入模式
with open("gen8moves_effect.json", "w") as f:
# 将新的招式名称到效果的字典以美观的格式写入到文件中
json.dump(move2effect_new, f, indent=4)
.PokeLLMonpoke_envdata__init__.py
代码语言：javascript
复制# 从 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",
]
.PokeLLMonpoke_envenvironmentabstract_battle.py
代码语言：javascript
复制# 导入必要的模块
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_tag
self._format: Optional[str] = None
self._max_team_size: Optional[int] = None
self._opponent_username: Optional[str] = None
self._player_role: Optional[str] = None
self._player_username: str = username
self._players: List[Dict[str, str]] = []
self._replay_data: List[List[str]] = []
self._save_replays: Union[str, bool] = save_replays
self._team_size: Dict[str, int] = {}
self._teampreview: bool = False
self._teampreview_opponent_team: Set[Pokemon] = set()
self._anybody_inactive: bool = False
self._reconnected: bool = True
self.logger: Optional[Logger] = logger
# 回合选择属性
self.in_team_preview: bool = False
self._move_on_next_request: bool = False
self._wait: Optional[bool] = None
# 战斗状态属性
self._dynamax_turn: Optional[int] = None
self._finished: bool = False
self._rqid = 0
self.rules: List[str] = []
self._turn: int = 0
self._opponent_can_terrastallize: bool = True
self._opponent_dynamax_turn: Optional[int] = None
self._opponent_rating: Optional[int] = None
self._rating: Optional[int] = None
self._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,
# 定义一个抽象方法用于清除所有精灵的增益效果
@abstractmethod
def clear_all_boosts(self):
pass
# 检查伤害信息中是否包含关于道具的信息
def _check_damage_message_for_item(self, split_message: List[str]):
# 捕获对方精灵受到道具伤害的情况
# 道具属于未受伤害的一方
if (
len(split_message) == 6
and 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: Ferrothorn
if (
len(split_message) == 6
and 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 Berry
if 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: Genesect
if 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: turn
for 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
# 抽象方法，用于解析请求
@abstractmethod
def 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_conditions
else:
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_conditions
else:
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")
# 切换精灵的抽象方法
@abstractmethod
def 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，否则为 False
if player_name == self._player_username:
self._won = True
else:
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
@abstractmethod
def active_pokemon(self) -> Any:
pass
@property
@abstractmethod
def all_active_pokemons(self) -> List[Optional[Pokemon]]:
pass
@property
@abstractmethod
def available_moves(self) -> Any:
pass
@property
@abstractmethod
def available_switches(self) -> Any:
pass
@property
def battle_tag(self) -> str:
"""
:return: The battle identifier.
:rtype: str
"""
return self._battle_tag
@property
@abstractmethod
def can_dynamax(self) -> Any:
pass
@property
@abstractmethod
def can_mega_evolve(self) -> Any:
pass
@property
@abstractmethod
def can_z_move(self) -> Any:
pass
@property
@abstractmethod
def can_tera(self) -> Any:
pass
@property
def 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)
@property
def fields(self) -> Dict[Field, int]:
"""
:return: A Dict mapping fields to the turn they have been activated.
:rtype: Dict[Field, int]
"""
return self._fields
@property
def finished(self) -> bool:
"""
:return: A boolean indicating whether the battle is finished.
:rtype: Optional[bool]
"""
return self._finished
@property
@abstractmethod
def force_switch(self) -> Any:
pass
@property
def lost(self) -> Optional[bool]:
"""
:return: If the battle is finished, a boolean indicating whether the battle is
lost. 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
@abstractmethod
def maybe_trapped(self) -> Any:
pass
# 抽象方法，对手的当前激活精灵
@property
@abstractmethod
def opponent_active_pokemon(self) -> Any:
pass
# 抽象方法，对手是否可以激活极巨化
@property
@abstractmethod
def opponent_can_dynamax(self) -> Any:
pass
# 设置对手是否可以激活极巨化
@opponent_can_dynamax.setter
@abstractmethod
def opponent_can_dynamax(self, value: bool) -> Any:
pass
# 返回对手的精灵剩余的极巨化回合数
@property
def 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
@property
def opponent_role(self) -> Optional[str]:
if self.player_role == "p1":
return "p2"
if self.player_role == "p2":
return "p1"
# 返回对手的场地状态
@property
def opponent_side_conditions(self) -> Dict[SideCondition, int]:
return self._opponent_side_conditions
def 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_team
else:
# 否则根据对手队伍预览信息生成对手队伍字典并返回
return {mon.species: mon for mon in self._teampreview_opponent_team}
@property
def opponent_username(self) -> Optional[str]:
"""
:return: The opponent's username, or None if unknown.
:rtype: str, optional.
"""
# 返回对手的用户名，如果未知则返回 None
return self._opponent_username
@opponent_username.setter
def opponent_username(self, value: str):
# 设置对手的用户名
self._opponent_username = value
@property
def player_role(self) -> Optional[str]:
"""
:return: Player's role in given battle. p1/p2
:rtype: str, optional
"""
# 返回玩家在战斗中的角色，可能是 p1 或 p2
return self._player_role
@player_role.setter
def player_role(self, value: Optional[str]):
# 设置玩家在战斗中的角色
self._player_role = value
@property
def player_username(self) -> str:
"""
:return: The player's username.
:rtype: str
"""
# 返回玩家的用户名
return self._player_username
@player_username.setter
def player_username(self, value: str):
# 设置玩家的用户名
self._player_username = value
@property
def 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.setter
def 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_1
else:
self._opponent_username = player_2
@property
def 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
@property
def 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
@property
def rqid(self) -> int:
"""
Should not be used.
:return: The last request's rqid.
:rtype: Tuple[str, str]
"""
# 不应该使用，返回最后一个请求的 rqid
return self._rqid
@property
def side_conditions(self) -> Dict[SideCondition, int]:
"""
:return: The player's side conditions. Keys are SideCondition objects, values
are:
- the number of layers of the side condition if the side condition is
stackable
- the turn where the SideCondition was setup otherwise
:rtype: Dict[SideCondition, int]
"""
# 返回玩家的边界条件，键为 SideCondition 对象，值为边界条件的层数或设置边界条件的回合数
return self._side_conditions
@property
def 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.setter
def team(self, value: Dict[str, Pokemon]):
# 设置玩家的队伍
self._team = value
@property
def 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]
# 如果没有分配玩家角色，则引发 ValueError
raise ValueError(
"Team size cannot be inferred without an assigned player role."
)
@property
def teampreview(self) -> bool:
"""
:return: Whether the battle is awaiting a teampreview order.
:rtype: bool
"""
# 返回战斗是否等待 teampreview 命令
return self._teampreview
@property
@abstractmethod
def trapped(self) -> Any:
pass
@trapped.setter
@abstractmethod
def trapped(self, value: Any):
pass
@property
def turn(self) -> int:
"""
:return: The current battle turn.
:rtype: int
"""
# 返回当前战斗回合数
return self._turn
@turn.setter
def turn(self, turn: int):
"""Sets the current turn counter to given value.
:param turn: Current turn value.
:type turn: int
"""
# 将当前回合计数器设置为给定值
self._turn = turn
@property
def 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
@property
def won(self) -> Optional[bool]:
"""
:return: If the battle is finished, a boolean indicating whether the battle is
won. Otherwise None.
:rtype: Optional[bool]
"""
# 如果战斗结束，返回一个布尔值指示战斗是否获胜，否则返回 None
return self._won
@property
def 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_request
self._move_on_next_request = value
# 获取是否正在恢复的属性
@property
def reviving(self) -> bool:
# 返回私有属性 _reviving 的布尔值
return self._reviving
.PokeLLMonpoke_envenvironmentbattle.py
代码语言：javascript
复制# 导入所需模块
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 = False
self._can_mega_evolve: bool = False
self._can_tera: Optional[PokemonType] = None
self._can_z_move: bool = False
self._opponent_can_dynamax = True
self._opponent_can_mega_evolve = True
self._opponent_can_z_move = True
self._opponent_can_tera: bool = False
self._force_switch: bool = False
self._maybe_trapped: bool = False
self._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):
# 根据角色名判断幻象状态的 Pokemon
if pokemon_name[:2] == self._player_role:
active = self.active_pokemon
else:
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)
@property
def active_pokemon(self) -> Optional[Pokemon]:
"""
:return: 活跃的精灵
:rtype: Optional[Pokemon]
"""
# 返回队伍中活跃的精灵
for pokemon in self.team.values():
if pokemon.active:
return pokemon
@property
def all_active_pokemons(self) -> List[Optional[Pokemon]]:
"""
:return: 包含所有活跃精灵和/或 None 的列表
:rtype: List[Optional[Pokemon]
"""
# 返回包含玩家和对手活跃精灵的列表
return [self.active_pokemon, self.opponent_active_pokemon]
@property
def available_moves(self) -> List[Move]:
"""
:return: 玩家可以在当前回合使用的招式列表
:rtype: List[Move]
"""
# 返回玩家可以使用的招式列表
return self._available_moves
@property
def available_switches(self) -> List[Pokemon]:
"""
:return: 玩家可以在当前回合进行的替换列表
:rtype: List[Pokemon]
"""
# 返回玩家可以进行的替换列表
return self._available_switches
@property
def can_dynamax(self) -> bool:
"""
:return: 当前活跃精灵是否可以极巨化
:rtype: bool
"""
# 返回当前活跃精灵是否可以进行极巨化
return self._can_dynamax
@property
def can_mega_evolve(self) -> bool:
"""
:return: 当前活跃精灵是否可以超级进化
:rtype: bool
"""
# 返回当前活跃精灵是否可以进行超级进化
return self._can_mega_evolve
@property
def can_tera(self) -> Optional[PokemonType]:
"""
:return: None, or the type the active pokemon can terastallize into.
:rtype: PokemonType, optional
"""
# 返回当前活跃宝可梦可以转变成的类型，如果不能则返回 None
return self._can_tera
@property
def can_z_move(self) -> bool:
"""
:return: Whether or not the current active pokemon can z-move.
:rtype: bool
"""
# 返回当前活跃宝可梦是否可以使用 Z 招式
return self._can_z_move
@property
def force_switch(self) -> bool:
"""
:return: A boolean indicating whether the active pokemon is forced to switch
out.
:rtype: Optional[bool]
"""
# 返回一个布尔值，指示当前活跃宝可梦是否被迫交换出场
return self._force_switch
@property
def maybe_trapped(self) -> bool:
"""
:return: A boolean indicating whether the active pokemon is maybe trapped by the
opponent.
:rtype: bool
"""
# 返回一个布尔值，指示当前活跃宝可梦是否可能被对手困住
return self._maybe_trapped
@property
def opponent_active_pokemon(self) -> Optional[Pokemon]:
"""
:return: The opponent active pokemon
:rtype: Pokemon
"""
# 返回对手当前活跃的宝可梦
for pokemon in self.opponent_team.values():
if pokemon.active:
return pokemon
return None
@property
def 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.setter
def opponent_can_dynamax(self, value: bool):
self._opponent_can_dynamax = value
@property
def 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.setter
def opponent_can_mega_evolve(self, value: bool):
self._opponent_can_mega_evolve = value
def opponent_can_tera(self) -> bool:
"""
:return: Whether or not opponent's current active pokemon can terastallize
:rtype: bool
"""
# 返回对手当前激活的宝可梦是否可以使用 terastallize
return self._opponent_can_tera
@property
def opponent_can_z_move(self) -> bool:
"""
:return: Whether or not opponent's current active pokemon can z-move
:rtype: bool
"""
# 返回对手当前激活的宝可梦是否可以使用 z-move
return self._opponent_can_z_move
@opponent_can_z_move.setter
def opponent_can_z_move(self, value: bool):
# 设置对手当前激活的宝可梦是否可以使用 z-move
self._opponent_can_z_move = value
@property
def trapped(self) -> bool:
"""
:return: A boolean indicating whether the active pokemon is trapped, either by
the opponent or as a side effect of one your moves.
:rtype: bool
"""
# 返回一个布尔值，指示激活的宝可梦是否被困住，无论是被对手困住还是作为你的招式的副作用
return self._trapped
@trapped.setter
def trapped(self, value: bool):
# 设置激活的宝可梦是否被困住
self._trapped = value
