diff --git a/installer/__init__.py b/installer/__init__.py new file mode 100644 index 0000000..c776e30 --- /dev/null +++ b/installer/__init__.py @@ -0,0 +1 @@ +"""Command-line installer.""" diff --git a/installer/__main__.py b/installer/__main__.py new file mode 100644 index 0000000..c04de60 --- /dev/null +++ b/installer/__main__.py @@ -0,0 +1,331 @@ +"""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 .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 .drivers import install_drivers +from .wake_word import configure_wake_word + +_LOGGER = logging.getLogger() + + +def main() -> None: + LOCAL_DIR.mkdir(parents=True, exist_ok=True) + logging.basicConfig( + filename=LOCAL_DIR / "installer.log", filemode="w", level=logging.DEBUG + ) + + 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) + + if choice == "satellite": + configure_satellite(settings) + elif choice == "microphone": + configure_microphone(settings) + elif choice == "speakers": + configure_speakers(settings) + elif choice == "wake": + configure_wake_word(settings) + elif choice == "drivers": + install_drivers(settings) + elif choice == "apply": + apply_settings(settings) + else: + break + + +def main_menu(last_choice: Optional[str]) -> Optional[str]: + items: List[ItemType] = [ + ("satellite", "Satellite"), + ("microphone", "Microphone"), + ("speakers", "Speakers"), + ("wake", "Wake Word"), + ("drivers", "Drivers"), + ("apply", "Apply Settings"), + ] + + return menu( + "Main", + items, + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Exit"], + ) + + +# ----------------------------------------------------------------------------- + + +def pip_install(*args) -> List[str]: + return [ + str(PROGRAM_DIR / ".venv" / "bin" / "pip3"), + "install", + "--extra-index-url", + "https://www.piwheels.org/simple", + "-f", + "https://synesthesiam.github.io/prebuilt-apps/", + ] + 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: + msgbox("Please configure microphone") + return + + # Satellite venv + venv_dir = PROGRAM_DIR / ".venv" + if not venv_dir.exists(): + result = run_with_gauge( + "Creating virtual environment...", [str(PROGRAM_DIR / "script" / "setup")] + ) + if not result: + error("creating virtual environment") + return + + # silero (vad) + if settings.satellite_type == SatelliteType.VAD: + result = run_with_gauge( + "Installing vad...", + [pip_install("-r", str(PROGRAM_DIR / "requirements_vad.txt"))], + ) + if not result: + error("installing vad") + return + + # webrtc (audio enhancements) + if (settings.noise_suppression_level > 0) or (settings.auto_gain > 0): + result = run_with_gauge( + "Installing audio enhancements...", + [ + pip_install( + "-r", str(PROGRAM_DIR / "requirements_audio_enhancement.txt") + ) + ], + ) + if not result: + error("installing audio enhancements") + return + + generate_services(settings) + + password = passwordbox("sudo password:") + if not password: + 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") + + +# ----------------------------------------------------------------------------- + + +if __name__ == "__main__": + main() diff --git a/installer/const.py b/installer/const.py new file mode 100644 index 0000000..e872a3c --- /dev/null +++ b/installer/const.py @@ -0,0 +1,85 @@ +"""Constants and dataclasses.""" +import logging +import json +from enum import Enum +from dataclasses import dataclass, fields, asdict, field +from typing import Any, Dict, Optional, List +from pathlib import Path + +_DIR = Path(__file__).parent +_LOGGER = logging.getLogger() + +PROGRAM_DIR = _DIR.parent +LOCAL_DIR = PROGRAM_DIR / "local" +SETTINGS_PATH = LOCAL_DIR / "settings.json" +SERVICES_DIR = LOCAL_DIR / "services" + +RECORD_SECONDS = 8 +RECORD_RMS_MIN = 30 + +TITLE = "Wyoming Satellite" +WIDTH = "75" +HEIGHT = "20" +LIST_HEIGHT = "12" + + +class SatelliteType(str, Enum): + ALWAYS_STREAMING = "always" + VAD = "vad" + WAKE = "wake" + + +class WakeWordSystem(str, Enum): + OPENWAKEWORD = "openWakeWord" + PORCUPINE1 = "porcupine1" + SNOWBOY = "snowboy" + + +@dataclass +class Settings: + microphone_device: Optional[str] = None + noise_suppression_level: int = 0 + auto_gain: int = 0 + mic_volume_multiplier: float = 1.0 + + 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 + + @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) + + 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) + + def save(self) -> None: + _LOGGER.debug("Saving settings to %s", SETTINGS_PATH) + + settings_dict = asdict(self) + SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) + + with open(SETTINGS_PATH, "w", encoding="utf-8") as settings_file: + json.dump(settings_dict, settings_file, ensure_ascii=False, indent=2) diff --git a/installer/drivers.py b/installer/drivers.py new file mode 100644 index 0000000..8500c36 --- /dev/null +++ b/installer/drivers.py @@ -0,0 +1,50 @@ +"""Hardware drivers.""" +import sys +from typing import Optional + +from .const import Settings, PROGRAM_DIR +from .whiptail import msgbox, run_with_gauge, menu, yesno, passwordbox, error + + +def install_drivers(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = menu( + "Main > Drivers", + [("respeaker", "Install ReSpeaker Drivers")], + selected_item=choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + if choice == "respeaker": + if yesno( + "ReSpeaker drivers for the Raspberry Pi will now be compiled and installed. " + "This will take a long time and require a reboot. " + "Continue?" + ): + password = passwordbox("sudo password:") + success = run_with_gauge( + "Installing drivers...", + [ + [ + "sudo", + "-S", + str(PROGRAM_DIR / "etc" / "install-respeaker-drivers.sh"), + ] + ], + sudo_password=password, + ) + + if not success: + error("installing ReSpeaker drivers") + return + + msgbox( + "Driver installation complete. " + "Please reboot your Raspberry Pi and re-run the installer. " + 'Once rebooted, select the "seeed" microphone.' + ) + + sys.exit(0) + else: + break diff --git a/installer/microphone.py b/installer/microphone.py new file mode 100644 index 0000000..8f98062 --- /dev/null +++ b/installer/microphone.py @@ -0,0 +1,191 @@ +"""Microphone settings.""" +import array +import math +import subprocess +import logging +from concurrent.futures import ThreadPoolExecutor, Future +from typing import Dict, Optional, List + +from .const import Settings, RECORD_SECONDS, RECORD_RMS_MIN +from .whiptail import gauge, msgbox, radiolist, menu, inputbox + +_LOGGER = logging.getLogger() + + +def configure_microphone(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = microphone_menu(choice) + + if choice == "detect": + # Automatically detect microphone with hightest RMS + best_device: Optional[str] = None + best_rms: Optional[float] = None + + devices = get_microphone_devices() + with ThreadPoolExecutor() as executor: + futures: Dict[str, Future] = {} + for device in devices: + futures[device] = executor.submit(_record_proc, device) + + gauge("Speak loudly into the microphone.", RECORD_SECONDS) + for device, future in futures.items(): + device_rms = future.result() + if device_rms < RECORD_RMS_MIN: + continue + + if (best_rms is None) or (device_rms > best_rms): + best_device = device + best_rms = device_rms + + if best_device is not None: + msgbox(f"Successfully detected microphone: {best_device}") + settings.microphone_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, + ) + if microphone_device: + settings.microphone_device = microphone_device + settings.save() + elif choice == "manual": + microphone_device = inputbox( + "Enter ALSA Device:", settings.microphone_device + ) + if microphone_device: + settings.microphone_device = microphone_device + settings.save() + elif choice == "enhancements": + configure_audio_enhancements(settings) + else: + break + + +def microphone_menu(last_choice: Optional[str]) -> Optional[str]: + return menu( + "Main > Microphone", + [ + ("detect", "Autodetect"), + ("list", "Select From List"), + ("manual", "Enter Manually"), + ("enhancements", "Audio Enhancements"), + ], + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + +def get_microphone_devices() -> List[str]: + devices = [] + lines = subprocess.check_output(["arecord", "-L"]).decode("utf-8").splitlines() + for line in lines: + line = line.strip() + if line.startswith("plughw:"): + devices.append(line) + + return devices + + +def _record_proc(device: str) -> float: + try: + audio = subprocess.check_output( + [ + "arecord", + "-q", + "-D", + device, + "-r", + "16000", + "-c", + "1", + "-f", + "S16_LE", + "-t", + "raw", + "-d", + str(RECORD_SECONDS), + ], + stderr=subprocess.DEVNULL, + ) + + # 16-bit mono + audio_array = array.array("h", audio) + rms = math.sqrt( + (1 / len(audio_array) * sum(x * x for x in audio_array.tolist())) + ) + return rms + except Exception: + _LOGGER.exception("Error recording from device: %s", device) + + return 0 + + +def configure_audio_enhancements(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = audio_enhancements_menu(choice) + + if choice == "noise": + result = radiolist( + "Noise Suppression Level", + [(0, "Off"), (1, "Low"), (2, "Medium"), (3, "High"), (4, "Maximum")], + settings.noise_suppression_level, + ) + if result is not None: + settings.noise_suppression_level = result + 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 + + if 0 <= auto_gain <= 31: + settings.auto_gain = auto_gain + settings.save() + break + + msgbox("Must be 0-31") + elif choice == "multiplier": + while True: + result = 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 > 0: + settings.mic_volume_multiplier = volume_multiplier + settings.save() + break + + msgbox("Must be > 0") + else: + break + + +def audio_enhancements_menu(last_choice: Optional[str]) -> Optional[str]: + return menu( + "Main > Microphone > Audio Enhancements", + [ + ("noise", "Noise Suppression"), + ("gain", "Auto Gain"), + ("multiplier", "Volume Multiplier"), + ], + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) diff --git a/installer/satellite.py b/installer/satellite.py new file mode 100644 index 0000000..6c236e3 --- /dev/null +++ b/installer/satellite.py @@ -0,0 +1,53 @@ +"""Satellite settings.""" +from typing import Optional + +from .const import Settings, SatelliteType +from .whiptail import menu, inputbox, radiolist, run_with_gauge + + +def configure_satellite(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = satellite_menu(choice) + + if choice == "name": + set_satellite_name(settings) + elif choice == "type": + set_satellite_type(settings) + else: + break + + +def satellite_menu(last_choice: Optional[str]) -> Optional[str]: + return menu( + "Main > Satellite", + [ + ("name", "Satellite Name"), + ("type", "Satellite Type"), + ], + 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/speakers.py b/installer/speakers.py new file mode 100644 index 0000000..2d2b132 --- /dev/null +++ b/installer/speakers.py @@ -0,0 +1,120 @@ +"""Speaker settings.""" +import subprocess +import logging +from typing import Optional, List + +from .const import Settings, PROGRAM_DIR +from .whiptail import radiolist, inputbox, msgbox, checklist, menu + +_LOGGER = logging.getLogger() + + +def configure_speakers(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = speakers_menu(choice) + + if choice == "test": + sound_device = test_speakers() + if sound_device is not None: + settings.sound_device = sound_device + settings.save() + elif choice == "list": + sound_device = radiolist( + "Select ALSA Device:", + get_sound_devices(), + settings.sound_device, + ) + if sound_device: + settings.sound_device = sound_device + settings.save() + elif choice == "manual": + sound_device = inputbox("Enter ALSA Device:", settings.sound_device) + if sound_device: + settings.sound_device = sound_device + settings.save() + elif choice == "disable": + settings.sound_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, + ) + + if feedback_sounds is not None: + settings.feedback_sounds = feedback_sounds + settings.save() + else: + break + + +def speakers_menu(last_choice: Optional[str]) -> Optional[str]: + return menu( + "Main > Speakers", + [ + ("test", "Test Speakers"), + ("list", "Select From List"), + ("manual", "Enter Manually"), + ("disable", "Disable Sound"), + ("feedback", "Toggle Feedback Sounds"), + ], + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + +def get_sound_devices() -> List[str]: + devices = [] + lines = subprocess.check_output(["aplay", "-L"]).decode("utf-8").splitlines() + for line in lines: + line = line.strip() + if line.startswith("plughw:"): + devices.append(line) + + return devices + + +def test_speakers() -> Optional[str]: + devices = get_sound_devices() + if not devices: + msgbox("No speakers found") + return None + + device = devices.pop() + + while True: + choice = menu( + f"Device: {device}", + [ + ("play", "Play Sound"), + ("next", "Next Device"), + ("choose", "Choose This Device"), + ], + ) + + if choice == "play": + test_sound_device(device) + elif choice == "next": + if not devices: + msgbox("No more devices") + break + + device = devices.pop() + elif choice == "choose": + return device + else: + break + + return None + + +def test_sound_device(device: str) -> None: + try: + subprocess.check_call( + ["aplay", "-q", "-D", device, str(PROGRAM_DIR / "sounds" / "awake.wav")] + ) + except Exception: + _LOGGER.exception("Error testing device: %s", device) diff --git a/installer/wake_word.py b/installer/wake_word.py new file mode 100644 index 0000000..1971446 --- /dev/null +++ b/installer/wake_word.py @@ -0,0 +1,308 @@ +"""Wake word settings.""" +import itertools +import re +import shutil +from pathlib import Path +from typing import Optional, Dict + +from .const import Settings, SatelliteType, WakeWordSystem, LOCAL_DIR +from .whiptail import msgbox, menu, radiolist, run_with_gauge, yesno, error + + +def configure_wake_word(settings: Settings) -> None: + if settings.satellite_type != SatelliteType.WAKE: + if not yesno("Set satellite type to local wake word?"): + return + + settings.satellite_type = SatelliteType.WAKE + settings.save() + + choice: Optional[str] = None + while True: + choice = menu( + "Main > Wake Word", + [("system", "Wake Word System"), ("wake_word", "Wake Word")], + selected_item=choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + if choice == "system": + wake_word_system = radiolist( + "Wake Word System:", + [v.value for v in WakeWordSystem], + settings.wake_word_system, + ) + if wake_word_system is None: + continue + + wake_word_system = WakeWordSystem(wake_word_system) + install_wake_word(settings, wake_word_system) + elif choice == "wake_word": + select_wake_word(settings) + else: + break + + +def install_wake_word(settings: Settings, wake_word_system: WakeWordSystem) -> None: + if wake_word_system == WakeWordSystem.OPENWAKEWORD: + oww_dir = LOCAL_DIR / "wyoming-openwakeword" + if not oww_dir.exists(): + if not yesno("Install openWakeWord?"): + return + + success = run_with_gauge( + "Installing openWakeWord", + [ + [ + "git", + "clone", + "https://github.com/rhasspy/wyoming-openwakeword.git", + str(oww_dir), + ], + [str(oww_dir / "script" / "setup")], + ], + ) + + if not success: + # Clean up + try: + shutil.rmtree(oww_dir) + except Exception: + pass + + error("installing openWakeWord") + return + + msgbox("openWakeWord installed successfully") + + settings.wake_word_system = wake_word_system + settings.wake_word.setdefault(WakeWordSystem.OPENWAKEWORD, "ok_nabu") + settings.save() + return + + if wake_word_system == WakeWordSystem.PORCUPINE1: + porcupine1_dir = LOCAL_DIR / "wyoming-porcupine1" + if not porcupine1_dir.exists(): + if not yesno("Install porcupine1?"): + return + + success = run_with_gauge( + "Installing porcupine1", + [ + [ + "git", + "clone", + "https://github.com/rhasspy/wyoming-porcupine1.git", + str(porcupine1_dir), + ], + [str(porcupine1_dir / "script" / "setup")], + ], + ) + + if not success: + # Clean up + try: + shutil.rmtree(porcupine1_dir) + except Exception: + pass + + error("installing porcupine1") + else: + msgbox("porcupine1 installed successfully") + + settings.wake_word_system = wake_word_system + settings.save() + return + + if wake_word_system == WakeWordSystem.SNOWBOY: + snowboy_dir = LOCAL_DIR / "wyoming-snowboy" + if not snowboy_dir.exists(): + if not yesno("Install snowboy?"): + return + + success = run_with_gauge( + "Installing snowboy", + [ + [ + "git", + "clone", + "https://github.com/rhasspy/wyoming-snowboy.git", + str(snowboy_dir), + ], + [str(snowboy_dir / "script" / "setup")], + ], + ) + + if not success: + # Clean up + try: + shutil.rmtree(snowboy_dir) + except Exception: + pass + + error("installing snowboy") + else: + msgbox("snowboy installed successfully") + + settings.wake_word_system = wake_word_system + settings.save() + return + + +def select_wake_word(settings: Settings) -> None: + if settings.wake_word_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.mkdir(parents=True, exist_ok=True) + + community_wake_word_dir = LOCAL_DIR / "home-assistant-wakewords-collection" + + while True: + ww_paths: Dict[str, Path] = { + p.stem: p for p in custom_wake_word_dir.glob("*.tflite") + } + + for ww_path in community_wake_word_dir.rglob("*.tflite"): + ww_name = ww_path.stem + if ww_name in ww_paths: + continue + + ww_paths[ww_name] = ww_path + + for ww_path in (oww_dir / "wyoming_openwakeword" / "models").glob( + "*.tflite" + ): + if not re.match("^.+_v[0-9].*$", ww_path.stem): + continue + + ww_name = ww_path.stem.rsplit("_", maxsplit=1)[0] + if ww_name in ww_paths: + continue + + ww_paths[ww_name] = ww_path + + items = [("__community__", "Download Community Wake Words")] + list( + sorted(ww_paths.keys()) + ) + + wake_word = radiolist( + "Wake Word:", items, settings.wake_word.get(WakeWordSystem.OPENWAKEWORD) + ) + + 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.save() + break + + return + + if settings.wake_word_system == WakeWordSystem.PORCUPINE1: + porcupine1_dir = LOCAL_DIR / "wyoming-porcupine1" + if not porcupine1_dir.exists(): + msgbox("porcupine1 is not installed") + return + + ww_names = sorted( + list( + set( + p.stem.rsplit("_", maxsplit=1)[0] + for p in ( + porcupine1_dir / "wyoming_porcupine1" / "data" / "resources" + ).rglob("*.ppn") + ) + ) + ) + + wake_word = radiolist( + "Wake Word:", + ww_names, + settings.wake_word.get(WakeWordSystem.PORCUPINE1), + ) + + if wake_word is not None: + settings.wake_word[WakeWordSystem.PORCUPINE1] = wake_word + settings.save() + + return + + if settings.wake_word_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.mkdir(parents=True, exist_ok=True) + + builtin_wake_words = (snowboy_dir / "wyoming_snowboy" / "data").glob("*.umdl") + custom_wake_words = itertools.chain( + custom_wake_word_dir.glob("*.pmdl"), custom_wake_word_dir.glob("*.umdl") + ) + ww_names = sorted( + list( + set( + p.stem + for p in itertools.chain(builtin_wake_words, custom_wake_words) + ) + ) + ) + + wake_word = radiolist( + "Wake Word:", ww_names, settings.wake_word.get(WakeWordSystem.SNOWBOY) + ) + + if wake_word is not None: + settings.wake_word[WakeWordSystem.SNOWBOY] = wake_word + settings.save() + + return diff --git a/installer/whiptail.py b/installer/whiptail.py new file mode 100644 index 0000000..1635a50 --- /dev/null +++ b/installer/whiptail.py @@ -0,0 +1,249 @@ +"""Python interface to whiptail command.""" +import subprocess +import time +import logging +import shlex +from concurrent.futures import ThreadPoolExecutor +from collections.abc import Sequence +from typing import Any, Dict, List, Tuple, Optional, Union + +from .const import TITLE, WIDTH, HEIGHT, LIST_HEIGHT + +ItemType = Union[str, Tuple[Any, str]] + +_LOGGER = logging.getLogger() + + +def whiptail(*args) -> Optional[str]: + proc = subprocess.Popen( + ["whiptail", "--title", TITLE] + list(args), + stderr=subprocess.PIPE, + ) + + _stdout, stderr = proc.communicate() + if proc.returncode != 0: + return None + + return stderr.decode("utf-8") + + +def menu( + text: str, + items: Sequence[ItemType], + selected_item: Optional[str] = None, + menu_args: Optional[Sequence[str]] = None, +) -> Optional[str]: + assert items, "No items" + + item_map: Dict[Optional[str], str] = {} + item_args: List[str] = [] + selected_tag: Optional[str] = None + for i, item in enumerate(items): + item_id = str(i) + item_args.append(item_id) + + if isinstance(item, str): + item_map[item_id] = item + item_args.append(item) + + if selected_item == item: + selected_tag = item_id + else: + item_map[item_id] = item[0] + item_args.append(item[1]) + + if selected_item == item[0]: + selected_tag = item_id + + menu_args = list(menu_args) if menu_args is not None else [] + if selected_tag is not None: + menu_args.extend(["--default-item", selected_tag]) + + result = whiptail( + "--notags", *menu_args, "--menu", text, HEIGHT, WIDTH, LIST_HEIGHT, *item_args + ) + return item_map.get(result) + + +def inputbox(text: str, init: Optional[Any] = None) -> Optional[str]: + return whiptail("--inputbox", text, HEIGHT, WIDTH, str(init) or "") + + +def passwordbox(text: str) -> Optional[str]: + return whiptail("--passwordbox", text, HEIGHT, WIDTH) + + +def radiolist( + text: str, + items: Sequence[ItemType], + selected_item: Any, + *args, +) -> Optional[Any]: + assert items, "No items" + + item_map: Dict[str, ItemType] = {} + item_args: List[str] = [] + for i, item in enumerate(items): + item_id = str(i) + item_args.append(item_id) + + if isinstance(item, str): + item_map[item_id] = item + item_args.append(item) + + if item == selected_item: + item_args.append("1") + else: + item_args.append("0") + else: + item_map[item_id] = item[0] + item_args.append(item[1]) + + if item[0] == selected_item: + item_args.append("1") + else: + item_args.append("0") + + result = whiptail( + "--notags", *args, "--radiolist", text, HEIGHT, WIDTH, LIST_HEIGHT, *item_args + ) + + if result is None: + return None + + return item_map.get(result, result) + + +def checklist( + text: str, + items: Sequence[ItemType], + selected_items: Sequence[Any], + *args, +) -> Optional[List[Any]]: + assert items, "No items" + + item_map: Dict[str, ItemType] = {} + item_args: List[str] = [] + for i, item in enumerate(items): + item_id = str(i) + item_args.append(item_id) + + if isinstance(item, str): + item_map[item_id] = item + item_args.append(item) + + if item in selected_items: + item_args.append("1") + else: + item_args.append("0") + else: + item_map[item_id] = item[0] + item_args.append(item[1]) + + if item[0] in selected_items: + item_args.append("1") + else: + item_args.append("0") + + result = whiptail( + "--notags", *args, "--checklist", text, HEIGHT, WIDTH, LIST_HEIGHT, *item_args + ) + + if result is None: + return None + + return [ + item_map.get(result_item, result_item) for result_item in shlex.split(result) + ] + + +def yesno(text: str) -> bool: + return whiptail("--yesno", text, HEIGHT, WIDTH) is not None + + +def msgbox(text: str) -> None: + whiptail("--msgbox", text, HEIGHT, WIDTH) + + +def gauge(text: str, seconds: int, parts: int = 20) -> None: + proc = subprocess.Popen( + ["whiptail", "--title", TITLE, "--gauge", text, HEIGHT, WIDTH, "0"], + stdin=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + assert proc.stdin is not None + + percent = 0 + while percent <= 100: + time.sleep(seconds / parts) + percent += int(100 / parts) + print(percent, file=proc.stdin, flush=True) + + proc.communicate() + + +def error(reason: str) -> None: + msgbox(f"An error occurred while {reason}.\n" "See local/installer.log for details.") + + +# ----------------------------------------------------------------------------- + + +def run_with_gauge( + text: str, commands: Sequence[Sequence[str]], sudo_password: Optional[str] = None +) -> bool: + proc = subprocess.Popen( + ["whiptail", "--title", TITLE, "--gauge", text, HEIGHT, WIDTH, "0"], + stdin=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + assert proc.stdin is not None + percent = 0 + seconds = 5 + parts = 20 + + with ThreadPoolExecutor() as executor: + for command in commands: + future = executor.submit(_run_command, command, sudo_password) + while not future.done(): + time.sleep(seconds / parts) + percent += int(100 / parts) + if percent > 100: + percent = 0 + + print(percent, file=proc.stdin, flush=True) + + if not future.result(): + # Error occurred + return False + + proc.communicate() + return True + + +def _run_command(command: Sequence[str], sudo_password: Optional[str] = None) -> bool: + try: + assert command + proc_input: Optional[str] = None + if (command[0] == "sudo") and (sudo_password is not None): + proc_input = sudo_password + + proc = subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + _stdout, stderr = proc.communicate(proc_input) + if proc.returncode != 0: + _LOGGER.error("Error running command: %s", command) + _LOGGER.error(stderr) + return False + except Exception: + _LOGGER.exception("Error running command: %s", command) + return False + + return True diff --git a/script/_installer.py b/script/_installer.py index 26ec882..e83fca2 100755 --- a/script/_installer.py +++ b/script/_installer.py @@ -3,12 +3,12 @@ import array import math import logging import json -import functools 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 @@ -29,7 +29,7 @@ WIDTH = "75" HEIGHT = "20" LIST_HEIGHT = "12" -ItemType = Union[str, Tuple[str, str]] +ItemType = Union[str, Tuple[Any, str]] class SatelliteType(str, Enum): @@ -47,6 +47,9 @@ 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 sound_device: Optional[str] = None feedback_sounds: List[str] = field(default_factory=list) @@ -57,6 +60,8 @@ class Settings: satellite_name: str = "Wyoming Satellite" satellite_type: SatelliteType = SatelliteType.ALWAYS_STREAMING + debug_enabled: bool = False + @staticmethod def load() -> "Settings": kwargs: Dict[str, Any] = {} @@ -100,65 +105,121 @@ def main() -> None: "Installing satellite base...", [[str(_PROGRAM_DIR / "script" / "setup")]] ) + choice: Optional[str] = None while True: - choice = main_menu(settings) + choice = main_menu(choice) - if choice == "mic": + if choice == "satellite": + configure_satellite(settings) + elif choice == "microphone": configure_microphone(settings) - elif choice == "snd": - configure_sound(settings) + elif choice == "speakers": + configure_speakers(settings) elif choice == "wake": configure_wake_word(settings) - elif choice == "name": - set_satellite_name(settings) - elif choice == "type": - set_satellite_type(settings) - elif choice == "generate": - generate_services(settings) - elif choice == "install": - install_services(settings) + elif choice == "services": + configure_services(settings) + elif choice == "drivers": + install_drivers(settings) else: break +def main_menu(last_choice: Optional[str]) -> Optional[str]: + items: List[ItemType] = [ + ("satellite", "Satellite"), + ("microphone", "Microphone"), + ("speakers", "Speakers"), + ("wake", "Wake Word"), + ("services", "Services"), + ("drivers", "Drivers"), + ] + + return menu( + "Main", + items, + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Exit"], + ) + + +# ----------------------------------------------------------------------------- +# Satellite # ----------------------------------------------------------------------------- -def main_menu(settings: Settings) -> Optional[str]: - items = [("mic", "Configure Microphone"), ("snd", "Configure Sound")] +def configure_satellite(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = satellite_menu(choice) - if settings.satellite_type == SatelliteType.WAKE: - items.append(("wake", "Configure Wake Word")) + if choice == "name": + set_satellite_name(settings) + elif choice == "type": + set_satellite_type(settings) + else: + break - items.extend( - [ - ("name", "Set Satellite Name"), - ("type", "Set Satellite Type"), - ("generate", "Generate Services"), - ("install", "Install Services"), - ] - ) +def satellite_menu(last_choice: Optional[str]) -> Optional[str]: return menu( - "Main Menu", - items, - "--ok-button", - "Select", - "--cancel-button", - "Exit", + "Main > Satellite", + [ + ("name", "Name"), + ("type", "Type"), + ], + 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 == SatelliteType.VAD: + result = run_with_gauge( + "Installing vad...", + [pip_install("-r", str(_PROGRAM_DIR / "requirements_vad.txt"))], + ) + if not result: + msgbox( + "An error occurred while installed vad. " + "See local/installer.log for details." + ) + return + + if satellite_type is not None: + settings.satellite_type = satellite_type + settings.save() + + # ----------------------------------------------------------------------------- # Microphone # ----------------------------------------------------------------------------- def configure_microphone(settings: Settings) -> None: + choice: Optional[str] = None while True: - choice = microphone_menu() + choice = microphone_menu(choice) if choice == "detect": + # Automatically detect microphone with hightest RMS best_device: Optional[str] = None best_rms: Optional[float] = None @@ -185,6 +246,7 @@ def configure_microphone(settings: Settings) -> None: else: msgbox("Audio was not detected from any microphone") elif choice == "list": + # arecord -L microphone_device = radiolist( "Select ALSA Device:", get_microphone_devices(), @@ -200,49 +262,49 @@ def configure_microphone(settings: Settings) -> None: if microphone_device: settings.microphone_device = microphone_device settings.save() - elif choice == "respeaker": - if yesno( - "ReSpeaker drivers for the Raspberry Pi will now be compiled and installed. " - "This will take a while and require a reboot. " - "Continue?" - ): - password = paswordbox("sudo password:") - run_with_gauge( - "Installing drivers...", - [ - [ - "sudo", - "-S", - str(_PROGRAM_DIR / "etc" / "install-respeaker-drivers.sh"), - ] - ], - sudo_password=password, - ) + elif choice == "enhancements": + configure_audio_enhancements(settings) + # elif choice == "respeaker": + # if yesno( + # "ReSpeaker drivers for the Raspberry Pi will now be compiled and installed. " + # "This will take a while and require a reboot. " + # "Continue?" + # ): + # password = paswordbox("sudo password:") + # run_with_gauge( + # "Installing drivers...", + # [ + # [ + # "sudo", + # "-S", + # str(_PROGRAM_DIR / "etc" / "install-respeaker-drivers.sh"), + # ] + # ], + # sudo_password=password, + # ) - msgbox( - "Driver installation complete. " - "Please reboot your Raspberry Pi and re-run the installer. " - 'Once rebooted, select the "seeed" microphone.' - ) + # msgbox( + # "Driver installation complete. " + # "Please reboot your Raspberry Pi and re-run the installer. " + # 'Once rebooted, select the "seeed" microphone.' + # ) - sys.exit(0) + # sys.exit(0) else: break -def microphone_menu() -> Optional[str]: +def microphone_menu(last_choice: Optional[str]) -> Optional[str]: return menu( - "Configure Microphone", + "Main > Microphone", [ ("detect", "Autodetect"), ("list", "Select From List"), ("manual", "Enter Manually"), - ("respeaker", "Install ReSpeaker Drivers"), + ("enhancements", "Audio Enhancements"), ], - "--ok-button", - "Select", - "--cancel-button", - "Back", + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], ) @@ -291,6 +353,69 @@ def _record_proc(device: str) -> float: return 0 +def configure_audio_enhancements(settings: Settings) -> None: + choice: Optional[str] = None + while True: + choice = audio_enhancements_menu(choice) + + if choice == "noise": + result = radiolist( + "Noise Suppression Level", + [(0, "Off"), (1, "Low"), (2, "Medium"), (3, "High"), (4, "Maximum")], + settings.noise_suppression_level, + ) + if result is not None: + settings.noise_suppression_level = result + 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 + + if 0 <= auto_gain <= 31: + settings.auto_gain = auto_gain + settings.save() + break + + msgbox("Must be 0-31") + elif choice == "multiplier": + while True: + result = 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 > 0: + settings.mic_volume_multiplier = volume_multiplier + settings.save() + break + + msgbox("Must be > 0") + else: + break + + +def audio_enhancements_menu(last_choice: Optional[str]) -> Optional[str]: + return menu( + "Main > Microphone > Audio Enhancements", + [ + ("noise", "Noise Suppression"), + ("gain", "Auto Gain"), + ("multiplier", "Volume Multiplier"), + ], + selected_item=last_choice, + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], + ) + + # ----------------------------------------------------------------------------- # Sound # ----------------------------------------------------------------------------- @@ -344,10 +469,7 @@ def sound_menu() -> Optional[str]: ("disable", "Disable Sound"), ("feedback", "Toggle Feedback Sounds"), ], - "--ok-button", - "Select", - "--cancel-button", - "Back", + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], ) @@ -368,6 +490,10 @@ def get_sound_devices() -> List[str]: def configure_wake_word(settings: Settings) -> None: + if settings.satellite_type != SatelliteType.WAKE: + msgbox("Satellite type is not set to local wake word.") + return + while True: choice = wake_word_menu(settings) @@ -378,7 +504,7 @@ def configure_wake_word(settings: Settings) -> None: settings.wake_word_system, ) if wake_word_system is not None: - install_wake_word(settings, wake_word_system) + install_wake_word(settings, WakeWordSystem(wake_word_system)) elif choice == "wake_word": select_wake_word(settings) else: @@ -393,10 +519,7 @@ def wake_word_menu(settings: Settings) -> Optional[str]: return menu( "Configure Wake Word", items, - "--ok-button", - "Select", - "--cancel-button", - "Back", + menu_args=["--ok-button", "Select", "--cancel-button", "Back"], ) @@ -574,50 +697,6 @@ def select_wake_word(settings: Settings) -> None: settings.save() -# ----------------------------------------------------------------------------- -# Satellite -# ----------------------------------------------------------------------------- - - -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 == SatelliteType.VAD: - result = run_with_gauge( - "Installing vad...", - [ - pip_install( - "-r", str(_PROGRAM_DIR / "requirements_vad.txt") - ) - ], - ) - if not result: - msgbox( - "An error occurred while installed vad. " - "See local/installer.log for details." - ) - return - - if satellite_type is not None: - settings.satellite_type = satellite_type - settings.save() - - # ----------------------------------------------------------------------------- # Services # ----------------------------------------------------------------------------- @@ -632,9 +711,8 @@ def generate_services(settings: Settings) -> None: services_dir.mkdir(parents=True, exist_ok=True) user = subprocess.check_output(["id", "--name", "-u"], text=True).strip() - group = subprocess.check_output(["id", "--name", "-g"], text=True).strip() - satellite_command = [ + satellite_command: List[str] = [ str(_PROGRAM_DIR / "script" / "run"), "--name", settings.satellite_name, @@ -643,9 +721,75 @@ def generate_services(settings: Settings) -> None: "--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.append( + f"aplay -D {settings.sound_device} -q -r 16000 -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.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) @@ -656,11 +800,13 @@ def generate_services(settings: Settings) -> None: 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"Group={group}", 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) @@ -745,11 +891,17 @@ def whiptail(*args) -> Optional[str]: return stderr.decode("utf-8") -def menu(text: str, items: List[ItemType], *args) -> Optional[str]: +def menu( + text: str, + items: Sequence[ItemType], + selected_item: Optional[str] = None, + menu_args: Optional[Sequence[str]] = None, +) -> Optional[str]: assert items, "No items" - item_map: Dict[str, ItemType] = {} + item_map: Dict[Optional[str], str] = {} item_args: List[str] = [] + selected_tag: Optional[str] = None for i, item in enumerate(items): item_id = str(i) item_args.append(item_id) @@ -757,18 +909,28 @@ def menu(text: str, items: List[ItemType], *args) -> Optional[str]: if isinstance(item, str): item_map[item_id] = item item_args.append(item) + + if selected_item == item: + selected_tag = item_id else: item_map[item_id] = item[0] item_args.append(item[1]) + if selected_item == item[0]: + selected_tag = item_id + + menu_args = menu_args or [] + if selected_tag is not None: + menu_args.extend(["--default-item", selected_tag]) + result = whiptail( - "--notags", *args, "--menu", text, HEIGHT, WIDTH, LIST_HEIGHT, *item_args + "--notags", *menu_args, "--menu", text, HEIGHT, WIDTH, LIST_HEIGHT, *item_args ) return item_map.get(result) -def inputbox(text: str, init: Optional[str] = None) -> Optional[str]: - return whiptail("--inputbox", text, HEIGHT, WIDTH, init or "") +def inputbox(text: str, init: Optional[Any] = None) -> Optional[str]: + return whiptail("--inputbox", text, HEIGHT, WIDTH, str(init) or "") def paswordbox(text: str) -> Optional[str]: @@ -777,7 +939,7 @@ def paswordbox(text: str) -> Optional[str]: def radiolist( text: str, - items: List[ItemType], + items: Sequence[ItemType], selected_item: Any, *args, ) -> Optional[str]: @@ -818,8 +980,8 @@ def radiolist( def checklist( text: str, - items: List[ItemType], - selected_items: List[Any], + items: Sequence[ItemType], + selected_items: Sequence[Any], *args, ) -> List[str]: assert items, "No items" @@ -886,7 +1048,7 @@ def gauge(text: str, seconds: int, parts: int = 20) -> None: def run_with_gauge( - text: str, commands: List[List[str]], sudo_password: Optional[str] = None + text: str, commands: Sequence[Sequence[str]], sudo_password: Optional[str] = None ) -> bool: proc = subprocess.Popen( ["whiptail", "--title", TITLE, "--gauge", text, HEIGHT, WIDTH, "0"], @@ -918,7 +1080,7 @@ def run_with_gauge( return True -def _run_command(command: List[str], sudo_password: Optional[str] = None) -> bool: +def _run_command(command: Sequence[str], sudo_password: Optional[str] = None) -> bool: try: assert command proc_input: Optional[str] = None