"""Settings web UI for the Jarvis bridge.
A small in-app page (served by the Flask bridge) to change models, language,
TTS and the LLM instructions WITHOUT editing files or rebuilding. Writes to the
runtime config JSON (JARVIS_CONFIG_PATH) that ``load_settings()`` reads, then
restarts the bridge (and TTS worker) via supervisord so changes take effect.
Internal-network use only (no auth, per deployment decision).
"""
from __future__ import annotations
import json
import os
import subprocess
import urllib.request
from pathlib import Path
from typing import Any, Dict
# Fields the UI manages. Each maps to a key in the runtime config JSON, with a
# label and an input kind for the form.
FIELDS = [
("ollama_chat_model", "LLM 모델", "model"),
("whisper_model", "STT(Whisper) 모델", "select:tiny,base,small,medium,large,large-v3"),
("tts_engine", "TTS 엔진", "select:edge,piper"),
("output_language", "출력 언어 (비우면 사용자 언어)", "text"),
("llm_thinking_enabled", "LLM 사고(thinking) 모드", "bool"),
("agentic_max_turns", "에이전트 최대 반복", "number:1:12:1"),
("llm_instructions", "LLM 추가 지침 (시스템 프롬프트에 덧붙임)", "textarea"),
]
_KEYS = [k for k, _, _ in FIELDS]
def _config_path() -> Path:
p = os.environ.get("JARVIS_CONFIG_PATH")
return Path(p).expanduser() if p else (Path.home() / ".config" / "jarvis" / "config.json")
def _persist_path() -> Path:
"""Persistent overrides on the data volume — survive container recreate.
entrypoint.sh merges this back onto the env-rendered config at startup."""
return Path(os.environ.get("JARVIS_SETTINGS_PATH") or "/data/jarvis-settings.json")
def _read_config() -> Dict[str, Any]:
try:
return json.loads(_config_path().read_text("utf-8"))
except Exception:
return {}
def _current() -> Dict[str, Any]:
cfg = _read_config()
out: Dict[str, Any] = {}
for k in _KEYS:
if k == "output_language":
out[k] = cfg.get("output_language", os.environ.get("OUTPUT_LANGUAGE", ""))
else:
out[k] = cfg.get(k, "")
return out
def _ollama_models() -> list[str]:
base = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434").rstrip("/")
try:
with urllib.request.urlopen(f"{base}/api/tags", timeout=4) as r:
data = json.loads(r.read())
return sorted(m.get("name", "") for m in data.get("models", []) if m.get("name"))
except Exception:
return []
def _coerce(updates: Dict[str, Any]) -> Dict[str, Any]:
clean: Dict[str, Any] = {}
for k, v in updates.items():
if k not in _KEYS:
continue
if k == "agentic_max_turns":
try:
v = int(v)
except (TypeError, ValueError):
continue
elif k == "llm_thinking_enabled":
v = str(v).lower() in ("1", "true", "on", "yes")
clean[k] = v
return clean
def _write_merge(path: Path, clean: Dict[str, Any]) -> None:
cur: Dict[str, Any] = {}
try:
cur = json.loads(path.read_text("utf-8"))
except Exception:
cur = {}
cur.update(clean)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(cur, ensure_ascii=False, indent=2), "utf-8")
def _save(updates: Dict[str, Any]) -> None:
clean = _coerce(updates)
# 1) persistent overrides (survive `docker compose up` recreate)
_write_merge(_persist_path(), clean)
# 2) runtime config so a bridge/worker restart picks it up immediately
_write_merge(_config_path(), clean)
def _apply() -> str:
# Restart the bridge AFTER this response is sent. Detached (new session) so
# the bridge being killed mid-restart doesn't drop the restart itself, and
# the HTTP client still receives this response. (Edge TTS has no worker.)
try:
subprocess.Popen(
["sh", "-c", "sleep 1; supervisorctl restart bridge"],
start_new_session=True,
)
return "1초 후 브리지가 재시작되어 반영됩니다."
except Exception as e: # pragma: no cover
return str(e)
_PAGE = """
Jarvis 설정