From 5ec90a5575f5fe36c1d58bab3643674b99389756 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 4 Sep 2026 14:51:12 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20ip=E9=9F=B3=E6=9F=B1=E6=8F=92?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- aibox_sophon_adapter/__init__.py | 9 +++- aibox_sophon_adapter/global_var.py | 6 +++ aibox_sophon_adapter/router/board_ping.py | 2 +- aibox_sophon_adapter/router/list_alarms.py | 2 +- aibox_sophon_adapter/routes.py | 58 +++++++++++++++++++++- doubao/__init__.py | 2 +- ip_sound_column/__init__.py | 52 +++++++++++++++++++ ip_sound_column/global_var.py | 41 +++++++++++++++ ip_sound_column/models/__init__.py | 14 ++++++ ip_sound_column/models/pydantic.py | 30 +++++++++++ ip_sound_column/router/__init__.py | 5 ++ ip_sound_column/router/tts.py | 41 +++++++++++++++ 12 files changed, 256 insertions(+), 6 deletions(-) create mode 100644 ip_sound_column/__init__.py create mode 100644 ip_sound_column/global_var.py create mode 100644 ip_sound_column/models/__init__.py create mode 100644 ip_sound_column/models/pydantic.py create mode 100644 ip_sound_column/router/__init__.py create mode 100644 ip_sound_column/router/tts.py diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 202ac9a..6bdf1bf 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -59,7 +59,7 @@ async def _check_offline_devices(interval: float): from .models.orm import BoardHeartbeat from sqlalchemy import select - stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online) + stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online) # type: ignore result = await session.execute(stmt) records = result.scalars().all() @@ -116,6 +116,11 @@ class AiboxSophonAdapterPlugin(Plugin): doubao_cfg = config.get("doubao_analysis", {}) doubao_enabled = doubao_cfg.get("enabled", True) + # ip 音柱 TTS 播放开关 + ip_sound_column_cfg = config.get("ip_sound_column", {}) + ip_sound_column_enabled = ip_sound_column_cfg.get("enabled", True) + ip_sound_column_volume = ip_sound_column_cfg.get("volume", 50) # 播放音量 0-100 + # 告警转发配置 alarm_fwd_cfg = config.get("alarm_forward", {}) alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False) @@ -152,6 +157,8 @@ class AiboxSophonAdapterPlugin(Plugin): set_global_var("upload_root", upload_root_path) set_global_var("alarm_max_records", alarm_max) set_global_var("doubao_analysis_enabled", doubao_enabled) + set_global_var("ip_sound_column_enabled", ip_sound_column_enabled) + set_global_var("ip_sound_column_volume", ip_sound_column_volume) set_global_var("alarm_forward_enabled", alarm_fwd_enabled) set_global_var("alarm_forward_url", alarm_fwd_url) set_global_var("alarm_forward_video_url", alarm_fwd_video_url) diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 3cb8ab4..6bda9fb 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -45,6 +45,12 @@ alarm_max_records: int = 0 # 0 表示不限制 # 豆包 AI 分析开关 doubao_analysis_enabled: bool = True +# ip 音柱 TTS 播放开关 +ip_sound_column_enabled: bool = True + +# ip 音柱 TTS 播放音量（0-100） +ip_sound_column_volume: int = 50 + # 告警转发配置 alarm_forward_enabled: bool = False alarm_forward_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receive" diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index 27bc378..34f6608 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/aibox_sophon_adapter/router/board_ping.py @@ -106,7 +106,7 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: stream_id = media["MediaName"] source_url = media["MediaUrl"].replace("/Link", link_path) try: - await stream_manager.start_stream(f"/{stream_id}", source_url) + await stream_manager.start_stream(f"/{stream_id}", source_url) # type: ignore except Exception as e: logger.warning(f"[Heartbeat] 通道 {stream_id} 拉流失败: {e}") diff --git a/aibox_sophon_adapter/router/list_alarms.py b/aibox_sophon_adapter/router/list_alarms.py index a9df204..e68acf2 100644 --- a/aibox_sophon_adapter/router/list_alarms.py +++ b/aibox_sophon_adapter/router/list_alarms.py @@ -49,7 +49,7 @@ async def list_alarms( total = count_result.scalar() or 0 # 分页数据 - stmt = select(AlarmRecord).order_by(desc(AlarmRecord.created_at)) + stmt = select(AlarmRecord).order_by(desc(AlarmRecord.created_at)) # type: ignore if conditions: from sqlalchemy import and_ stmt = stmt.where(and_(*conditions)) diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index ae9e1b9..967affe 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -70,6 +70,41 @@ class _FileStream: assert self._file is not None, "必须在 with 上下文内使用" return self._file.seek(offset, whence) +# ============================================================ +# 辅助函数 - 播放 tts 语音 +# ============================================================ + +async def ip_sound_column_tts_play(zenoh_session, payload: dict) -> dict: + """ + 播放 ip 音柱 tts 语音 + + 可通过配置 ip_sound_column.enabled 关闭。失败时返回 {"command": ..., "result": -1}。 + + Args: + zenoh_session: Zenoh 会话（用于调用 ip_sound_column Worker） + payload: 请求体字典，包含 command / volume / play_count / queue / sync_response / + play_interval / tts_pitch / tts_speed / tts_voice_name / tts_text + """ + from .global_var import ip_sound_column_enabled + + if not ip_sound_column_enabled: + logger.info("[TTS] ip_sound_column 播放已关闭 (ip_sound_column.enabled=false)") + return {"command": payload.get("command", ""), "result": -1} + + async with Client(session=zenoh_session, logger=logger) as client: + try: + result = await client.call( + worker_type="ip_sound_column", + subject="ip_sound_column.tts", + method="POST", + payload=payload, + timeout=10.0, + retries=1, + ) + return result + except Exception as e: + logger.error(f"[TTS] ip_sound_column 调用失败: {type(e).__name__}: {e}") + return {"command": payload.get("command", ""), "result": -1} # ============================================================ # 辅助函数 — 告警处理各步骤拆分为独立函数 @@ -463,10 +498,29 @@ async def alarm_report( Result=AlarmReportResult(Code=0, Desc="Doubao 分析判定为 False，忽略告警") ) - # 2. 写入本地数据库（按 AlarmId 去重 + 裁剪旧记录） + from .global_var import ip_sound_column_volume + + # 2. ip 音柱 TTS 播放（可通过配置开关关闭），失败时仅记录日志，不影响主流程 + tts_payload = { + "command": "play_local_tts", + "volume": ip_sound_column_volume, + "play_count": 1, + "queue": True, + "sync_response": False, + "play_interval": 2, + "tts_pitch": 50, + "tts_speed": 50, + "tts_voice_name": "xiaoyan", + "tts_text": f"告警: {full_data.get('Summary', '')}", + } + tts_result = await ip_sound_column_tts_play(request.app.state.zenoh_session, tts_payload) + if tts_result.get("result") != 0: + logger.warning(f"[AlarmReport] ip_sound_column TTS 播放失败: {tts_result}") + + # 3. 写入本地数据库（按 AlarmId 去重 + 裁剪旧记录） await _save_alarm_record(full_data) - # 3. 转发外部平台（先上传视频，再上报） + # 4. 转发外部平台（先上传视频，再上报） await _forward_alarm_to_external(full_data) return AlarmReportResponse( diff --git a/doubao/__init__.py b/doubao/__init__.py index a8b682b..569d01c 100644 --- a/doubao/__init__.py +++ b/doubao/__init__.py @@ -79,7 +79,7 @@ class DoubaoPlugin(Plugin): client = AsyncOpenAI( api_key=api_key, base_url=base_url, - http_client=AsyncSecureHttpClient(aicc_config=aicc_config), + http_client=AsyncSecureHttpClient(aicc_config=aicc_config), # type: ignore ) set_aicc_client(client) logger.info("豆包 AICC 客户端已创建") diff --git a/ip_sound_column/__init__.py b/ip_sound_column/__init__.py new file mode 100644 index 0000000..3be0e8a --- /dev/null +++ b/ip_sound_column/__init__.py @@ -0,0 +1,52 @@ +"""Ip_sound_column Worker 插件 + +统一风格模板 — 基于 chongming_worker.plugin.Plugin 基类。 + +工作流: + 1. 编辑 config/workers/ip_sound_column/*.toml 定义 RPC 端点 + 2. uv run gen_models config/workers/ip_sound_column src/plugins/ip_sound_column + 3. 在 router/<操作>.py 中实现业务逻辑 + 4. 在 on_init / on_cleanup 中初始化 / 清理资源 +""" +import logging + +from chongming_worker.plugin import Plugin + +from . import router # noqa: F401 触发 @rpc_method 注册 +# from .routes import router # 可选: 自定义 FastAPI 路由（网关自动发现） + +logger = logging.getLogger(__name__) + + +class IpSoundColumnPlugin(Plugin): + """ip_sound_column 插件 — 统一生命周期。""" + + type = "ip_sound_column" + db_name = "ip_sound_column.db" # 自动在 data_dir 下建 SQLite 库（不需要则置 None） + auto_bind = True # 首次初始化时自动 auto_bind_models(self.type) + + async def on_init(self): + """首次 create_workers 时调用一次：初始化资源、启动后台任务。 + + 数据库已由基类 _bootstrap 自动初始化（self.db）。 + """ + config = self.config + logger.info( + "[ip_sound_column] 初始化开始 (count=%s, ip=%s)", + config.get("count", 1), + config.get("ip", "unknown") + ) + ip = config.get("ip", "unknown") + self.set_var("ip", ip) + + logger.info("[ip_sound_column] 初始化完成") + + async def on_cleanup(self): + """cleanup 时调用：关闭自定义客户端等（后台任务与数据库由基类处理）。""" + logger.info("[ip_sound_column] 清理完成") + + +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +_plugin = IpSoundColumnPlugin() +create_workers = _plugin.create_workers +cleanup = _plugin.cleanup diff --git a/ip_sound_column/global_var.py b/ip_sound_column/global_var.py new file mode 100644 index 0000000..dc4ecec --- /dev/null +++ b/ip_sound_column/global_var.py @@ -0,0 +1,41 @@ +"""Ip_sound_column 插件全局运行时状态 — 统一容器 + +约定: + - 模块只做"声明默认值"，不产生任何导入期副作用 + - 所有值统一在 Plugin.on_init 中通过 set_global_var 显式写入 + - set_global_var 仅允许写入 _ALLOWED 白名单，拼写错误会立即抛错 + - require() 用于取"必须已初始化"的值，未初始化时抛清晰错误 +""" +from typing import Any + +from chongming_database import Database + +# 允许写入的全局变量白名单（防止拼写错误静默创建无用的新变量） +_ALLOWED = frozenset({"ip"}) + +# 显式声明所有运行时状态（默认 None，运行时由 on_init 写入） +# db: Database | None = None + + +def set_global_var(name: str, value: Any) -> None: + """白名单校验版 setter — 设置模块级运行时状态。""" + if name not in _ALLOWED: + raise KeyError( + f"global_var: 未知变量 '{name}'（允许: {sorted(_ALLOWED)}）" + ) + globals()[name] = value + + +def require(name: str) -> Any: + """获取必须已初始化的全局变量，未初始化时抛清晰错误。""" + if name not in _ALLOWED: + raise KeyError(f"global_var: 未知变量 '{name}'（允许: {sorted(_ALLOWED)}）") + value = globals().get(name) + if value is None: + raise RuntimeError( + f"global_var: '{name}' 未初始化 —— 请确认插件 create_workers 已执行且配置正确" + ) + return value + + +__all__ = sorted(_ALLOWED) + ["set_global_var", "require"] diff --git a/ip_sound_column/models/__init__.py b/ip_sound_column/models/__init__.py new file mode 100644 index 0000000..3df3e1f --- /dev/null +++ b/ip_sound_column/models/__init__.py @@ -0,0 +1,14 @@ +""" +模型导出 — ip_sound_column +自动生成，请勿手动修改 +""" + +from .pydantic import IpSoundColumnResponse +from .pydantic import TtsPostRequest +from .pydantic import TtsPostResponse + +__all__ = [ + "IpSoundColumnResponse", + "TtsPostRequest", + "TtsPostResponse", +] diff --git a/ip_sound_column/models/pydantic.py b/ip_sound_column/models/pydantic.py new file mode 100644 index 0000000..a81b724 --- /dev/null +++ b/ip_sound_column/models/pydantic.py @@ -0,0 +1,30 @@ +""" +自动生成的 Pydantic 模型 — ip_sound_column +来源: config/workers/ip_sound_column/ +""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + +class TtsPostRequest(BaseModel): + """ip 音柱 tts 播放 — 请求体""" + command: str = Field(...) + volume: int = Field(...) + play_count: int = Field(...) + queue: bool = Field(...) + sync_response: bool = Field(...) + play_interval: int = Field(...) + tts_pitch: int = Field(...) + tts_speed: int = Field(...) + tts_voice_name: str = Field(...) + tts_text: str = Field(...) + +class TtsPostResponse(BaseModel): + """ip 音柱 tts 播放 — 响应体""" + command: str = Field(...) + result: int = Field(...) + + +# 所有响应模型的联合类型 +IpSoundColumnResponse = TtsPostResponse diff --git a/ip_sound_column/router/__init__.py b/ip_sound_column/router/__init__.py new file mode 100644 index 0000000..dceef31 --- /dev/null +++ b/ip_sound_column/router/__init__.py @@ -0,0 +1,5 @@ +"""Ip_sound_column RPC 端点""" + +from . import tts + +__all__ = ['tts'] diff --git a/ip_sound_column/router/tts.py b/ip_sound_column/router/tts.py new file mode 100644 index 0000000..7bbb051 --- /dev/null +++ b/ip_sound_column/router/tts.py @@ -0,0 +1,41 @@ +"""RPC 端点: ip_sound_column.ping + +模型由 gen_models 生成（编辑 config/workers/ip_sound_column/ping.toml 后重新生成）。 +""" +import httpx + +from chongming_worker.decorators import rpc_method + +from ..models.pydantic import TtsPostRequest, TtsPostResponse +from ..global_var import require + + +@rpc_method( + subject="ip_sound_column.tts", + method="POST", + summary="tts 操作", + response_model=TtsPostResponse, +) +async def tts(request: TtsPostRequest) -> TtsPostResponse: + """业务逻辑实现。""" + + ip_addr: str = require("ip") + if ip_addr == "unknown": + raise RuntimeError("ip_sound_column: 未初始化 ip 地址，请检查配置") + + tts_url = f"http://{ip_addr}/v1/api" + try: + async with httpx.AsyncClient() as client: + resp = await client.post( + tts_url, + json=request.model_dump(), + headers={"Content-Type": "application/json"}, + ) + if resp.is_success: + return TtsPostResponse(**resp.json()) + else: + raise RuntimeError( + f"ip_sound_column: tts 请求失败，状态码 {resp.status_code}，响应内容: {resp.text}" + ) + except httpx.RequestError as e: + raise RuntimeError(f"ip_sound_column: tts 请求异常: {str(e)}") from e