""" AI Box Sophon Adapter — IN2 数字输入控制任务启停 监听盒子 IN2 DI 输入（复用 diwatch 的 DIChannel + SysfsBackend，sysfs 轮询）， 电平变化时通过盒子本地 HTTP 接口控制所有算法任务： - IN2 有电（电平 1）→ 启动全部任务（ControlCommand = start_command，默认 1） - IN2 断电（电平 0）→ 停止全部任务（ControlCommand = stop_command，默认 0） 盒子接口（POST，JSON）： - /api/alg_task_fetch 查询全部任务：{} → {"Content": [{"AlgTaskSession": ...}], "Result": {"Code": 0}} - /api/alg_task_control 控制单个任务：{"AlgTaskSession": ..., "ControlCommand": 0/1} → {"Result": {"Code": 0, "Desc": "已停止"}} 说明： - 只响应电平跳变，不按启动时的初始电平动作（避免插件重启时误触发全量启停）。 - DIChannel 阻塞轮询跑在独立线程，回调经 run_coroutine_threadsafe 调度回事件循环， HTTP 控制在事件循环内串行执行（apply_lock 防止快速抖动导致并发穿插）。 """ import asyncio import logging import threading import httpx from .diwatch.backends import EventBackend, make_in2_backend from .diwatch.core import DIChannel logger = logging.getLogger(__name__) class In2TaskSwitch: """IN2 数字输入 → 全部算法任务启停开关。 后端自适应（make_in2_backend）：SE9 走 sysfs gpio_status 轮询； SE5 无 gpio-in-2 时回退 gpio-keys event 设备（默认键码 KEY_DOWN=108， 可用 device/key 参数覆盖，键码用 diwatch detect 查询）。 """ def __init__( self, api_base: str = "http://127.0.0.1", start_command: int = 1, stop_command: int = 0, poll_interval: float = 0.05, debounce: int = 3, active_low: bool = False, device: str | None = None, key: int | None = None, timeout: float = 10.0, ): self.api_base = api_base.rstrip("/") self.start_command = start_command self.stop_command = stop_command self.poll_interval = poll_interval self.debounce = debounce self.active_low = active_low self.device = device self.key = key self.timeout = timeout self._channel: DIChannel | None = None self._thread: threading.Thread | None = None self._loop: asyncio.AbstractEventLoop | None = None self._apply_lock = asyncio.Lock() # ---- 生命周期 ---- def start(self, loop: asyncio.AbstractEventLoop) -> bool: """启动监听线程；找不到 IN2 可用后端时返回 False（非盒子环境属正常）。""" backend = make_in2_backend( device=self.device, key=self.key, interval=self.poll_interval, debounce=self.debounce, ) if backend is None: logger.warning( "[TaskSwitch] 找不到 IN2 可用后端（sysfs 状态文件与 gpio-keys " "event 设备均不存在），任务启停开关未启动（非盒子环境属正常现象）" ) return False self._channel = DIChannel( "IN2", backend, active_low=self.active_low, on_change=self._on_level_change, log=lambda m: logger.debug(f"[TaskSwitch] {m}"), ) self._loop = loop self._thread = threading.Thread( target=self._run_channel, name="in2-task-switch", daemon=True, ) self._thread.start() backend_desc = ( f"event:{getattr(backend, 'device', '?')}, 键码={backend.key}" if isinstance(backend, EventBackend) else f"sysfs:{backend.path}" ) logger.info( f"[TaskSwitch] IN2 任务启停开关已启动: 后端={backend_desc}, " f"api_base={self.api_base}, " f"启动命令={self.start_command}, 停止命令={self.stop_command}" ) if isinstance(backend, EventBackend): logger.info( "[TaskSwitch] event 后端无初始电平，以首个事件为准；" "若操作 IN2 无反应，用 diwatch detect 查实际键码并配置 task_switch.key" ) return True def stop(self) -> None: if self._channel is not None: self._channel.request_stop() if self._thread is not None: self._thread.join(timeout=2.0) self._thread = None logger.info("[TaskSwitch] IN2 任务启停开关已停止") def _run_channel(self) -> None: try: self._channel.run() # type: ignore except Exception as e: logger.error(f"[TaskSwitch] IN2 监听线程异常退出: {type(e).__name__}: {e}") # ---- DI 回调（轮询线程内执行）---- def _on_level_change(self, old: int, new: int, ts: float) -> None: logger.info(f"[TaskSwitch] IN2 电平变化: {old} -> {new}") if self._loop is not None: asyncio.run_coroutine_threadsafe(self.apply(new), self._loop) # ---- 任务控制（事件循环内执行）---- async def apply(self, level: int) -> None: """按目标电平控制全部任务：1=启动，0=停止。""" async with self._apply_lock: command = self.start_command if level == 1 else self.stop_command action = "启动" if level == 1 else "停止" sessions = await self._fetch_task_sessions() if not sessions: logger.warning("[TaskSwitch] 未获取到任何任务，跳过控制") return logger.info( f"[TaskSwitch] IN2 {'有电' if level else '断电'}，" f"{action}全部 {len(sessions)} 个任务: {sessions}" ) ok = 0 failed = 0 async with httpx.AsyncClient(timeout=self.timeout) as client: for session in sessions: try: resp = await client.post( f"{self.api_base}/api/alg_task_control", json={"AlgTaskSession": session, "ControlCommand": command}, ) result = resp.json().get("Result", {}) if resp.is_success and result.get("Code") == 0: ok += 1 logger.info( f"[TaskSwitch] 任务 {session} {action}成功: " f"{result.get('Desc', '')}" ) else: failed += 1 logger.warning( f"[TaskSwitch] 任务 {session} {action}失败: " f"Status={resp.status_code}, Body={resp.text[:200]}" ) except Exception as e: failed += 1 logger.error( f"[TaskSwitch] 任务 {session} {action}异常: " f"{type(e).__name__}: {e}" ) logger.info(f"[TaskSwitch] {action}完成: 成功 {ok}, 失败 {failed}") async def _fetch_task_sessions(self) -> list[str]: """查询全部任务的 AlgTaskSession 列表；失败返回空列表。""" try: async with httpx.AsyncClient(timeout=self.timeout) as client: # resp = await client.post(f"{self.api_base}/api/alg_task_fetch", json={}) resp = await client.get(f"{self.api_base}/api/alg_task_fetch") if not resp.is_success: logger.warning( f"[TaskSwitch] 查询任务失败: Status={resp.status_code}, " f"Body={resp.text[:200]}" ) return [] data = resp.json() result = data.get("Result", {}) if result.get("Code") != 0: logger.warning(f"[TaskSwitch] 查询任务返回错误: {result}") return [] content = data.get("Content") or [] return [t.get("AlgTaskSession") for t in content if t.get("AlgTaskSession")] except Exception as e: logger.error(f"[TaskSwitch] 查询任务异常: {type(e).__name__}: {e}") return []