import subprocess import os.path import signal 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" class PipeWireRecord(InputHandler): def cleanup(self): if os.path.exists(FIFO_PATH): os.remove(FIFO_PATH) def get_input(self): 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)