Merge pull request #34 from imonlinux/centralized-configuration-file

Refactor: Centralize Configuration into config.json
Type: Refactor Issue: Closes #XYZ (if you have one)

Description
This pull request completes a major refactoring goal by moving all (20+) command-line arguments into a single, centralized config.json file.

The application is now launched with a single -c or --config flag pointing to this new file, dramatically simplifying startup scripts and configuration management.

Changes
NEW: linux_voice_assistant/config.py > * Adds new dataclasses (AppConfig, AudioConfig, LedConfig, etc.) to define the application's configuration structure. > * Includes a load_config_from_json() helper to parse the JSON file and provide default values for any omitted sections.

MODIFIED: linux_voice_assistant/__main__.py > * Removed all argparse.add_argument() calls, except for the new --config flag. > * The main() function now calls load_config_from_json() at startup. > * All internal logic now references the config object (e.g., config.app.name) instead of the old args object.
This commit is contained in:
imonlinux
2025-11-17 18:40:01 -08:00
committed by GitHub
11 changed files with 346 additions and 266 deletions
+53 -75
View File
@@ -77,7 +77,7 @@ mkdir -p ~/.config/systemd/user
``` ```
```bash ```bash
cp ~/linux-voice-assistant/service/user-pw-linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service cp ~/linux-voice-assistant/service/linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service
``` ```
**Enable & start:** **Enable & start:**
@@ -108,7 +108,7 @@ mkdir -p ~/.config/systemd/user
``` ```
```bash ```bash
cp ~/linux-voice-assistant/service/user-pa-linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service cp ~/linux-voice-assistant/service/linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service
``` ```
**Enable & start:** **Enable & start:**
@@ -144,7 +144,7 @@ mkdir -p ~/.config/systemd/user
``` ```
```bash ```bash
cp ~/linux-voice-assistant/service/user-linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service cp ~/linux-voice-assistant/service/linux-voice-assistant.service ~/.config/systemd/user/linux-voice-assistant.service
``` ```
**Enable & start:** **Enable & start:**
@@ -239,57 +239,43 @@ The following effects can be selected via the `[State Name] Effect` (select) ent
| **Fast Blink** | LEDs rapidly turn on and off. | | **Fast Blink** | LEDs rapidly turn on and off. |
| **Spin** | A single LED "spins" around the strip. | | **Spin** | A single LED "spins" around the strip. |
**Edit LVA user-mode service:** **Edit LVA config.json file:**
```bash ```bash
systemctl --user edit --force --full linux-voice-assistant nano ~/linux-voice-assistant/linux_voice_assistant/config.json
``` ```
**Add MQTT configuration entries to LVA user-mode service:** **Add MQTT configuration entries to LVA config.json file:**
```bash ```bash
--mqtt-host <IP_Adress> \ "mqtt": {
--mqtt-port 1883 \ "host": "192.168.1.2",
--mqtt-username <user_name> \ "port": 1883,
--mqtt-password <password> \ "username": "mqtt_username",
"password": "mqtt_password"
}
``` ```
**Example (Alsa LVA User Mode Service File)** **Example of complete config.json file**
***Note: Change the MQTT values to match your system!*** ***Note: Change the MQTT values to match your system!***
```bash ```bash
[Unit] {
Description=Linux Voice Assistant "app": {
"name": "Linux Voice Assistant"
[Service] },
Type=simple "mqtt": {
WorkingDirectory=/home/pi/linux-voice-assistant "host": "192.168.1.2",
Environment=PYTHONUNBUFFERED=1 "port": 1883,
"username": "mqtt_username",
ExecStart=/home/pi/linux-voice-assistant/script/run \ "password": "mqtt_password"
--name 'Linux Voice Assistant' \ }
--mqtt-host 192.168.1.2 \ }
--mqtt-port 1883 \
--mqtt-username mqtt \
--mqtt-password Ch@ngeMe_1st \
--wake-word-dir wakewords/openWakeWord \
--wake-word-dir wakewords \
--stop-model stop \
--audio-input-device seeed-2mic-voicecard \
--audio-output-device alsa/plughw:CARD=seeed2micvoicec \
--debug
Restart=always
RestartSec=2
[Install]
WantedBy=default.target
``` ```
**Enable & start:** **Enable & start:**
```bash ```bash
systemctl --user daemon-reload
systemctl --user restart linux-voice-assistant.service systemctl --user restart linux-voice-assistant.service
``` ```
@@ -311,59 +297,51 @@ This optional configuration support the use of the ReSpeaker 2Mic Grove Port wit
| 🟡 **Yellow Wire** | GPIO12 (Signal 1) | **DI** (Data Input) | | 🟡 **Yellow Wire** | GPIO12 (Signal 1) | **DI** (Data Input) |
| ⚪ **White Wire** | GPIO13 (Signal 2) | **CI** (Clock Input) | | ⚪ **White Wire** | GPIO13 (Signal 2) | **CI** (Clock Input) |
**Edit LVA user-mode service:** **Edit LVA config.json file:**
```bash ```bash
systemctl --user edit --force --full linux-voice-assistant nano ~/linux-voice-assistant/linux_voice_assistant/config.json
``` ```
**Add Grove (GPIO) configuration entries to LVA user-mode service:** **Add Grove (GPIO) configuration entries to LVA config.json file:**
```bash ```bash
--led-interface gpio \ "led": {
--led-data-pin 12 \ "led_type": "dotstar",
--led-clock-pin 13 \ "interface": "gpio",
"clock_pin": 13,
"data_pin": 12,
"num_leds": 10
}
``` ```
**Example (Alsa LVA User Mode Service File with MQTT Enabled)** **Example (LVA config.json file with MQTT and Grove Port enabled)**
***Note: Change the GPIO values to match your system!*** ***Note: Change the GPIO values to match your system!***
```bash ```bash
[Unit] {
Description=Linux Voice Assistant "app": {
"name": "Linux Voice Assistant"
[Service] },
Type=simple "mqtt": {
WorkingDirectory=/home/pi/linux-voice-assistant "host": "192.168.1.2",
Environment=PYTHONUNBUFFERED=1 "port": 1883,
"username": "mqtt_username",
ExecStart=/home/pi/linux-voice-assistant/script/run \ "password": "mqtt_password"
--name 'Linux Voice Assistant' \ },
--mqtt-host 192.168.1.2 \ "led": {
--mqtt-port 1883 \ "led_type": "dotstar",
--mqtt-username mqtt \ "interface": "gpio",
--mqtt-password Ch@ngeMe_1st \ "clock_pin": 13,
--led-interface gpio \ "data_pin": 12,
--led-data-pin 12 \ "num_leds": 10
--led-clock-pin 13 \ }
--wake-word-dir wakewords/openWakeWord \ }
--wake-word-dir wakewords \
--stop-model stop \
--audio-input-device seeed-2mic-voicecard \
--audio-output-device alsa/plughw:CARD=seeed2micvoicec \
--debug
Restart=always
RestartSec=2
[Install]
WantedBy=default.target
``` ```
**Enable & start:** **Enable & start:**
```bash ```bash
systemctl --user daemon-reload
systemctl --user restart linux-voice-assistant.service systemctl --user restart linux-voice-assistant.service
``` ```
+87 -119
View File
@@ -12,6 +12,7 @@ from typing import Dict, List, Optional, Union
import numpy as np import numpy as np
import sounddevice as sd import sounddevice as sd
from .config import Config, load_config_from_json # <-- NEW
from .event_bus import EventBus, EventHandler, subscribe from .event_bus import EventBus, EventHandler, subscribe
from .led_controller import LedController from .led_controller import LedController
from .mqtt_controller import MqttController from .mqtt_controller import MqttController
@@ -26,9 +27,6 @@ from .zeroconf import HomeAssistantZeroconf
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_MODULE_DIR = Path(__file__).parent _MODULE_DIR = Path(__file__).parent
_REPO_DIR = _MODULE_DIR.parent _REPO_DIR = _MODULE_DIR.parent
_WAKEWORDS_DIR = _REPO_DIR / "wakewords"
_OWW_DIR = _WAKEWORDS_DIR / "openWakeWord"
_SOUNDS_DIR = _REPO_DIR / "sounds"
if is_arm(): if is_arm():
_LIB_DIR = _REPO_DIR / "lib" / "linux_arm64" _LIB_DIR = _REPO_DIR / "lib" / "linux_arm64"
@@ -69,100 +67,27 @@ class MicMuteHandler(EventHandler):
async def main() -> None: async def main() -> None:
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("--name", required=True)
parser.add_argument( parser.add_argument(
"--audio-input-device", "-c",
default="default", "--config",
help="sounddevice name for input device", type=Path,
) required=True,
parser.add_argument("--audio-input-block-size", type=int, default=1024) help="Path to configuration.json file",
parser.add_argument("--audio-output-device", help="mpv name for output device")
parser.add_argument(
"--wake-word-dir",
default=[_WAKEWORDS_DIR],
action="append",
help="Directory with wake word models (.tflite) and configs (.json)",
)
parser.add_argument(
"--wake-model", default="okay_nabu", help="Id of active wake model"
)
parser.add_argument("--stop-model", default="stop", help="Id of stop model")
parser.add_argument(
"--refractory-seconds",
default=2.0,
type=float,
help="Seconds before wake word can be activated again",
)
parser.add_argument(
"--oww-melspectrogram-model",
default=_OWW_DIR / "melspectrogram.tflite",
help="Path to openWakeWord melspectrogram model",
)
parser.add_argument(
"--oww-embedding-model",
default=_OWW_DIR / "embedding_model.tflite",
help="Path to openWakeWord embedding model",
)
parser.add_argument(
"--wakeup-sound", default=str(_SOUNDS_DIR / "wake_word_triggered.flac")
)
parser.add_argument(
"--timer-finished-sound", default=str(_SOUNDS_DIR / "timer_finished.flac")
)
parser.add_argument("--preferences-file", default=_REPO_DIR / "preferences.json")
parser.add_argument(
"--host",
default="0.0.0.0",
help="Address for ESPHome server (default: 0.0.0.0)",
)
parser.add_argument(
"--port", type=int, default=6053, help="Port for ESPHome server (default: 6053)"
)
parser.add_argument(
"--led-type",
choices=["dotstar", "neopixel"],
default="dotstar",
help="Type of LED strip (default: dotstar for APA102)",
)
parser.add_argument(
"--led-interface",
choices=["spi", "gpio"],
default="spi",
help="Interface for LEDs (default: spi)",
)
parser.add_argument(
"--led-clock-pin",
type=int,
default=13,
help="GPIO pin for LED clock (for Grove)",
)
parser.add_argument(
"--led-data-pin",
type=int,
default=12,
help="GPIO pin for LED data (for Grove)",
)
parser.add_argument(
"--num-leds",
type=int,
default=3,
help="Number of LEDs in the strip",
)
parser.add_argument("--mqtt-host", help="MQTT broker host")
parser.add_argument("--mqtt-port", type=int, default=1883, help="MQTT broker port")
parser.add_argument("--mqtt-username", help="MQTT broker username")
parser.add_argument("--mqtt-password", help="MQTT broker password")
parser.add_argument(
"--debug", action="store_true", help="Print DEBUG messages to console"
) )
args = parser.parse_args() args = parser.parse_args()
# --- Load Configuration ---
config = load_config_from_json(args.config)
# --- Setup Logging ---
logging.basicConfig(level=logging.DEBUG if config.app.debug else logging.INFO)
_LOGGER.debug("Configuration loaded: %s", config)
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
_LOGGER.debug(args)
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
event_bus = EventBus() event_bus = EventBus()
preferences_path = Path(args.preferences_file) # --- Load Preferences ---
preferences_path = _REPO_DIR / config.app.preferences_file
if preferences_path.exists(): if preferences_path.exists():
with open(preferences_path, "r", encoding="utf-8") as preferences_file: with open(preferences_path, "r", encoding="utf-8") as preferences_file:
preferences_dict = json.load(preferences_file) preferences_dict = json.load(preferences_file)
@@ -170,13 +95,21 @@ async def main() -> None:
else: else:
preferences = Preferences() preferences = Preferences()
preferences.num_leds = getattr(preferences, 'num_leds', args.num_leds) # Set default num_leds from config, allowing preferences to override
preferences.num_leds = getattr(preferences, 'num_leds', config.led.num_leds)
# --- Load Wake Words ---
available_wake_words: Dict[str, AvailableWakeWord] = {} available_wake_words: Dict[str, AvailableWakeWord] = {}
for wake_word_dir in [Path(d) for d in args.wake_word_dir]:
# Add default directories if none are specified
if not config.wake_word.directories:
config.wake_word.directories = ["wakewords", "wakewords/openWakeWord"]
for ww_dir_str in config.wake_word.directories:
wake_word_dir = _REPO_DIR / ww_dir_str
for model_config_path in wake_word_dir.glob("*.json"): for model_config_path in wake_word_dir.glob("*.json"):
model_id = model_config_path.stem model_id = model_config_path.stem
if model_id == args.stop_model: if model_id == config.wake_word.stop_model:
continue continue
with open(model_config_path, "r", encoding="utf-8") as model_config_file: with open(model_config_path, "r", encoding="utf-8") as model_config_file:
model_config = json.load(model_config_file) model_config = json.load(model_config_file)
@@ -193,39 +126,46 @@ async def main() -> None:
libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so" libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so"
# --- Load Active Wake Word Models ---
wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {} wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {}
if preferences.active_wake_words: if preferences.active_wake_words:
for wake_word_id in preferences.active_wake_words: for wake_word_id in preferences.active_wake_words:
wake_word = available_wake_words.get(wake_word_id) wake_word = available_wake_words.get(wake_word_id)
if wake_word is None: continue if wake_word is None: continue
wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path)
# Load default model if no preferences are set
if not wake_models: if not wake_models:
wake_word_id = args.wake_model wake_word_id = config.wake_word.model
wake_word = available_wake_words.get(wake_word_id) wake_word = available_wake_words.get(wake_word_id)
if wake_word: if wake_word:
wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path)
# --- Load Stop Model ---
stop_model: Optional[MicroWakeWord] = None stop_model: Optional[MicroWakeWord] = None
for wake_word_dir in [Path(d) for d in args.wake_word_dir]: for ww_dir_str in config.wake_word.directories:
stop_config_path = wake_word_dir / f"{args.stop_model}.json" wake_word_dir = _REPO_DIR / ww_dir_str
stop_config_path = wake_word_dir / f"{config.wake_word.stop_model}.json"
if not stop_config_path.exists(): continue if not stop_config_path.exists(): continue
stop_model = MicroWakeWord.from_config(stop_config_path, libtensorflowlite_c_path) stop_model = MicroWakeWord.from_config(stop_config_path, libtensorflowlite_c_path)
break break
assert stop_model is not None assert stop_model is not None
# --- Initialize Media Players ---
music_player = MpvMediaPlayer( music_player = MpvMediaPlayer(
loop=loop, # <-- MODIFIED loop=loop,
device=args.audio_output_device, device=config.audio.output_device,
initial_volume=preferences.volume_level initial_volume=preferences.volume_level
) )
tts_player = MpvMediaPlayer( tts_player = MpvMediaPlayer(
loop=loop, # <-- MODIFIED loop=loop,
device=args.audio_output_device, device=config.audio.output_device,
initial_volume=preferences.volume_level initial_volume=preferences.volume_level
) )
# --- Create Global Server State ---
state = ServerState( state = ServerState(
name=args.name, name=config.app.name,
mac_address=get_mac(), mac_address=get_mac(),
audio_queue=Queue(), audio_queue=Queue(),
entities=[], entities=[],
@@ -234,50 +174,78 @@ async def main() -> None:
stop_word=stop_model, stop_word=stop_model,
music_player=music_player, music_player=music_player,
tts_player=tts_player, tts_player=tts_player,
wakeup_sound=args.wakeup_sound, wakeup_sound=str(_REPO_DIR / config.app.wakeup_sound),
timer_finished_sound=args.timer_finished_sound, timer_finished_sound=str(_REPO_DIR / config.app.timer_finished_sound),
preferences=preferences, preferences=preferences,
preferences_path=preferences_path, preferences_path=preferences_path,
libtensorflowlite_c_path=libtensorflowlite_c_path, libtensorflowlite_c_path=libtensorflowlite_c_path,
event_bus=event_bus, event_bus=event_bus,
loop=loop, loop=loop,
oww_melspectrogram_path=Path(args.oww_melspectrogram_model), oww_melspectrogram_path=(_REPO_DIR / config.wake_word.oww_melspectrogram_model),
oww_embedding_path=Path(args.oww_embedding_model), oww_embedding_path=(_REPO_DIR / config.wake_word.oww_embedding_model),
refractory_seconds=args.refractory_seconds, refractory_seconds=config.wake_word.refractory_seconds,
) )
# --- Initialize Controllers ---
led_controller = LedController( led_controller = LedController(
state, state,
led_type=args.led_type, led_type=config.led.led_type,
interface=args.led_interface, interface=config.led.interface,
clock_pin=args.led_clock_pin, clock_pin=config.led.clock_pin,
data_pin=args.led_data_pin, data_pin=config.led.data_pin,
num_leds=state.preferences.num_leds, num_leds=state.preferences.num_leds,
) )
mqtt_controller: Optional[MqttController] = None mqtt_controller: Optional[MqttController] = None
if args.mqtt_host: if config.mqtt.enabled:
mqtt_controller = MqttController(state=state, host=args.mqtt_host, port=args.mqtt_port, username=args.mqtt_username, password=args.mqtt_password) mqtt_controller = MqttController(
state=state,
host=config.mqtt.host,
port=config.mqtt.port,
username=config.mqtt.username,
password=config.mqtt.password
)
mqtt_controller.start() mqtt_controller.start()
mic_mute_handler = MicMuteHandler(state, mqtt_controller) mic_mute_handler = MicMuteHandler(state, mqtt_controller)
# --- Start Audio Processing Thread ---
process_audio_thread = threading.Thread(target=process_audio, args=(state,), daemon=True) process_audio_thread = threading.Thread(target=process_audio, args=(state,), daemon=True)
process_audio_thread.start() process_audio_thread.start()
def sd_callback(indata, _frames, _time, _status): state.audio_queue.put_nowait(bytes(indata)) def sd_callback(indata, _frames, _time, _status):
state.audio_queue.put_nowait(bytes(indata))
server = await loop.create_server(lambda: VoiceSatelliteProtocol(state), host=args.host, port=args.port) # --- Start ESPHome Server ---
discovery = HomeAssistantZeroconf(port=args.port, name=args.name) server = await loop.create_server(
lambda: VoiceSatelliteProtocol(state),
host=config.esphome.host,
port=config.esphome.port
)
discovery = HomeAssistantZeroconf(port=config.esphome.port, name=config.app.name)
await discovery.register_server() await discovery.register_server()
# --- Run Forever ---
try: try:
with sd.RawInputStream(samplerate=16000, blocksize=args.audio_input_block_size, device=args.audio_input_device, dtype="int16", channels=1, callback=sd_callback): input_device = config.audio.input_device if config.audio.input_device else "default"
with sd.RawInputStream(
samplerate=16000,
blocksize=config.audio.input_block_size,
device=input_device,
dtype="int16",
channels=1,
callback=sd_callback
):
async with server: async with server:
_LOGGER.info("Server started (host=%s, port=%s)", args.host, args.port) _LOGGER.info("Server started (host=%s, port=%s)", config.esphome.host, config.esphome.port)
await server.serve_forever() await server.serve_forever()
except KeyboardInterrupt: pass except KeyboardInterrupt:
pass
except sd.PortAudioError as e:
_LOGGER.critical("Failed to open audio input device '%s': %s", input_device, e)
finally: finally:
if mqtt_controller: mqtt_controller.stop() if mqtt_controller:
mqtt_controller.stop()
state.audio_queue.put_nowait(None) state.audio_queue.put_nowait(None)
process_audio_thread.join() process_audio_thread.join()
+5
View File
@@ -0,0 +1,5 @@
{
"app": {
"name": "Linux Voice Assistant"
}
}
+61
View File
@@ -0,0 +1,61 @@
{
"app": {
"name": "Linux Voice Assistant",
"debug": true,
"wakeup_sound": "sounds/wake_word_triggered.flac",
"timer_finished_sound": "sounds/timer_finished.flac",
"preferences_file": "preferences.json"
},
"audio": {
// This is the "manual bypass" for audio.
// Set to "null" to use auto-detection (recommended).
// Set to specific device names to force them.
"input_device": "seeed-2mic-voicecard",
"output_device": "alsa/plughw:CARD=seeed2micvoicec",
"input_block_size": 1024
},
"wake_word": {
// A list of directories to scan for wake word models
"directories": [
"wakewords",
"wakewords/openWakeWord"
],
// The default wake word to use if none is set in preferences
"model": "okay_nabu",
"stop_model": "stop",
"refractory_seconds": 2.0,
// Paths are relative to the project root
"oww_melspectrogram_model": "wakewords/openWakeWord/melspectrogram.tflite",
"oww_embedding_model": "wakewords/openWakeWord/embedding_model.tflite"
},
"esphome": {
// Network settings for the satellite server
"host": "0.0.0.0",
"port": 6053
},
"led": {
// "led_type" can be "dotstar" (APA102) or "neopixel" (WS2812B)
"led_type": "dotstar",
// "interface" can be "spi" (hardware SPI) or "gpio" (bit-banging)
"interface": "spi",
// --- GPIO Settings ---
// These are ONLY used if interface is "gpio"
// For Grove on Pi: clock=13, data=12
"clock_pin": 13,
"data_pin": 12,
// Default number of LEDs. This is overridden by
// "num_leds" in preferences.json after first run.
"num_leds": 3
},
"mqtt": {
// "host" is the only required field to enable MQTT.
// If "host" is null or this section is omitted, MQTT is disabled.
"host": "192.168.1.2",
"port": 1883,
"username": "mqtt_username",
"password": "mqtt_password"
}
}
+122
View File
@@ -0,0 +1,122 @@
"""Configuration models for the application."""
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
# -----------------------------------------------------------------------------
# Configuration Dataclasses
# -----------------------------------------------------------------------------
@dataclass
class AudioConfig:
"""Settings for audio input and output."""
input_device: Optional[str] = None
input_block_size: int = 1024
output_device: Optional[str] = None
@dataclass
class WakeWordConfig:
"""Settings for wake word detection."""
directories: List[str] = field(default_factory=list)
model: str = "okay_nabu"
stop_model: str = "stop"
refractory_seconds: float = 2.0
oww_melspectrogram_model: str = "wakewords/openWakeWord/melspectrogram.tflite"
oww_embedding_model: str = "wakewords/openWakeWord/embedding_model.tflite"
@dataclass
class ESPHomeConfig:
"""Settings for the ESPHome API server."""
host: str = "0.0.0.0"
port: int = 6053
@dataclass
class LedConfig:
"""Settings for the LED strip."""
led_type: str = "dotstar"
interface: str = "spi"
clock_pin: int = 13
data_pin: int = 12
num_leds: int = 3 # Note: This will be overridden by 'preferences.json' if it exists
@dataclass
class MqttConfig:
"""Settings for the MQTT client."""
enabled: bool = False
host: Optional[str] = None
port: int = 1883
username: Optional[str] = None
password: Optional[str] = None
@dataclass
class AppConfig:
"""General application settings."""
name: str
wakeup_sound: str = "sounds/wake_word_triggered.flac"
timer_finished_sound: str = "sounds/timer_finished.flac"
preferences_file: str = "preferences.json"
debug: bool = False
@dataclass
class Config:
"""Main configuration object."""
app: AppConfig
audio: AudioConfig = field(default_factory=AudioConfig)
wake_word: WakeWordConfig = field(default_factory=WakeWordConfig)
esphome: ESPHomeConfig = field(default_factory=ESPHomeConfig)
led: LedConfig = field(default_factory=LedConfig)
mqtt: MqttConfig = field(default_factory=MqttConfig)
# -----------------------------------------------------------------------------
# Helper Function
# -----------------------------------------------------------------------------
def load_config_from_json(config_path: Path) -> Config:
"""Loads configuration from a JSON file and populates dataclasses."""
# --- Step 1: Load raw JSON data ---
try:
with open(config_path, "r", encoding="utf-8") as f:
raw_data = json.load(f)
except FileNotFoundError:
_LOGGER.critical(f"Configuration file not found at: {config_path}")
raise
except json.JSONDecodeError as e:
_LOGGER.critical(f"Error parsing configuration file: {e}")
raise
# --- Step 2: Create config objects from raw data ---
# This nested loading allows users to omit entire sections (like 'audio')
# and still get the defaults.
# 'app' is the only required section
if "app" not in raw_data:
raise ValueError("Configuration file must contain an 'app' section with a 'name'.")
app_config = AppConfig(**raw_data.get("app", {}))
audio_config = AudioConfig(**raw_data.get("audio", {}))
wake_word_config = WakeWordConfig(**raw_data.get("wake_word", {}))
esphome_config = ESPHomeConfig(**raw_data.get("esphome", {}))
led_config = LedConfig(**raw_data.get("led", {}))
mqtt_config = MqttConfig(**raw_data.get("mqtt", {}))
# --- Step 3: Set MQTT 'enabled' flag ---
# If host is provided, automatically enable MQTT
if mqtt_config.host:
mqtt_config.enabled = True
# --- Step 4: Return the main Config object ---
return Config(
app=app_config,
audio=audio_config,
wake_word=wake_word_config,
esphome=esphome_config,
led=led_config,
mqtt=mqtt_config,
)
# Logger is needed for the load function
import logging
_LOGGER = logging.getLogger(__name__)
+1
View File
@@ -23,6 +23,7 @@ _LOGGER = logging.getLogger(__name__)
class SatelliteState(str, Enum): class SatelliteState(str, Enum):
"""Voice satellite state.""" """Voice satellite state."""
STARTING = "starting" # <-- ADDED
IDLE = "idle" IDLE = "idle"
LISTENING = "listening" # Wake word detected LISTENING = "listening" # Wake word detected
THINKING = "thinking" # VAD, waiting for STT/intent THINKING = "thinking" # VAD, waiting for STT/intent
+2 -2
View File
@@ -60,7 +60,7 @@ class VoiceSatelliteProtocol(APIServer):
self.state.entities.append(self.media_player_entity) self.state.entities.append(self.media_player_entity)
# State machine # State machine
self._state: SatelliteState = SatelliteState.IDLE self._state: SatelliteState = SatelliteState.STARTING # <-- MODIFIED
self._is_streaming_audio: bool = False # Is HA requesting audio? self._is_streaming_audio: bool = False # Is HA requesting audio?
# Announce/TTS state # Announce/TTS state
@@ -255,7 +255,7 @@ class VoiceSatelliteProtocol(APIServer):
self.send_messages([VoiceAssistantAudio(data=audio_chunk)]) self.send_messages([VoiceAssistantAudio(data=audio_chunk)])
def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None: def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
if self._state != SatelliteState.IDLE: if self._state not in (SatelliteState.IDLE, SatelliteState.STARTING):
return # Already awake return # Already awake
if self._timer_finished: if self._timer_finished:
+15
View File
@@ -0,0 +1,15 @@
[Unit]
Description=Linux Voice Assistant
[Service]
Type=simple
WorkingDirectory=/home/pi/linux-voice-assistant
Environment=PYTHONUNBUFFERED=1
ExecStart=/home/pi/linux-voice-assistant/script/run
Restart=always
RestartSec=2
[Install]
WantedBy=default.target
@@ -1,22 +0,0 @@
[Unit]
Description=Linux Voice Assistant
[Service]
Type=simple
WorkingDirectory=/home/pi/linux-voice-assistant
Environment=PYTHONUNBUFFERED=1
ExecStart=/home/pi/linux-voice-assistant/script/run \
--name 'Linux Voice Assistant' \
--wake-word-dir wakewords/openWakeWord \
--wake-word-dir wakewords \
--stop-model stop \
--audio-input-device seeed-2mic-voicecard \
--audio-output-device alsa/plughw:CARD=seeed2micvoicec \
--debug
Restart=always
RestartSec=2
[Install]
WantedBy=default.target
@@ -1,24 +0,0 @@
[Unit]
Description=Linux Voice Assistant (PulseAudio)
After=pulseaudio.service
Wants=pulseaudio.service
[Service]
Type=simple
WorkingDirectory=/home/pi/linux-voice-assistant
Environment=PYTHONUNBUFFERED=1
ExecStart=/home/pi/linux-voice-assistant/script/run \
--name 'Linux Voice Assistant' \
--wake-word-dir wakewords/openWakeWord \
--wake-word-dir wakewords \
--stop-model stop \
--audio-input-device default \
--audio-output-device pulse/alsa_output.platform-soc_sound.stereo-fallback \
--debug
Restart=always
RestartSec=2
[Install]
WantedBy=default.target
@@ -1,24 +0,0 @@
[Unit]
Description=Linux Voice Assistant (PipeWire)
After=pipewire.service pipewire-pulse.service wireplumber.service
Wants=pipewire-pulse.service wireplumber.service
[Service]
Type=simple
WorkingDirectory=/home/pi/linux-voice-assistant
Environment=PYTHONUNBUFFERED=1
ExecStart=/home/pi/linux-voice-assistant/script/run \
--name 'Linux Voice Assistant' \
--wake-word-dir wakewords/openWakeWord \
--wake-word-dir wakewords \
--stop-model stop \
--audio-input-device default \
--audio-output-device pipewire/default \
--debug
Restart=always
RestartSec=2
[Install]
WantedBy=default.target