From 8063f5b773e43a4f361ce80228f42a2841236c54 Mon Sep 17 00:00:00 2001 From: Michael Hansen Date: Sun, 14 Jan 2024 22:19:01 -0600 Subject: [PATCH] First working version --- installer/__main__.py | 229 ++------------------------- installer/const.py | 92 +++++++---- installer/dataclasses_json.py | 70 +++++++++ installer/drivers.py | 4 +- installer/microphone.py | 104 +++++++------ installer/satellite.py | 76 +++++---- installer/services.py | 231 ++++++++++++++++++++++++++++ installer/speakers.py | 39 +++-- installer/wake_word.py | 282 +++++++++++++++++++++++++--------- installer/whiptail.py | 18 ++- script/format | 11 +- script/lint | 23 +-- 12 files changed, 740 insertions(+), 439 deletions(-) create mode 100644 installer/dataclasses_json.py create mode 100644 installer/services.py diff --git a/installer/__main__.py b/installer/__main__.py index c04de60..87a5352 100644 --- a/installer/__main__.py +++ b/installer/__main__.py @@ -1,34 +1,15 @@ """Command-line installer.""" -import array -import math import logging -import json -import shlex -import subprocess -import shutil -import sys -import time -from collections.abc import Sequence -from concurrent.futures import ThreadPoolExecutor, Future -from enum import Enum -from dataclasses import dataclass, fields, asdict, field -from pathlib import Path -from typing import Any, Dict, List, Tuple, Optional, Union +from typing import List, Optional -from .const import ( - Settings, - PROGRAM_DIR, - LOCAL_DIR, - SatelliteType, - SERVICES_DIR, - WakeWordSystem, -) -from .whiptail import run_with_gauge, menu, ItemType, msgbox, error, passwordbox -from .satellite import configure_satellite -from .microphone import configure_microphone -from .speakers import configure_speakers +from .const import LOCAL_DIR, PROGRAM_DIR, SatelliteType, Settings from .drivers import install_drivers +from .microphone import configure_microphone +from .satellite import configure_satellite +from .services import generate_services, install_services, stop_services +from .speakers import configure_speakers from .wake_word import configure_wake_word +from .whiptail import ItemType, error, menu, msgbox, passwordbox, run_with_gauge _LOGGER = logging.getLogger() @@ -41,12 +22,6 @@ def main() -> None: settings = Settings.load() - venv_dir = PROGRAM_DIR / ".venv" - if not venv_dir.exists(): - run_with_gauge( - "Installing satellite base...", [[str(PROGRAM_DIR / "script" / "setup")]] - ) - choice: Optional[str] = None while True: choice = main_menu(choice) @@ -99,151 +74,17 @@ def pip_install(*args) -> List[str]: ] + list(args) -def stop_services(password: str) -> None: - stop_commands = [] - for service in ("satellite", "openwakeword", "porcupine1", "snowboy"): - service_filename = f"wyoming-{service}.service" - service_path = Path("/etc/systemd/system") / service_filename - if not service_path.exists(): - continue - - stop_commands.append(["sudo", "-S", "systemctl", "stop", service_filename]) - stop_commands.append(["sudo", "-S", "systemctl", "disable", service_filename]) - - run_with_gauge("Stopping Services...", stop_commands, sudo_password=password) - - -def generate_services(settings: Settings) -> None: - SERVICES_DIR.mkdir(parents=True, exist_ok=True) - - user = subprocess.check_output(["id", "--name", "-u"], text=True).strip() - satellite_command: List[str] = [ - str(PROGRAM_DIR / "script" / "run"), - "--name", - settings.satellite_name, - "--uri", - "tcp://0.0.0.0:10700", - "--mic-command", - f"arecord -D {settings.microphone_device} -q -r 16000 -c 1 -f S16_LE -t raw", - ] - satellite_requires: List[str] = [] - - if settings.sound_device is not None: - # Audio output - satellite_command.extend( - [ - "--snd-command", - f"aplay -D {settings.sound_device} -q -r 22050 -c 1 -f S16_LE -t raw", - ] - ) - - for sound_name in settings.feedback_sounds: - # Try local/sounds first - sound_path = LOCAL_DIR / "sounds" / f"{sound_name}.wav" - if not sound_path.exists(): - sound_path = PROGRAM_DIR / "sounds" / f"{sound_name}.wav" - - satellite_command.extend([f"--{sound_name}-wav", str(sound_path)]) - - if settings.satellite_type == SatelliteType.VAD: - # Voice activity detect - satellite_command.append("--vad") - elif settings.satellite_type == SatelliteType.WAKE: - # Local wake word detection - assert settings.wake_word_system is not None, "Wake word system not set" - wake_word = settings.wake_word.get(settings.wake_word_system) - assert wake_word, "No wake word set" - - wake_word_service = "wyoming-" + str(settings.wake_word_system).lower() - wake_word_dir = LOCAL_DIR / wake_word_service - wake_word_command = [ - str(wake_word_dir / "script" / "run"), - "--uri", - "tcp://127.0.0.1:10400", - ] - - if settings.wake_word_system in ( - WakeWordSystem.OPENWAKEWORD, - WakeWordSystem.SNOWBOY, - ): - wake_word_command.extend( - [ - "--custom-model-dir", - str( - LOCAL_DIR - / "custom-wake-words" - / WakeWordSystem(settings.wake_word_system).value - ), - ] - ) - - if settings.debug_enabled: - wake_word_command.append("--debug") - - wake_word_command_str = shlex.join(wake_word_command) - - with open( - SERVICES_DIR / f"{wake_word_service}.service", "w", encoding="utf-8" - ) as service_file: - print("[Unit]", file=service_file) - print(f"Description={settings.wake_word_system}", file=service_file) - print("", file=service_file) - print("[Service]", file=service_file) - print("Type=simple", file=service_file) - print(f"User={user}", file=service_file) - print(f"ExecStart={wake_word_command_str}", file=service_file) - print(f"WorkingDirectory={wake_word_dir}", file=service_file) - print("Restart=always", file=service_file) - print("RestartSec=1", file=service_file) - print("", file=service_file) - print("[Install]", file=service_file) - print("WantedBy=default.target", file=service_file) - - satellite_command.extend( - [ - "--wake-uri", - "tcp://127.0.0.1:10400", - "--wake-word-name", - wake_word, - ] - ) - satellite_requires.append(f"{wake_word_service}.service") - - if settings.debug_enabled: - satellite_command.extend( - ["--debug", "--debug-recording-dir", str(LOCAL_DIR / "debug-recording")] - ) - - satellite_command_str = shlex.join(satellite_command) - - with open( - SERVICES_DIR / "wyoming-satellite.service", "w", encoding="utf-8" - ) as service_file: - print("[Unit]", file=service_file) - print("Description=Wyoming Satellite", file=service_file) - print("Wants=network-online.target", file=service_file) - print("After=network-online.target", file=service_file) - for requires in satellite_requires: - print(f"Requires={requires}", file=service_file) - - print("", file=service_file) - print("[Service]", file=service_file) - print("Type=simple", file=service_file) - print(f"User={user}", file=service_file) - print(f"ExecStart={satellite_command_str}", file=service_file) - print(f"WorkingDirectory={PROGRAM_DIR}", file=service_file) - print("Restart=always", file=service_file) - print("RestartSec=1", file=service_file) - print("", file=service_file) - print("[Install]", file=service_file) - print("WantedBy=default.target", file=service_file) - - def apply_settings(settings: Settings) -> None: - if settings.microphone_device is None: + if settings.mic.device is None: msgbox("Please configure microphone") return + if (settings.satellite.type == SatelliteType.WAKE) and ( + settings.wake.system is None + ): + msgbox("Please set wake word system") + return + # Satellite venv venv_dir = PROGRAM_DIR / ".venv" if not venv_dir.exists(): @@ -255,7 +96,7 @@ def apply_settings(settings: Settings) -> None: return # silero (vad) - if settings.satellite_type == SatelliteType.VAD: + if settings.satellite.type == SatelliteType.VAD: result = run_with_gauge( "Installing vad...", [pip_install("-r", str(PROGRAM_DIR / "requirements_vad.txt"))], @@ -265,7 +106,7 @@ def apply_settings(settings: Settings) -> None: return # webrtc (audio enhancements) - if (settings.noise_suppression_level > 0) or (settings.auto_gain > 0): + if (settings.mic.noise_suppression > 0) or (settings.mic.auto_gain > 0): result = run_with_gauge( "Installing audio enhancements...", [ @@ -285,43 +126,7 @@ def apply_settings(settings: Settings) -> None: return stop_services(password) - - # Install and run - installed_services = ["satellite"] - if settings.satellite_type == SatelliteType.WAKE: - assert settings.wake_word_system is not None, "No wake word system" - installed_services.append( - WakeWordSystem(settings.wake_word_system).value.lower() - ) - - install_commands = [] - for service in installed_services: - service_filename = f"wyoming-{service}.service" - install_commands.append( - [ - "sudo", - "-S", - "cp", - str(SERVICES_DIR / service_filename), - "/etc/systemd/system/", - ] - ) - - install_commands.append(["sudo", "-S", "systemctl", "daemon-reload"]) - - # Copy first, then enable and start - for service in installed_services: - install_commands.append(["sudo", "-S", "systemctl", "enable", service_filename]) - - install_commands.append(["sudo", "-S", "systemctl", "start", "wyoming-satellite.service"]) - - success = run_with_gauge( - "Installing Services...", install_commands, sudo_password=password - ) - if success: - msgbox("Successfully installed services") - else: - error("installing services") + install_services(settings, password) # ----------------------------------------------------------------------------- diff --git a/installer/const.py b/installer/const.py index e872a3c..ef49e96 100644 --- a/installer/const.py +++ b/installer/const.py @@ -1,10 +1,12 @@ """Constants and dataclasses.""" -import logging import json +import logging +from dataclasses import dataclass, field from enum import Enum -from dataclasses import dataclass, fields, asdict, field -from typing import Any, Dict, Optional, List from pathlib import Path +from typing import List, Optional + +from .dataclasses_json import DataClassJsonMixin _DIR = Path(__file__).parent _LOGGER = logging.getLogger() @@ -29,6 +31,47 @@ class SatelliteType(str, Enum): WAKE = "wake" +@dataclass +class SatelliteSettings(DataClassJsonMixin): + name: str = "Wyoming Satellite" + type: SatelliteType = SatelliteType.ALWAYS_STREAMING + debug: bool = False + + +@dataclass +class MicrophoneSettings(DataClassJsonMixin): + device: Optional[str] = None + noise_suppression: int = 0 + auto_gain: int = 0 + volume_multiplier: float = 1.0 + + +@dataclass +class SpeakerSettings(DataClassJsonMixin): + device: Optional[str] = None + volume_multiplier: float = 1.0 + feedback_sounds: List[str] = field(default_factory=list) + + +@dataclass +class OpenWakeWordSettings(DataClassJsonMixin): + wake_word: str = "ok_nabu" + threshold: float = 0.5 + trigger_level: int = 1 + + +@dataclass +class Porcupine1Settings(DataClassJsonMixin): + wake_word: str = "porcupine" + sensitivity: float = 0.5 + + +@dataclass +class SnowboySettings(DataClassJsonMixin): + wake_word: str = "snowboy" + sensitivity: float = 0.5 + + class WakeWordSystem(str, Enum): OPENWAKEWORD = "openWakeWord" PORCUPINE1 = "porcupine1" @@ -36,49 +79,34 @@ class WakeWordSystem(str, Enum): @dataclass -class Settings: - microphone_device: Optional[str] = None - noise_suppression_level: int = 0 - auto_gain: int = 0 - mic_volume_multiplier: float = 1.0 +class WakeWordSettings(DataClassJsonMixin): + system: Optional[WakeWordSystem] = None + openwakeword: OpenWakeWordSettings = field(default_factory=OpenWakeWordSettings) + porcupine1: Porcupine1Settings = field(default_factory=Porcupine1Settings) + snowboy: SnowboySettings = field(default_factory=SnowboySettings) - sound_device: Optional[str] = None - feedback_sounds: List[str] = field(default_factory=list) - wake_word_system: Optional[WakeWordSystem] = None - wake_word: Dict[WakeWordSystem, str] = field( - default_factory=lambda: { - WakeWordSystem.OPENWAKEWORD: "ok_nabu", - WakeWordSystem.PORCUPINE1: "porcupine", - WakeWordSystem.SNOWBOY: "snowboy", - } - ) - - satellite_name: str = "Wyoming Satellite" - satellite_type: SatelliteType = SatelliteType.ALWAYS_STREAMING - - debug_enabled: bool = False +@dataclass +class Settings(DataClassJsonMixin): + satellite: SatelliteSettings = field(default_factory=SatelliteSettings) + mic: MicrophoneSettings = field(default_factory=MicrophoneSettings) + snd: SpeakerSettings = field(default_factory=SpeakerSettings) + wake: WakeWordSettings = field(default_factory=WakeWordSettings) @staticmethod def load() -> "Settings": - kwargs: Dict[str, Any] = {} - if SETTINGS_PATH.exists(): _LOGGER.debug("Loading settings from %s", SETTINGS_PATH) with open(SETTINGS_PATH, "r", encoding="utf-8") as settings_file: settings_dict = json.load(settings_file) + return Settings.from_dict(settings_dict) - for settings_field in fields(Settings): - value = settings_dict.get(settings_field.name) - if value is not None: - kwargs[settings_field.name] = value - - return Settings(**kwargs) + return Settings() def save(self) -> None: _LOGGER.debug("Saving settings to %s", SETTINGS_PATH) - settings_dict = asdict(self) + settings_dict = self.to_dict() SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) with open(SETTINGS_PATH, "w", encoding="utf-8") as settings_file: diff --git a/installer/dataclasses_json.py b/installer/dataclasses_json.py new file mode 100644 index 0000000..c2d1035 --- /dev/null +++ b/installer/dataclasses_json.py @@ -0,0 +1,70 @@ +"""Implement a tiny subset of dataclasses_json for config.""" +from collections.abc import Mapping, Sequence +from dataclasses import asdict, fields, is_dataclass +from typing import Any, Dict, Type + + +class DataClassJsonMixin: + """Adds from_dict to dataclass.""" + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> Any: + """Parse dataclasses recursively.""" + kwargs: Dict[str, Any] = {} + + cls_fields = {field.name: field for field in fields(cls)} + for key, value in data.items(): + if key not in cls_fields: + # Skip unknown fields + continue + + field = cls_fields[key] + if is_dataclass(field.type): + assert issubclass(field.type, DataClassJsonMixin), field.type + kwargs[key] = field.type.from_dict(value) + else: + kwargs[key] = _decode(value, field.type) + + # Fill in optional fields with None + for field in cls_fields.values(): + if (field.name not in kwargs) and _is_optional(field.type): + kwargs[field.name] = None + + return cls(**kwargs) + + def to_dict(self) -> Dict[str, Any]: + """Alias for asdict.""" + return asdict(self) + + +def _decode(value: Any, target_type: Type) -> Any: + """Decode value using (possibly generic) type.""" + if is_dataclass(target_type): + assert issubclass(target_type, DataClassJsonMixin), target_type + return target_type.from_dict(value) if value is not None else None + + if hasattr(target_type, "__args__"): + # Optional[T] + if type(None) in target_type.__args__: + optional_type = target_type.__args__[0] + return _decode(value, optional_type) + + # List[T] + if isinstance(value, Sequence): + list_type = target_type.__args__[0] + return [_decode(item, list_type) for item in value] + + # Dict[str, T] + if isinstance(value, Mapping): + value_type = target_type.__args__[1] + return { + map_key: _decode(map_value, value_type) + for map_key, map_value in value.items() + } + + return value + + +def _is_optional(target_type: Type): + """True if type is Optional""" + return hasattr(target_type, "__args__") and (type(None) in target_type.__args__) diff --git a/installer/drivers.py b/installer/drivers.py index 8500c36..1d62227 100644 --- a/installer/drivers.py +++ b/installer/drivers.py @@ -2,8 +2,8 @@ import sys from typing import Optional -from .const import Settings, PROGRAM_DIR -from .whiptail import msgbox, run_with_gauge, menu, yesno, passwordbox, error +from .const import PROGRAM_DIR, Settings +from .whiptail import error, menu, msgbox, passwordbox, run_with_gauge, yesno def install_drivers(settings: Settings) -> None: diff --git a/installer/microphone.py b/installer/microphone.py index 8f98062..27d39cc 100644 --- a/installer/microphone.py +++ b/installer/microphone.py @@ -1,13 +1,13 @@ """Microphone settings.""" import array +import logging import math import subprocess -import logging -from concurrent.futures import ThreadPoolExecutor, Future -from typing import Dict, Optional, List +from concurrent.futures import Future, ThreadPoolExecutor +from typing import Dict, List, Optional -from .const import Settings, RECORD_SECONDS, RECORD_RMS_MIN -from .whiptail import gauge, msgbox, radiolist, menu, inputbox +from .const import RECORD_RMS_MIN, RECORD_SECONDS, Settings +from .whiptail import gauge, inputbox, menu, msgbox, radiolist _LOGGER = logging.getLogger() @@ -40,29 +40,25 @@ def configure_microphone(settings: Settings) -> None: if best_device is not None: msgbox(f"Successfully detected microphone: {best_device}") - settings.microphone_device = best_device + settings.mic.device = best_device settings.save() else: msgbox("Audio was not detected from any microphone") elif choice == "list": # arecord -L microphone_device = radiolist( - "Select ALSA Device:", - get_microphone_devices(), - settings.microphone_device, + "Select ALSA Device:", get_microphone_devices(), settings.mic.device ) if microphone_device: - settings.microphone_device = microphone_device + settings.mic.device = microphone_device settings.save() elif choice == "manual": - microphone_device = inputbox( - "Enter ALSA Device:", settings.microphone_device - ) + microphone_device = inputbox("Enter ALSA Device:", settings.mic.device) if microphone_device: - settings.microphone_device = microphone_device + settings.mic.device = microphone_device settings.save() - elif choice == "enhancements": - configure_audio_enhancements(settings) + elif choice == "settings": + configure_audio_settings(settings) else: break @@ -74,7 +70,7 @@ def microphone_menu(last_choice: Optional[str]) -> Optional[str]: ("detect", "Autodetect"), ("list", "Select From List"), ("manual", "Enter Manually"), - ("enhancements", "Audio Enhancements"), + ("settings", "Audio Settings"), ], selected_item=last_choice, menu_args=["--ok-button", "Select", "--cancel-button", "Back"], @@ -86,7 +82,9 @@ def get_microphone_devices() -> List[str]: lines = subprocess.check_output(["arecord", "-L"]).decode("utf-8").splitlines() for line in lines: line = line.strip() - if line.startswith("plughw:"): + + # default = PulseAudio + if (line == "default") or line.startswith("plughw:"): devices.append(line) return devices @@ -126,61 +124,65 @@ def _record_proc(device: str) -> float: return 0 -def configure_audio_enhancements(settings: Settings) -> None: +def configure_audio_settings(settings: Settings) -> None: choice: Optional[str] = None while True: - choice = audio_enhancements_menu(choice) + choice = audio_settings_menu(choice) if choice == "noise": - result = radiolist( + noise_suppression = radiolist( "Noise Suppression Level", [(0, "Off"), (1, "Low"), (2, "Medium"), (3, "High"), (4, "Maximum")], - settings.noise_suppression_level, + settings.mic.noise_suppression, ) - if result is not None: - settings.noise_suppression_level = result + if noise_suppression is not None: + settings.mic.noise_suppression = noise_suppression settings.save() elif choice == "gain": while True: - result = inputbox("Auto Gain (0-31 dbFS)", settings.auto_gain) - if result is not None: - try: - auto_gain = int(result) - except ValueError: - msgbox("Invalid value") - continue + auto_gain = inputbox("Auto Gain (0-31 dbFS)", settings.mic.auto_gain) + if auto_gain is None: + break - if 0 <= auto_gain <= 31: - settings.auto_gain = auto_gain - settings.save() - break + try: + auto_gain_int = int(auto_gain) + except ValueError: + msgbox("Invalid value") + continue - msgbox("Must be 0-31") + if 0 <= auto_gain_int <= 31: + settings.mic.auto_gain = auto_gain_int + settings.save() + break + + msgbox("Must be 0-31") elif choice == "multiplier": while True: - result = inputbox( - "Volume Multipler (1 = default)", settings.mic_volume_multiplier + volume_multiplier = inputbox( + "Volume Multipler (1 = default)", settings.mic.volume_multiplier ) - if result is not None: - try: - volume_multiplier = float(result) - except ValueError: - msgbox("Invalid value") - continue + if volume_multiplier is None: + break - if volume_multiplier > 0: - settings.mic_volume_multiplier = volume_multiplier - settings.save() - break + try: + volume_multiplier_float = float(volume_multiplier) + except ValueError: + msgbox("Invalid value") + continue - msgbox("Must be > 0") + if volume_multiplier_float > 0: + settings.mic.volume_multiplier = volume_multiplier_float + settings.save() + break + + msgbox("Must be > 0") else: break -def audio_enhancements_menu(last_choice: Optional[str]) -> Optional[str]: +def audio_settings_menu(last_choice: Optional[str]) -> Optional[str]: return menu( - "Main > Microphone > Audio Enhancements", + "Main > Microphone > Audio Settings", [ ("noise", "Noise Suppression"), ("gain", "Auto Gain"), diff --git a/installer/satellite.py b/installer/satellite.py index 6c236e3..d1f2be1 100644 --- a/installer/satellite.py +++ b/installer/satellite.py @@ -1,8 +1,8 @@ """Satellite settings.""" from typing import Optional -from .const import Settings, SatelliteType -from .whiptail import menu, inputbox, radiolist, run_with_gauge +from .const import SatelliteType, Settings +from .whiptail import error, inputbox, menu, passwordbox, radiolist, run_with_gauge def configure_satellite(settings: Settings) -> None: @@ -11,9 +11,51 @@ def configure_satellite(settings: Settings) -> None: choice = satellite_menu(choice) if choice == "name": - set_satellite_name(settings) + name = inputbox("Satellite Name:", settings.satellite.name) + if name: + settings.satellite.name = name + settings.save() elif choice == "type": - set_satellite_type(settings) + satellite_type = radiolist( + "Satellite Type:", + [ + (SatelliteType.ALWAYS_STREAMING, "Always streaming"), + (SatelliteType.VAD, "Voice activity detection"), + (SatelliteType.WAKE, "Local wake word detection"), + ], + settings.satellite.type, + ) + + if satellite_type is not None: + settings.satellite.type = SatelliteType(satellite_type) + settings.save() + elif choice in ("stop", "start"): + password = passwordbox("sudo password:") + if not password: + continue + + command = ["sudo", "-S", "systemctl", choice, "wyoming-satellite.service"] + text = ( + "Stopping satellite..." if choice == "stop" else "Starting satellite..." + ) + success = run_with_gauge(text, [command], sudo_password=password) + if not success: + error( + "stopping satellite" if choice == "stop" else "starting satellite" + ) + elif choice == "debug": + debug = radiolist( + "Debug Mode:", + [ + ("disabled", "Disabled"), + ("enabled", "Enabled"), + ], + "enabled" if settings.satellite.debug else "disabled", + ) + + if debug is not None: + settings.satellite.debug = debug == "enabled" + settings.save() else: break @@ -24,30 +66,10 @@ def satellite_menu(last_choice: Optional[str]) -> Optional[str]: [ ("name", "Satellite Name"), ("type", "Satellite Type"), + ("stop", "Stop Service"), + ("start", "Start Service"), + ("debug", "Set Debug Mode"), ], selected_item=last_choice, menu_args=["--ok-button", "Select", "--cancel-button", "Back"], ) - - -def set_satellite_name(settings: Settings) -> None: - name = inputbox("Satellite Name:", settings.satellite_name) - if name: - settings.satellite_name = name - settings.save() - - -def set_satellite_type(settings: Settings) -> None: - satellite_type = radiolist( - "Satellite Type:", - [ - (SatelliteType.ALWAYS_STREAMING, "Always streaming"), - (SatelliteType.VAD, "Voice activity detection"), - (SatelliteType.WAKE, "Local wake word detection"), - ], - settings.satellite_type, - ) - - if satellite_type is not None: - settings.satellite_type = SatelliteType(satellite_type) - settings.save() diff --git a/installer/services.py b/installer/services.py new file mode 100644 index 0000000..1e82999 --- /dev/null +++ b/installer/services.py @@ -0,0 +1,231 @@ +"""Systemd service management.""" +import shlex +import subprocess +from pathlib import Path +from typing import List + +from .const import ( + LOCAL_DIR, + PROGRAM_DIR, + SERVICES_DIR, + SatelliteType, + Settings, + WakeWordSystem, +) +from .whiptail import error, msgbox, run_with_gauge + + +def stop_services(password: str) -> None: + stop_commands = [] + for service in ("satellite", "openwakeword", "porcupine1", "snowboy"): + service_filename = f"wyoming-{service}.service" + service_path = Path("/etc/systemd/system") / service_filename + if not service_path.exists(): + continue + + stop_commands.append(["sudo", "-S", "systemctl", "stop", service_filename]) + stop_commands.append(["sudo", "-S", "systemctl", "disable", service_filename]) + + run_with_gauge("Stopping Services...", stop_commands, sudo_password=password) + + +def generate_services(settings: Settings) -> None: + SERVICES_DIR.mkdir(parents=True, exist_ok=True) + + user_id = subprocess.check_output(["id", "-u"], text=True).strip() + user_name = subprocess.check_output(["id", "--name", "-u"], text=True).strip() + + satellite_command: List[str] = [ + str(PROGRAM_DIR / "script" / "run"), + "--name", + settings.satellite.name, + "--uri", + "tcp://0.0.0.0:10700", + "--mic-command", + f"arecord -D {settings.mic.device} -q -r 16000 -c 1 -f S16_LE -t raw", + ] + satellite_requires: List[str] = [] + + if settings.snd.device is not None: + # Audio output + satellite_command.extend( + [ + "--snd-command", + f"aplay -D {settings.snd.device} -q -r 22050 -c 1 -f S16_LE -t raw", + ] + ) + + for sound_name in settings.snd.feedback_sounds: + # Try local/sounds first + sound_path = LOCAL_DIR / "sounds" / f"{sound_name}.wav" + if not sound_path.exists(): + sound_path = PROGRAM_DIR / "sounds" / f"{sound_name}.wav" + + satellite_command.extend([f"--{sound_name}-wav", str(sound_path)]) + + if settings.mic.noise_suppression > 0: + satellite_command.extend( + ["--mic-noise-suppression", str(settings.mic.noise_suppression)] + ) + + if settings.mic.auto_gain > 0: + satellite_command.extend(["--mic-auto-gain", str(settings.mic.auto_gain)]) + + if settings.mic.volume_multiplier != 1.0: + satellite_command.extend( + ["--mic-volume-multiplier", str(settings.mic.volume_multiplier)] + ) + + if settings.satellite.type == SatelliteType.VAD: + # Voice activity detect + satellite_command.append("--vad") + elif settings.satellite.type == SatelliteType.WAKE: + # Local wake word detection + assert settings.wake.system is not None, "Wake word system not set" + + wake_word_service = ( + "wyoming-" + WakeWordSystem(settings.wake.system).value.lower() + ) + wake_word_dir = LOCAL_DIR / wake_word_service + wake_word_command = [ + str(wake_word_dir / "script" / "run"), + "--uri", + "tcp://127.0.0.1:10400", + ] + + if settings.wake.system == WakeWordSystem.OPENWAKEWORD: + wake_word = settings.wake.openwakeword.wake_word + wake_word_command.extend( + [ + "--threshold", + str(settings.wake.openwakeword.threshold), + "--trigger-level", + str(settings.wake.openwakeword.trigger_level), + "--custom-model-dir", + str(LOCAL_DIR / "custom-wake-words" / "openWakeWord"), + ] + ) + elif settings.wake.system == WakeWordSystem.PORCUPINE1: + wake_word = settings.wake.porcupine1.wake_word + wake_word_command.extend( + ["--sensitivity", str(settings.wake.porcupine1.sensitivity)] + ) + elif settings.wake.system == WakeWordSystem.SNOWBOY: + wake_word = settings.wake.snowboy.wake_word + wake_word_command.extend( + [ + "--sensitivity", + str(settings.wake.snowboy.sensitivity), + "--custom-model-dir", + str(LOCAL_DIR / "custom-wake-words" / "snowboy"), + ] + ) + else: + raise ValueError(settings.wake.system) + + if settings.satellite.debug: + wake_word_command.append("--debug") + + wake_word_command_str = shlex.join(wake_word_command) + + with open( + SERVICES_DIR / f"{wake_word_service}.service", "w", encoding="utf-8" + ) as service_file: + print("[Unit]", file=service_file) + print( + f"Description={WakeWordSystem(settings.wake.system).value}", + file=service_file, + ) + print("", file=service_file) + print("[Service]", file=service_file) + print("Type=simple", file=service_file) + print(f"User={user_name}", file=service_file) + print(f"ExecStart={wake_word_command_str}", file=service_file) + print(f"WorkingDirectory={wake_word_dir}", file=service_file) + print("Restart=always", file=service_file) + print("RestartSec=1", file=service_file) + print("", file=service_file) + print("[Install]", file=service_file) + print("WantedBy=default.target", file=service_file) + + satellite_command.extend( + [ + "--wake-uri", + "tcp://127.0.0.1:10400", + "--wake-word-name", + wake_word, + ] + ) + satellite_requires.append(f"{wake_word_service}.service") + + if settings.satellite.debug: + satellite_command.extend( + ["--debug", "--debug-recording-dir", str(LOCAL_DIR / "debug-recording")] + ) + + satellite_command_str = shlex.join(satellite_command) + + with open( + SERVICES_DIR / "wyoming-satellite.service", "w", encoding="utf-8" + ) as service_file: + print("[Unit]", file=service_file) + print("Description=Wyoming Satellite", file=service_file) + print("Wants=network-online.target", file=service_file) + print("After=network-online.target", file=service_file) + for requires in satellite_requires: + print(f"Requires={requires}", file=service_file) + + print("", file=service_file) + print("[Service]", file=service_file) + print("Type=simple", file=service_file) + print(f"User={user_name}", file=service_file) + + # For PulseAudio + print(f"Environment=XDG_RUNTIME_DIR=/run/user/{user_id}", file=service_file) + + print(f"ExecStart={satellite_command_str}", file=service_file) + print(f"WorkingDirectory={PROGRAM_DIR}", file=service_file) + print("Restart=always", file=service_file) + print("RestartSec=1", file=service_file) + print("", file=service_file) + print("[Install]", file=service_file) + print("WantedBy=default.target", file=service_file) + + +def install_services(settings: Settings, password: str): + # Install and run + installed_services = ["satellite"] + if settings.satellite.type == SatelliteType.WAKE: + assert settings.wake.system is not None, "No wake word system" + installed_services.append(WakeWordSystem(settings.wake.system).value.lower()) + + install_commands = [] + for service in installed_services: + service_filename = f"wyoming-{service}.service" + install_commands.append( + [ + "sudo", + "-S", + "cp", + str(SERVICES_DIR / service_filename), + "/etc/systemd/system/", + ] + ) + + install_commands.append(["sudo", "-S", "systemctl", "daemon-reload"]) + + # Copy first, then enable and start + for service in installed_services: + install_commands.append(["sudo", "-S", "systemctl", "enable", service_filename]) + + install_commands.append( + ["sudo", "-S", "systemctl", "start", "wyoming-satellite.service"] + ) + + success = run_with_gauge( + "Installing Services...", install_commands, sudo_password=password + ) + if success: + msgbox("Successfully installed services") + else: + error("installing services") diff --git a/installer/speakers.py b/installer/speakers.py index 2d2b132..c0033ad 100644 --- a/installer/speakers.py +++ b/installer/speakers.py @@ -1,10 +1,10 @@ """Speaker settings.""" -import subprocess import logging -from typing import Optional, List +import subprocess +from typing import List, Optional -from .const import Settings, PROGRAM_DIR -from .whiptail import radiolist, inputbox, msgbox, checklist, menu +from .const import PROGRAM_DIR, Settings +from .whiptail import checklist, inputbox, menu, msgbox, radiolist _LOGGER = logging.getLogger() @@ -14,38 +14,44 @@ def configure_speakers(settings: Settings) -> None: while True: choice = speakers_menu(choice) - if choice == "test": + if choice == "play": + if settings.snd.device is None: + msgbox("No speaker device set") + continue + + test_sound_device(settings.snd.device) + elif choice == "test": sound_device = test_speakers() if sound_device is not None: - settings.sound_device = sound_device + settings.snd.device = sound_device settings.save() elif choice == "list": sound_device = radiolist( "Select ALSA Device:", get_sound_devices(), - settings.sound_device, + settings.snd.device, ) if sound_device: - settings.sound_device = sound_device + settings.snd.device = sound_device settings.save() elif choice == "manual": - sound_device = inputbox("Enter ALSA Device:", settings.sound_device) + sound_device = inputbox("Enter ALSA Device:", settings.snd.device) if sound_device: - settings.sound_device = sound_device + settings.snd.device = sound_device settings.save() elif choice == "disable": - settings.sound_device = None + settings.snd.device = None settings.save() msgbox("Sound disabled") elif choice == "feedback": feedback_sounds = checklist( "Enabled Sounds:", [("awake", "On wake-up"), ("done", "After voice command")], - settings.feedback_sounds, + settings.snd.feedback_sounds, ) if feedback_sounds is not None: - settings.feedback_sounds = feedback_sounds + settings.snd.feedback_sounds = feedback_sounds settings.save() else: break @@ -55,7 +61,8 @@ def speakers_menu(last_choice: Optional[str]) -> Optional[str]: return menu( "Main > Speakers", [ - ("test", "Test Speakers"), + ("play", "Play Sound"), + ("test", "Test All Speakers"), ("list", "Select From List"), ("manual", "Enter Manually"), ("disable", "Disable Sound"), @@ -71,7 +78,9 @@ def get_sound_devices() -> List[str]: lines = subprocess.check_output(["aplay", "-L"]).decode("utf-8").splitlines() for line in lines: line = line.strip() - if line.startswith("plughw:"): + + # default = PulseAudio + if (line == "default") or line.startswith("plughw:"): devices.append(line) return devices diff --git a/installer/wake_word.py b/installer/wake_word.py index 1971446..eb91164 100644 --- a/installer/wake_word.py +++ b/installer/wake_word.py @@ -3,25 +3,38 @@ import itertools import re import shutil from pathlib import Path -from typing import Optional, Dict +from typing import Dict, Optional -from .const import Settings, SatelliteType, WakeWordSystem, LOCAL_DIR -from .whiptail import msgbox, menu, radiolist, run_with_gauge, yesno, error +from .const import LOCAL_DIR, SatelliteType, Settings, WakeWordSystem +from .whiptail import error, inputbox, menu, msgbox, radiolist, run_with_gauge, yesno def configure_wake_word(settings: Settings) -> None: - if settings.satellite_type != SatelliteType.WAKE: + if settings.satellite.type != SatelliteType.WAKE: if not yesno("Set satellite type to local wake word?"): return - settings.satellite_type = SatelliteType.WAKE + settings.satellite.type = SatelliteType.WAKE settings.save() + def star(ww_system: WakeWordSystem) -> str: + return ( + f"{ww_system.value} [*]" + if settings.wake.system == ww_system + else ww_system.value + ) + choice: Optional[str] = None while True: choice = menu( "Main > Wake Word", - [("system", "Wake Word System"), ("wake_word", "Wake Word")], + [ + ("system", "Wake Word System"), + ("wake_word", "Choose Wake Word"), + ("openWakeWord", star(WakeWordSystem.OPENWAKEWORD)), + ("porcupine1", star(WakeWordSystem.PORCUPINE1)), + ("snowboy", star(WakeWordSystem.SNOWBOY)), + ], selected_item=choice, menu_args=["--ok-button", "Select", "--cancel-button", "Back"], ) @@ -30,7 +43,7 @@ def configure_wake_word(settings: Settings) -> None: wake_word_system = radiolist( "Wake Word System:", [v.value for v in WakeWordSystem], - settings.wake_word_system, + settings.wake.system, ) if wake_word_system is None: continue @@ -39,6 +52,12 @@ def configure_wake_word(settings: Settings) -> None: install_wake_word(settings, wake_word_system) elif choice == "wake_word": select_wake_word(settings) + elif choice == "openWakeWord": + configure_openWakeWord(settings) + elif choice == "porcupine1": + configure_porcupine1(settings) + elif choice == "snowboy": + configure_snowboy(settings) else: break @@ -75,8 +94,7 @@ def install_wake_word(settings: Settings, wake_word_system: WakeWordSystem) -> N msgbox("openWakeWord installed successfully") - settings.wake_word_system = wake_word_system - settings.wake_word.setdefault(WakeWordSystem.OPENWAKEWORD, "ok_nabu") + settings.wake.system = wake_word_system settings.save() return @@ -110,7 +128,7 @@ def install_wake_word(settings: Settings, wake_word_system: WakeWordSystem) -> N else: msgbox("porcupine1 installed successfully") - settings.wake_word_system = wake_word_system + settings.wake.system = wake_word_system settings.save() return @@ -144,21 +162,19 @@ def install_wake_word(settings: Settings, wake_word_system: WakeWordSystem) -> N else: msgbox("snowboy installed successfully") - settings.wake_word_system = wake_word_system + settings.wake.system = wake_word_system settings.save() return def select_wake_word(settings: Settings) -> None: - if settings.wake_word_system == WakeWordSystem.OPENWAKEWORD: + if settings.wake.system == WakeWordSystem.OPENWAKEWORD: oww_dir = LOCAL_DIR / "wyoming-openwakeword" if not oww_dir.exists(): msgbox("openWakeWord is not installed") return - custom_wake_word_dir = ( - LOCAL_DIR / "custom-wake-words" / WakeWordSystem.OPENWAKEWORD.value - ) + custom_wake_word_dir = LOCAL_DIR / "custom-wake-words" / "openWakeWord" custom_wake_word_dir.mkdir(parents=True, exist_ok=True) community_wake_word_dir = LOCAL_DIR / "home-assistant-wakewords-collection" @@ -187,64 +203,25 @@ def select_wake_word(settings: Settings) -> None: ww_paths[ww_name] = ww_path - items = [("__community__", "Download Community Wake Words")] + list( - sorted(ww_paths.keys()) - ) - + items = sorted(list(ww_paths.keys())) wake_word = radiolist( - "Wake Word:", items, settings.wake_word.get(WakeWordSystem.OPENWAKEWORD) + "Wake Word:", items, settings.wake.openwakeword.wake_word ) - if wake_word is None: break - if wake_word == "__community__": - if not community_wake_word_dir.exists(): - success = run_with_gauge( - "Downloading community wake words...", - [ - [ - "git", - "clone", - "https://github.com/fwartner/home-assistant-wakewords-collection.git", - str(community_wake_word_dir), - ] - ], - ) - - if not success: - error("downloading community wake words") - else: - success = run_with_gauge( - "Updating community wake words...", - [ - [ - "git", - "-C", - str(community_wake_word_dir), - "pull", - "origin", - "main", - ] - ], - ) - - if not success: - error("updating community wake words") - continue - wake_word_path = ww_paths[wake_word] if wake_word_path.is_relative_to(community_wake_word_dir): # Copy to custom directory shutil.copy(wake_word_path, custom_wake_word_dir) - settings.wake_word[WakeWordSystem.OPENWAKEWORD] = wake_word + settings.wake.openwakeword.wake_word = wake_word settings.save() break return - if settings.wake_word_system == WakeWordSystem.PORCUPINE1: + if settings.wake.system == WakeWordSystem.PORCUPINE1: porcupine1_dir = LOCAL_DIR / "wyoming-porcupine1" if not porcupine1_dir.exists(): msgbox("porcupine1 is not installed") @@ -262,26 +239,21 @@ def select_wake_word(settings: Settings) -> None: ) wake_word = radiolist( - "Wake Word:", - ww_names, - settings.wake_word.get(WakeWordSystem.PORCUPINE1), + "Wake Word:", ww_names, settings.wake.porcupine1.wake_word ) - if wake_word is not None: - settings.wake_word[WakeWordSystem.PORCUPINE1] = wake_word + settings.wake.porcupine1.wake_word = wake_word settings.save() return - if settings.wake_word_system == WakeWordSystem.SNOWBOY: + if settings.wake.system == WakeWordSystem.SNOWBOY: snowboy_dir = LOCAL_DIR / "wyoming-snowboy" if not snowboy_dir.exists(): msgbox("snowboy is not installed") return - custom_wake_word_dir = ( - LOCAL_DIR / "custom-wake-words" / WakeWordSystem.SNOWBOY.value - ) + custom_wake_word_dir = LOCAL_DIR / "custom-wake-words" / "snowboy" custom_wake_word_dir.mkdir(parents=True, exist_ok=True) builtin_wake_words = (snowboy_dir / "wyoming_snowboy" / "data").glob("*.umdl") @@ -297,12 +269,178 @@ def select_wake_word(settings: Settings) -> None: ) ) - wake_word = radiolist( - "Wake Word:", ww_names, settings.wake_word.get(WakeWordSystem.SNOWBOY) - ) - + wake_word = radiolist("Wake Word:", ww_names, settings.wake.snowboy.wake_word) if wake_word is not None: - settings.wake_word[WakeWordSystem.SNOWBOY] = wake_word + settings.wake.snowboy.wake_word = wake_word settings.save() return + + +def configure_openWakeWord(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = menu( + "Main > Wake Word > openWakeWord", + [ + ("community", "Download Community Wake Words"), + ("threshold", "Set Threshold"), + ("trigger_level", "Set Trigger Level"), + ], + selected_item=choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + if choice == "community": + community_wake_word_dir = LOCAL_DIR / "home-assistant-wakewords-collection" + + if not community_wake_word_dir.exists(): + success = run_with_gauge( + "Downloading community wake words...", + [ + [ + "git", + "clone", + "https://github.com/fwartner/home-assistant-wakewords-collection.git", + str(community_wake_word_dir), + ] + ], + ) + + if not success: + error("downloading community wake words") + else: + success = run_with_gauge( + "Updating community wake words...", + [ + [ + "git", + "-C", + str(community_wake_word_dir), + "pull", + "origin", + "main", + ] + ], + ) + + if not success: + error("updating community wake words") + elif choice == "threshold": + while True: + threshold = inputbox( + "Threshold (0-1, 0.5 = default):", + settings.wake.openwakeword.threshold, + ) + if threshold is None: + break + + try: + threshold_float = float(threshold) + except ValueError: + msgbox("Invalid value") + continue + + if 0 < threshold_float < 1: + settings.wake.openwakeword.threshold = threshold_float + settings.save() + break + + msgbox("Threshold must be in (0, 1)") + elif choice == "trigger_level": + while True: + trigger_level = inputbox( + "Trigger Level (> 0, 1 = default):", + settings.wake.openwakeword.trigger_level, + ) + if trigger_level is None: + break + + try: + trigger_level_int = int(trigger_level) + except ValueError: + msgbox("Invalid value") + continue + + if trigger_level_int > 0: + settings.wake.openwakeword.trigger_level = trigger_level_int + settings.save() + break + + msgbox("Trigger level must be > 0") + else: + break + + +def configure_porcupine1(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = menu( + "Main > Wake Word > porcupine1", + [ + ("sensitivity", "Set Sensitivity"), + ], + selected_item=choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + if choice == "sensitivity": + while True: + sensitivity = inputbox( + "Sensitivity (0-1, 0.5 = default):", + settings.wake.porcupine1.sensitivity, + ) + if sensitivity is None: + break + + try: + sensitivity_float = float(sensitivity) + except ValueError: + msgbox("Invalid value") + continue + + if 0 < sensitivity_float < 1: + settings.wake.porcupine1.sensitivity = sensitivity_float + settings.save() + break + + msgbox("Sensitivity must be in (0, 1)") + else: + break + + +def configure_snowboy(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = menu( + "Main > Wake Word > snowboy", + [ + ("sensitivity", "Set Sensitivity"), + ], + selected_item=choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + if choice == "sensitivity": + while True: + sensitivity = inputbox( + "Sensitivity (0-1, 0.5 = default):", + settings.wake.snowboy.sensitivity, + ) + if sensitivity is None: + break + + try: + sensitivity_float = float(sensitivity) + except ValueError: + msgbox("Invalid value") + continue + + if 0 < sensitivity_float < 1: + settings.wake.snowboy.sensitivity = sensitivity_float + settings.save() + break + + msgbox("Sensitivity must be in (0, 1)") + else: + break diff --git a/installer/whiptail.py b/installer/whiptail.py index 1635a50..4c4d911 100644 --- a/installer/whiptail.py +++ b/installer/whiptail.py @@ -1,13 +1,13 @@ """Python interface to whiptail command.""" -import subprocess -import time import logging import shlex -from concurrent.futures import ThreadPoolExecutor +import subprocess +import time from collections.abc import Sequence -from typing import Any, Dict, List, Tuple, Optional, Union +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Dict, List, Optional, Tuple, Union -from .const import TITLE, WIDTH, HEIGHT, LIST_HEIGHT +from .const import HEIGHT, LIST_HEIGHT, TITLE, WIDTH ItemType = Union[str, Tuple[Any, str]] @@ -66,7 +66,9 @@ def menu( def inputbox(text: str, init: Optional[Any] = None) -> Optional[str]: - return whiptail("--inputbox", text, HEIGHT, WIDTH, str(init) or "") + return whiptail( + "--inputbox", text, HEIGHT, WIDTH, str(init) if init is not None else "" + ) def passwordbox(text: str) -> Optional[str]: @@ -184,7 +186,9 @@ def gauge(text: str, seconds: int, parts: int = 20) -> None: def error(reason: str) -> None: - msgbox(f"An error occurred while {reason}.\n" "See local/installer.log for details.") + msgbox( + f"An error occurred while {reason}.\n" "See local/installer.log for details." + ) # ----------------------------------------------------------------------------- diff --git a/script/format b/script/format index 3f8a9b2..2bb48d6 100755 --- a/script/format +++ b/script/format @@ -8,11 +8,10 @@ _PROGRAM_DIR = _DIR.parent _VENV_DIR = _PROGRAM_DIR / ".venv" _MODULE_DIR = _PROGRAM_DIR / "wyoming_satellite" _TESTS_DIR = _PROGRAM_DIR / "tests" +_INSTALLER_DIR = _PROGRAM_DIR / "installer" + +_FORMAT_DIRS = [_MODULE_DIR, _TESTS_DIR, _INSTALLER_DIR] context = venv.EnvBuilder().ensure_directories(_VENV_DIR) -subprocess.check_call( - [context.env_exe, "-m", "black", str(_MODULE_DIR), str(_TESTS_DIR)] -) -subprocess.check_call( - [context.env_exe, "-m", "isort", str(_MODULE_DIR), str(_TESTS_DIR)] -) +subprocess.check_call([context.env_exe, "-m", "black"] + _FORMAT_DIRS) +subprocess.check_call([context.env_exe, "-m", "isort"] + _FORMAT_DIRS) diff --git a/script/lint b/script/lint index 46ba277..ceabca0 100755 --- a/script/lint +++ b/script/lint @@ -8,20 +8,13 @@ _PROGRAM_DIR = _DIR.parent _VENV_DIR = _PROGRAM_DIR / ".venv" _MODULE_DIR = _PROGRAM_DIR / "wyoming_satellite" _TESTS_DIR = _PROGRAM_DIR / "tests" +_INSTALLER_DIR = _PROGRAM_DIR / "installer" + +_LINT_DIRS = [_MODULE_DIR, _TESTS_DIR, _INSTALLER_DIR] context = venv.EnvBuilder().ensure_directories(_VENV_DIR) -subprocess.check_call( - [context.env_exe, "-m", "black", str(_MODULE_DIR), str(_TESTS_DIR), "--check"] -) -subprocess.check_call( - [context.env_exe, "-m", "isort", str(_MODULE_DIR), str(_TESTS_DIR), "--check"] -) -subprocess.check_call( - [context.env_exe, "-m", "flake8", str(_MODULE_DIR), str(_TESTS_DIR)] -) -subprocess.check_call( - [context.env_exe, "-m", "pylint", str(_MODULE_DIR), str(_TESTS_DIR)] -) -subprocess.check_call( - [context.env_exe, "-m", "mypy", str(_MODULE_DIR), str(_TESTS_DIR)] -) +subprocess.check_call([context.env_exe, "-m", "black"] + _LINT_DIRS + ["--check"]) +subprocess.check_call([context.env_exe, "-m", "isort"] + _LINT_DIRS + ["--check"]) +subprocess.check_call([context.env_exe, "-m", "flake8"] + _LINT_DIRS) +subprocess.check_call([context.env_exe, "-m", "pylint"] + _LINT_DIRS) +subprocess.check_call([context.env_exe, "-m", "mypy"] + _LINT_DIRS)