添加扩展

在 SRA 中添加扩展

工作原理

SRA 的扩展系统是一套“泛型声明 + 装饰器注册 + 动态发现”的可插拔架构：

1. 配置泛型：每个扩展通过继承 BaseExtension[YourConfig] 声明自己的配置类型，YourConfig 是一个 Pydantic BaseModel。泛型参数让框架在编译期就知道扩展需要哪些配置，并能自动生成 JSON Schema 供前端渲染配置表单。

2. 装饰器注册：@extension 装饰器在模块导入时自动将扩展类及其配置模型注册到全局 ExtensionRegistry。扩展标识（ID）、展示名称和描述都可在此声明。

3. 动态发现：SRA 启动时调用 load_extensions()，扫描 extensions/ 目录下所有 .py 文件并逐一导入，触发各模块顶层的 @extension 装饰器完成注册。这与 tasks/ 目录的动态导入机制完全一致。

4. 运行与配置：ExtensionRunner 负责实例化扩展并注入 IOperator，扩展可通过 self.operator 执行截图、点击、OCR 等实际操作。ExtensionConfigManager 负责从 extensions.json 加载和保存各扩展的配置。

简单来说：在 extensions/ 目录下创建一个 Python 文件，定义配置模型并继承 BaseExtension[Config]，用 @extension 装饰器注册，SRA 启动时就会自动发现并注册你的扩展。

扩展 vs 任务 vs 命令

SRA 有三种可插拔模块，它们共享“动态发现”机制但定位不同：

	扩展（Extension）	任务（Task）	命令（CommandSet）
目录	extensions/	tasks/	tasks/
基类	BaseExtension[Config]	BaseTask	cmd2.CommandSet[SRACli]
注册方式	@extension 装饰器，泛型声明配置类	@task 装饰器	cmd2 自动发现
用途	可独立运行的功能模块（截图、OCR 等）	自动化任务流水线中的一个步骤	CLI 命令
配置	Pydantic BaseModel，自动生成 Schema，前端可渲染配置表单	通过 self.settings 读取全局设置，无独立配置模型	无内置配置系统
运行方式	extension run <id> / extension run-all	task single <index> / task single <ClassName>	CLI 命令直接调用
执行顺序	无固定顺序，按 ID 独立调用	按文件名排序决定索引，task run 依次执行	无固定顺序
IOperator	自动注入，可执行游戏内操作	自动注入，可执行游戏内操作	不可用
生命周期回调	on_start / on_completed / on_failed	由 TaskManager 管理，任务失败会停止后续任务	无

简单来说：

• 如果你想做一个可独立运行、有配置界面的功能模块，用扩展。
• 如果你想参与自动化的任务流水线（如每日任务链），用任务。
• 如果你想添加一个CLI 命令供手动调用，用命令。

先决条件

• 已安装 SRA 或已获取 SRA 源码
• 了解 Python 编程基础
• 了解 Pydantic BaseModel 的基本用法
• 文本编辑器（如 VSCode、Notepad++）

分步教程

1. 打开 SRA 的安装目录或源码目录，找到 extensions 文件夹。 这里存放了所有的扩展脚本。

2. 创建一个新的 Python 文件，例如 MyExtension.py。

3. 定义配置模型。配置模型是一个继承自 pydantic.BaseModel 的类，使用 Field 声明每个配置项的默认值、描述和约束：

   from pydantic import BaseModel, Field
   
   class MyConfig(BaseModel):
       """我的扩展配置。"""
       target: str = Field(default="星穹列车", description="目标名称")
       count: int = Field(default=1, ge=1, le=100, description="执行次数")
       verbose: bool = Field(default=False, description="是否输出详细日志")

   解释：

   • Field(default=...) — 设置默认值，未配置时使用此值。
   • Field(description=...) — 描述文本，会显示在前端配置弹窗中。
   • Field(ge=1, le=100) — 约束条件，ge = 最小值，le = 最大值。前端会自动为 int 类型渲染带上下限的数字输入框。

4. 定义扩展类并用 @extension 装饰器注册：

   from SRACore.extension import BaseExtension, extension
   
   @extension(name="我的扩展", description="这是一个自定义扩展")
   class MyExtension(BaseExtension[MyConfig]):
       """一个自定义扩展示例。"""
   
       def run(self) -> bool:
           for i in range(self.config.count):
               print(f"[MyExtension] ({i + 1}/{self.config.count}) 目标: {self.config.target}")
           return True

   解释：

   • class MyExtension(BaseExtension[MyConfig]) — 通过泛型参数 MyConfig 声明配置类型。框架会自动从中提取配置模型并注册。
   • @extension(name="...", description="...") — 注册装饰器。name 是展示名称，description 是功能描述。不传时 name 默认使用类名，description 默认使用 docstring 首行。
   • def run(self) -> bool: — 抽象方法，必须实现。返回 True 表示成功，False 表示失败。
   • self.config — 类型为 MyConfig，可直接访问配置字段，IDE 能提供完整的自动补全。

5. 使用 IOperator 执行实际操作。扩展通过 self.operator 可以执行游戏内的各种操作：

   @extension(name="截图示例", description="演示 IOperator 的基本用法")
   class ScreenshotDemo(BaseExtension[MyConfig]):
       """演示 operator 的使用。"""
   
       def run(self) -> bool:
           # 截图
           screenshot = self.operator.screenshot()
   
           # 定位模板图片
           box = self.operator.locate("resources/template.png")
           if box:
               self.operator.click_box(box)
   
           # OCR 文字识别
           text = self.operator.ocr(screenshot, x=100, y=200, w=300, h=50)
           print(f"识别到文字: {text}")
   
           # 发送通知
           self.send_notification("截图完成", f"识别结果: {text}")
   
           return True

   self.operator 的常用方法：

   • screenshot() — 截取当前屏幕
   • locate(template) — 定位模板图片在屏幕中的位置
   • click_box(box) — 点击指定区域
   • ocr(...) — 对指定区域进行 OCR 文字识别

6. （可选）重写生命周期回调：

   @extension(name="生命周期示例", description="演示生命周期回调")
   class LifecycleDemo(BaseExtension[MyConfig]):
       """演示生命周期回调。"""
   
       def on_start(self) -> None:
           print("扩展开始执行前调用，可用于初始化资源")
   
       def run(self) -> bool:
           print("执行主要逻辑")
           return True
   
       def on_completed(self) -> None:
           print("执行成功后调用，可用于清理资源")
   
       def on_failed(self) -> None:
           print("执行失败后调用，可用于错误恢复")

7. 运行 SRA 测试你的扩展。

   确保你的文件已保存到 extensions/ 目录下。运行 SRA-cli：

   sra> extension list
   已注册 2 个扩展：
     Hello     问候               HelloExtension (config: HelloConfig)    简单的问候示例...
     MyExtension  我的扩展          MyExtension (config: MyConfig)          这是一个自定义扩展
   
   sra> extension run MyExtension
   [MyExtension] (1/1) 目标: 星穹列车
   扩展 'MyExtension' 执行成功
   
   sra> extension run MyExtension --count 3 --target 黑塔空间站
   [MyExtension] (1/3) 目标: 黑塔空间站
   [MyExtension] (2/3) 目标: 黑塔空间站
   [MyExtension] (3/3) 目标: 黑塔空间站
   扩展 'MyExtension' 执行成功

   你还可以查看扩展的配置 Schema：

   sra> extension info MyExtension --json
   {"properties": {"target": {"default": "星穹列车", "description": "目标名称", "type": "string"}, ...}}

8. 恭喜！ 你已经学会了如何在 SRA 中添加扩展。回顾一下关键点：

   • 文件放在 extensions/ 目录下
   • 定义 Pydantic BaseModel 作为配置模型，用 Field 声明默认值和约束
   • 继承 BaseExtension[YourConfig] 并实现 run() 方法
   • 用 @extension 装饰器注册，可附带 name 和 description
   • 通过 self.operator 执行游戏内操作（截图、OCR、点击等）
   • 通过 self.config 访问配置，IDE 完整支持类型补全
   • SRA 启动时自动发现并注册，无需手动修改任何注册代码
   • 前端会根据配置 Schema 自动生成配置表单，无需编写 UI 代码

无配置类扩展

如果扩展不需要用户配置，可以直接继承 BaseExtension（不带泛型参数）：

from SRACore.extension import BaseExtension, extension

@extension(name="简单问候", description="不需要配置的简单扩展")
class HelloExtension(BaseExtension):
    """简单的问候扩展。"""

    def run(self) -> bool:
        print("Hello, SRA!")
        return True

与带配置类型的写法相比，差异如下：

	BaseExtension	BaseExtension[MyConfig]
self.config	始终为 None	类型为 MyConfig，IDE 自动补全
JSON Schema	不生成	自动生成
前端配置表单	无	根据字段自动渲染
extension run <id> 参数覆盖	不可用	可用（如 --count 3）

注意

@extension 装饰器会自动检测泛型参数：有则提取配置类，无则注册为无配置扩展，两种写法的注册方式完全一致。

后台扩展

前面的扩展都是“单次运行”模式：extension run <id> 启动后执行一次 run() 就结束。但有些场景需要扩展持续运行（如监控游戏状态、定时轮询、实时响应事件），这时可以使用后台扩展。

声明方式

在 @extension 装饰器中传入 background=True：

import time
from pydantic import BaseModel, Field
from SRACore.extension import BaseExtension, extension
from SRACore.util.errors import ThreadStoppedError

class MonitorConfig(BaseModel):
    """后台监控扩展配置。"""
    interval: float = Field(default=5.0, ge=0.5, le=3600, description="轮询间隔（秒）")

@extension(name="状态监控", description="定期检查游戏状态", background=True)
class MonitorExtension(BaseExtension[MonitorConfig]):
    """后台轮询扩展示例。"""

    def run(self) -> bool:
        cfg = self.config or MonitorConfig()

        # 后台扩展的 run() 应做一轮工作后返回，
        # 框架会周期性地再次调用它
        screenshot = self.operator.screenshot()
        # ... 执行检查逻辑 ...
        print(f"[Monitor] 检查完成，等待 {cfg.interval} 秒后下一轮")

        # 利用 stop_event 实现可中断的等待
        if self.operator.stop_event.wait(cfg.interval):
            raise ThreadStoppedError("监控停止", "线程已停止")

        return True

运行机制

后台扩展与单次扩展的运行方式完全不同：

	单次扩展	后台扩展
声明	@extension(...)	@extension(..., background=True)
启动命令	extension run <id>	extension run <id>（自动分发）
线程模型	独占线程，与任务互斥	共享轮询线程，多个后台扩展可同时运行
run() 调用	执行一次后结束	被周期性重复调用，每轮调用间间隔约 500ms
停止命令	extension stop（无参数）	extension stop <id>（指定 ID）
配置热更新	不支持（需重新运行）	支持，修改配置后自动替换实例，不中断线程

关键设计：

1. 共享线程 — 所有后台扩展共享一个轮询线程，_background_loop 每 500ms 遍历所有已启用的后台扩展并调用其 run()。这意味着多个后台扩展是串行轮询的，一个扩展的 run() 阻塞会影响后续扩展的调度。
2. run() 应快速返回 — 每次 run() 调用应完成一轮工作后及时返回，将长时间等待交给 stop_event.wait() 而非 time.sleep()，这样停止请求能被立即响应。
3. 自动分发 — extension run <id> 会检查扩展的 background 标志：后台扩展走 start_extension()，单次扩展走 run_in_thread()，用户无需区分。

CLI 操作

sra> extension list
已注册 3 个扩展：
  AutoPlayer  自动演奏    AutoPlayerExtension (config: AutoPlayerConfig)    按 JSON 乐谱自动演奏...
  Monitor     状态监控    MonitorExtension (config: MonitorConfig)        定期检查游戏状态

sra> extension run Monitor
已启动后台扩展 'Monitor'
[Monitor] 检查完成，等待 5.0 秒后下一轮
[Monitor] 检查完成，等待 5.0 秒后下一轮

sra> extension stop Monitor
已停止后台扩展 'Monitor'

注意

extension stop 的行为取决于是否传参：

• extension stop <id> — 停止指定的后台扩展（必须是 background=True 的扩展）
• extension stop（无参数）— 停止当前正在运行的单次扩展（非后台扩展）

对非后台扩展使用 extension stop <id> 会报错：扩展 '<id>' 不是后台扩展，不能通过 stop 指定停止。

配置热更新

后台扩展支持配置热更新——修改配置后，框架会自动替换实例，无需手动重启：

1. 用户通过前端或 extension config 修改了某个后台扩展的配置
2. ExtensionConfigManager.set() 触发 reload_extension 回调
3. 框架用新配置创建一个全新的扩展实例，替换 self.extensions[id] 中的旧实例
4. 共享轮询线程不中断，下一轮调用自然使用新实例

这意味着扩展开发者无需为配置更新做任何特殊处理——每次 run() 调用时读取 self.config 即可自动使用最新配置。

extension reload 的行为

extension reload 会重新扫描 extensions/ 目录、重新导入所有模块。对于正在运行的后台扩展：

1. 记录当前活跃的后台扩展 ID 列表
2. 停止所有后台扩展
3. 重新导入模块（触发 @extension 装饰器重新注册）
4. 对仍存在（未被删除）的后台扩展自动重新启动

因此开发者在修改后台扩展代码后，直接 extension reload 即可热加载，无需重启 SRA。

完整示例：实时事件监控

下面是一个更完整的后台扩展示例，演示了轮询模式与异常处理：

from pydantic import BaseModel, Field
from SRACore.extension import BaseExtension, extension
from SRACore.util.errors import ThreadStoppedError

class WatcherConfig(BaseModel):
    """事件监控配置。"""
    check_interval: float = Field(default=10.0, ge=1.0, le=300, description="检查间隔（秒）")
    notify_on_event: bool = Field(default=True, description="检测到事件时发送通知")

@extension(name="事件监控", description="后台监控游戏内事件", background=True)
class EventWatcher(BaseExtension[WatcherConfig]):
    """后台事件监控扩展。"""

    def on_start(self) -> None:
        self._event_count = 0

    def run(self) -> bool:
        cfg = self.config or WatcherConfig()

        try:
            # 每轮做一次检查
            screenshot = self.operator.screenshot()
            if self._detect_event(screenshot):
                self._event_count += 1
                if cfg.notify_on_event:
                    self.send_notification("检测到事件", f"累计 {self._event_count} 次")
            # 利用 stop_event 实现可中断的等待
            if self.operator.stop_event.wait(cfg.check_interval):
                raise ThreadStoppedError("监控停止", "线程已停止")
        except ThreadStoppedError:
            raise  # 不要吞掉停止信号
        except Exception as e:
            # 后台扩展的 run() 抛出异常会被框架捕获并自动停止该扩展
            # （见 _background_loop 中的 except 分支）
            # 如需容错（不因单轮异常而退出），在此 try/except 并 return True
            raise

        return True

    def _detect_event(self, screenshot) -> bool:
        # ... 实际的事件检测逻辑 ...
        return False

警告

后台扩展的 run() 抛出未捕获异常时，框架会自动停止该扩展（从轮询列表中移除并记录错误日志）。如果希望单轮异常不影响后续轮次，在 run() 内部 try/except 并返回 True 即可。

在扩展中使用热键监听

SRA 内置了全局键盘监听器，扩展可以注册热键，让用户在扩展运行期间通过按键控制行为（如暂停/继续、重置等）。内置的“自动演奏”扩展（extensions/auto_player.py）就是一个典型的热键使用范例。

工作原理

1. 宿主创建单实例：KeyboardListener 由宿主（CLI）创建并启动，同时负责注册全局停止热键（默认 F9，可在设置中修改）。
2. 注入给扩展：监听器实例会自动注入到扩展的 self.event_listener 属性，扩展无需自己创建。
3. 扩展只注册/注销：扩展只应调用 register_key_event / unregister_key_event 操作按键注册表，不要调用 start() / stop()，监听线程的生命周期由宿主管理。

基本 API

# 注册按键按下事件（重复注册同一按键会抛出 ValueError）
self.event_listener.register_key_event(key, callback, args=None)

# 注销指定按键的事件
self.event_listener.unregister_key_event(key)

• key — 按键字符串。普通按键为字符本身（如 'a'、'1'），特殊按键使用 pynput 的名称（如 'f8'、'enter'、'ctrl'、'space'）。
• callback — 按键按下时执行的回调函数，签名必须为 callback(args)，不需要参数时可写为 def callback(_args): ...。
• args — 传给回调的参数（可选）。

注意

热键带有防抖机制：按键按住期间只触发一次回调，释放按键后才会重新武装。回调在独立的守护线程中执行，不会阻塞键盘输入，但编写回调时需注意线程安全。

完整示例

下面是一个带“暂停/继续”热键的扩展，演示了热键监听的推荐写法：

import threading
import time
from pydantic import BaseModel, Field
from SRACore.extension import BaseExtension, extension
from SRACore.util.errors import ThreadStoppedError

class CounterConfig(BaseModel):
    """热键扩展示例配置。"""
    hotkey_pause: str = Field(default="f8", description="暂停/继续热键")

@extension(name="热键示例", description="演示在扩展中使用热键监听")
class HotkeyDemo(BaseExtension[CounterConfig]):
    """带暂停热键的计数扩展示例。"""

    def run(self) -> bool:
        cfg = self.config or CounterConfig()

        # 1. 检查监听器是否可用（可能未被宿主注入）
        if self.event_listener is None:
            print("未注入全局事件监听器，热键不可用，将直接运行")

        # 2. 共享状态：热键线程与运行线程都会访问
        paused = threading.Event()

        # 3. 回调签名必须为 callback(args)
        def on_pause(_args=None) -> None:
            if paused.is_set():
                paused.clear()
                print("继续")
            else:
                paused.set()
                print("已暂停（按热键继续）")

        # 4. 注册热键
        pause_key = cfg.hotkey_pause.strip().lower()
        if self.event_listener is not None:
            self.event_listener.register_key_event(pause_key, on_pause)
            print(f"热键已启用：暂停/继续 [{pause_key.upper()}]")

        # 5. 主循环：finally 保证注销热键
        try:
            for i in range(1, 11):
                if self.operator.stop_event.is_set():
                    raise ThreadStoppedError("任务停止", "线程已停止")
                while paused.is_set():
                    if self.operator.stop_event.wait(0.2):
                        raise ThreadStoppedError("任务停止", "线程已停止")
                print(f"计数 {i}/10")
                time.sleep(1)
        finally:
            if self.event_listener is not None:
                self.event_listener.unregister_key_event(pause_key)

        return True

关键点解释：

• 判空检查 — self.event_listener 类型为 KeyboardListener | None，在某些运行环境（如通过 SRA-server 运行）下可能为 None，使用前必须判断。
• 可配置热键 — 将热键定义成配置字段（hotkey_pause），用户可以在前端配置表单中修改，避免与其他扩展或全局热键冲突。
• 线程安全 — 回调运行在独立线程中，与 run() 的主线程共享状态时使用 threading.Event 等线程安全原语。
• finally 注销 — 无论扩展正常结束还是被停止（ThreadStoppedError），都必须注销热键，否则会残留已注册的按键，导致后续重复注册时报 ValueError。
• 响应停止 — self.operator.stop_event 是全局停止事件（用户按 F9 时被设置），长时间运行的扩展应周期性检查它，否则任务停止后热键监听还会残留。

注意事项

警告

注册热键前请确认按键未被占用：

• 全局停止热键默认为 F9，注册它会与任务停止功能冲突。
• 同一按键只能注册一个回调，重复注册会抛出 ValueError: 按键 {key} 已被注册。
• 如果热键与扩展自身的模拟按键冲突（例如演奏类扩展），触发热键时可能会同时影响正常操作。

建议像“自动演奏”扩展一样，把热键做成可配置项，并在启动时检测与已知按键的冲突，给出警告日志。