run.py

"""
Ctrlable Voice Assistant entry point.

Wraps linux-voice-assistant and optionally replaces the TTS player with
SnapcastTtsPlayer, routing announcements to the Ctrlable Snapcast TTS
Streamer instead of playing them locally.

Environment variables (set in satellites/<name>.env):
  AUDIO_OUTPUT          "snapcast" (default) or "local"
  SNAPCAST_SATELLITE_ID Snapcast satellite name; defaults to INSTANCE_NAME (%i)
  INSTANCE_NAME         Injected by systemd as the template instance name (%i)

All other CLI arguments (--port, --audio-input-device, wake word flags …)
are forwarded to the upstream linux-voice-assistant unchanged.
"""
import asyncio
import hashlib
import os
import sys

try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass

AUDIO_OUTPUT = os.environ.get("AUDIO_OUTPUT", "snapcast").lower()

# INSTANCE_NAME is injected by the systemd template as %i (e.g. "living-room").
# SNAPCAST_SATELLITE_ID falls back to the instance name so you rarely need to
# set it explicitly — the satellite ID matches the systemd instance name.
INSTANCE_NAME = os.environ.get("INSTANCE_NAME", "")
SNAPCAST_SATELLITE_ID = os.environ.get("SNAPCAST_SATELLITE_ID", INSTANCE_NAME or "linux-va")

# Derive a stable locally-administered MAC from the instance name so HA treats
# each satellite on the same host as a distinct device. Both the mDNS TXT record
# and the ESPHome API DeviceInfo response must report this MAC — otherwise HA
# matches all instances to the real NIC MAC and collapses them into one entry.
if INSTANCE_NAME:
    _mac_raw = hashlib.md5(INSTANCE_NAME.encode()).digest()[:6]
    _mac_bytes = bytearray(_mac_raw)
    _mac_bytes[0] = (_mac_bytes[0] | 0x02) & 0xFE  # locally administered, unicast
    _lva_instance_mac = ':'.join(f'{b:02x}' for b in _mac_bytes)
else:
    _lva_instance_mac = None

if AUDIO_OUTPUT == "snapcast":
    from snapcast_player import SnapcastTtsPlayer
    from linux_voice_assistant.satellite import VoiceSatelliteProtocol

    _snapcast_player = SnapcastTtsPlayer(satellite_id=SNAPCAST_SATELLITE_ID)
    _orig_init = VoiceSatelliteProtocol.__init__

    def _patched_init(self, state):
        _orig_init(self, state)
        state.tts_player = _snapcast_player
        _snapcast_player.set_satellite(self)
        if INSTANCE_NAME:
            state.name = f"lva-{INSTANCE_NAME}"
            state.mac_address = _lva_instance_mac

    VoiceSatelliteProtocol.__init__ = _patched_init

elif _lva_instance_mac:
    from linux_voice_assistant.satellite import VoiceSatelliteProtocol
    _orig_init = VoiceSatelliteProtocol.__init__

    def _patched_init(self, state):
        _orig_init(self, state)
        state.name = f"lva-{INSTANCE_NAME}"
        state.mac_address = _lva_instance_mac

    VoiceSatelliteProtocol.__init__ = _patched_init

# Per-instance mDNS: each satellite on the same host would otherwise register
# the same lva-<mac> hostname and only one wins. Patch HomeAssistantZeroconf
# to use lva-<instance-name> so every instance gets its own unique mDNS record.
if INSTANCE_NAME:
    from linux_voice_assistant.zeroconf import HomeAssistantZeroconf
    _lva_mdns_name = f"lva-{INSTANCE_NAME}"
    _orig_zc_init = HomeAssistantZeroconf.__init__

    def _patched_zc_init(self, port, mac_address, host_ip_address, name=None):
        _orig_zc_init(self, port=port, mac_address=_lva_instance_mac,
                      host_ip_address=host_ip_address, name=_lva_mdns_name)

    HomeAssistantZeroconf.__init__ = _patched_zc_init

    if "--name" not in sys.argv:
        sys.argv += ["--name", f"LVA - {INSTANCE_NAME}"]

# Auto-inject the bundled pymicro_wakeword models dir if no --wake-word-dir
# was given explicitly. This lets the service start without needing separate
# model downloads while still being overridable via CLI flags.
# hey_jarvis is used as the stop model so okay_nabu remains the wake word.
if "--wake-word-dir" not in sys.argv:
    try:
        import importlib.util
        spec = importlib.util.find_spec("pymicro_wakeword")
        if spec and spec.submodule_search_locations:
            models_dir = str(next(iter(spec.submodule_search_locations))) + "/models"
            sys.argv += ["--wake-word-dir", models_dir]
            if "--stop-model" not in sys.argv:
                sys.argv += ["--stop-model", "hey_jarvis"]
    except Exception:
        pass

from linux_voice_assistant.__main__ import main  # noqa: E402

sys.exit(asyncio.run(main()) or 0)