Read settings from config.toml over built-in defaults

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 70 insertions, 18 deletions

diff --git a/assistant.py b/assistant.py
index 00c1081..5dde49b 100644
--- a/assistant.py
+++ b/assistant.py
@@ -30,20 +30,58 @@ import subprocess
 import sys
 import threading
 import time
+import tomllib
 import unicodedata
 import urllib.request
 import wave
 
-CARD = "Microphone"                       # ALSA card id of the USB mic
-MODELS = "/data/voice-models"
-WHISPER = "http://127.0.0.1:8182"
-LLM = "http://localhost:8181/v1"
+CONFIG_DIR = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"),
+                          "desktop-assistant")
 RAW = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "desktop-assistant.raw")
 WAV = RAW[:-3] + "wav"
-SYSTEM = ("You are a voice assistant. Your replies are spoken aloud, so answer "
-          "briefly in plain sentences: no markdown, no lists, no emoji. "
-          "Reply in the language the user speaks, Italian or English.")
-KLANG = {"it": "it", "en": "en-us"}
+DEFAULTS = {
+    "card": "Microphone",                 # ALSA card id of the USB mic
+    "llm": "http://localhost:8181/v1",
+    "system": ("You are a voice assistant. Your replies are spoken aloud, so answer "
+               "briefly in plain sentences: no markdown, no lists, no emoji. "
+               "Reply in the language the user speaks, Italian or English."),
+    "tool_rounds": 3,
+    "whisper": {"model": "/data/voice-models/whisper/ggml-large-v3-turbo-q5_0.bin",
+                "device": 0, "port": 8182},
+    "kokoro": {"model": "/data/voice-models/kokoro/kokoro-v1.0.onnx",
+               "voices": "/data/voice-models/kokoro/voices-v1.0.bin"},
+    "voice": {"it": {"lang": "it", "blend": {"if_sara": 0.8, "af_bella": 0.2}},
+              "en": {"lang": "en-us", "blend": {"af_heart": 1.0}}},
+    "env": {},
+}
+OPEN = {"env", "blend"}                   # free-form tables: replaced whole, not merged
+
+
+def merge(base, over, where=""):
+    # A typo must fail loudly, not fall back to the default in silence.
+    out = dict(base)
+    for k, v in over.items():
+        if k not in base:
+            raise SystemExit(f"config: unknown key {where}{k}")
+        if isinstance(base[k], dict) and k not in OPEN:
+            if not isinstance(v, dict):
+                raise SystemExit(f"config: {where}{k} must be a table")
+            v = merge(base[k], v, f"{where}{k}.")
+        out[k] = v
+    return out
+
+
+def load_config(path):
+    try:
+        with open(path, "rb") as f:
+            return merge(DEFAULTS, tomllib.load(f))
+    except FileNotFoundError:
+        return DEFAULTS
+
+
+CFG = load_config(os.path.join(CONFIG_DIR, "config.toml"))
+WHISPER = f"http://127.0.0.1:{CFG['whisper']['port']}"
+KLANG = {l: v["lang"] for l, v in CFG["voice"].items()}
 
 
 def log(*a):
@@ -51,7 +89,7 @@ def log(*a):
 
 
 def mic_on():
-    out = subprocess.run(["amixer", "-c", CARD, "cget", "numid=2"],
+    out = subprocess.run(["amixer", "-c", CFG["card"], "cget", "numid=2"],
                          capture_output=True, text=True).stdout
     return out.rstrip().endswith("values=on")
 
@@ -113,7 +151,7 @@ def split_sentences(buf):
 def llm_model():
     # The router keeps one model resident; asking for another would swap it
     # out under other clients, so use whatever is not unloaded.
-    with urllib.request.urlopen(LLM + "/models") as r:
+    with urllib.request.urlopen(CFG["llm"] + "/models") as r:
         models = json.load(r)["data"]
     return next((m["id"] for m in models if m["status"]["value"] != "unloaded"),
                 models[0]["id"])
@@ -122,7 +160,7 @@ def llm_model():
 def ask(messages):
     body = {"model": llm_model(), "messages": messages, "stream": True,
             "chat_template_kwargs": {"enable_thinking": False}}
-    req = urllib.request.Request(LLM + "/chat/completions", json.dumps(body).encode(),
+    req = urllib.request.Request(CFG["llm"] + "/chat/completions", json.dumps(body).encode(),
                                  {"Content-Type": "application/json"})
     with urllib.request.urlopen(req) as r:
         for line in r:
@@ -135,9 +173,10 @@ def ask(messages):
 
 def speaker(q):
     from kokoro_onnx import Kokoro
-    k = Kokoro(f"{MODELS}/kokoro/kokoro-v1.0.onnx", f"{MODELS}/kokoro/voices-v1.0.bin")
-    voices = {"it": 0.8 * k.get_voice_style("if_sara") + 0.2 * k.get_voice_style("af_bella"),
-              "en": k.get_voice_style("af_heart")}
+    k = Kokoro(CFG["kokoro"]["model"], CFG["kokoro"]["voices"])
+    # Each voice is a weighted blend of Kokoro style vectors.
+    voices = {l: sum(w * k.get_voice_style(n) for n, w in v["blend"].items())
+              for l, v in CFG["voice"].items()}
     play = None
     while True:
         item = q.get()
@@ -174,8 +213,9 @@ def start_whisper():
             return False
     if up():
         return
-    p = subprocess.Popen(["whisper-server", "-m", f"{MODELS}/whisper/ggml-large-v3-turbo-q5_0.bin",
-                          "-dev", "0", "--host", "127.0.0.1", "--port", WHISPER.rsplit(":", 1)[1]],
+    p = subprocess.Popen(["whisper-server", "-m", CFG["whisper"]["model"],
+                          "-dev", str(CFG["whisper"]["device"]),
+                          "--host", "127.0.0.1", "--port", str(CFG["whisper"]["port"])],
                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                          preexec_fn=die_with_parent)
     for _ in range(60):
@@ -204,7 +244,7 @@ def reply(history, q):
     log(f"you ({lang}, {time.perf_counter() - t:.2f}s): {text}")
     history.append({"role": "user", "content": text})
     t, said, buf = time.perf_counter(), [], ""
-    for delta in ask([{"role": "system", "content": SYSTEM}] + history[-20:]):
+    for delta in ask([{"role": "system", "content": CFG["system"]}] + history[-20:]):
         # ponytail: last 20 messages, summarise if long sessions need memory
         buf += delta
         done, buf = split_sentences(buf)
@@ -231,7 +271,7 @@ def main():
     q = queue.Queue()
     threading.Thread(target=speaker, args=(q,), daemon=True).start()
     history, rec = [], None
-    events = subprocess.Popen(["stdbuf", "-oL", "amixer", "-c", CARD, "events"],
+    events = subprocess.Popen(["stdbuf", "-oL", "amixer", "-c", CFG["card"], "events"],
                               stdout=subprocess.PIPE, text=True, preexec_fn=die_with_parent)
     prev = mic_on()
     log(f"ready, mic is {'on' if prev else 'off'}")
@@ -270,6 +310,18 @@ def test():
     assert lang_of("Non posso fornire il meteo in tempo reale.", "en") == "it"
     assert lang_of("That is a very exciting place to be.", "it") == "en"
     assert lang_of("Tokyo!", "it") == "it"
+    c = merge(DEFAULTS, {"card": "X", "whisper": {"port": 9},
+                         "voice": {"it": {"blend": {"a": 1.0}}}, "env": {"K": "v"}})
+    assert c["card"] == "X" and c["whisper"] == {**DEFAULTS["whisper"], "port": 9}
+    assert c["voice"]["it"] == {"lang": "it", "blend": {"a": 1.0}}      # blend replaced whole
+    assert c["voice"]["en"] == DEFAULTS["voice"]["en"] and c["env"] == {"K": "v"}
+    assert DEFAULTS["card"] == "Microphone"                             # defaults untouched
+    for bad in ({"whisper": {"prot": 1}}, {"cards": "X"}, {"whisper": 3}):
+        try:
+            merge(DEFAULTS, bad)
+            assert False, bad
+        except SystemExit as e:
+            assert "config:" in str(e)
     print("ok")