diff options
Diffstat (limited to 'modules/input_handlers')
| -rw-r--r-- | modules/input_handlers/input_handler.py | 11 | ||||
| -rw-r--r-- | modules/input_handlers/pipewire_record.py | 91 | ||||
| -rw-r--r-- | modules/input_handlers/stdin_input.py | 19 |
3 files changed, 71 insertions, 50 deletions
diff --git a/modules/input_handlers/input_handler.py b/modules/input_handlers/input_handler.py new file mode 100644 index 0000000..82a2830 --- /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): + pass + + @abstractmethod + def cleanup(self): + pass
\ No newline at end of file diff --git a/modules/input_handlers/pipewire_record.py b/modules/input_handlers/pipewire_record.py index 147d77b..3ad295c 100644 --- a/modules/input_handlers/pipewire_record.py +++ b/modules/input_handlers/pipewire_record.py @@ -1,69 +1,72 @@ import subprocess import os.path import signal -import sys from time import sleep import whisper +from modules.input_handlers.input_handler import InputHandler + FIFO_PATH = "/tmp/hestia-listening" RECORD_PATH = "/tmp/hestia-record.mp3" -def cleanup(): - if os.path.exists(FIFO_PATH): - os.remove(FIFO_PATH) - +class PipeWireRecord(InputHandler): + def cleanup(self): + if os.path.exists(FIFO_PATH): + os.remove(FIFO_PATH) -def get_input_pw_record(): - device = get_device() - cleanup() - os.mkfifo(FIFO_PATH) + def get_input(self): + device = PipeWireRecord.get_device() - while True: - with open(FIFO_PATH): - pass - # TODO "I'm listening" + self.cleanup() + os.mkfifo(FIFO_PATH) - try: - ps = subprocess.Popen((f"pw-record --target {device} {RECORD_PATH}",), shell=True) + while 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 + 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") + model = whisper.load_model("base") - audio = whisper.load_audio(RECORD_PATH) - audio = whisper.pad_or_trim(audio) + 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() + 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) + print(result_text) - yield result_text + yield result_text -def get_device() -> str: - already_warned = False + @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() + 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() + if ps.returncode == 0: + return ps.stdout.read().decode().strip() - elif not already_warned: - already_warned = True - # TODO warn about device not found + elif not already_warned: + already_warned = True + # TODO warn about device not found - sleep(3)
\ No newline at end of file + sleep(3)
\ No newline at end of file diff --git a/modules/input_handlers/stdin_input.py b/modules/input_handlers/stdin_input.py index 044ac0d..5c50511 100644 --- a/modules/input_handlers/stdin_input.py +++ b/modules/input_handlers/stdin_input.py @@ -1,9 +1,16 @@ import sys -def get_input_stdin(): - for line in sys.stdin: - line = line.strip() - if not line: - continue +from modules.input_handlers.input_handler import InputHandler - yield line
\ No newline at end of file + +class StdinInput(InputHandler): + def cleanup(self): + pass + + def get_input(self): + for line in sys.stdin: + line = line.strip() + if not line: + continue + + yield line
\ No newline at end of file |