前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >PokéLLMon 源码解析(一)

PokéLLMon 源码解析(一)

作者头像
ApacheCN_飞龙
发布2024-03-08 09:14:24
830
发布2024-03-08 09:14:24
举报
文章被收录于专栏:信数据得永生信数据得永生

.\PokeLLMon\poke_env\concurrency.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)

.\PokeLLMon\poke_env\data\gen_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)

.\PokeLLMon\poke_env\data\normalize.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()

.\PokeLLMon\poke_env\data\replay_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()

.\PokeLLMon\poke_env\data\static\abilities\construct_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")

.\PokeLLMon\poke_env\data\static\items\construct_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")

.\PokeLLMon\poke_env\data\static\moves\extract_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)

.\PokeLLMon\poke_env\data\__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",
]

.\PokeLLMon\poke_env\environment\abstract_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

.\PokeLLMon\poke_env\environment\battle.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
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2024-03-07,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • .\PokeLLMon\poke_env\concurrency.py
  • .\PokeLLMon\poke_env\data\gen_data.py
  • .\PokeLLMon\poke_env\data\normalize.py
  • .\PokeLLMon\poke_env\data\replay_template.py
  • .\PokeLLMon\poke_env\data\static\abilities\construct_ability_json.py
  • .\PokeLLMon\poke_env\data\static\items\construct_item_json.py
  • .\PokeLLMon\poke_env\data\static\moves\extract_gen8_moves.py
  • .\PokeLLMon\poke_env\data\__init__.py
  • .\PokeLLMon\poke_env\environment\abstract_battle.py
  • .\PokeLLMon\poke_env\environment\battle.py
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档