103 lines
4.4 KiB
Python
103 lines
4.4 KiB
Python
"""Environment-backed configuration for the Meeting Assistant application."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
class ConfigurationError(ValueError):
|
|
"""Raised when required processing configuration is unavailable."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AppSettings:
|
|
"""Machine-specific Meeting Lab settings kept outside the UI."""
|
|
|
|
data_root: Path
|
|
whisper_model: Path | None
|
|
whisper_executable: str = "whisper-cli"
|
|
ffmpeg_executable: str = "ffmpeg"
|
|
protocol_model: str = "qwen3.8:27b"
|
|
ollama_endpoint: str = "http://127.0.0.1:11434"
|
|
protocol_num_ctx: int = 32_768
|
|
protocol_safe_input_token_budget: int = 29_000
|
|
language: str = "de"
|
|
threads: str | int = "auto"
|
|
diarization_mode: str = "auto"
|
|
diarization_runtime: str = "native"
|
|
diarization_container_image: str | None = None
|
|
diarization_container_args: tuple[str, ...] = ()
|
|
|
|
@classmethod
|
|
def from_environment(cls) -> AppSettings:
|
|
"""Load settings from MKA_* environment variables."""
|
|
whisper_model = os.getenv("MKA_WHISPER_MODEL")
|
|
threads = os.getenv("MKA_WHISPER_THREADS", "auto")
|
|
parsed_threads: str | int = int(threads) if threads.isdigit() else threads
|
|
return cls(
|
|
data_root=Path(os.getenv("MKA_DATA_ROOT", "data/meetings")),
|
|
whisper_model=Path(whisper_model) if whisper_model else None,
|
|
whisper_executable=os.getenv("MKA_WHISPER_EXECUTABLE", "whisper-cli"),
|
|
ffmpeg_executable=os.getenv("MKA_FFMPEG_EXECUTABLE", "ffmpeg"),
|
|
protocol_model=os.getenv("MKA_PROTOCOL_MODEL", "qwen3.8:27B"),
|
|
ollama_endpoint=os.getenv("MKA_OLLAMA_ENDPOINT", "http://127.0.0.1:11434"),
|
|
language=os.getenv("MKA_LANGUAGE", "de"),
|
|
threads=parsed_threads,
|
|
diarization_mode=os.getenv("MKA_DIARIZATION_MODE", "auto"),
|
|
diarization_runtime=os.getenv("MKA_DIARIZATION_RUNTIME", "native"),
|
|
diarization_container_image=os.getenv("MKA_DIARIZATION_CONTAINER_IMAGE"),
|
|
diarization_container_args=_parse_container_args(
|
|
os.getenv("MKA_DIARIZATION_CONTAINER_ARGS")
|
|
),
|
|
)
|
|
|
|
def validate_for_processing(self) -> None:
|
|
"""Validate settings needed before a Meeting Lab run starts."""
|
|
if self.whisper_model is None:
|
|
raise ConfigurationError("MKA_WHISPER_MODEL is not configured.")
|
|
if not self.whisper_model.is_file():
|
|
raise ConfigurationError(
|
|
f"Configured Whisper model does not exist: {self.whisper_model}"
|
|
)
|
|
if not self.whisper_executable.strip():
|
|
raise ConfigurationError("MKA_WHISPER_EXECUTABLE must not be empty.")
|
|
if not self.ffmpeg_executable.strip():
|
|
raise ConfigurationError("MKA_FFMPEG_EXECUTABLE must not be empty.")
|
|
if self.diarization_mode not in {"auto", "cpu", "gpu"}:
|
|
raise ConfigurationError("MKA_DIARIZATION_MODE must be one of: auto, cpu, gpu.")
|
|
if self.diarization_runtime not in {"native", "container"}:
|
|
raise ConfigurationError("MKA_DIARIZATION_RUNTIME must be native or container.")
|
|
if self.diarization_runtime == "container" and not self.diarization_container_image:
|
|
raise ConfigurationError(
|
|
"MKA_DIARIZATION_CONTAINER_IMAGE is required for container runtime."
|
|
)
|
|
if any(
|
|
not isinstance(argument, str) or not argument
|
|
for argument in self.diarization_container_args
|
|
):
|
|
raise ConfigurationError(
|
|
"MKA_DIARIZATION_CONTAINER_ARGS must contain only non-empty strings."
|
|
)
|
|
|
|
|
|
def _parse_container_args(value: str | None) -> tuple[str, ...]:
|
|
"""Parse an ordered JSON array of opaque container command arguments."""
|
|
if value is None or not value.strip():
|
|
return ()
|
|
try:
|
|
parsed = json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigurationError(
|
|
"MKA_DIARIZATION_CONTAINER_ARGS must be a JSON array of strings."
|
|
) from exc
|
|
if not isinstance(parsed, list) or any(
|
|
not isinstance(argument, str) or not argument for argument in parsed
|
|
):
|
|
raise ConfigurationError(
|
|
"MKA_DIARIZATION_CONTAINER_ARGS must be a JSON array of non-empty strings."
|
|
)
|
|
return tuple(parsed)
|