"""Load configuration from environment / .env file.""" import os from dataclasses import dataclass, field from pathlib import Path from typing import Optional def _load_dotenv(path: Path) -> None: if not path.is_file(): return for line in path.read_text().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") os.environ.setdefault(key.strip(), value.strip()) @dataclass class Config: # connection url: str = "" api_key: str = "" api_secret: str = "" room: str = "cameras" participant_prefix: str = "cam" # video width: int = 640 height: int = 360 fps: int = 30 min_fps: int = 5 video_bitrate: int = 800_000 video_codec: str = "h264" video_encoder: str = "vaapi" # auto|software|hardware|nvenc|vaapi vaapi_device: str = "/dev/dri/renderD129" # audio audio_rate: int = 16000 audio_channels: int = 1 # speechbrain cleanup enhance_mode: str = "auto" # auto | force | off enhance_model: str = "speechbrain/metricgan-plus-voicebank" enhance_chunk_s: float = 1.0 enhance_hop_s: float = 0.1 enhance_infer: str = "auto" # auto | openvino | torch # SpeechBrain Delay-and-Sum (GCC-PHAT) on C920 stereo before enhance beamform_mode: str = "auto" # auto | force | off torch_num_threads: int = 1 beamform_tdoa_every: int = 5 # misc publish_timeout_s: float = 60.0 log_level: str = "INFO" cameras: list[str] = field(default_factory=lambda: ["all"]) # headless displays (LiveKit subscribe -> GStreamer kmssink) display_count: int = 3 display_identities: list[str] = field( default_factory=lambda: ["cam-01", "cam-02", "cam-03"]) display_connectors: list[str] = field(default_factory=list) display_identity: str = "display-wall" display_roles: list[str] = field( default_factory=lambda: ["grid", "speaker", "screenshare"]) display_speaker_camera: str = "cam-01" # cam-NN or "active" display_speaker_participant: str = "speaker" display_grid_cols: int = 5 display_grid_rows: int = 4 display_grid_width: int = 1920 display_grid_height: int = 1080 display_grid_fps: int = 30 livekit_public_url: str = "" def _parse_cameras(spec: str) -> list[str]: """Expand 'all' / '1,3,5' / '0-19' into a list of indices ('' = all).""" spec = (spec or "all").strip().lower() if spec in ("all", ""): return ["all"] out = [] for part in spec.split(","): part = part.strip() if not part: continue if "-" in part: lo, hi = part.split("-", 1) out.extend(str(i) for i in range(int(lo), int(hi) + 1)) else: out.append(part) return out def _parse_list(spec: str) -> list[str]: """Comma-separated names, preserving case (HDMI-A-7, cam-01).""" spec = (spec or "").strip() if not spec: return [] return [p.strip() for p in spec.split(",") if p.strip()] def load_config(env_path: Optional[Path] = None) -> Config: _load_dotenv(Path(env_path or Path(__file__).parent / ".env")) cfg = Config( url=os.environ.get("LIVEKIT_URL", ""), api_key=os.environ.get("LIVEKIT_API_KEY", ""), api_secret=os.environ.get("LIVEKIT_API_SECRET", ""), room=os.environ.get("LIVEKIT_ROOM", "cameras"), participant_prefix=os.environ.get("PARTICIPANT_PREFIX", "cam"), width=int(os.environ.get("VIDEO_WIDTH", "640")), height=int(os.environ.get("VIDEO_HEIGHT", "360")), fps=int(os.environ.get("VIDEO_FPS", "30")), min_fps=int(os.environ.get("VIDEO_MIN_FPS", "5")), video_bitrate=int(os.environ.get("VIDEO_BITRATE", "800000")), video_codec=os.environ.get("VIDEO_CODEC", "h264").lower(), video_encoder=os.environ.get("VIDEO_ENCODER", "vaapi").lower(), vaapi_device=os.environ.get("LIBVA_DRM_DEVICE", "/dev/dri/renderD129"), audio_rate=int(os.environ.get("AUDIO_SAMPLE_RATE", "16000")), audio_channels=int(os.environ.get("AUDIO_CHANNELS", "1")), enhance_mode=os.environ.get("ENHANCE_MODE", "auto").lower(), enhance_model=os.environ.get( "ENHANCE_MODEL", "speechbrain/metricgan-plus-voicebank" ), enhance_chunk_s=float(os.environ.get("ENHANCE_CHUNK_S", "1.0")), enhance_hop_s=float(os.environ.get("ENHANCE_HOP_S", "0.1")), enhance_infer=os.environ.get("ENHANCE_INFER", "auto").lower(), beamform_mode=os.environ.get("BEAMFORM_MODE", "auto").lower(), torch_num_threads=int(os.environ.get("TORCH_NUM_THREADS", "1")), beamform_tdoa_every=int(os.environ.get("BEAMFORM_TDOA_EVERY", "5")), publish_timeout_s=float(os.environ.get("PUBLISH_TIMEOUT_S", "60")), log_level=os.environ.get("LOG_LEVEL", "INFO").upper(), cameras=_parse_cameras(os.environ.get("CAMERAS", "all")), display_count=int(os.environ.get("DISPLAY_COUNT", "3")), display_identities=_parse_list( os.environ.get("DISPLAY_IDENTITIES", "cam-01,cam-02,cam-03")), display_connectors=_parse_list(os.environ.get("DISPLAY_CONNECTORS", "")), display_identity=os.environ.get("DISPLAY_IDENTITY", "display-wall"), display_roles=_parse_list( os.environ.get("DISPLAY_ROLES", "grid,speaker,screenshare")) or ["grid", "speaker", "screenshare"], display_speaker_camera=os.environ.get("DISPLAY_SPEAKER_CAMERA", "cam-01"), display_speaker_participant=os.environ.get( "DISPLAY_SPEAKER_PARTICIPANT", "speaker"), display_grid_cols=int(os.environ.get("DISPLAY_GRID_COLS", "5")), display_grid_rows=int(os.environ.get("DISPLAY_GRID_ROWS", "4")), display_grid_width=int(os.environ.get("DISPLAY_GRID_WIDTH", "1920")), display_grid_height=int(os.environ.get("DISPLAY_GRID_HEIGHT", "1080")), display_grid_fps=int(os.environ.get("DISPLAY_GRID_FPS", "30")), livekit_public_url=os.environ.get("LIVEKIT_PUBLIC_URL", ""), ) if cfg.enhance_mode not in ("auto", "force", "off"): raise ValueError(f"ENHANCE_MODE must be auto|force|off, got {cfg.enhance_mode}") if cfg.enhance_infer not in ("auto", "openvino", "torch"): raise ValueError( f"ENHANCE_INFER must be auto|openvino|torch, got {cfg.enhance_infer}") if cfg.beamform_mode not in ("auto", "force", "off"): raise ValueError( f"BEAMFORM_MODE must be auto|force|off, got {cfg.beamform_mode}") if cfg.torch_num_threads < 1: raise ValueError("TORCH_NUM_THREADS must be >= 1") if cfg.beamform_tdoa_every < 1: raise ValueError("BEAMFORM_TDOA_EVERY must be >= 1") return cfg