SentimentAnalysisSpeech.lf

/**
 * @file SentimentAnalysisSpeech.lf
 * @author Vincenzo Barbuto
 * @brief Examples of how to use the NLP library to perform real time speech to text sentiment
 * analysis.
 */
target Python

import Microphone from "lib/Input.lf"
import NLClassifier from "lib/NLP.lf"

/**
 * The `SpeechToText` reactor is responsible for performing real-time speech recognition using the
 * Vosk speech recognition model. It takes an `audio` input coming from `Microphone` reactor,
 * processes the audio data, and outputs the recognized text to the `NLClassifier` reactor to
 * perform text classification.
 *
 * The `ResultPrinter` reactor is responsible for printing the results of the sentiment analysis
 * performed by the `NLClassifier` reactor. It takes the `results` and `text` inputs, and prints the
 * recognized text and the sentiment analysis results.
 *
 * The `main` reactor creates instances of the `Microphone`, `SpeechToText`, `NLClassifier`, and
 * `ResultPrinter` reactors, and connects them together to form the complete speech-to-text
 * sentiment analysis application.
 */
reactor SpeechToText {
  input audio
  output text

  state model
  state rec

  preamble {=
    import vosk
    import json
    import os
  =}

  # The Vosk speech recognition model path is in the models directory of the project.
  state model_path = {=
    self.os.path.join(self.os.getcwd(), "models", "vosk-model")
  =}

  reaction(startup) {=
    # Load the model
    print(self.model_path)
    self.model = self.vosk.Model(self.model_path)
    # Create the recognizer
    self.rec = self.vosk.KaldiRecognizer(self.model, 16000)
  =}

  reaction(audio) -> text {=
    # Process the audio data
    data = bytes(audio.value)
    if(self.rec.AcceptWaveform(data)):
        result = self.json.loads(self.rec.Result())
        recognized_text = result["text"]
        text.set(recognized_text)
  =}
}

reactor ResultPrinter {
  input results
  input text

  reaction(results, text) {=
    if(text.is_present and text.value!= ""):
      print("-"*70)
      print(f"Text: {text.value}")
      for result in results.value:
          print(f"Result: {result['name']}, Score: {result['score']:.2f}")
  =}
}

main reactor {
  mic = new Microphone(buffer_size=8000, dtype="int16")
  stt = new SpeechToText()
  cls = new NLClassifier(
      model = {= os.path.join(os.getcwd(),"models/nlp/classification/bert_classifier.tflite") =})
  printer = new ResultPrinter()
  mic.audio_data -> stt.audio
  stt.text -> cls.input_data
  stt.text -> printer.text
  cls.results -> printer.results
}