From 71578620955b30d1b574b5e80fef368be31c02d7 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Sat, 26 Sep 2026 08:00:04 +0200 Subject: Read settings from config.toml over built-in defaults Co-Authored-By: Claude Opus 5.5 --- assistant.py | 88 +++++++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 70 insertions(+), 18 deletions(-) (limited to 'assistant.py') 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") -- cgit v1.2.3