"""One camera -> one LiveKit participant. Run as a subprocess per camera: python -m cameras.run_worker --index 0 ... """ from __future__ import annotations import argparse import asyncio import contextlib import json import logging import os import signal import subprocess import sys import time import numpy as np log = logging.getLogger("cameras.worker") def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--camera-json", required=True, help="JSON of one discovered camera dict") ap.add_argument("--room", required=True) ap.add_argument("--url", required=True) ap.add_argument("--token", required=True) ap.add_argument("--width", type=int, default=1280) ap.add_argument("--height", type=int, default=720) ap.add_argument("--fps", type=int, default=30) ap.add_argument("--audio-rate", type=int, default=16000) ap.add_argument("--enhance-mode", default="auto") ap.add_argument("--enhance-model", default="speechbrain/metricgan-plus-voicebank") ap.add_argument("--enhance-chunk-s", type=float, default=1.0) ap.add_argument("--enhance-hop-s", type=float, default=0.2) args = ap.parse_args() cam = json.loads(args.camera_json) logging.basicConfig( level=logging.INFO, format=f"%(asctime)s [{cam['identity']}] %(levelname)s %(message)s", ) from config import Config # noqa: F401 (ensures .env is loaded) import livekit.rtc as rtc from audio_cleanup import AudioCleaner identity = cam["identity"] video_dev = cam["video_device"] audio_card = cam.get("audio_card") async def run() -> int: cleaner = AudioCleaner( mode=args.enhance_mode, model_source=args.enhance_model, sample_rate=args.audio_rate, chunk_s=args.enhance_chunk_s, hop_s=args.enhance_hop_s, ) log.info("audio backend: %s", cleaner.backend) room = rtc.Room() await room.connect(args.url, args.token) log.info("connected to room %r as %r", room.name, identity) # ---------------- video ---------------- video_source = rtc.VideoSource(args.width, args.height) video_track = rtc.LocalVideoTrack.create_video_track( f"{identity}-video", video_source) vproc = subprocess.Popen( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "v4l2", "-input_format", "mjpeg", "-video_size", f"{args.width}x{args.height}", "-framerate", str(args.fps), "-i", video_dev, "-f", "rawvideo", "-pix_fmt", "bgra", "pipe:1"], stdout=subprocess.PIPE, ) frame_bytes = args.width * args.height * 4 video_frames = 0 async def video_loop(): nonlocal video_frames loop = asyncio.get_running_loop() last = time.monotonic() while vproc.poll() is None: data = await loop.run_in_executor( None, vproc.stdout.read, frame_bytes) if not data or len(data) < frame_bytes: if not data: break # partial read: accumulate (v4l2/ffmpeg shouldn't do this) keep = b"" while len(keep) < frame_bytes: more = await loop.run_in_executor( None, vproc.stdout.read, frame_bytes - len(keep)) if not more: return keep += more data = keep now = time.monotonic() ts_us = int((now - last) * 1_000_000) try: video_source.capture_frame( rtc.VideoFrame( args.width, args.height, rtc.VideoBufferType.BGRA, data), timestamp_us=ts_us, ) video_frames += 1 except Exception: # noqa: BLE001 log.exception("video capture_frame failed") # ---------------- audio ---------------- aproc: subprocess.Popen | None = None atrack: rtc.LocalAudioTrack | None = None if audio_card is not None: audio_source = rtc.AudioSource( sample_rate=args.audio_rate, num_channels=1) atrack = rtc.LocalAudioTrack.create_audio_track( f"{identity}-audio", audio_source) hop_s = args.enhance_hop_s chunk = int(hop_s * args.audio_rate) # samples per hop async def audio_loop(): assert aproc is not None and atrack is not None loop = asyncio.get_running_loop() # mic delivers stereo s16le: bytes per chunk = # chunk * 2 ch * 2 bytes need = chunk * 2 * 2 buf = b"" while aproc.poll() is None: data = await loop.run_in_executor( None, aproc.stdout.read, 4096) if not data: break buf += data while len(buf) >= need: piece, buf = buf[:need], buf[need:] pcm_stereo = np.frombuffer(piece, dtype=np.int16) pcm_mono = pcm_stereo.reshape(-1, 2).mean( axis=1).astype(np.int16) # align to full chunks usable = (len(pcm_mono) // chunk) * chunk if usable: cleaned = await loop.run_in_executor( None, cleaner.process, pcm_mono[:usable]) await audio_source.capture_frame( rtc.AudioFrame( cleaned.tobytes(), args.audio_rate, 1, len(cleaned))) aproc = subprocess.Popen( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "alsa", "-i", f"hw:{audio_card},0", "-ar", str(args.audio_rate), "-ac", "2", "-f", "s16le", "pipe:1"], stdout=subprocess.PIPE, ) # ---------------- publish ---------------- from livekit.rtc import TrackSource, TrackPublishOptions pub_opts = TrackPublishOptions( source=TrackSource.SOURCE_CAMERA, simulcast=False) await room.local_participant.publish_track(video_track, pub_opts) if atrack is not None: await room.local_participant.publish_track( atrack, TrackPublishOptions( source=TrackSource.SOURCE_MICROPHONE, dtx=False)) log.info("published video+audio (audio backend: %s)", cleaner.backend) else: log.info("published video only (no audio card matched)") log.info("status: UP identity=%s video=%s audio=%s", identity, video_dev, f"hw:{audio_card},0" if audio_card is not None else "none") # ---------------- run ---------------- tasks = [asyncio.create_task(video_loop(), name="video")] if aproc is not None: tasks.append(asyncio.create_task(audio_loop(), name="audio")) stop = asyncio.Event() loop = asyncio.get_running_loop() for sig in (signal.SIGTERM, signal.SIGINT): with contextlib.suppress(NotImplementedError): loop.add_signal_handler(sig, stop.set) await stop.wait() for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptions=True) for p in (vproc, aproc): if p is not None and p.poll() is None: with contextlib.suppress(Exception): p.terminate() with contextlib.suppress(Exception): p.wait(timeout=3) log.info("stopped; delivered %d video frames", video_frames) with contextlib.suppress(Exception): await room.disconnect() return 0 try: return asyncio.run(run()) except Exception: # noqa: BLE001 log.exception("worker fatal") return 1 if __name__ == "__main__": sys.exit(main())