#!/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()