ykcheungab
132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
"""Capture configuration, health state, and reconnect policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any, Optional, Sequence, Tuple, Union
|
|
|
|
MAX_SIMULTANEOUS_CAPTURES = 10
|
|
|
|
|
|
class CaptureHealth(str, Enum):
|
|
"""Lifecycle / health state for a capture source."""
|
|
|
|
CLOSED = "closed"
|
|
CONNECTING = "connecting"
|
|
CONNECTED = "connected"
|
|
RECONNECTING = "reconnecting"
|
|
FAILED = "failed"
|
|
|
|
|
|
SourceType = Union[int, str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReconnectPolicy:
|
|
"""Auto-reconnect settings for live cameras.
|
|
|
|
Parameters
|
|
----------
|
|
enabled:
|
|
When False, failed reads do not attempt reopen.
|
|
max_attempts:
|
|
Maximum reopen attempts per disconnect episode. ``None`` means unlimited.
|
|
initial_backoff_s:
|
|
First wait before a reconnect attempt.
|
|
max_backoff_s:
|
|
Upper bound for exponential backoff.
|
|
fail_threshold:
|
|
Consecutive failed reads that trigger reconnect.
|
|
stale_timeout_s:
|
|
If no frame arrives for this many seconds while connected, treat as stale
|
|
and reconnect. ``None`` disables stale detection.
|
|
"""
|
|
|
|
enabled: bool = False
|
|
max_attempts: Optional[int] = None
|
|
initial_backoff_s: float = 0.25
|
|
max_backoff_s: float = 5.0
|
|
fail_threshold: int = 5
|
|
stale_timeout_s: Optional[float] = 2.0
|
|
|
|
def next_backoff(self, attempt: int) -> float:
|
|
"""Exponential backoff capped at ``max_backoff_s`` (attempt is 0-based)."""
|
|
delay = self.initial_backoff_s * (2 ** max(0, attempt))
|
|
return min(delay, self.max_backoff_s)
|
|
|
|
|
|
@dataclass
|
|
class CaptureConfig:
|
|
"""Configuration for opening and running a capture source."""
|
|
|
|
source: SourceType
|
|
backend: str = "auto"
|
|
width: Optional[int] = None
|
|
height: Optional[int] = None
|
|
fps: Optional[float] = None
|
|
fourcc: Optional[str] = "MJPG"
|
|
background: bool = True
|
|
reconnect: ReconnectPolicy = field(default_factory=ReconnectPolicy)
|
|
verbose: bool = False
|
|
# Optional discovery filters (used by list/open helpers, not open itself).
|
|
name_include: Optional[Sequence[str]] = None
|
|
name_exclude: Optional[Sequence[str]] = None
|
|
vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None
|
|
min_width: Optional[int] = None
|
|
min_height: Optional[int] = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not isinstance(self.reconnect, ReconnectPolicy):
|
|
raise TypeError("reconnect must be a ReconnectPolicy")
|
|
|
|
@classmethod
|
|
def from_camera_dict(
|
|
cls,
|
|
source: SourceType,
|
|
camera: dict[str, Any],
|
|
*,
|
|
reconnect: Optional[ReconnectPolicy] = None,
|
|
verbose: bool = False,
|
|
**kwargs: Any,
|
|
) -> "CaptureConfig":
|
|
"""Build config from a sensor JSON ``camera`` section."""
|
|
# Accept deprecated queue/worker kwargs from older callers without error.
|
|
kwargs.pop("mode", None)
|
|
kwargs.pop("queue_size", None)
|
|
kwargs.pop("overflow", None)
|
|
if "start_worker" in kwargs:
|
|
kwargs.setdefault("background", bool(kwargs.pop("start_worker")))
|
|
res = camera.get("resolution_raw") or [None, None]
|
|
width = int(res[0]) if res and res[0] else None
|
|
height = int(res[1]) if res and res[1] else None
|
|
fps = camera.get("fps")
|
|
return cls(
|
|
source=source,
|
|
width=width,
|
|
height=height,
|
|
fps=float(fps) if fps is not None else None,
|
|
reconnect=reconnect or ReconnectPolicy(),
|
|
verbose=verbose,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CaptureStatus:
|
|
"""Snapshot of capture health for diagnostics / multi-camera UIs."""
|
|
|
|
health: CaptureHealth = CaptureHealth.CLOSED
|
|
source: Optional[SourceType] = None
|
|
last_error: Optional[str] = None
|
|
last_frame_time: Optional[float] = None
|
|
last_seq: int = 0
|
|
dropped_frames: int = 0
|
|
reconnect_attempts: int = 0
|
|
consecutive_failures: int = 0
|
|
width: int = 0
|
|
height: int = 0
|
|
fps: float = 0.0
|
|
backend_name: str = ""
|
|
fourcc: str = ""
|