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

Merge branch 'main' into feature/add-docker

Browse Source
This commit is contained in:
Florian Asche authored and GitHub committed 2026-02-08 17:23:39 +01:00
2 parents a9335bf6c7 31eeac8f68
commit 761bcea5ac
4 files changed
+71 -43
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

linux_voice_assistant/__main__.py
+21 -9
View File
@@ -33,7 +33,7 @@ _SOUNDS_DIR = _REPO_DIR / "sounds"
async def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--name", required=True)
parser.add_argument("--name")
parser.add_argument(
"--audio-input-device",
help="soundcard name for input device (see --list-input-devices)",
@@ -82,16 +82,19 @@ async def main() -> None:
"--timer-finished-sound", default=str(_SOUNDS_DIR / "timer_finished.flac")
)
parser.add_argument(
"--processing-sound", default=str(_SOUNDS_DIR / "processing.wav"),
help="Short sound to play while assistant is processing (thinking)"
"--processing-sound",
default=str(_SOUNDS_DIR / "processing.wav"),
help="Short sound to play while assistant is processing (thinking)",
)
parser.add_argument(
"--mute-sound", default=str(_SOUNDS_DIR / "mute_switch_on.flac"),
help="Sound to play when muting the assistant"
"--mute-sound",
default=str(_SOUNDS_DIR / "mute_switch_on.flac"),
help="Sound to play when muting the assistant",
)
parser.add_argument(
"--unmute-sound", default=str(_SOUNDS_DIR / "mute_switch_off.flac"),
help="Sound to play when unmuting the assistant"
"--unmute-sound",
default=str(_SOUNDS_DIR / "mute_switch_off.flac"),
help="Sound to play when unmuting the assistant",
)
#
parser.add_argument("--preferences-file", default=_REPO_DIR / "preferences.json")
@@ -106,7 +109,9 @@ async def main() -> None:
"--port", type=int, default=6053, help="Port for ESPHome server (default: 6053)"
)
parser.add_argument(
"--enable-thinking-sound", action="store_true", help="Enable thinking sound on startup"
"--enable-thinking-sound",
action="store_true",
help="Enable thinking sound on startup",
)
parser.add_argument(
"--debug", action="store_true", help="Print DEBUG messages to console"
@@ -134,6 +139,9 @@ async def main() -> None:
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
_LOGGER.debug(args)
if not args.name:
parser.error("the following arguments are required: --name")
args.download_dir = Path(args.download_dir)
args.download_dir.mkdir(parents=True, exist_ok=True)
@@ -371,7 +379,11 @@ def process_audio(state: ServerState, mic, block_size: int):
if state.stop_word.process_streaming(micro_input):
stopped = True
if stopped and (state.stop_word.id in state.active_wake_words) and not state.muted:
if (
stopped
and (state.stop_word.id in state.active_wake_words)
and not state.muted
):
state.satellite.stop()
except Exception:
_LOGGER.exception("Unexpected error handling audio")
linux_voice_assistant/entity.py
+16 -6
View File
@@ -7,17 +7,17 @@ from aioesphomeapi.api_pb2 import ( # type: ignore[attr-defined]
ListEntitiesMediaPlayerResponse,
ListEntitiesRequest,
ListEntitiesSwitchResponse,
SwitchCommandRequest,
SwitchStateResponse,
MediaPlayerCommandRequest,
MediaPlayerStateResponse,
SubscribeHomeAssistantStatesRequest,
SwitchCommandRequest,
SwitchStateResponse,
)
from aioesphomeapi.model import (
EntityCategory,
MediaPlayerCommand,
MediaPlayerEntityFeature,
MediaPlayerState,
EntityCategory,
)
from google.protobuf import message
@@ -35,6 +35,7 @@ SUPPORTED_MEDIA_PLAYER_FEATURES = (
| MediaPlayerEntityFeature.MEDIA_ANNOUNCE
)
class ESPHomeEntity:
def __init__(self, server: APIServer) -> None:
self.server = server
@@ -167,8 +168,10 @@ class MediaPlayerEntity(ESPHomeEntity):
muted=self.muted,
)
# -----------------------------------------------------------------------------
class MuteSwitchEntity(ESPHomeEntity):
def __init__(
self,
@@ -186,7 +189,9 @@ class MuteSwitchEntity(ESPHomeEntity):
self.object_id = object_id
self._get_muted = get_muted
self._set_muted = set_muted
self._switch_state = self._get_muted() # Sync internal state with actual muted value on init
self._switch_state = (
self._get_muted()
) # Sync internal state with actual muted value on init
def update_set_muted(self, set_muted: Callable[[bool], None]) -> None:
# Update the callback used to change the mute state.
@@ -221,6 +226,7 @@ class MuteSwitchEntity(ESPHomeEntity):
self.sync_with_state()
yield SwitchStateResponse(key=self.key, state=self._switch_state)
class ThinkingSoundEntity(ESPHomeEntity):
def __init__(
self,
@@ -240,11 +246,15 @@ class ThinkingSoundEntity(ESPHomeEntity):
self._set_thinking_sound_enabled = set_thinking_sound_enabled
self._switch_state = self._get_thinking_sound_enabled() # Sync internal state
def update_get_thinking_sound_enabled(self, get_thinking_sound_enabled: Callable[[], bool]) -> None:
def update_get_thinking_sound_enabled(
self, get_thinking_sound_enabled: Callable[[], bool]
) -> None:
# Update the callback used to read the thinking sound enabled state.
self._get_thinking_sound_enabled = get_thinking_sound_enabled
def update_set_thinking_sound_enabled(self, set_thinking_sound_enabled: Callable[[bool], None]) -> None:
def update_set_thinking_sound_enabled(
self, set_thinking_sound_enabled: Callable[[bool], None]
) -> None:
# Update the callback used to change the thinking sound enabled state.
self._set_thinking_sound_enabled = set_thinking_sound_enabled
linux_voice_assistant/models.py
+7 -2
View File
@@ -12,8 +12,12 @@ if TYPE_CHECKING:
from pymicro_wakeword import MicroWakeWord
from pyopen_wakeword import OpenWakeWord
from .entity import ESPHomeEntity, MediaPlayerEntity
from .entity import ESPHomeEntity, MediaPlayerEntity, MuteSwitchEntity, ThinkingSoundEntity
from .entity import (
ESPHomeEntity,
MediaPlayerEntity,
MuteSwitchEntity,
ThinkingSoundEntity,
)
from .mpv_player import MpvMediaPlayer
from .satellite import VoiceSatelliteProtocol
@@ -55,6 +59,7 @@ class Preferences:
active_wake_words: List[str] = field(default_factory=list)
thinking_sound: int = 0 # 0 = disabled, 1 = enabled
@dataclass
class ServerState:
name: str
linux_voice_assistant/satellite.py
+27 -26
View File
@@ -1,10 +1,10 @@
"""Voice satellite protocol."""
import asyncio
import re
import hashlib
import logging
import posixpath
import re
import shutil
import time
from collections.abc import Iterable
@@ -14,6 +14,7 @@ from urllib.request import urlopen
# pylint: disable=no-name-in-module
from aioesphomeapi.api_pb2 import ( # type: ignore[attr-defined]
AuthenticationRequest,
DeviceInfoRequest,
DeviceInfoResponse,
ListEntitiesDoneResponse,
@@ -32,7 +33,6 @@ from aioesphomeapi.api_pb2 import ( # type: ignore[attr-defined]
VoiceAssistantSetConfiguration,
VoiceAssistantTimerEventResponse,
VoiceAssistantWakeWord,
AuthenticationRequest,
)
from aioesphomeapi.core import MESSAGE_TYPE_TO_PROTO
from aioesphomeapi.model import (
@@ -53,6 +53,7 @@ _LOGGER = logging.getLogger(__name__)
PROTO_TO_MESSAGE_TYPE = {v: k for k, v in MESSAGE_TYPE_TO_PROTO.items()}
class VoiceSatelliteProtocol(APIServer):
def __init__(self, state: ServerState) -> None:
@@ -146,14 +147,22 @@ class VoiceSatelliteProtocol(APIServer):
self.state.entities.append(thinking_sound_switch)
# Load thinking sound enabled state from preferences (default to False if not set or unknown)
if hasattr(self.state.preferences, 'thinking_sound') and self.state.preferences.thinking_sound in (0, 1):
self.state.thinking_sound_enabled = bool(self.state.preferences.thinking_sound)
if hasattr(
self.state.preferences, "thinking_sound"
) and self.state.preferences.thinking_sound in (0, 1):
self.state.thinking_sound_enabled = bool(
self.state.preferences.thinking_sound
)
else:
self.state.thinking_sound_enabled = False
thinking_sound_switch.server = self
thinking_sound_switch.update_get_thinking_sound_enabled(lambda: self.state.thinking_sound_enabled)
thinking_sound_switch.update_set_thinking_sound_enabled(self._set_thinking_sound_enabled)
thinking_sound_switch.update_get_thinking_sound_enabled(
lambda: self.state.thinking_sound_enabled
)
thinking_sound_switch.update_set_thinking_sound_enabled(
self._set_thinking_sound_enabled
)
thinking_sound_switch.sync_with_state()
self._is_streaming_audio = False
@@ -168,7 +177,9 @@ class VoiceSatelliteProtocol(APIServer):
def _set_thinking_sound_enabled(self, new_state: bool) -> None:
self.state.thinking_sound_enabled = bool(new_state)
self.state.preferences.thinking_sound = 1 if self.state.thinking_sound_enabled else 0
self.state.preferences.thinking_sound = (
1 if self.state.thinking_sound_enabled else 0
)
if self.state.thinking_sound_enabled:
_LOGGER.debug("Thinking sound enabled")
@@ -204,7 +215,10 @@ class VoiceSatelliteProtocol(APIServer):
self._tts_url = data.get("url")
self._tts_played = False
self._continue_conversation = False
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_START and self.state.thinking_sound_enabled:
elif (
event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_START
and self.state.thinking_sound_enabled
):
# Play short "thinking/processing" sound if configured
processing = getattr(self.state, "processing_sound", None)
if processing:
@@ -273,14 +287,12 @@ class VoiceSatelliteProtocol(APIServer):
self._continue_conversation = msg.start_conversation
self.duck()
yield from self.state.media_player_entity.play(
urls, announcement=True, done_callback=self._tts_finished
)
self.state.tts_player.play(urls, done_callback=self._tts_finished)
elif isinstance(msg, VoiceAssistantTimerEventResponse):
self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg)
elif isinstance(msg, DeviceInfoRequest):
# Compute dynamic device name
base_name = re.sub(r'[\s-]+', '-', self.state.name.lower()).strip('-')
base_name = re.sub(r"[\s-]+", "-", self.state.name.lower()).strip("-")
mac_no_colon = self.state.mac_address.replace(":", "").lower()
mac_last6 = mac_no_colon[-6:]
device_name = f"{base_name}-{mac_last6}"
@@ -505,10 +517,11 @@ class VoiceSatelliteProtocol(APIServer):
# Send states after connect
states = []
for entity in self.state.entities:
states.extend(entity.handle_message(SubscribeHomeAssistantStatesRequest()))
states.extend(
entity.handle_message(SubscribeHomeAssistantStatesRequest())
)
self.send_messages(states)
_LOGGER.debug("Sent entity states after connect")
_LOGGER.info("Disconnected from Home Assistant")
def _download_external_wake_word(
self, external_wake_word: VoiceAssistantExternalWakeWord
@@ -576,15 +589,3 @@ class VoiceSatelliteProtocol(APIServer):
trained_languages=external_wake_word.trained_languages,
wake_word_path=config_path,
)
def process_packet(self, msg_type: int, packet_data: bytes) -> None:
super().process_packet(msg_type, packet_data)
if msg_type == PROTO_TO_MESSAGE_TYPE[AuthenticationRequest]:
self.state.connected = True
# Send states after connect
states = []
for entity in self.state.entities:
states.extend(entity.handle_message(SubscribeHomeAssistantStatesRequest()))
self.send_messages(states)
_LOGGER.debug("Sent entity states after connect")