.\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 = 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
# 导入必要的模块和函数 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)
PokéLLMon 源码解析（一）(2)https://developer.aliyun.com/article/1483587
