feat: 添加上下文事件驱动的InfoBar提醒

src/one_dragon/base/operation/context_event_bus.py

@@ -1,23 +1,24 @@
-from concurrent.futures import ThreadPoolExecutor, Future
-
-from typing import Callable, Any, List
+from collections.abc import Callable
+from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import suppress
+from dataclasses import dataclass
+from typing import Any
 
 from one_dragon.utils import thread_utils
 
 _od_event_bus_executor = ThreadPoolExecutor(thread_name_prefix='od_event_bus', max_workers=32)
 
 
+@dataclass
 class ContextEventItem:
-
-    def __init__(self, event_id: str, data: Any):
-        self.event_id: str = event_id
-        self.data: Any = data
+    event_id: str
+    data: Any
 
 
 class ContextEventBus:
 
     def __init__(self):
-        self.callbacks: dict[str, List[Callable[[Any], None]]] = {}
+        self.callbacks: dict[str, list[Callable[[Any], None]]] = {}
 
     def dispatch_event(self, event_id: str, event_obj: Any = None):
         """
@@ -55,10 +56,8 @@ def unlisten_event(self, event_id: str, callback: Callable[[Any], None]):
         """
         if event_id not in self.callbacks:
             return
-        try:
+        with suppress(Exception):
             self.callbacks[event_id].remove(callback)
-        except Exception as e:
-            pass
 
     def unlisten_all_event(self, obj: Any):
         """

src/one_dragon/base/operation/context_notify_event.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import ClassVar
+
+
+class ContextNotifyLevelEnum(Enum):
+    """上下文通知级别。"""
+
+    INFORMATION = "information"
+    SUCCESS = "success"
+    WARNING = "warning"
+    ERROR = "error"
+
+
+@dataclass(slots=True)
+class ContextNotifyEvent:
+    """上下文通知事件体。"""
+
+    EVENT_ID: ClassVar[str] = 'context_notify'
+
+    title: str
+    content: str
+    level: ContextNotifyLevelEnum = ContextNotifyLevelEnum.INFORMATION
+
+    @classmethod
+    def info(cls, title: str, content: str) -> ContextNotifyEvent:
+        """创建普通通知。"""
+        return cls(title=title, content=content, level=ContextNotifyLevelEnum.INFORMATION)
+
+    @classmethod
+    def success(cls, title: str, content: str) -> ContextNotifyEvent:
+        """创建成功通知。"""
+        return cls(title=title, content=content, level=ContextNotifyLevelEnum.SUCCESS)
+
+    @classmethod
+    def warning(cls, title: str, content: str) -> ContextNotifyEvent:
+        """创建警告通知。"""
+        return cls(title=title, content=content, level=ContextNotifyLevelEnum.WARNING)
+
+    @classmethod
+    def error(cls, title: str, content: str) -> ContextNotifyEvent:
+        """创建错误通知。"""
+        return cls(title=title, content=content, level=ContextNotifyLevelEnum.ERROR)

src/one_dragon_qt/windows/main_app_window_base.py

@@ -3,7 +3,13 @@
 import contextlib
 from typing import TYPE_CHECKING
 
+from PySide6.QtCore import Signal
+from qfluentwidgets import InfoBar, InfoBarIcon, InfoBarPosition
+
+from one_dragon.base.operation.context_event_bus import ContextEventItem
+from one_dragon.base.operation.context_notify_event import ContextNotifyEvent
 from one_dragon.envs.project_config import ProjectConfig
+from one_dragon.utils.i18_utils import gt
 from one_dragon_qt.services.app_setting.app_setting_manager import AppSettingManager
 from one_dragon_qt.widgets.back_navigation_button import BackNavigationButton
 from one_dragon_qt.widgets.base_interface import BaseInterface
@@ -23,6 +29,8 @@ class MainAppWindowBase(AppWindowBase):
     - 导航栏返回按钮
     """
 
+    context_notify_signal = Signal(object)
+
     def __init__(
         self,
         ctx: OneDragonContext,
@@ -31,6 +39,7 @@ def __init__(
         app_icon: str | None = None,
         parent=None,
     ):
+        self.ctx: OneDragonContext = ctx
         self.app_setting_manager = AppSettingManager(ctx)
         self._connected_pivot_navi: PivotNavigatorInterface | None = None
 
@@ -42,6 +51,9 @@ def __init__(
             parent=parent,
         )
 
+        self.context_notify_signal.connect(self._show_context_notify)
+        self.ctx.listen_event(ContextNotifyEvent.EVENT_ID, self._emit_context_notify)
+
     def create_sub_interface(self) -> None:
         # 导航栏返回按钮（最上方，在子界面之前添加）
         self._back_nav_btn = BackNavigationButton(on_click=self._on_back_nav_clicked, parent=self)
@@ -80,6 +92,23 @@ def _update_back_btn_for_interface(self, interface: BaseInterface) -> None:
         else:
             self._back_nav_btn.set_active(False)
 
+    def _emit_context_notify(self, event: ContextEventItem) -> None:
+        """将上下文通知事件通过信号传递到主线程。"""
+        if isinstance(event.data, ContextNotifyEvent):
+            self.context_notify_signal.emit(event.data)
+
+    def _show_context_notify(self, event: ContextNotifyEvent) -> None:
+        """在主窗口展示上下文通知。"""
+        InfoBar.new(
+            icon=InfoBarIcon[event.level.name],
+            title=gt(event.title),
+            content=gt(event.content),
+            isClosable=True,
+            duration=5000,
+            position=InfoBarPosition.TOP_RIGHT,
+            parent=self,
+        )
+
     def on_ctx_ready(self) -> None:
         """在 ctx.init() 完成后调用，执行设置提供者扫描"""
         self.app_setting_manager.discover()

src/zzz_od/auto_battle/auto_battle_dodge_context.py

@@ -2,6 +2,7 @@
 
 import os
 import threading
+from collections.abc import Callable
 from concurrent.futures import Future, ThreadPoolExecutor
 from enum import Enum
 from typing import TYPE_CHECKING
@@ -13,6 +14,7 @@
 from sklearn.preprocessing import scale
 
 from one_dragon.base.conditional_operation.state_recorder import StateRecord
+from one_dragon.base.operation.context_notify_event import ContextNotifyEvent
 from one_dragon.utils import cal_utils, os_utils, thread_utils, yolo_config_utils
 from one_dragon.utils.log_utils import log
 from zzz_od.context.zzz_context import ZContext
@@ -31,9 +33,10 @@ class AudioRecorder:
     音频录制类，用于录制和处理音频数据。
     """
 
-    def __init__(self):
+    def __init__(self, error_callback: Callable[[RuntimeError], None] | None = None):
         self.running: bool = False  # 标记录制是否正在运行
         self._run_lock = threading.Lock()  # 用于线程安全的锁
+        self._error_callback: Callable[[RuntimeError], None] | None = error_callback
 
         self._sample_rate = 32000  # 采样率
         self._used_channel = 2  # 使用的音频通道数
@@ -76,26 +79,34 @@ def _record_loop(self) -> None:
         音频录制循环，持续录制音频数据。
         """
         # 这个在全局导入的话 会导致QT的选择文件无法使用
-        import soundcard as sc
-        from soundcard.mediafoundation import SoundcardRuntimeWarning
         import warnings
-        warnings.filterwarnings('ignore', category=SoundcardRuntimeWarning)
 
-        _mic = sc.get_microphone(id=str(sc.default_speaker().name), include_loopback=True)
-        _recorder = _mic.recorder(samplerate=self._sample_rate, channels=self._used_channel)
+        import soundcard as sc
+        from soundcard.mediafoundation import SoundcardRuntimeWarning
 
-        with _recorder as audio_recorder:
-            while self.running:
-                stream_data = audio_recorder.record(numframes=self._chunk_size)
-                if self._used_channel > 1:
-                    stream_data = librosa.to_mono(stream_data.T)
-                else:
-                    stream_data = stream_data.T
+        warnings.filterwarnings('ignore', category=SoundcardRuntimeWarning)
 
-                with self._update_audio_lock:
-                    # 更新 latest_audio
-                    self.latest_audio[:-len(stream_data)] = self.latest_audio[len(stream_data):]
-                    self.latest_audio[-len(stream_data):] = stream_data
+        try:
+            _mic = sc.get_microphone(id=str(sc.default_speaker().name), include_loopback=True)
+            _recorder = _mic.recorder(samplerate=self._sample_rate, channels=self._used_channel)
+            with _recorder as audio_recorder:
+                while self.running:
+                    stream_data = audio_recorder.record(numframes=self._chunk_size)
+                    if self._used_channel > 1:
+                        stream_data = librosa.to_mono(stream_data.T)
+                    else:
+                        stream_data = stream_data.T
+
+                    with self._update_audio_lock:
+                        # 更新 latest_audio
+                        self.latest_audio[:-len(stream_data)] = self.latest_audio[len(stream_data):]
+                        self.latest_audio[-len(stream_data):] = stream_data
+        except RuntimeError as e:
+            log.warning('音频录制异常，已停止声音闪避识别', exc_info=True)
+            if self._error_callback is not None:
+                self._error_callback(e)
+        finally:
+            self.running = False
 
     def stop_running(self) -> None:
         """
@@ -129,7 +140,7 @@ def __init__(self, ctx: ZContext):
         self.ctx: ZContext = ctx  # 上下文对象
 
         self._flash_model: FlashClassifier | None = None  # 闪避分类器
-        self._audio_recorder: AudioRecorder = AudioRecorder()  # 音频录制器
+        self._audio_recorder: AudioRecorder = AudioRecorder(self._on_audio_record_error)  # 音频录制器
         self._audio_template: np.ndarray | None = None  # 音频模板
 
         # 识别锁，保证每种类型只有一个实例在进行识别
@@ -148,6 +159,16 @@ def __init__(self, ctx: ZContext):
         self._audio_event_interval: float = 0.1
         self._last_audio_event_time: float = 0
 
+    def _on_audio_record_error(self, error: RuntimeError) -> None:
+        """音频录制异常时通知当前运行界面。"""
+        self.ctx.dispatch_event(
+            ContextNotifyEvent.EVENT_ID,
+            ContextNotifyEvent.warning(
+                title='声音闪避已停用',
+                content=f'音频录制异常，请检查音频设备/独占模式/默认输出设备：{error}',
+            ),
+        )
+
     def init_auto_op(
             self,
             auto_op: AutoBattleOperator,