"""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