diff options
Diffstat (limited to 'modules')
| -rw-r--r-- | modules/config.py | 37 | ||||
| -rw-r--r-- | modules/input_handlers/__init__.py | 0 | ||||
| -rw-r--r-- | modules/input_handlers/input_handler.py | 11 | ||||
| -rw-r--r-- | modules/input_handlers/pipewire_record.py | 72 | ||||
| -rw-r--r-- | modules/input_handlers/stdin_input.py | 16 | ||||
| -rw-r--r-- | modules/intents/process.py | 6 | ||||
| -rw-r--r-- | modules/intents/sway.py | 8 | ||||
| -rw-r--r-- | modules/responses/__init__.py | 0 | ||||
| -rw-r--r-- | modules/responses/libnotify.py | 15 | ||||
| -rw-r--r-- | modules/responses/response_handler.py | 6 | ||||
| -rw-r--r-- | modules/responses/response_manager.py | 32 | ||||
| -rw-r--r-- | modules/responses/responses.py | 29 |
12 files changed, 224 insertions, 8 deletions
diff --git a/modules/config.py b/modules/config.py index 602ec8d..c5e8d1d 100644 --- a/modules/config.py +++ b/modules/config.py @@ -1,20 +1,48 @@ +import sys + import __main__ import os.path import yaml class Config: + INPUT_PW = "pw-record" + INPUT_STDIN = "stdin" + INPUT_MODES = [INPUT_PW, INPUT_STDIN] + def __init__(self): + self.input_mode = "" + self.intents_dir = "" + self.responses_dir = "" + self.resources_dir = "" self.applications_dir = "" - def update(self, **entries): + self.lock = "" + + self.hestia_images = "" + + def update(self, **entries) -> None: self.__dict__.update(entries) - if not self.intents_dir.startswith("/"): - self.intents_dir = os.path.join(os.path.dirname(__main__.__file__), self.intents_dir) + self.intents_dir = Config.__convert_to_absolute_path(self.intents_dir) + self.responses_dir = Config.__convert_to_absolute_path(self.responses_dir) + self.resources_dir = Config.__convert_to_absolute_path(self.resources_dir) + + self.hestia_images = os.path.join(self.resources_dir, "hestia_images") + + @staticmethod + def __convert_to_absolute_path(path: str) -> str: + if not path.startswith("/"): + path = os.path.join(os.path.dirname(__main__.__file__), path) + + return path + + def validate(self) -> None: + if self.input_mode not in self.INPUT_MODES: + sys.exit(f"Invalid input_mode '{self.input_mode}', valid options: {', '.join(self.INPUT_MODES)}") -def load_config(): +def load_config() -> Config: config = Config() with open(os.path.join(os.path.dirname(__main__.__file__), "config.yaml")) as stream: config.update(**yaml.safe_load(stream)) @@ -24,4 +52,5 @@ def load_config(): with open(user_config) as stream: config.update(**yaml.safe_load(stream)) + config.validate() return config
\ No newline at end of file diff --git a/modules/input_handlers/__init__.py b/modules/input_handlers/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/modules/input_handlers/__init__.py diff --git a/modules/input_handlers/input_handler.py b/modules/input_handlers/input_handler.py new file mode 100644 index 0000000..3db042c --- /dev/null +++ b/modules/input_handlers/input_handler.py @@ -0,0 +1,11 @@ +from abc import ABC, abstractmethod + +class InputHandler(ABC): + + @abstractmethod + def get_input(self) -> str: + pass + + @abstractmethod + def cleanup(self) -> None: + pass
\ No newline at end of file diff --git a/modules/input_handlers/pipewire_record.py b/modules/input_handlers/pipewire_record.py new file mode 100644 index 0000000..0a767e1 --- /dev/null +++ b/modules/input_handlers/pipewire_record.py @@ -0,0 +1,72 @@ +import subprocess +import os.path +import signal +from time import sleep + +import whisper + +from .input_handler import InputHandler + +FIFO_PATH = "/tmp/hestia-listening" +RECORD_PATH = "/tmp/hestia-record.mp3" + +class PipeWireRecord(InputHandler): + def cleanup(self) -> None: + if os.path.exists(FIFO_PATH): + os.remove(FIFO_PATH) + + + def get_input(self) -> str: + device = PipeWireRecord.get_device() + + self.cleanup() + os.mkfifo(FIFO_PATH) + + while True: + with open(FIFO_PATH): + pass + # TODO "I'm listening" + + try: + ps = subprocess.Popen((f"pw-record --target {device} {RECORD_PATH}",), shell=True) + with open(FIFO_PATH): + print("finished") + ps.send_signal(signal.SIGINT) + # TODO "acknowledged" + except: + if "ps" in locals(): + ps.kill() + # TODO "error" + # TODO exit gracefully or try to recover + raise StopIteration + + model = whisper.load_model("base") + + audio = whisper.load_audio(RECORD_PATH) + audio = whisper.pad_or_trim(audio) + + mel = whisper.log_mel_spectrogram(audio).to(model.device) + options = whisper.DecodingOptions(language="en", fp16=False) + result = whisper.decode(model, mel, options) + result_text = result.text.replace(",", "").replace(".", "").lower() + + print(result_text) + + yield result_text + + @staticmethod + def get_device() -> str: + already_warned = False + + while True: + ps = subprocess.Popen(('pw-cli ls | \\grep -Poi "(?<=node.name = \\").*mic.*(?=\\")"',), shell=True, stdout=subprocess.PIPE) + ps.wait() + + if ps.returncode == 0: + return ps.stdout.read().decode().strip() + + elif not already_warned: + already_warned = True + # TODO warn about device not found + + sleep(3)
\ No newline at end of file diff --git a/modules/input_handlers/stdin_input.py b/modules/input_handlers/stdin_input.py new file mode 100644 index 0000000..3d338c1 --- /dev/null +++ b/modules/input_handlers/stdin_input.py @@ -0,0 +1,16 @@ +import sys + +from .input_handler import InputHandler + + +class StdinInput(InputHandler): + def cleanup(self) -> None: + pass + + def get_input(self) -> str: + for line in sys.stdin: + line = line.strip() + if not line: + continue + + yield line
\ No newline at end of file diff --git a/modules/intents/process.py b/modules/intents/process.py index bc834b6..48ade48 100644 --- a/modules/intents/process.py +++ b/modules/intents/process.py @@ -1,9 +1,11 @@ import os from typing import Dict from .sway import HesNavigate +from ..config import Config -def HesExecuteProcess(data: Dict): + +def HesExecuteProcess(data: Dict, config: Config) -> None: if "workspace" in data: - HesNavigate(data) + HesNavigate(data, config) os.popen(data["process"].strip()) diff --git a/modules/intents/sway.py b/modules/intents/sway.py index 5ab1b25..b5bd9fe 100644 --- a/modules/intents/sway.py +++ b/modules/intents/sway.py @@ -1,5 +1,9 @@ import os from typing import Dict +from ..config import Config -def HesNavigate(data: Dict): - os.popen(f"swaymsg workspace {data["workspace"]}") +def HesNavigate(data: Dict, config: Config) -> None: + os.popen(f"swaymsg workspace {data['workspace']}") + +def HesLock(data: Dict, config: Config) -> None: + os.popen(config.lock)
\ No newline at end of file diff --git a/modules/responses/__init__.py b/modules/responses/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/modules/responses/__init__.py diff --git a/modules/responses/libnotify.py b/modules/responses/libnotify.py new file mode 100644 index 0000000..0bc774f --- /dev/null +++ b/modules/responses/libnotify.py @@ -0,0 +1,15 @@ +import os + +from .response_handler import ResponseHandler +from ..config import Config + + +class LibNotify(ResponseHandler): + def __init__(self, config: Config): + self.config = config + self.hestia_images = {} + for path in os.listdir(self.config.hestia_images): + self.hestia_images[os.path.splitext(os.path.basename(path))[0]] = os.path.join(self.config.hestia_images, path) + + def respond(self, text: str) -> None: + os.popen(f'notify-send "Hestia " "{text}" -i "{self.hestia_images["hestia"]}" --app-name=hestia')
\ No newline at end of file diff --git a/modules/responses/response_handler.py b/modules/responses/response_handler.py new file mode 100644 index 0000000..9cc0cf0 --- /dev/null +++ b/modules/responses/response_handler.py @@ -0,0 +1,6 @@ +from abc import ABC, abstractmethod + +class ResponseHandler(ABC): + @abstractmethod + def respond(self, response: str): + pass
\ No newline at end of file diff --git a/modules/responses/response_manager.py b/modules/responses/response_manager.py new file mode 100644 index 0000000..34c0a41 --- /dev/null +++ b/modules/responses/response_manager.py @@ -0,0 +1,32 @@ +import yaml + +from pathlib import Path + +from ..config import Config +from ..responses.responses import Responses +from ..hassil.util import merge_dict +from .response_handler import ResponseHandler +from .libnotify import LibNotify + +class ResponseManager: + def __init__(self, config: Config): + response_dict = {"responses": {}} + + response_yaml_path = Path(config.responses_dir) + response_yaml_file_paths = response_yaml_path.glob("*.yaml") + for yaml_file_path in response_yaml_file_paths: + with open(yaml_file_path, "r", encoding="utf-8") as yaml_file: + merge_dict(response_dict, yaml.safe_load(yaml_file)) + + self.responses = Responses.from_dict(response_dict) + self.respond_handlers: list[ResponseHandler] = [LibNotify(config)] + + def respond(self, response: str, intent_name: str): + response_key = intent_name if response == "default" else response + if response_key not in self.responses.responses: + print(f"No response found for: {response_key}") + return + + response_text = self.responses.responses[response_key].sentence_texts[0] + for handler in self.respond_handlers: + handler.respond(response_text) diff --git a/modules/responses/responses.py b/modules/responses/responses.py new file mode 100644 index 0000000..f89a1fe --- /dev/null +++ b/modules/responses/responses.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass, field +from typing import List, Dict, Any + +@dataclass(frozen=True) +class Response: + name: str + sentence_texts: List[str] + +@dataclass(frozen=True) +class Responses: + responses: Dict[str, Response] + + @staticmethod + def from_dict(input_dict: Dict[str, Any]) -> "Responses": + # responses: + # ResponseName: + # data: + # - sentences: + # - "<sentence>" + return Responses( + responses={ + response_name: Response( + name=response_name, + sentence_texts=response_dict["sentences"] + ) + for response_name, response_dict in input_dict["responses"].items() + }, + ) + |