dschlueter
d45e2c9ae3
Lokales LLM: Call-Dauer gemessen -> Stromkosten je Job ((max-idle)*load*n/ 3.6e6*Preis(EUR/kWh)*usd_per_eur), Kategorie llm_local. Cartesia-TTS: Zeichen x cartesia_usd_per_char (tts). Notruf: pro Webhook-Alarm 0,20€*usd_per_eur (alert), nur Doku, nie Sperre. Helfer in costs.py; neue Settings (GPU-Watt/Last/ Strompreis/usd_per_eur/cartesia/alert). Statistik: "exakt vs. geschätzt" (Tooltip + Summenzeile, Kategorien tts/alert/llm_local). Tests: 318. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
279 lines
13 KiB
Python
279 lines
13 KiB
Python
import os
|
||
from pathlib import Path
|
||
|
||
try:
|
||
import tomllib # Python >= 3.11 (stdlib)
|
||
except ModuleNotFoundError: # pragma: no cover - Fallback fuer aeltere Interpreter
|
||
import tomli as tomllib # type: ignore
|
||
|
||
from pydantic.fields import FieldInfo
|
||
from pydantic_settings import (
|
||
BaseSettings,
|
||
PydanticBaseSettingsSource,
|
||
SettingsConfigDict,
|
||
)
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||
ENV_FILE = BASE_DIR / ".env"
|
||
DEFAULT_CONFIG_FILE = BASE_DIR / "config" / "voice-assistant.toml"
|
||
|
||
|
||
def _setting_lookup(key: str) -> str | None:
|
||
"""Liest einen Steuer-Schluessel: echte Umgebung zuerst, dann die .env-Datei.
|
||
|
||
Noetig fuer VA_PROFILE/VA_CONFIG_FILE, weil diese gebraucht werden, BEVOR
|
||
pydantic-settings die .env laedt - und .env-Werte sonst nicht in os.environ stehen.
|
||
"""
|
||
value = os.getenv(key)
|
||
if value is not None:
|
||
return value
|
||
try:
|
||
from dotenv import dotenv_values
|
||
except ModuleNotFoundError: # pragma: no cover
|
||
return None
|
||
if ENV_FILE.is_file():
|
||
return dotenv_values(ENV_FILE).get(key)
|
||
return None
|
||
|
||
|
||
def _config_file_path() -> Path:
|
||
return Path(_setting_lookup("VA_CONFIG_FILE") or str(DEFAULT_CONFIG_FILE))
|
||
|
||
|
||
def active_profile() -> str | None:
|
||
"""Name des aktiven Profils (VA_PROFILE) aus Umgebung oder .env, falls gesetzt."""
|
||
profile = _setting_lookup("VA_PROFILE")
|
||
return profile.strip() or None if profile else None
|
||
|
||
|
||
def load_profile_config() -> dict:
|
||
"""Liest die zentrale TOML-Config und merged [defaults] + [profiles.<VA_PROFILE>].
|
||
|
||
- Fehlt die Datei, gilt ein leeres dict (nur ENV/Defaults greifen) - kein Fehler,
|
||
damit reine Cloud-Deployments ohne Datei (nur ENV) funktionieren.
|
||
- Ein gesetztes, aber unbekanntes VA_PROFILE ist ein Konfigurationsfehler.
|
||
"""
|
||
path = _config_file_path()
|
||
if not path.is_file():
|
||
return {}
|
||
|
||
with path.open("rb") as handle:
|
||
data = tomllib.load(handle)
|
||
|
||
merged: dict = dict(data.get("defaults", {}))
|
||
|
||
profile = active_profile()
|
||
if profile:
|
||
profiles = data.get("profiles", {})
|
||
if profile not in profiles:
|
||
raise ValueError(
|
||
f"Unbekanntes VA_PROFILE {profile!r}. "
|
||
f"Verfuegbar: {sorted(profiles)}"
|
||
)
|
||
merged.update(profiles[profile])
|
||
|
||
return merged
|
||
|
||
|
||
class TomlProfileSource(PydanticBaseSettingsSource):
|
||
"""Settings-Quelle aus der zentralen TOML-Config (inkl. aktivem Profil).
|
||
|
||
Liegt in der Praezedenz unter ENV/.env, aber ueber den eingebauten Defaults.
|
||
Es werden nur Schluessel durchgereicht, die auch als Settings-Feld existieren.
|
||
"""
|
||
|
||
def __init__(self, settings_cls):
|
||
super().__init__(settings_cls)
|
||
raw = load_profile_config()
|
||
known = set(settings_cls.model_fields)
|
||
self._values = {
|
||
key.lower(): value
|
||
for key, value in raw.items()
|
||
if key.lower() in known
|
||
}
|
||
|
||
def get_field_value(self, field: FieldInfo, field_name: str):
|
||
if field_name in self._values:
|
||
return self._values[field_name], field_name, False
|
||
return None, field_name, False
|
||
|
||
def __call__(self) -> dict:
|
||
return dict(self._values)
|
||
|
||
|
||
class Settings(BaseSettings):
|
||
app_env: str = "dev"
|
||
host: str = "0.0.0.0"
|
||
port: int = 8080
|
||
log_level: str = "info"
|
||
openrouter_api_key: str = ""
|
||
openrouter_stt_model: str = "openai/whisper-large-v3"
|
||
openrouter_tts_model: str = "openai/gpt-4o-mini-tts"
|
||
openrouter_tts_voice: str = "alloy"
|
||
openrouter_llm_model: str = "openai/gpt-4.1-mini"
|
||
default_language: str = "de"
|
||
default_input_endpoint: str = "local-default"
|
||
default_output_endpoint: str = "local-default"
|
||
default_stt_provider: str = "openrouter"
|
||
default_llm_provider: str = "local-openai-compatible"
|
||
default_tts_provider: str = "openrouter"
|
||
# Lokaler llama.cpp-Server (OpenAI-kompatibel), siehe scripts/llm-server/.
|
||
local_llm_base_url: str = "http://127.0.0.1:8001/v1"
|
||
local_llm_api_key: str = "dummy"
|
||
local_llm_model: str = "va_llm" # = --alias des llama.cpp-Servers
|
||
# Sprach-Assistent: knappe, vorlesbare Antworten + Reasoning aus = deutlich schneller.
|
||
local_llm_system_prompt: str = (
|
||
"Du bist ein gesprochener Sprachassistent. Antworte kurz und natuerlich "
|
||
"(in der Regel 1-3 Saetze), in reinem Fliesstext ohne Markdown, ohne "
|
||
"Aufzaehlungen, ohne Emojis. Formuliere so, wie man es laut vorliest."
|
||
)
|
||
local_llm_disable_reasoning: bool = True # Qwen3 /no_think: spart die Denkphase
|
||
local_llm_max_tokens: int = 0 # 0 = serverseitiges Limit (-n)
|
||
local_llm_temperature: float = 0.3
|
||
local_llm_top_p: float = 0.9 # Nucleus-Sampling (0.0–1.0)
|
||
faster_whisper_model: str = "small" # tiny|base|small|medium|large-v3
|
||
faster_whisper_device: str = "auto" # auto|cpu|cuda
|
||
faster_whisper_compute_type: str = "int8" # default|int8|float16|int8_float16
|
||
stt_force_language: bool = True # STT mit fester Zielsprache (robust); False = Auto-Detect + Übersetzung
|
||
# GPU-Sichtbarkeit: "auto" = erkennen (respektiert vorhandenes CUDA_VISIBLE_DEVICES),
|
||
# "none"/leer = keine GPU erzwingen, "0,1,2" = nur diese Indizes. Wird beim Start an
|
||
# CUDA_VISIBLE_DEVICES gekoppelt (siehe _apply_gpu_visibility), damit das, was das Menü
|
||
# anzeigt, auch real von torch/faster-whisper genutzt wird.
|
||
gpus: str = "auto"
|
||
# --- Lokales TTS (piper) -------------------------------------------------
|
||
piper_bin: str = "piper" # Pfad/Name des piper-Binaries
|
||
piper_voices_dir: str = str(Path.home() / ".local" / "share" / "piper" / "voices")
|
||
piper_voice: str = "de_DE-thorsten-high" # Stimmmodell-Name (ohne .onnx) oder voller Pfad
|
||
tts_sample_rate: int = 24000 # Ziel-Sample-Rate (das Gateway erwartet 24000 Hz)
|
||
# --- Chatterbox-TTS (hohe Qualitaet + Voice-Cloning, eigener HTTP-Dienst) -
|
||
# --- Cartesia TTS (Cloud, sehr niedrige Latenz) -------------------------
|
||
cartesia_api_key: str = ""
|
||
cartesia_voice_id: str = "" # UUID aus cartesia.ai/voices (Fallback)
|
||
cartesia_tts_model: str = "sonic-turbo"
|
||
cartesia_voices: str = "" # "lang:gender:uuid:name" je Zeile, # = Kommentar # sonic-2 | sonic-multilingual
|
||
chatterbox_base_url: str = "http://127.0.0.1:9999"
|
||
chatterbox_voice: str = "" # Pfad zu Referenz-WAV (Voice-Cloning) oder leer
|
||
chatterbox_lang: str = "de"
|
||
chatterbox_speed: float = 1.0
|
||
chatterbox_timeout: int = 180
|
||
# Verzeichnis mit nativen Referenz-WAVs je Sprache (Konvention <lang>.wav, z. B. fr.wav).
|
||
# Greift cross-lingual: pro Sprache eine muttersprachliche Stimme. Leer -> nur chatterbox_voice.
|
||
chatterbox_voices_dir: str = str(BASE_DIR / "config" / "voices")
|
||
db_path: str = str(BASE_DIR / "data" / "voice-assistant.db")
|
||
admin_api_key: str = ""
|
||
auth_enabled: bool = True
|
||
# --- Forward-/Trusted-Header-Auth (Reverse-Proxy / YunoHost-SSO) ---------
|
||
# Ist trusted_auth_header gesetzt UND die Quell-IP in trusted_proxy_ips, wird die
|
||
# Identitaet aus diesem Header gelesen (SSO-User) und ein interner Nutzer
|
||
# automatisch angelegt. Sonst gilt die normale Token-/Anonymous-Auth.
|
||
trusted_auth_header: str = ""
|
||
# Alternativ zur Header-Variante: Identitaet aus einem (signierten) JWT-Cookie lesen.
|
||
# YunoHost reicht den Usernamen nicht als Header durch, sondern im Cookie
|
||
# "yunohost.portal" (JWT, Claim "user"). Nur von der Proxy-IP akzeptiert.
|
||
trusted_auth_cookie: str = "" # Cookie-Name (z. B. yunohost.portal)
|
||
trusted_auth_cookie_claim: str = "user" # JWT-Claim mit dem Usernamen
|
||
trusted_auth_jwt_secret: str = "" # optional: HS256-Secret -> Signatur pruefen
|
||
trusted_proxy_ips: str = "" # kommasepariert; IP(s) des Reverse-Proxys
|
||
admin_users: str = "" # kommaseparierte SSO-Usernamen mit Admin-Rechten
|
||
sso_logout_url: str = "" # Logout-Link fuers Frontend (SSO-Portal)
|
||
# Login-/Portal-URL: unauthentifizierte Seitenaufrufe werden hierhin umgeleitet
|
||
# (Defense-in-Depth zusaetzlich zu SSOwat). Leer -> stattdessen HTTP 401.
|
||
sso_login_url: str = ""
|
||
history_max_messages: int = 10
|
||
# Automatische Erinnerungs-Extraktion: das LLM destilliert dauerhafte Fakten
|
||
# aus dem Gespraech und legt sie als Nutzer-Erinnerungen ab (best-effort,
|
||
# nicht-blockierend). Leerer Provider = Default-LLM-Provider.
|
||
memory_extraction_enabled: bool = True
|
||
memory_extraction_every_n_turns: int = 3
|
||
memory_extraction_max: int = 50
|
||
memory_extraction_provider: str = ""
|
||
audio_stream_default: bool = True # satzweises TTS als Default (Admin kann abschalten)
|
||
web_search_enabled: bool = True # Web-Suche via Tool-Calling (Weg 2); global an, pro Nutzer abschaltbar
|
||
web_search_daily_limit: int = 0 # max. Web-Suchen pro Nutzer/Tag (0 = unbegrenzt); pro Nutzer überschreibbar
|
||
cost_daily_limit_usd: float = 0 # max. Kosten pro Nutzer/Tag in USD (0 = unbegrenzt); pro Nutzer überschreibbar
|
||
# Kostenerfassung Stufe 2 (geschätzte Quellen). Siehe Docs/kosten-erfassung.md.
|
||
usd_per_eur: float = 1.08 # Umrechnung EUR-naher Schätzungen in die USD-Anzeige
|
||
cartesia_usd_per_char: float = 0.00005 # Cartesia: 100k Credits ≈ 100.000 Zeichen → $5
|
||
emergency_alert_eur: float = 0.20 # SMS+Voice pro Alarm (Schätzung)
|
||
# Lokales LLM: Stromkosten pro Anfrage (marginal über Leerlauf), 2× RTX 3090.
|
||
gpu_idle_watts: float = 40.0 # Leerlauf je GPU = Infrastruktur (nicht pro Anfrage)
|
||
gpu_max_watts: float = 390.0 # Volllast je GPU
|
||
gpu_avg_load: float = 0.85 # mittlere Auslastung während einer Anfrage
|
||
electricity_price_eur_per_kwh: float = 0.33
|
||
# TTS-Text-Normalisierung: auto|full|light|off. "auto" = piper -> full, Cloud -> light.
|
||
tts_normalize_level: str = "auto"
|
||
stt_fallback: str = "" # kommaseparierte Provider-Namen (Fallback-Kette)
|
||
llm_fallback: str = ""
|
||
tts_fallback: str = ""
|
||
daily_request_limit: int = 0 # 0 = unbegrenzt; Anfragen pro Nutzer pro Tag
|
||
# Notruf-Benachrichtigung per E-Mail.
|
||
emergency_contact_email: str = "dieter.schlueter@linix.de" # Default-Kontaktperson
|
||
smtp_host: str = "" # ohne Host wird KEINE E-Mail versendet
|
||
smtp_port: int = 587
|
||
smtp_user: str = ""
|
||
smtp_password: str = ""
|
||
smtp_from: str = "" # Absenderadresse; leer -> smtp_user
|
||
smtp_starttls: bool = True
|
||
|
||
# Notruf-Eskalation per SMS/Anruf: provider-agnostischer Webhook. Bei einem Notruf
|
||
# geht ein JSON-POST an emergency_webhook_url (Kanäle, Telefonnummern, fertige
|
||
# SMS-/Anruf-Texte). Den konkreten Gateway (Twilio, seven.io, …) verdrahtet der
|
||
# Empfänger dahinter. Ohne URL wird nichts gesendet.
|
||
emergency_webhook_url: str = ""
|
||
emergency_webhook_token: str = "" # optional -> Authorization: Bearer <token>
|
||
emergency_contact_phone: str = "" # Default-Telefonnummer(n), CSV, E.164 (+49…)
|
||
emergency_webhook_channels: str = "sms,call" # angefragte Kanäle (CSV)
|
||
# Sperre nach ausgelöstem Notruf: erneuter Alarm ist N Minuten blockiert.
|
||
# Verhindert, dass verängstigte/verwirrte Senioren mehrfach hintereinander
|
||
# alarmieren. 0 = keine Sperre. Im Admin-Bereich zur Laufzeit änderbar.
|
||
emergency_cooldown_minutes: int = 5
|
||
model_config = SettingsConfigDict(
|
||
env_file=ENV_FILE, case_sensitive=False, extra="ignore"
|
||
)
|
||
|
||
@classmethod
|
||
def settings_customise_sources(
|
||
cls,
|
||
settings_cls,
|
||
init_settings,
|
||
env_settings,
|
||
dotenv_settings,
|
||
file_secret_settings,
|
||
):
|
||
# Praezedenz (frueher = hoeher): init > ENV > .env > TOML/Profil > Defaults
|
||
return (
|
||
init_settings,
|
||
env_settings,
|
||
dotenv_settings,
|
||
TomlProfileSource(settings_cls),
|
||
file_secret_settings,
|
||
)
|
||
|
||
|
||
settings = Settings()
|
||
|
||
|
||
def _parse_gpu_indices(value: str) -> list[int]:
|
||
"""'0,2' -> [0, 2]; leer/ungültig -> []."""
|
||
return sorted({int(t) for t in (value or "").split(",") if t.strip().isdigit()})
|
||
|
||
|
||
def _apply_gpu_visibility() -> None:
|
||
"""Koppelt settings.gpus an CUDA_VISIBLE_DEVICES, BEVOR torch/CUDA initialisiert.
|
||
|
||
- "auto": nichts erzwingen (ein bereits gesetztes CUDA_VISIBLE_DEVICES bleibt gültig).
|
||
- "none"/leer/"-1": CUDA_VISIBLE_DEVICES="" -> faster-whisper "auto" fällt auf CPU.
|
||
- "0,1,...": genau diese Indizes sichtbar machen (explizit gewinnt).
|
||
"""
|
||
mode = (settings.gpus or "auto").strip().lower()
|
||
if mode == "auto":
|
||
return
|
||
if mode in ("none", "off", "false", "-1", ""):
|
||
os.environ["CUDA_VISIBLE_DEVICES"] = ""
|
||
return
|
||
indices = _parse_gpu_indices(settings.gpus)
|
||
if indices:
|
||
os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(str(i) for i in indices)
|
||
|
||
|
||
_apply_gpu_visibility()
|