rust-whisper/whisper-2023/hotwords.py

import speech_recognition as sr
import time
import threading
import queue
import signal
import sys
from os import environ

def main():
   managerToParserQ = queue.Queue(maxsize=1)
   readerToParserQ = queue.Queue(maxsize=10)
   parserToReactorQ = queue.Queue(maxsize=10)
   reactorToActorQ = queue.Queue(maxsize=10)
   threads = [
      Manager(managerToParserQ),
      Reader(managerToParserQ, readerToParserQ),
      Parser(readerToParserQ, parserToReactorQ),
      Reactor(parserToReactorQ, reactorToActorQ),
      Actor(reactorToActorQ),
   ]
   [t.start() for t in threads]
   [t.join() for t in threads]

def log(*args):
   print(">", *args, file=sys.stderr)

class Manager(threading.Thread):
   def __init__(self, outq):
      threading.Thread.__init__(self)
      self.outq = outq
      inq = queue.Queue(maxsize=1)
      def catcher(sig, frame):
         inq.put(None)
      self.inq = inq
      signal.signal(signal.SIGINT, catcher)

   def run(self):
      log("Manager.run: start")
      self.inq.get()
      self.outq.put(None)
      log("Manager.run: stop")

class Reader(threading.Thread):
   def __init__(self, inq, outq):
      threading.Thread.__init__(self)
      self.name = environ.get("MIC_NAME", "pulse_monitor")
      self.inq = inq
      self.outq = outq

   def run(self):
      log("Reader.run: start")
      idx = [
         idx for idx,v in enumerate(
            sr.Microphone.list_microphone_names(),
         ) if v in self.name.split(",")
      ][0]
      with sr.Microphone(device_index=idx) as mic:
         while not self.should_stop():
            try:
               self.outq.put(self._run(mic))
            except Exception as e:
               log("Reader.run: error:", e)
      self.outq.put(None)
      log("Reader.run: stop")

   def should_stop(self):
      return not self.inq.empty()

   def _run(self, mic):
      mic_timeout = int(environ.get("MIC_TIMEOUT", 5))
      r = sr.Recognizer()
      return r.listen(
         mic,
         timeout=mic_timeout,
         phrase_time_limit=mic_timeout,
      )

class Parser(threading.Thread):
   def __init__(self, inq, outq):
      threading.Thread.__init__(self)
      self.inq = inq
      self.outq = outq

   def run(self):
      log("Parser.run: start")
      while True:
         try:
            clip = self.inq.get()
            if not clip:
               break
            self.outq.put(self._run(clip).strip())
         except Exception as e:
            log("Parser.run: error:", e)
      self.outq.put(None)
      log("Parser.run: stop")

   def _run(self, clip):
      r = sr.Recognizer()
      return r.recognize_whisper(clip, language="english")


class Reactor(threading.Thread):
   def __init__(self, inq, outq):
      threading.Thread.__init__(self)
      self.inq = inq
      self.outq = outq
      self.load_hotwords = Reactor.new_load_hotwords()
      [i for i in self.load_hotwords()]

   def new_load_hotwords():
      p = environ.get("HOTWORDS", None)
      if not p:
         def load_nothing():
            return []
         return load_nothing

      try:
         def load_hotwords_in_file():
            with open(p, "r") as f:
               return ["".join(i.strip().lower().split()) for i in f.readlines()]
         load_hotwords_in_file()
         return load_hotwords_in_file
      except Exception:
         pass

      hotwords = ["".join(i.lower().strip().split()) for i in p.split(",")]
      log(f'$HOTWORDS: {hotwords}')
      def load_hotwords_as_literal():
         return hotwords
      return load_hotwords_as_literal

   def run(self):
      log("Reactor.run: start")
      while True:
         text = self.inq.get()
         if text is None:
            break
         self.handle(text)
      self.outq.put(None)
      log("Reactor.run: stop")

   def handle(self, text):
      hotwords = self.load_hotwords()
      if not hotwords:
         self.outq.put(text)
         return
      cleantext = "".join([i for i in "".join(text.lower().split()) if i.isalpha()])
      for i in hotwords:
         if i in cleantext:
            log(f"Reactor.handle: found hotword '{i}' in '{text}' as '{cleantext}'")
            self.outq.put(i)

class Actor(threading.Thread):
   def __init__(self, inq):
      threading.Thread.__init__(self)
      self.inq = inq

   def run(self):
      log("Actor.run: start")
      while True:
         hotword = self.inq.get()
         if hotword is None:
            break
         self.handle(hotword)
      log("Actor.run: stop")

   def handle(self, hotword):
      print(hotword)

if __name__ == "__main__":
   main()