from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path from urllib.parse import urlparse class ConfigurationError(ValueError): """The ASR service is deliberately unusable until configured safely.""" def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except ValueError as exc: raise ConfigurationError(f"{name} must be an integer") from exc def _env_float(name: str, default: float) -> float: try: return float(os.getenv(name, str(default))) except ValueError as exc: raise ConfigurationError(f"{name} must be a number") from exc def _env_bool(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default normalized = value.strip().casefold() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise ConfigurationError(f"{name} must be a boolean") def _require_loopback_http(value: str) -> str: parsed = urlparse(value) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise ConfigurationError("OMNI_ASR_ENDPOINT must be an explicit HTTP(S) URL") if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: raise ConfigurationError("OMNI_ASR_ENDPOINT must resolve through loopback") if parsed.username or parsed.password or parsed.fragment: raise ConfigurationError("OMNI_ASR_ENDPOINT must not contain credentials or a fragment") return value.rstrip("/") def _require_local_model_path(value: str) -> str: if "://" in value or value.startswith(("hf:", "s3:", "gs:")): raise ConfigurationError("OMNI_ASR_MODEL must be a local directory, not a URL") path = Path(value) if not path.is_absolute(): raise ConfigurationError("OMNI_ASR_MODEL must be an absolute local directory") try: resolved = path.resolve(strict=True) except OSError as exc: raise ConfigurationError("OMNI_ASR_MODEL does not exist locally") from exc if not resolved.is_dir(): raise ConfigurationError("OMNI_ASR_MODEL must be a local directory") return str(resolved) @dataclass(slots=True) class ASRSettings: """Bounds and provider selection for an isolated side-tap process.""" host: str = "127.0.0.1" port: int = 8101 allow_container_bind: bool = False provider: str = "" endpoint: str | None = None model: str | None = None api_key: str | None = None device: str = "cuda:0" dtype: str = "auto" max_connections: int = 8 max_provider_concurrency: int = 2 max_audio_bytes: int = 8 * 1024 * 1024 max_audio_seconds: float = 90.0 max_frame_bytes: int = 256 * 1024 max_response_bytes: int = 256 * 1024 start_timeout_s: float = 3.0 idle_timeout_s: float = 15.0 session_timeout_s: float = 120.0 provider_queue_timeout_s: float = 0.05 transcribe_timeout_s: float = 30.0 connect_timeout_s: float = 1.0 write_timeout_s: float = 5.0 partial_interval_ms: int = 320 @classmethod def from_env(cls) -> "ASRSettings": return cls( host=os.getenv("OMNI_ASR_HOST", "127.0.0.1"), port=_env_int("OMNI_ASR_PORT", 8101), allow_container_bind=_env_bool("OMNI_ASR_ALLOW_CONTAINER_BIND"), provider=os.getenv("OMNI_ASR_PROVIDER", "").strip().lower(), endpoint=os.getenv("OMNI_ASR_ENDPOINT") or None, model=os.getenv("OMNI_ASR_MODEL") or None, api_key=os.getenv("OMNI_ASR_API_KEY") or None, device=os.getenv("OMNI_ASR_DEVICE", "cuda:0").strip().lower(), dtype=os.getenv("OMNI_ASR_DTYPE", "auto").strip().lower(), max_connections=_env_int("OMNI_ASR_MAX_CONNECTIONS", 8), max_provider_concurrency=_env_int("OMNI_ASR_MAX_PROVIDER_CONCURRENCY", 2), max_audio_bytes=_env_int("OMNI_ASR_MAX_AUDIO_BYTES", 8 * 1024 * 1024), max_audio_seconds=_env_float("OMNI_ASR_MAX_AUDIO_SECONDS", 90.0), max_frame_bytes=_env_int("OMNI_ASR_MAX_FRAME_BYTES", 256 * 1024), max_response_bytes=_env_int("OMNI_ASR_MAX_RESPONSE_BYTES", 256 * 1024), start_timeout_s=_env_float("OMNI_ASR_START_TIMEOUT_S", 3.0), idle_timeout_s=_env_float("OMNI_ASR_IDLE_TIMEOUT_S", 15.0), session_timeout_s=_env_float("OMNI_ASR_SESSION_TIMEOUT_S", 120.0), provider_queue_timeout_s=_env_float("OMNI_ASR_QUEUE_TIMEOUT_S", 0.05), transcribe_timeout_s=_env_float("OMNI_ASR_TRANSCRIBE_TIMEOUT_S", 30.0), connect_timeout_s=_env_float("OMNI_ASR_CONNECT_TIMEOUT_S", 1.0), write_timeout_s=_env_float("OMNI_ASR_WRITE_TIMEOUT_S", 5.0), partial_interval_ms=_env_int("OMNI_ASR_PARTIAL_INTERVAL_MS", 320), ).validated() def validated(self) -> "ASRSettings": if self.host not in {"127.0.0.1", "localhost", "::1"}: if self.host not in {"0.0.0.0", "::"} or not self.allow_container_bind: raise ConfigurationError( "OMNI_ASR_HOST must be loopback; a wildcard container bind " "requires OMNI_ASR_ALLOW_CONTAINER_BIND=1" ) if not 1 <= self.port <= 65_535: raise ConfigurationError("OMNI_ASR_PORT must be a valid port") if self.provider not in {"mock", "openai", "whisper_cpp", "transformers_local"}: raise ConfigurationError( "OMNI_ASR_PROVIDER must explicitly be mock, openai, whisper_cpp, " "or transformers_local" ) if self.provider in {"openai", "whisper_cpp"}: if not self.endpoint: raise ConfigurationError("OMNI_ASR_ENDPOINT is required") if not self.model or not self.model.strip(): raise ConfigurationError("OMNI_ASR_MODEL is required") self.endpoint = _require_loopback_http(self.endpoint) self.model = self.model.strip() elif self.provider == "transformers_local": if not self.model or not self.model.strip(): raise ConfigurationError("OMNI_ASR_MODEL is required") self.model = _require_local_model_path(self.model.strip()) if self.endpoint: raise ConfigurationError( "transformers_local runs in-process and must not configure OMNI_ASR_ENDPOINT" ) if self.device != "cpu": prefix, separator, index = self.device.partition(":") if prefix != "cuda" or not separator or not index.isdigit(): raise ConfigurationError("OMNI_ASR_DEVICE must be cpu or cuda:") if self.dtype not in {"auto", "float16", "bfloat16", "float32"}: raise ConfigurationError( "OMNI_ASR_DTYPE must be auto, float16, bfloat16, or float32" ) if self.device == "cpu" and self.dtype not in {"auto", "float32"}: raise ConfigurationError( "CPU transformers_local ASR requires OMNI_ASR_DTYPE=auto or float32" ) positive_ints = { "OMNI_ASR_MAX_CONNECTIONS": self.max_connections, "OMNI_ASR_MAX_PROVIDER_CONCURRENCY": self.max_provider_concurrency, "OMNI_ASR_MAX_AUDIO_BYTES": self.max_audio_bytes, "OMNI_ASR_MAX_FRAME_BYTES": self.max_frame_bytes, "OMNI_ASR_MAX_RESPONSE_BYTES": self.max_response_bytes, "OMNI_ASR_PARTIAL_INTERVAL_MS": self.partial_interval_ms, } for name, value in positive_ints.items(): if value <= 0: raise ConfigurationError(f"{name} must be positive") positive_times = { "OMNI_ASR_MAX_AUDIO_SECONDS": self.max_audio_seconds, "OMNI_ASR_START_TIMEOUT_S": self.start_timeout_s, "OMNI_ASR_IDLE_TIMEOUT_S": self.idle_timeout_s, "OMNI_ASR_SESSION_TIMEOUT_S": self.session_timeout_s, "OMNI_ASR_QUEUE_TIMEOUT_S": self.provider_queue_timeout_s, "OMNI_ASR_TRANSCRIBE_TIMEOUT_S": self.transcribe_timeout_s, "OMNI_ASR_CONNECT_TIMEOUT_S": self.connect_timeout_s, "OMNI_ASR_WRITE_TIMEOUT_S": self.write_timeout_s, } for name, value in positive_times.items(): if value <= 0: raise ConfigurationError(f"{name} must be positive") if self.max_frame_bytes > self.max_audio_bytes: raise ConfigurationError("OMNI_ASR_MAX_FRAME_BYTES cannot exceed the audio buffer") return self