voice_bridge/mqtt_audio_bridge.py

import paho.mqtt.client as mqtt
import requests
import wave
import io
import os
import numpy as np
import openwakeword
from openwakeword.model import Model
import collections
import threading


def _load_secret(name):
    """Resolve a secret from the environment, then from the systemd secrets file.

    environment.d reaches the systemd user manager, not a fresh SSH shell
    (AGENTS.md section 4.4), so the file fallback is required. Secret values are
    never logged.
    """
    value = os.environ.get(name)
    if value:
        return value
    path = os.path.expanduser(
        os.environ.get("SECRETS_FILE", "~/.config/environment.d/10-secrets.conf")
    )
    try:
        with open(path) as handle:
            for raw in handle:
                line = raw.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, val = line.partition("=")
                if key.strip() == name:
                    return val.strip().strip('"').strip("'")
    except OSError:
        pass
    return None


# --- CONFIG ---
# 1. VOICE SERVER (Audio IO)
VOICE_BROKER = "192.168.20.13"
VOICE_TOPIC_AUDIO = "voice/audio_stream"
VOICE_TOPIC_STATUS = "voice/status"

# 2. HOME ASSISTANT (Text Output)
HA_BROKER = "192.168.20.30"
HA_TOPIC_TEXT = "homeassistant/voice/text"

# MQTT credentials come from the environment or
# ~/.config/environment.d/10-secrets.conf (mode 600). They were previously
# hardcoded in this file and that commit is in Gitea; the password must be
# rotated. See the project brief section 1.6 / 1.9.
MQTT_USER = _load_secret("MQTT_USER")
MQTT_PASS = _load_secret("MQTT_PASS")
if not MQTT_USER or not MQTT_PASS:
    raise SystemExit(
        "MQTT_USER and MQTT_PASS must be set in the environment or in "
        "~/.config/environment.d/10-secrets.conf"
    )

WHISPER_URL = "http://127.0.0.1:5000/transcribe"
WAKE_WORD   = "hey_jarvis"

# --- OpenWakeWord expects 128-sample frames (16kHz, 0.008s per frame) ---
# We accumulate at least MIN_SAMPLES before calling predict()
MIN_SAMPLES = 128          # Minimum samples for a valid predict call
BUFFER_SIZE = 1280 * 8     # ~10 seconds of 16kHz int16 audio (20KB)

# --- STATE ---
audio_buffer = collections.deque(maxlen=BUFFER_SIZE)
command_buffer = bytearray()
is_command_mode = False
leftover_bytes = bytearray()  # Accumulates partial/odd-byte chunks from MQTT

print("Loading OpenWakeWord model...")
owwModel = Model()
print("Model loaded. Ready.")

# --- HELPER FUNCTIONS ---
def transcribe_and_send(raw_data):
    wav_io = io.BytesIO()
    with wave.open(wav_io, "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(16000)
        wav_file.writeframes(raw_data)
    wav_io.seek(0)

    try:
        files = {'audio': ('cmd.wav', wav_io, 'audio/wav')}
        res = requests.post(WHISPER_URL, files=files)
        if res.status_code == 200:
            text = res.json().get("transcript", "").strip()
            # --- known mishears (homophones whisper gets wrong) ---
            CORRECTIONS = {
                "fade the fish": "feed the fish",
                "fade fish": "feed the fish",
                "feed fish": "feed the fish",
            }
            _t = text.strip().lower()
            if _t in CORRECTIONS:
                text = CORRECTIONS[_t]
                print(f"TEXT (corrected): {text}")
            else:
                print(f"TEXT: {text}")
            if text:
                ha_client.publish(HA_TOPIC_TEXT, text)
                voice_client.publish(VOICE_TOPIC_STATUS, "OK")
    except Exception as e:
        print(f"API Error: {e}")


def safe_predict(samples):
    """Run OpenWakeWord predict with length validation."""
    if len(samples) < MIN_SAMPLES:
        return None
    # Truncate to nearest multiple of MIN_SAMPLES for model compatibility
    valid_len = (len(samples) // MIN_SAMPLES) * MIN_SAMPLES
    if valid_len == 0:
        return None
    try:
        return owwModel.predict(samples[:valid_len])
    except Exception as e:
        print(f"Predict error (skipping chunk): {e}")
        return None


# --- CALLBACKS ---
def on_voice_message(client, userdata, msg):
    global is_command_mode, command_buffer, leftover_bytes

    if msg.topic == VOICE_TOPIC_AUDIO:
        payload = msg.payload

        # 1. COMMAND MODE (Recording for Whisper)
        if is_command_mode:
            if len(command_buffer) > 3000000:
                print("Error: Buffer overflow. Resetting.")
                is_command_mode = False
                command_buffer = bytearray()
                owwModel.reset()
                leftover_bytes = bytearray()
                return
            command_buffer.extend(payload)
            return

        # 2. LISTENING MODE (Wake Word Detection)
        # Prepend any leftover bytes from a previous truncated chunk
        if leftover_bytes:
            payload = leftover_bytes + payload
            leftover_bytes = bytearray()

        # Discard odd byte if payload length is not a multiple of 2 (int16)
        if len(payload) % 2 != 0:
            leftover_bytes = bytearray([payload[-1]])
            payload = payload[:-1]

        if len(payload) < 2:
            return  # Not enough data yet

        try:
            audio_int16 = np.frombuffer(payload, dtype=np.int16)
        except Exception as e:
            print(f"Buffer conversion error: {e}")
            leftover_bytes = bytearray()
            return

        if len(audio_int16) < MIN_SAMPLES:
            # Save for next chunk
            leftover_bytes = payload
            return

        prediction = safe_predict(audio_int16)
        if prediction is None:
            return

        if WAKE_WORD in prediction and prediction[WAKE_WORD] > 0.5:
            print(f"--- WAKE WORD DETECTED ({WAKE_WORD}) ---")
            is_command_mode = True
            command_buffer = bytearray()
            # Save leftover samples that weren't used in prediction
            valid_len = (len(audio_int16) // MIN_SAMPLES) * MIN_SAMPLES
            if valid_len < len(audio_int16):
                command_buffer.extend(audio_int16[valid_len:].tobytes())
            voice_client.publish(VOICE_TOPIC_STATUS, "WAKE")

    elif msg.topic == "voice/status" and msg.payload.decode() == "processing":
        if is_command_mode:
            print("Processing command audio...")
            transcribe_and_send(command_buffer)
            is_command_mode = False
            command_buffer = bytearray()
            owwModel.reset()
            leftover_bytes = bytearray()


# --- MAIN ---
voice_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id="Bridge_Voice")
voice_client.username_pw_set(MQTT_USER, MQTT_PASS)
voice_client.on_message = on_voice_message
voice_client.connect(VOICE_BROKER, 1883, 60)
voice_client.subscribe([(VOICE_TOPIC_AUDIO, 0), ("voice/status", 0)])

ha_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id="Bridge_HA")
ha_client.username_pw_set(MQTT_USER, MQTT_PASS)
ha_client.connect(HA_BROKER, 1883, 60)

print("Bridge Running (Dual Broker)...")

ha_thread = threading.Thread(target=ha_client.loop_forever, daemon=True)
ha_thread.start()

voice_client.loop_forever()