155 phase1

pyproject.toml

@@ -64,7 +64,8 @@ exclude = ["build", "dist", ".venv", "__pypackages__"]
 per-file-ignores = { "__init__.py" = ["F401"] }
 
 [[tool.uv.index]]
-url = "https://pypi.tuna.tsinghua.edu.cn/simple"
+# url = "https://pypi.tuna.tsinghua.edu.cn/simple"        2025.10后，清华源限制了API访问，需要改用官方源
+url = "https://pypi.org/simple"
 default = true
 
 [tool.pytest.ini_options]

zsim/define.py

@@ -2,11 +2,27 @@
 import shutil
 import sys
 import tomllib
+from enum import Enum
 from pathlib import Path
 from typing import Callable, Literal
 
+
+# ZSim事件类型定义
+class ZSimEventTypes(str, Enum):
+    DEFAULT = "default"
+    SKILL_EVENT = "skill_event"
+    BUFF_EVENT = "buff_event"
+
+
+class SkillSubEventTypes(str, Enum):
+    HIT = "hit"
+    START = "start"
+    END = "end"
+
+
 # 属性类型：
 ElementType = Literal[0, 1, 2, 3, 4, 5, 6]
+SkillType = Literal[1, 2, 3, 4, 5]
 Number = int | float
 
 INVALID_ELEMENT_ERROR = "Invalid element type"
@@ -34,7 +50,6 @@
 saved_char_config = {}
 
 
-
 # 修复：将char_config_file作为参数传递给initialize_config_files
 def initialize_config_files_with_paths(char_file, data_dir, config_path):
     """
@@ -126,7 +141,7 @@ def update_json_config(template: dict, user: dict) -> bool:
 # 该字典的key为CID，value为招架动作的skill_tag
 # 注意，不同的招架策略有时候存在着影画或是其他的限制条件，
 # 所以若是在不满足这些条件的情况下强行使用这些招架策略，那么character中的审查函数会报错而中断程序运行。
-CHAR_PARRY_STRATEGY_MAP: dict = {1411: "1411_Assault_Aid_A"}
+CHAR_PARRY_STRATEGY_MAP: dict[int, str] = {1411: "1411_Assault_Aid_A"}
 
 # debug参数，用于检查APL在窗口期间的想法
 APL_THOUGHT_CHECK: bool = _config["apl_mode"].get("apl_thought_check", False)
@@ -191,6 +206,7 @@ def update_json_config(template: dict, user: dict) -> bool:
 
 # 开发变量
 NEW_SIM_BOOT: bool = _config.get("dev", {}).get("new_sim_boot", True)
+ZSIM_EVENT_SYSTEM_DEV: bool = _config.get("dev", {}).get("zsim_event_system_dev", False)
 
 compare_methods_mapping: dict[str, Callable[[float | int, float | int], bool]] = {
     "<": lambda a, b: a < b,

zsim/sim_progress/Character/skill_class.py

@@ -8,14 +8,15 @@
     DEFAULT_SKILL_PATH,
     SKILL_DATA_PATH,
     ElementType,
+    SkillType,
 )
 from zsim.sim_progress import Report
 
 try:
     # 读取角色数据
     char_lf = pl.scan_csv(CHARACTER_DATA_PATH)
 except Exception as e:
-    raise IOError(f"无法读取文件 {CHARACTER_DATA_PATH}: {e}")
+    raise IOError(f"无法读取文件 {CHARACTER_DATA_PATH}: {e}") from e
 
 
 @lru_cache(maxsize=64)
@@ -302,11 +303,11 @@ def __init__(
             self.char_name: str = char_name
             # 储存技能Tag
             self.cid = CID
-            self.skill_tag = f"{CID}_{key}" if str(CID) not in key else key
+            self.skill_tag: str = f"{CID}_{key}" if str(CID) not in key else key
             self.CN_skill_tag: str = _raw_skill_data["CN_skill_tag"]
             self.skill_text: str = _raw_skill_data["skill_text"]
             # 确定使用的技能等级
-            self.skill_type: int = int(_raw_skill_data["skill_type"])
+            self.skill_type: SkillType = _raw_skill_data["skill_type"]
             self.skill_level: int = self.__init_skill_level(
                 self.skill_type,
                 normal_level,
@@ -403,6 +404,7 @@ def __init__(
                     _raw_skill_data["aid_lag_ticks"]
                 )  # 技能激活快速支援的滞后时间
             tick_value = _raw_skill_data["tick_list"]
+            self.tick_list: list[int | float] | None = None
             if tick_value is None:
                 self.tick_list = None
             elif isinstance(tick_value, str):
@@ -415,7 +417,7 @@ def __init__(
                         self.tick_list = ast.literal_eval(str(tick_value).strip())
                         # self.tick_list = [int(v.strip()) for v in split_values]
                     except ValueError as e:
-                        raise ValueError(f"{self.skill_tag} 的 tick_list 包含无效整数: {e}")
+                        raise ValueError(f"{self.skill_tag} 的 tick_list 包含无效整数: {e}") from e
             else:
                 # 处理非字符串类型（如意外数值）
                 self.tick_list = None

zsim/sim_progress/zsim_event_system/Handler/__init__.py

@@ -0,0 +1,3 @@
+from .zsim_event_handler_registry import ZSimEventHandlerRegistry
+
+__all__ = ["ZSimEventHandlerRegistry"]

zsim/sim_progress/zsim_event_system/Handler/base_handler_class.py

@@ -0,0 +1,20 @@
+from abc import ABC, abstractmethod
+from typing import Generic, Iterable, TypeVar
+
+from ..zsim_events import BaseZSimEventContext, EventMessage, ZSimEventABC
+
+T = TypeVar("T", bound=EventMessage)
+
+
+class ZSimEventHandler(Generic[T], ABC):
+    @abstractmethod
+    def supports(self, event: ZSimEventABC[T]) -> bool:
+        """检查该处理器是否支持处理给定的事件"""
+        ...
+
+    @abstractmethod
+    def handle(
+        self, event: ZSimEventABC[T], context: BaseZSimEventContext
+    ) -> Iterable[ZSimEventABC[EventMessage]]:
+        """处理给定的事件, 并可能产生新的事件"""
+        ...

zsim/sim_progress/zsim_event_system/Handler/decorator.py

@@ -0,0 +1,107 @@
+from functools import wraps
+from typing import Any, Callable, Iterable, Protocol, TypeVar, cast
+
+from ....define import ZSimEventTypes
+from ..zsim_events import BaseZSimEventContext, EventMessage, ZSimEventABC
+from .base_handler_class import ZSimEventHandler
+from .zsim_event_handler_registry import ZSimEventHandlerRegistry
+
+# 全局的注册表实例
+_global_hander_registry: ZSimEventHandlerRegistry | None = None
+T = TypeVar("T", bound=EventMessage)
+C = TypeVar("C", bound=BaseZSimEventContext, contravariant=True)
+"""
+笔记：
+原有方案中,我们指定了HandlerFunc接收一个抽象基类BaseZSimEventContext,
+但实际上,各个handler传入的是更加具体的类型,比如SkillEventContext或者其他。
+这些更具体的Context都是从BaseZSimEventContext继承而来的,是它的子类，
+所以,在接收这些Context的HandlerFunc中,这种子类关系会被转化为一种逆变关系：
+即：
+SkillEventContext <: BaseZSimEventContext
+HandlerFunc[BaseZSimEventContext] <: HandlerFunc[SkillEventContext]
+解释转译：
+HandlerFunc[BaseZSimEventContext]宣称能够处理所有类型的Context,自然也能处理SkillEventContext类型,
+但HandlerFunc[SkillEventContext]并不一定能处理BaseZSimEventContext类型,
+从这个角度来看,功能更全面的那个应该是子类,功能更狭隘的那个应该是父类
+所以, HandlerFunc[SkillEventContext]是父类,HandlerFunc[BaseZSimEventContext]是子类。
+
+因此,我们需要修改HandlerFunc的类型定义,让contravariant参数为True,
+"""
+
+
+class HandlerFunc(Protocol[C]):
+    """事件处理函数协议"""
+
+    def __call__(
+        self, event: ZSimEventABC[T], context: C
+    ) -> Iterable[ZSimEventABC[EventMessage]]: ...
+
+    __name__: str
+
+
+F = TypeVar("F", bound=HandlerFunc[Any])
+
+
+def get_global_handler_registry() -> ZSimEventHandlerRegistry:
+    """获取全局的事件处理器注册表实例"""
+    global _global_hander_registry
+    if _global_hander_registry is None:
+        _global_hander_registry = ZSimEventHandlerRegistry()
+    return _global_hander_registry
+
+
+def event_handler(event_type: ZSimEventTypes) -> Callable[[F], F]:
+    """
+    事件处理装饰器, 用于自动注册Handler到全局注册表
+
+    Args:
+        event_type (ZSimEventTypes): 事件类型
+
+    Returns:
+        装饰器函数对象
+
+    """
+
+    def decorator(func: F) -> F:
+        @wraps(func)
+        def wrapper(
+            event: ZSimEventABC[T], context: BaseZSimEventContext
+        ) -> Iterable[ZSimEventABC[EventMessage]]:
+            return func(event, context)
+
+        class FunctionEventHandler(ZSimEventHandler[EventMessage]):
+            def __init__(self, handler_func: F):
+                self._handler_func = handler_func
+                self._event_type = event_type
+
+            def supports(self, event: ZSimEventABC[T]) -> bool:
+                return event.event_type == self._event_type
+
+            def handle(
+                self, event: ZSimEventABC[T], context: BaseZSimEventContext
+            ) -> list[ZSimEventABC[EventMessage]]:
+                return list(self._handler_func(event, context))
+
+            def __repr__(self) -> str:
+                return f"构造了FunctionEventHandler对象, 参数为：(event_type={self._event_type}, handler_func={self._handler_func.__name__})"
+
+        registry = get_global_handler_registry()
+        handler_instance: ZSimEventHandler[EventMessage] = FunctionEventHandler(func)
+        registry.register(event_type, handler_instance)
+
+        # 将处理器实例(FunctionEventHandler)和事件类型附加到包装函数上
+        setattr(wrapper, "_event_handler", handler_instance)  # noqa: B010
+        setattr(wrapper, "_event_type", event_type)  # noqa: B010
+
+        return cast(F, wrapper)
+
+    return decorator
+
+
+"""注意,考虑到业务尚未拓展,暂时不提供另外的注册接口,仅实现event_handler这一个装饰器。"""
+
+
+def clear_registry() -> None:
+    """清空全局事件处理器注册表"""
+    global _global_hander_registry
+    _global_hander_registry = None

zsim/sim_progress/zsim_event_system/Handler/skill_start_handler.py

@@ -0,0 +1,38 @@
+from typing import TYPE_CHECKING, Iterable, TypeVar
+
+from ....define import ZSimEventTypes
+from ..zsim_events import EventMessage, ZSimEventABC
+from ..zsim_events.skill_event import (
+    SkillEvent,
+    SkillEventContext,
+    SkillEventMessage,
+    SkillExecutionEvent,
+)
+from .decorator import event_handler
+
+if TYPE_CHECKING:
+    from ...Preload import SkillNode
+
+T = TypeVar("T", bound=EventMessage)
+
+
+@event_handler(ZSimEventTypes.SKILL_EVENT)
+def skill_start_handler(
+    event: ZSimEventABC[EventMessage], context: SkillEventContext
+) -> Iterable[SkillExecutionEvent]:
+    """
+    技能开始事件Handler,该Handler接收一个SkillEvent事件,返回一个ExecutionEvent 事件
+    """
+    assert isinstance(event, SkillEvent), (
+        f"skill_start_handler接收的event类型只能是SkillEvent，当前event类型为{type(event).__name__}"
+    )
+    assert isinstance(event.event_origin, SkillNode)
+    assert context.preload_tick is not None, "preload_tick不能为None"
+    event_message = SkillEventMessage(skill_tag=event.event_origin.skill_tag)
+    event_message.preload_tick = context.preload_tick
+    skill_execution_event = SkillExecutionEvent(
+        event_type=ZSimEventTypes.SKILL_EVENT, event_origin=event, event_message=event_message
+    )
+
+    print(f"技能开始事件Handler处理技能{event.event_origin.skill_tag}开始执行")
+    return [skill_execution_event]

zsim/sim_progress/zsim_event_system/Handler/zsim_event_handler_registry.py

@@ -0,0 +1,32 @@
+from collections import defaultdict
+from typing import Any, DefaultDict, Generator, Iterable, TypeVar
+
+from ....define import ZSimEventTypes
+from ..zsim_events import BaseZSimEventContext, EventMessage, ZSimEventABC
+from .base_handler_class import ZSimEventHandler
+
+T = TypeVar("T", bound=EventMessage)
+
+
+class ZSimEventHandlerRegistry:
+    """事件Handler注册表, 用于管理和检索事件Handler类"""
+
+    def __init__(self):
+        self._handlers: DefaultDict[ZSimEventTypes, list[ZSimEventHandler[Any]]] = defaultdict(list)
+
+    def register(self, event_type: ZSimEventTypes, handler: ZSimEventHandler[Any]) -> None:
+        """注册事件处理器类到指定事件类型"""
+        self._handlers[event_type].append(handler)
+
+    def iter_handlers(self, event: ZSimEventABC[T]) -> Iterable[ZSimEventHandler[Any]]:
+        """迭代所有支持处理给定事件的处理器"""
+        for handler in self._handlers.get(event.event_type, []):
+            if handler.supports(event):
+                yield handler
+
+    def handle(
+        self, event: ZSimEventABC[T], context: BaseZSimEventContext
+    ) -> Generator[ZSimEventABC[EventMessage]]:
+        """处理给定的事件, 并可能产生新的事件"""
+        for handler in self.iter_handlers(event):
+            yield from handler.handle(event, context)

zsim/sim_progress/zsim_event_system/__init__.py

@@ -0,0 +1 @@
+

zsim/sim_progress/zsim_event_system/tree/__init__.py

@@ -0,0 +1,4 @@
+from .nodes import DynamicLeafNode, EventTreeNode, LeafConfiguration
+from .state_tree import EventStateTree
+
+__all__ = ["EventTreeNode", "DynamicLeafNode", "EventStateTree", "LeafConfiguration"]