livekit-cameras/capture_va.py

"""GStreamer capture: C920 MJPEG -> jpegdec -> I420.

Shared capture daemon hosts a small pool of gst-launch processes (not
one per worker). No sidecar H.264: LiveKit VAAPI is the only encoder.
"""
from __future__ import annotations

import shutil

GST_LAUNCH = shutil.which("gst-launch-1.0") or "gst-launch-1.0"

# 5 cameras per gst-launch. group=1 does not raise fps (UVC still 15);
# a small pool still cuts 20 worker pipelines.
GST_GROUP_SIZE = 5

_QUEUE = (
    "queue",
    "max-size-buffers=1",
    "max-size-bytes=0",
    "max-size-time=0",
    "leaky=downstream",
)


def _caps(width: int, height: int, fps: int) -> str:
    fps = max(1, int(fps))
    return f"image/jpeg,width={int(width)},height={int(height)},framerate={fps}/1"


JPEG_MCU_FLICKER_CAMS = frozenset()


def jpeg_mcu_bottom_crop(height: int) -> int:
    """Rows in the last 16-tall 4:2:0 JPEG MCU (0 if height is aligned).

    C920 MJPEG 640x360 leaves 8 leftover rows. Some sensors fill that
    pad with garbage that flickers on the wall (cam-07/14/15).
    """
    h = int(height)
    if h < 32:
        return 0
    rem = h % 16
    if rem == 0:
        return 0
    return rem + 16


def needs_jpeg_mcu_fix(ident: str) -> bool:
    return ident in JPEG_MCU_FLICKER_CAMS


def crop_scale_i420_drop_bottom(
    buf: bytearray | bytes, width: int, height: int, crop: int
) -> bytearray:
    """Drop the bottom `crop` I420 rows; fill with studio black (Y=16).

    Unused on the live path: Y=16 is a raised bar once encode is 16-aligned.
    Clone is the live fix on JPEG_MCU_FLICKER_CAMS. Kept for tests.
    """
    crop = int(crop)
    width, height = int(width), int(height)
    need = width * height * 3 // 2
    out = bytearray(buf[:need] if len(buf) >= need else buf)
    if crop <= 0 or height <= crop or width < 2 or height % 2 or crop % 2:
        return out
    y_stride = width
    y_size = width * height
    uv_w = width // 2
    uv_h = height // 2
    for r in range(height - crop, height):
        dst = r * y_stride
        out[dst:dst + y_stride] = bytes([16]) * y_stride
    uv_crop = crop // 2
    u_base = y_size
    v_base = y_size + uv_w * uv_h
    z = bytes([128]) * uv_w
    for r in range(uv_h - uv_crop, uv_h):
        out[u_base + r * uv_w:u_base + (r + 1) * uv_w] = z
        out[v_base + r * uv_w:v_base + (r + 1) * uv_w] = z
    return out


def repeat_i420_bottom_from_last_good(
    buf: bytearray, width: int, height: int, crop: int
) -> None:
    """Overwrite the bottom `crop` I420 rows by cloning the last good line.

    Used after videocrop+videobox (studio-black pad). Cloning beats a
    visible 8px bar and beats videoscale (which re-edged cam-07).
    """
    crop = int(crop)
    if crop <= 0:
        return
    width, height = int(width), int(height)
    if width <= 0 or height <= crop or width % 2 or height % 2 or crop % 2:
        return
    y_stride = width
    y_size = width * height
    uv_w = width // 2
    uv_h = height // 2
    uv_crop = crop // 2
    last_y = height - crop - 1
    src_y = last_y * y_stride
    src_row = buf[src_y:src_y + y_stride]
    for r in range(height - crop, height):
        dst = r * y_stride
        buf[dst:dst + y_stride] = src_row
    last_uv = uv_h - uv_crop - 1
    u_base = y_size
    v_base = y_size + uv_w * uv_h
    src_u = buf[u_base + last_uv * uv_w:u_base + (last_uv + 1) * uv_w]
    src_v = buf[v_base + last_uv * uv_w:v_base + (last_uv + 1) * uv_w]
    for r in range(uv_h - uv_crop, uv_h):
        buf[u_base + r * uv_w:u_base + (r + 1) * uv_w] = src_u
        buf[v_base + r * uv_w:v_base + (r + 1) * uv_w] = src_v


def _one_cam(device: str, width: int, height: int, fps: int, fd: int) -> list[str]:
    width, height = int(width), int(height)
    tail: list[str] = [
        "videoconvert", "qos=false", "n-threads=2",
        "!", "video/x-raw,format=I420",
        "!", *_QUEUE,
        "!", "fdsink", f"fd={int(fd)}", "sync=false",
    ]
    return [
        "v4l2src", f"device={device}", "do-timestamp=false",
        "!", _caps(width, height, fps),
        "!", "jpegparse",
        "!", "jpegdec", "qos=false",
        "!", *tail,
    ]


def video_capture_cmd(
    device: str,
    width: int,
    height: int,
    fps: int,
    h264_fifo: str | None = None,  # ignored: no sidecar encode
    bitrate_kbps: int = 800,
    fd: int = 1,
) -> list[str]:
    """Single-camera pipeline: jpegdec -> I420, latest-frame queue, fdsink."""
    del h264_fifo, bitrate_kbps
    return [GST_LAUNCH, "-q", *_one_cam(device, width, height, fps, fd)]


def shared_capture_cmd(
    cameras: list[tuple[str, str]],
    width: int,
    height: int,
    fps: int,
    fds: list[int],
) -> list[str]:
    """One gst-launch with N jpegdec->I420->fdsink graphs (no h264enc)."""
    if len(cameras) != len(fds):
        raise ValueError("cameras and fds length mismatch")
    cmd: list[str] = [GST_LAUNCH, "-q"]
    for (_ident, device), fd in zip(cameras, fds):
        cmd += _one_cam(device, width, height, fps, fd)
    return cmd


def grouped_cameras(
    cameras: list[tuple[str, str]],
    fds: list[int],
    group_size: int = GST_GROUP_SIZE,
) -> list[tuple[list[tuple[str, str]], list[int]]]:
    if group_size < 1:
        raise ValueError("group_size must be >= 1")
    out: list[tuple[list[tuple[str, str]], list[int]]] = []
    for i in range(0, len(cameras), group_size):
        out.append((cameras[i:i + group_size], fds[i:i + group_size]))
    return out