livekit-cameras/latency.py

"""Pure helpers for env-flagged capture/display latency knobs."""
from __future__ import annotations


def audio_queue_size_ms(hop_s: float, audio_queue_ms: int | None = None) -> int:
    """LiveKit AudioSource queue. Unset AUDIO_QUEUE_MS keeps the old 150 ms floor."""
    hop_ms = int(round(float(hop_s) * 1000.0))
    if audio_queue_ms is not None:
        return max(1, int(audio_queue_ms))
    return max(150, hop_ms + 50)


def video_hold_seconds(
    hop_s: float,
    has_audio: bool,
    video_hold_s: float | None = None,
) -> float:
    """Delay published video to match audio hops. VIDEO_HOLD_S=0 disables."""
    if not has_audio:
        return 0.0
    if video_hold_s is not None:
        return max(0.0, float(video_hold_s))
    return float(hop_s)


def effective_enhance_mode(
    mode: str,
    identity: str,
    speaker_only: bool,
    speaker_camera: str,
) -> str:
    """ENHANCE_SPEAKER_ONLY=1 keeps enhance on DISPLAY_SPEAKER_CAMERA only.

    ``speaker_camera=active`` has no single identity, so every camera keeps
    the configured mode.
    """
    if not speaker_only:
        return mode
    target = (speaker_camera or "").strip()
    if not target or target.lower() == "active":
        return mode
    if identity != target:
        return "off"
    return mode


def next_video_capture(now: float, due: float, period: float) -> tuple[bool, float]:
    """Pace VideoSource.capture_frame.

    When the loop is late, capture once and jump the next due time to
    ``now + period`` so we never catch up by pushing extra frames into
    libwebrtc's EncoderQueue (that path leaked ~2.6 MiB/s native RSS).
    """
    period = float(period)
    if period <= 0:
        period = 1.0 / 30.0
    if now < due:
        return False, due
    return True, now + period


def should_drop_stale_vaapi_frame(
    now_us: int,
    capture_us: int,
    fps: int,
    max_frames_behind: int = 3,
    is_keyframe: bool = False,
) -> bool:
    """True when VAAPI Encode should drop before ToI420 (encoder behind).

    LiveKit's VAAPIH264EncoderWrapper::Encode never dropped; EncoderQueue
    kept I420 copies (~0.37 extra frames/s/cam → OOM). Keyframes always
    pass. Missing timestamps are not treated as stale.
    """
    if is_keyframe:
        return False
    if int(capture_us) <= 0:
        return False
    fps = max(1, int(fps))
    budget = max(1, int(max_frames_behind))
    max_delay_us = budget * 1_000_000 // fps
    return int(now_us) - int(capture_us) > max_delay_us


def should_drop_encoder_queue_depth(queued: int, max_queued: int = 1) -> bool:
    """True when VideoTrackSource must not push another frame (latest-wins)."""
    return int(queued) >= max(1, int(max_queued))


def video_capture_gate_acquire(gate: dict) -> bool:
    """Admit a capture_frame if in-flight slots remain (AudioSource-like).

    ``gate`` is ``{in_flight, dropped, max_in_flight}``. Caller must
    ``video_capture_gate_release`` when the encoder slot is free (or after
    one frame period as a stand-in when FFI has no completion callback).
    """
    max_in_flight = max(1, int(gate.get("max_in_flight") or 3))
    if int(gate.get("in_flight") or 0) >= max_in_flight:
        gate["dropped"] = int(gate.get("dropped") or 0) + 1
        return False
    gate["in_flight"] = int(gate.get("in_flight") or 0) + 1
    return True


def video_capture_gate_release(gate: dict) -> None:
    """Free one in-flight capture slot."""
    n = int(gate.get("in_flight") or 0)
    gate["in_flight"] = n - 1 if n > 0 else 0


def should_recycle_encoders(
    rss_anon_kb: int,
    limit_kb: int,
    now: float,
    last_recycle_at: float,
    min_interval_s: float = 60.0,
) -> bool:
    """True when native RSS is over the cap and a recycle is allowed."""
    if int(limit_kb) <= 0:
        return False
    if int(rss_anon_kb) < int(limit_kb):
        return False
    if last_recycle_at > 0 and (now - last_recycle_at) < float(min_interval_s):
        return False
    return True


def video_stream_kwargs(capacity: int = 0, format_name: str = "") -> dict:
    """LiveKit VideoStream options. capacity=0 is unbounded (SDK default)."""
    out: dict = {}
    if int(capacity) > 0:
        out["capacity"] = int(capacity)
    fmt = (format_name or "").strip().lower()
    if fmt:
        out["format"] = fmt
    return out