sirmrak/linux-voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update config.py

Browse Source 
What this fixes:

On boot, if PipeWire/Pulse starts with the sink at ~40%, LVA now pushes the sink volume to match the persisted LVA volume (0.0–1.0).

When you move the HA media_player volume slider, LVA now updates:

wpctl (PipeWire) if available

pactl (PulseAudio / PipeWire-Pulse) if available

amixer (ALSA) as a fallback

No reboot / alsactl store required — LVA simply re-applies the persisted volume at start.
This commit is contained in:
imonlinux
2026-01-05 21:51:21 -08:00
committed by GitHub
parent a1b460391d
commit eb4c4c26c8
1 changed files with 132 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/config.py
+138 -148
View File
@@ -1,161 +1,151 @@
"""Configuration models for the application.""" from abc import abstractmethod
from collections.abc import Iterable
from typing import TYPE_CHECKING, Callable, List, Optional, Union
import json # pylint: disable=no-name-in-module
from dataclasses import dataclass, field from aioesphomeapi.api_pb2 import ( # type: ignore[attr-defined]
from pathlib import Path ListEntitiesMediaPlayerResponse,
from typing import Dict, List, Optional ListEntitiesRequest,
MediaPlayerCommandRequest,
MediaPlayerStateResponse,
SubscribeHomeAssistantStatesRequest,
)
from aioesphomeapi.model import MediaPlayerCommand, MediaPlayerState
from google.protobuf import message
from .api_server import APIServer
from .mpv_player import MpvMediaPlayer
from .util import call_all
if TYPE_CHECKING:
from .models import ServerState
class ESPHomeEntity:
def __init__(self, server: APIServer, state: "ServerState") -> None:
self.server = server
self.state = state
@abstractmethod
def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
pass
import logging
_LOGGER = logging.getLogger(__name__)
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# 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 MediaPlayerEntity(ESPHomeEntity):
class WakeWordConfig: def __init__(
"""Settings for wake word detection.""" self,
directories: List[str] = field(default_factory=list) server: APIServer,
model: str = "okay_nabu" state: "ServerState",
stop_model: str = "stop" key: int,
refractory_seconds: float = 2.0 name: str,
download_dir: str = "local" object_id: str,
music_player: MpvMediaPlayer,
announce_player: MpvMediaPlayer,
) -> None:
super().__init__(server, state)
self.key = key
self.name = name
self.object_id = object_id
self.state_enum = MediaPlayerState.IDLE
self.volume = state.preferences.volume_level # Initialize with saved volume
self.muted = False
self.music_player = music_player
self.announce_player = announce_player
@dataclass def play(
class ESPHomeConfig: self,
"""Settings for the ESPHome API server.""" url: Union[str, List[str]],
host: str = "0.0.0.0" announcement: bool = False,
port: int = 6053 done_callback: Optional[Callable[[], None]] = None,
) -> Iterable[message.Message]:
if announcement:
@dataclass if self.music_player.is_playing:
class LedConfig: # Announce, resume music
"""Settings for the LED strip. self.music_player.pause()
self.announce_player.play(
Fields: url,
enabled -> master on/off for hardware LEDs. If False, LedController done_callback=lambda: call_all(
still subscribes to events and publishes MQTT state, but self.music_player.resume, done_callback
never touches hardware. ),
led_type -> "dotstar", "neopixel", or "xvf3800" )
interface -> "spi", "gpio", "usb", or "i2c" (future) else:
""" # Announce, idle
enabled: bool = True self.announce_player.play(
led_type: str = "dotstar" url,
interface: str = "spi" done_callback=lambda: call_all(
clock_pin: int = 13 self.server.send_messages(
data_pin: int = 12 [self._update_state(MediaPlayerState.IDLE)]
# Note: overridden by 'preferences.json' if it exists ),
num_leds: int = 3 done_callback,
),
)
@dataclass else:
class MqttConfig: # Music
"""Settings for the MQTT client.""" self.music_player.play(
enabled: bool = False url,
host: Optional[str] = None done_callback=lambda: call_all(
port: int = 1883 self.server.send_messages(
username: Optional[str] = None [self._update_state(MediaPlayerState.IDLE)]
password: Optional[str] = None ),
done_callback,
),
@dataclass
class ButtonConfig:
"""
Settings for a hardware momentary button.
mode:
- "gpio" -> legacy GPIO button (e.g. ReSpeaker 2-Mic HAT)
- "xvf3800"-> USB-based mute integration for the ReSpeaker XVF3800
For mode="gpio":
- 'pin' and 'long_press_seconds' are used (short vs long press).
For mode="xvf3800":
- The XVF3800ButtonController uses the built-in mute button purely
as a mute toggle; 'pin' and 'long_press_seconds' are ignored.
"""
enabled: bool = False
mode: str = "gpio" # "gpio" or "xvf3800"
# BCM GPIO pin number for the button input (gpio mode only).
pin: int = 17
# Press duration (in seconds) to be considered a "long press" (gpio mode).
long_press_seconds: float = 1.0
# Poll interval used by both GPIO and XVF3800 controllers.
poll_interval_seconds: float = 0.01
@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)
button: ButtonConfig = field(default_factory=ButtonConfig)
# -----------------------------------------------------------------------------
# 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 ---
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", {})) yield self._update_state(MediaPlayerState.PLAYING)
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", {}))
button_config = ButtonConfig(**raw_data.get("button", {}))
# --- Step 3: Set MQTT 'enabled' flag --- def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
if mqtt_config.host: if isinstance(msg, MediaPlayerCommandRequest) and (msg.key == self.key):
mqtt_config.enabled = True if msg.has_media_url:
announcement = msg.has_announcement and msg.announcement
yield from self.play(msg.media_url, announcement=announcement)
elif msg.has_command:
if msg.command == MediaPlayerCommand.PAUSE:
self.music_player.pause()
yield self._update_state(MediaPlayerState.PAUSED)
elif msg.command == MediaPlayerCommand.PLAY:
self.music_player.resume()
yield self._update_state(MediaPlayerState.PLAYING)
# --- Step 4: Return the main Config object --- if msg.has_volume:
return Config( # HA sends volume as 0.0-1.0. We treat this as the *master*
app=app_config, # output volume and sync it to the OS audio backend.
audio=audio_config, self.volume = msg.volume
wake_word=wake_word_config,
esphome=esphome_config, # Keep mpv at 100% for normal playback; ducking uses mpv volume.
led=led_config, self.music_player.set_volume(100)
mqtt=mqtt_config, self.announce_player.set_volume(100)
button=button_config,
# Apply master volume to the system sink (PipeWire/Pulse/ALSA).
self.music_player.set_master_volume(self.volume)
# Save the new volume level to preferences
self.state.preferences.volume_level = self.volume
self.state.save_preferences()
yield self._update_state(self.state_enum)
elif isinstance(msg, ListEntitiesRequest):
yield ListEntitiesMediaPlayerResponse(
object_id=self.object_id,
key=self.key,
name=self.name,
supports_pause=True,
)
elif isinstance(msg, SubscribeHomeAssistantStatesRequest):
yield self._get_state_message()
def _update_state(self, new_state: MediaPlayerState) -> MediaPlayerStateResponse:
self.state_enum = new_state
return self._get_state_message()
def _get_state_message(self) -> MediaPlayerStateResponse:
return MediaPlayerStateResponse(
key=self.key,
state=self.state_enum,
volume=self.volume,
muted=self.muted,
) )