root
b02138669a
Delay-and-Sum still uses the tutorial STFT/cov/GCC-PHAT/DelaySum/ISTFT path. TDOA is refreshed every BEAMFORM_TDOA_EVERY hops (default 5).
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
"""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 = 1280
|
|
height: int = 720
|
|
fps: int = 30
|
|
min_fps: int = 5
|
|
# 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.2
|
|
# 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"])
|
|
|
|
|
|
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 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", "1280")),
|
|
height=int(os.environ.get("VIDEO_HEIGHT", "720")),
|
|
fps=int(os.environ.get("VIDEO_FPS", "30")),
|
|
min_fps=int(os.environ.get("VIDEO_MIN_FPS", "5")),
|
|
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.2")),
|
|
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")),
|
|
)
|
|
|
|
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.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
|