From 49df03426f00085b8a0c1503243ba6adf5f97df9 Mon Sep 17 00:00:00 2001 From: Michael Hansen Date: Mon, 15 Jan 2024 11:15:08 -0600 Subject: [PATCH] Clean up --- script/_installer.py | 1124 ------------------------------------------ script/installer.sh | 482 ------------------ 2 files changed, 1606 deletions(-) delete mode 100755 script/_installer.py delete mode 100755 script/installer.sh diff --git a/script/_installer.py b/script/_installer.py deleted file mode 100755 index e83fca2..0000000 --- a/script/_installer.py +++ /dev/null @@ -1,1124 +0,0 @@ -#!/usr/bin/env python3 -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 - -_DIR = Path(__file__).parent -_PROGRAM_DIR = _DIR.parent -_LOCAL_DIR = _PROGRAM_DIR / "local" -_SETTINGS_PATH = _LOCAL_DIR / "settings.json" -_LOGGER = logging.getLogger() - -_RECORD_SECONDS = 8 -_RECORD_RMS_MIN = 30 - -TITLE = "Wyoming Satellite" -WIDTH = "75" -HEIGHT = "20" -LIST_HEIGHT = "12" - -ItemType = Union[str, Tuple[Any, str]] - - -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=dict) - - 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) - - -# ----------------------------------------------------------------------------- - - -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 == "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 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", "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) - - 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) - # 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.' - # ) - - # sys.exit(0) - 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"], - ) - - -# ----------------------------------------------------------------------------- -# Sound -# ----------------------------------------------------------------------------- - - -def configure_sound(settings: Settings) -> None: - while True: - choice = sound_menu() - - if choice == "detect": - pass - 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 sound_menu() -> Optional[str]: - return menu( - "Configure Sound", - [ - ("detect", "Autodetect"), - ("list", "Select From List"), - ("manual", "Enter Manually"), - ("disable", "Disable Sound"), - ("feedback", "Toggle Feedback Sounds"), - ], - 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 - - -# ----------------------------------------------------------------------------- -# Wake Word -# ----------------------------------------------------------------------------- - - -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) - - 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 not None: - install_wake_word(settings, WakeWordSystem(wake_word_system)) - elif choice == "wake_word": - select_wake_word(settings) - else: - break - - -def wake_word_menu(settings: Settings) -> Optional[str]: - items = [("system", "Select System")] - if settings.wake_word_system is not None: - items.append(("wake_word", "Select Wake Word")) - - return menu( - "Configure Wake Word", - items, - menu_args=["--ok-button", "Select", "--cancel-button", "Back"], - ) - - -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(): - success = run_with_gauge( - "Installing openWakeWord", - [ - [ - "git", - "clone", - "https://github.com/rhasspy/wyoming-openwakeword.git", - str(oww_dir), - ], - [str(oww_dir / "script" / "setup")], - ], - ) - - if success: - msgbox("openWakeWord installed successfully") - else: - msgbox( - "An error occurred while installing openWakeWord. " - "See local/installer.log for details." - ) - - try: - shutil.rmtree(oww_dir) - except Exception: - pass - - return - - settings.wake_word_system = wake_word_system - settings.wake_word.setdefault(WakeWordSystem.OPENWAKEWORD, "ok_nabu") - settings.save() - elif wake_word_system == WakeWordSystem.PORCUPINE1: - porcupine1_dir = _LOCAL_DIR / "wyoming-porcupine1" - if not porcupine1_dir.exists(): - success = run_with_gauge( - "Installing porcupine1", - [ - [ - "git", - "clone", - "https://github.com/rhasspy/wyoming-porcupine1.git", - str(porcupine1_dir), - ], - [str(porcupine1_dir / "script" / "setup")], - ], - ) - - if success: - msgbox("porcupine1 installed successfully") - else: - msgbox( - "An error occurred while installing porcupine1. " - "See local/installer.log for details." - ) - - try: - shutil.rmtree(porcupine1_dir) - except Exception: - pass - - return - - settings.wake_word_system = wake_word_system - settings.wake_word.setdefault(WakeWordSystem.PORCUPINE1, "porcupine") - settings.save() - elif settings.wake_word_system == WakeWordSystem.SNOWBOY: - snowboy_dir = _LOCAL_DIR / "wyoming-snowboy" - if not snowboy_dir.exists(): - success = run_with_gauge( - "Installing snowboy", - [ - [ - "git", - "clone", - "https://github.com/rhasspy/wyoming-snowboy.git", - str(snowboy_dir), - ], - [str(snowboy_dir / "script" / "setup")], - ], - ) - - if success: - msgbox("snowboy installed successfully") - else: - msgbox( - "An error occurred while installing snowboy. " - "See local/installer.log for details." - ) - - try: - shutil.rmtree(snowboy_dir) - except Exception: - pass - - return - - settings.wake_word_system = wake_word_system - settings.wake_word.setdefault(WakeWordSystem.SNOWBOY, "snowboy") - settings.save() - - -def select_wake_word(settings: Settings) -> None: - if settings.wake_word_system == WakeWordSystem.OPENWAKEWORD: - wake_word = radiolist( - "Wake Word:", - [ - ("ok_nabu", "ok nabu"), - ("hey_jarvis", "hey jarvis"), - ("alexa", "alexa"), - ("hey_mycroft", "hey mycroft"), - ("community", "Community Wake Words"), - ], - settings.wake_word.get(WakeWordSystem.OPENWAKEWORD), - ) - - if wake_word is not None: - settings.wake_word[WakeWordSystem.OPENWAKEWORD] = wake_word - settings.save() - - elif settings.wake_word_system == WakeWordSystem.PORCUPINE1: - wake_word = radiolist( - "Wake Word:", - [ - ("ok_nabu", "ok nabu"), - ("hey_jarvis", "hey jarvis"), - ("alexa", "alexa"), - ("hey_mycroft", "hey mycroft"), - ("community", "Community Wake Words"), - ], - settings.wake_word.get(WakeWordSystem.OPENWAKEWORD), - ) - - if wake_word is not None: - settings.wake_word[WakeWordSystem.OPENWAKEWORD] = wake_word - settings.save() - - elif settings.wake_word_system == WakeWordSystem.PORCUPINE1: - wake_word = radiolist( - "Wake Word:", - [ - ("porcupine", "porcupine"), - ], - settings.wake_word.get(WakeWordSystem.PORCUPINE1), - ) - - if wake_word is not None: - settings.wake_word[WakeWordSystem.PORCUPINE1] = wake_word - settings.save() - - elif settings.wake_word_system == WakeWordSystem.SNOWBOY: - wake_word = radiolist( - "Wake Word:", - [ - ("snowboy", "snowboy"), - ("jarvis", "jarvis"), - ("alexa", "alexa"), - ("smart_mirror", "smart mirror"), - ("view_glass", "view glass"), - ("hey_extreme", "hey extreme"), - ("neoya", "neoya"), - ("subex", "subex"), - ], - settings.wake_word.get(WakeWordSystem.SNOWBOY), - ) - - if wake_word is not None: - settings.wake_word[WakeWordSystem.SNOWBOY] = wake_word - settings.save() - - -# ----------------------------------------------------------------------------- -# Services -# ----------------------------------------------------------------------------- - - -def generate_services(settings: Settings) -> None: - if settings.microphone_device is None: - msgbox("Please configure microphone") - return - - services_dir = _LOCAL_DIR / "services" - 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.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) - - 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) - - msgbox("Services generated") - - -def install_services(settings: Settings) -> None: - services_dir = _LOCAL_DIR / "services" - satellite_service_path = services_dir / "wyoming-satellite.service" - - if not satellite_service_path.exists(): - msgbox("Please generate services") - return - - password = paswordbox("sudo password:") - - run_with_gauge( - "Stopping Services...", - [ - [ - "sudo", - "-S", - "systemctl", - "disable", - "--now", - f"wyoming-{service}.service", - ] - for service in ("satellite", "openwakeword", "porcupine1", "snowboy") - ], - sudo_password=password, - ) - - installed_services = ["satellite"] - install_commands = [["sudo", "-S", "systemctl", "daemon-reload"]] - 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", "enable", "--now", service_filename] - ) - - success = run_with_gauge( - "Installing Services...", install_commands, sudo_password=password - ) - if success: - msgbox("Successfully installed services") - else: - msgbox( - "An error occurred while installing services. " - "See local/installer.log for details." - ) - - -# ----------------------------------------------------------------------------- -# whiptail -# ----------------------------------------------------------------------------- - - -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 = menu_args or [] - 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 paswordbox(text: str) -> Optional[str]: - return whiptail("--passwordbox", text, HEIGHT, WIDTH) - - -def radiolist( - text: str, - items: Sequence[ItemType], - selected_item: Any, - *args, -) -> Optional[str]: - 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, -) -> List[str]: - 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 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 - - -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) - - -# ----------------------------------------------------------------------------- - - -if __name__ == "__main__": - main() diff --git a/script/installer.sh b/script/installer.sh deleted file mode 100755 index 1fabe58..0000000 --- a/script/installer.sh +++ /dev/null @@ -1,482 +0,0 @@ -#!/usr/bin/env bash -set -eu - -title='Wyoming Satellite' - -# Directory of *this* script -this_dir="$( cd "$( dirname "$0" )" && pwd )" -program_dir="$(realpath "${this_dir}/..")" - -# Directory to store stuff that won't interfere with git -local_dir="${program_dir}/local" -mkdir -p "${local_dir}" - -debug_enabled='' -system_deps='' - -satellite_name='' -satellite_type='' - -wake_word_system='' -wake_word='' -custom_wake_word_dir="${local_dir}/custom-wake-words" -mkdir -p "${custom_wake_word_dir}" - -noise_suppression='' - -microphone_command='arecord -q -r 16000 -c 1 -f S16_LE -t raw' -speaker_command='aplay -q -r 22050 -c 1 -f S16_LE -t raw' -speaker_sounds=() - -pushd () { - command pushd "$@" > /dev/null -} - -popd () { - command popd "$@" > /dev/null -} - -# Args -for arg in "$@"; do - if [ "${arg}" = '--debug' ]; then - debug_enabled='1' - fi -done - -if [ -n "${debug_enabled}" ]; then - whiptail --title "${title}" --msgbox \ - 'Debug mode is enabled. This will cause all services to log verbosely, and audio to be recorded in local/debug-recording. Make sure to re-run the installer without debug mode before deploying the satellite.' \ - 10 75 -fi - -# ----------------------------------------------------------------------------- -# System dependencies -# ----------------------------------------------------------------------------- - -commands=('whiptail' 'git' 'python3' 'pip3') -for command in "${commands[@]}"; -do - if ! command -v "${command}" &> /dev/null; then - system_deps='1' - break - fi -done - -if command -v python3 &> /dev/null; then - if ! python3 -c 'import venv'; then - system_deps='1' - fi -fi - -if [ -n "${system_deps}" ]; then - echo 'Installing system dependencies' - sudo apt-get update - sudo apt-get install --yes --no-install-recommends \ - whiptail python3-venv python3-pip git -fi - -# ----------------------------------------------------------------------------- -# Microphone -# ----------------------------------------------------------------------------- - -function get_microphone { - microphones=() - while read -r microphone; do - microphones+=("${microphone}" "${microphone}" 0) - done < <(arecord -L | grep 'plughw:') - - whiptail --title "${title}" --notags \ - --radiolist 'Select Microphone:' 20 70 \ - 12 "${microphones[@]}" \ - 'custom' 'Custom ALSA Device' 0 \ - 'respeaker' 'Install ReSpeaker Drivers' 0 -} - -function get_custom_microphone { - whiptail --title "${title}" --inputbox \ - 'ALSA Input Device Name:' \ - 8 75 -} - -microphone='' -while [ -z "${microphone}" ]; do - microphone="$(get_microphone 3>&1 1>&2 2>&3)" -done - -if [ "${microphone}" = 'custom' ]; then - # Custom ALSA device name from "arecord -L" - microphone='' - while [ -z "${microphone}" ]; do - microphone="$(get_custom_microphone 3>&1 1>&2 2>&3)" - done -elif [ "${microphone}" = 'respeaker' ]; then - # ReSpeaker driver installation (requires reboot) - whiptail --title "${title}" --yesno \ - 'ReSpeaker drivers for the Raspberry Pi will now be compiled and installed. This will take a while and require a reboot. Continue?' \ - 8 75 - - echo 'Installing ReSpeaker drivers' - "${program_dir}/etc/install-respeaker-drivers.sh" - - whiptail --title "${title}" --msgbox \ - 'Driver installation complete. The system will now reboot. Re-run the installer and select the "seeed" microphone.' \ - 8 75 - - sudo reboot -fi - -# Add device -microphone_command="${microphone_command} -D ${microphone}" - -# ----------------------------------------------------------------------------- -# Speaker -# ----------------------------------------------------------------------------- - -function get_speaker { - speakers=() - while read -r speaker; do - speakers+=("${speaker}" "${speaker}" 0) - done < <(aplay -L | grep 'plughw:') - - whiptail --title "${title}" --notags \ - --radiolist 'Select Speaker:' 20 70 \ - 12 "${speakers[@]}" \ - 'custom' 'Custom ALSA Device' 0 \ - 'no_speaker' 'No Audio Output' 0 -} - -function get_custom_speaker { - whiptail --title "${title}" --inputbox \ - 'ALSA Output Device Name:' \ - 8 75 -} - -function get_speaker_sounds { - whiptail --title "${title}" --checklist \ - 'Play Sounds?' \ - 10 50 2 \ - 'awake' 'On wake-up' 0 \ - 'done' 'After voice command' 0 -} - -speaker='' -while [ -z "${speaker}" ]; do - speaker="$(get_speaker 3>&1 1>&2 2>&3)" -done - -if [ "${speaker}" = 'custom' ]; then - # Custom ALSA device name from "aplay -L" - speaker='' - while [ -z "${speaker}" ]; do - speaker="$(get_custom_speaker 3>&1 1>&2 2>&3)" - done -fi - -if [ "${speaker}" = 'no_speaker' ]; then - # No audio output - speaker_command='' -else - # Add device - speaker_command="${speaker_command} -D ${speaker}" - - # awake/done sounds - speaker_sounds="$(get_speaker_sounds 3>&1 1>&2 2>&3)" -fi - -# ----------------------------------------------------------------------------- -# wyoming-satellite -# ----------------------------------------------------------------------------- - -if [ ! -d "${program_dir}/wyoming_satellite" ]; then - if [ ! -d "${this_dir}/wyoming-satellite" ]; then - # Download source here - git clone 'https://github.com/rhasspy/wyoming-satellite.git' - fi - - program_dir="${this_dir}/wyoming-satellite" -fi - -# Virtual environment -venv="${program_dir}/.venv" -pip="${venv}/bin/pip3" -if [ ! -d "${venv}" ]; then - echo 'Creating virtual environment' - "${program_dir}/script/setup" -fi - -# Satellite name -function get_satellite_name { - whiptail --title "${title}" --inputbox \ - 'Satellite Name:' \ - 8 75 \ - 'Wyoming Satellite' -} - -while [ -z "${satellite_name}" ]; do - satellite_name="$(get_satellite_name 3>&1 1>&2 2>&3)" -done - -# ----------------------------------------------------------------------------- -# Satellite type -# ----------------------------------------------------------------------------- - -function get_satellite_type { - whiptail --title "${title}" --notags \ - --radiolist 'Select Satellite Type:' 10 50 \ - 3 \ - always 'Always streaming' 1 \ - vad 'Voice activity detection' 0 \ - wake 'Local wake word detection' 0 -} - -function get_wake_word_system { - # TODO: Support porcupine1 and snowboy - echo 'openWakeWord' - # whiptail --title "${title}" --notags \ - # --radiolist 'Select Wake Word System:' 10 50 \ - # 3 \ - # openWakeWord 'openWakeWord' 1 \ - # porcupine1 'porcupine1' 0 \ - # snowboy 'snowboy' 0 -} - -function install_openWakeWord { - if [ ! -d '/usr/share/doc/libopenblas-dev' ]; then - echo 'Installing system dependencies' - sudo apt-get install --yes --no-install-recommends \ - libopenblas-dev - fi - - oww_dir="${local_dir}/wyoming-openwakeword" - if [ ! -d "${oww_dir}" ]; then - echo 'Installing openWakeWord' - - # Download - pushd "${local_dir}" - git clone 'https://github.com/rhasspy/wyoming-openwakeword.git' - popd - - # Install - "${oww_dir}/script/setup" - fi -} - -function list_openWakeWord { - whiptail --title "${title}" --notags \ - --radiolist 'Select Wake Word:' 15 50 \ - 6 \ - ok_nabu 'ok nabu' 1 \ - hey_jarvis 'hey jarvis' 0 \ - alexa 'alexa' 0 \ - hey_rhasspy 'hey rhasspy' 0 \ - hey_mycroft 'hey mycroft' 0 \ - community 'Community Wake Words' 0 -} - -satellite_type="$(get_satellite_type 3>&1 1>&2 2>&3)" - -if [ "${satellite_type}" = 'vad' ]; then - # Install vad requirements - "${pip}" install \ - --extra-index-url 'https://www.piwheels.org/simple' \ - -f 'https://synesthesiam.github.io/prebuilt-apps/' \ - -r "${program_dir}/requirements_vad.txt" -elif [ "${satellite_type}" = 'wake' ]; then - # Install openWakeWord - wake_word_system="$(get_wake_word_system 3>&1 1>&2 2>&3)" - - if [ "${wake_word_system}" = 'openWakeWord' ]; then - install_openWakeWord - mkdir -p "${custom_wake_word_dir}/openWakeWord" - - # Choose wake word - wake_word="$(list_openWakeWord 3>&1 1>&2 2>&3)" - - if [ "${wake_word}" = 'community' ]; then - # Download community wake words - community_wake_word_dir="${local_dir}/home-assistant-wakewords-collection" - if [ ! -d "${community_wake_word_dir}" ]; then - echo 'Downloading community wake words' - pushd "${local_dir}" - git clone 'https://github.com/fwartner/home-assistant-wakewords-collection.git' - popd - fi - - # Select community wake word - community_wake_words=() - - pushd "${community_wake_word_dir}" - while read -r wake_word_model; do - wake_word_name="$(basename "${wake_word_model}" .tflite)" - community_wake_words+=("${wake_word_model}" "${wake_word_name}" 0) - done < <(find . -name '*.tflite' -type f | sort) - popd - - function list_community_wake_words { - whiptail --title "${title}" --notags \ - --radiolist 'Select Wake Word:' 25 80 \ - 20 ${community_wake_words[@]} - } - - wake_word="$(list_community_wake_words 3>&1 1>&2 2>&3)" - cp "${community_wake_word_dir}/${wake_word}" \ - "${custom_wake_word_dir}/openWakeWord/" - - wake_word="$(basename "${wake_word}" .tflite)" - fi # community wake word - fi # openWakeWord - -fi # local wake word detection - -# ----------------------------------------------------------------------------- -# Audio enhancements -# ----------------------------------------------------------------------------- - -function get_noise_suppression { - whiptail --title "${title}" --notags --radiolist \ - 'Noise Suppression Level:' \ - 15 75 6 \ - '' 'Off' 1 \ - '1' 'Low' 0 \ - '2' 'Medium' 0 \ - '3' 'High' 0 \ - '4' 'Max' 0 -} - -noise_suppression="$(get_noise_suppression 3>&1 1>&2 2>&3)" - -if [ -n "${noise_suppression}" ]; then - # Install audio enhancement requirements - "${pip}" install \ - --extra-index-url 'https://www.piwheels.org/simple' \ - -f 'https://synesthesiam.github.io/prebuilt-apps/' \ - -r "${program_dir}/requirements_audio_enhancement.txt" -fi - -# ----------------------------------------------------------------------------- -# Services -# ----------------------------------------------------------------------------- - -services_dir="${local_dir}/services" -mkdir -p "${services_dir}" - -wake_service_requires='' -event_service_requires='' - -# Disable any previous services -echo 'Stopping previous services' -for service in 'wyoming-satellite' 'wyoming-openwakeword'; do - sudo systemctl disable --now "${service}.service" 2>/dev/null || true -done - -echo 'Creating systemd unit files' -new_services=('wyoming-satellite.service') - -# openWakeWord -if [ "${wake_word_system}" = 'openWakeWord' ]; then - new_services+=('wyoming-openwakeword.service') - wake_service_requires='Requires=wyoming-openwakeword.service' - - wake_word_command=("${local_dir}/wyoming-openwakeword/script/run") - wake_word_command+=('--uri' 'tcp://127.0.0.1:10400') - wake_word_command+=('--custom-model-dir' "'${custom_wake_word_dir}/openWakeWord'") - - if [ -n "${debug_enabled}" ]; then - wake_word_command+=('--debug') - fi - - cat > "${services_dir}/wyoming-openwakeword.service" < "${services_dir}/wyoming-satellite.service" <