| """Bounded vision inference queue — drop stale frames, skip, single worker.""" |
| from __future__ import annotations |
|
|
| import asyncio |
| import logging |
| import time |
| from dataclasses import dataclass, field |
| from typing import Awaitable, Callable |
|
|
| import cv2 |
| import numpy as np |
|
|
| logger = logging.getLogger(__name__) |
|
|
| FRAME_SKIP_MOD = max(1, int(__import__("os").getenv("CEPHEUS_FRAME_SKIP", "4") or "4")) |
| INFER_WIDTH = max(160, int(__import__("os").getenv("CEPHEUS_INFER_WIDTH", "320") or "320")) |
| LIVE_INFER_WIDTH = max(INFER_WIDTH, int(__import__("os").getenv("CEPHEUS_LIVE_INFER_WIDTH", "480") or "480")) |
| PIPELINE_INFER_TIMEOUT = float(__import__("os").getenv("CEPHEUS_PIPELINE_TIMEOUT", "20") or "20") |
|
|
|
|
| @dataclass |
| class VisionPipelineStats: |
| submitted: int = 0 |
| processed: int = 0 |
| dropped: int = 0 |
| skipped: int = 0 |
| last_inference_ms: float = 0.0 |
| fps: float = 0.0 |
| queue_depth: int = 0 |
| _counter: int = 0 |
| _last_process_ts: float = field(default_factory=time.time) |
|
|
| def as_dict(self) -> dict: |
| return { |
| "submitted": self.submitted, |
| "processed": self.processed, |
| "dropped": self.dropped, |
| "skipped": self.skipped, |
| "last_inference_ms": round(self.last_inference_ms, 2), |
| "fps": round(self.fps, 2), |
| "queue_depth": self.queue_depth, |
| "frame_skip_mod": FRAME_SKIP_MOD, |
| "infer_width": INFER_WIDTH, |
| "live_infer_width": LIVE_INFER_WIDTH, |
| } |
|
|
|
|
| def scale_match_bboxes(matches: list, from_frame, to_frame) -> list: |
| """Map detection boxes from infer resolution back to full frame.""" |
| if not matches or from_frame is None or to_frame is None: |
| return matches |
| fh, fw = from_frame.shape[:2] |
| th, tw = to_frame.shape[:2] |
| if fh == th and fw == tw: |
| return matches |
| sx, sy = tw / max(fw, 1), th / max(fh, 1) |
| for m in matches: |
| b = m.get("bbox") |
| if b and len(b) >= 4: |
| m["bbox"] = [b[0] * sx, b[1] * sy, b[2] * sx, b[3] * sy] |
| return matches |
|
|
|
|
| def resize_for_infer(frame: np.ndarray, width: int = INFER_WIDTH) -> np.ndarray: |
| if frame is None or frame.size == 0: |
| return frame |
| h, w = frame.shape[:2] |
| if w <= width: |
| return frame |
| nh = max(1, int(h * (width / w))) |
| return cv2.resize(frame, (width, nh), interpolation=cv2.INTER_AREA) |
|
|
|
|
| AsyncInferRunner = Callable[..., Awaitable] |
|
|
|
|
| class VisionFramePipeline: |
| """Frame skip + bounded async infer via shared face executor (no extra thread).""" |
|
|
| def __init__(self, infer_fn, async_runner: AsyncInferRunner | None = None): |
| self._infer_fn = infer_fn |
| self._async_runner = async_runner |
| self.stats = VisionPipelineStats() |
|
|
| async def infer(self, cam_id: str, frame: np.ndarray) -> tuple[list, bool]: |
| """Returns (matches, skipped). skipped=True means frame was not inferred.""" |
| self.stats._counter += 1 |
| if self.stats._counter % FRAME_SKIP_MOD != 0: |
| self.stats.skipped += 1 |
| return [], True |
|
|
| if self._async_runner is None: |
| logger.error("VisionFramePipeline missing async_runner") |
| return [], True |
|
|
| self.stats.submitted += 1 |
| self.stats.queue_depth = 1 |
| small = resize_for_infer(frame) |
| t0 = time.perf_counter() |
| try: |
| matches = await self._async_runner( |
| self._infer_fn, |
| cam_id, |
| small, |
| timeout=PIPELINE_INFER_TIMEOUT, |
| ) |
| if not isinstance(matches, list): |
| matches = [] |
| except asyncio.TimeoutError: |
| logger.error("Vision pipeline infer timed out after %.0fs", PIPELINE_INFER_TIMEOUT) |
| matches = [] |
| except Exception as exc: |
| logger.warning("vision pipeline infer error: %s", exc) |
| matches = [] |
|
|
| elapsed_ms = (time.perf_counter() - t0) * 1000.0 |
| self.stats.last_inference_ms = elapsed_ms |
| self.stats.processed += 1 |
| now = time.time() |
| dt = now - self.stats._last_process_ts |
| if dt > 0: |
| self.stats.fps = 1.0 / dt |
| self.stats._last_process_ts = now |
| self.stats.queue_depth = 0 |
| return matches, False |
|
|
