Trigger-Talk/engine_vosk.py at master · ManiAm/Trigger-Talk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

import sys
import json
import queue
import threading
import gc
import sounddevice as sd
from vosk import Model, KaldiRecognizer

import utility


class VoskEngine:

    def __init__(self):

        self.q = queue.Queue(maxsize=50)

        self.vosk_model = None
        self.vosk_recognizer = None

        self.dev_index = None
        self.dev_sample_rate = None
        self.dev_channels = None


    def init_model(self, model_name, dev_index, dev_sample_rate, dev_channels):

        self.dev_index = dev_index
        self.dev_sample_rate = dev_sample_rate
        self.dev_channels = dev_channels

        print(f"\n🔄 Loading Vosk model '{model_name}'...")

        try:

            self.vosk_model = Model(model_name=model_name)
            self.vosk_recognizer = KaldiRecognizer(self.vosk_model, self.dev_sample_rate)
            self.vosk_recognizer.SetWords(True) # enable word-level recognition output

        except Exception as e:
            return False, str(e)

        return True, None


    def start_hotword_detection(self, hotword_list, target_latency_ms, script_state, on_hotword_callback=None):

        blocksize = utility.choose_blocksize(target_latency_ms, self.dev_sample_rate)

        self.__empty_queue()
        detected_hotword = None

        with sd.RawInputStream(
            device=self.dev_index,
            samplerate=self.dev_sample_rate,
            blocksize=blocksize,
            dtype='int16',
            channels=self.dev_channels,
            callback=self.__audio_callback):

            while not script_state["interrupted"]:

                data = self.q.get()

                if self.vosk_recognizer.AcceptWaveform(data):

                    result = json.loads(self.vosk_recognizer.Result())
                    text = result.get("text", "").lower()

                    if not text:
                        continue

                    print(f"[VOICE] {text}")

                    for word in hotword_list:
                        if word in text:
                            print(f"🔊 Hotword detected: {word}")
                            detected_hotword = word
                            break

                    if detected_hotword:
                        break  # break while loop

                else:
                    # partial results:
                    # partial = json.loads(recognizer.PartialResult())["partial"]
                    pass

        if detected_hotword:
            if on_hotword_callback:
                on_hotword_callback(detected_hotword)

        return True, None


    def stop_hotword_detection(self):

        self.vosk_recognizer = None
        self.vosk_model = None
        gc.collect()


    def __audio_callback(self, indata, frames, time_info, status):

        if status:
            print(f"[STATUS] {status}", file=sys.stderr)

        try:
            self.q.put_nowait(bytes(indata))
        except queue.Full:
            print("[WARN] Audio queue full — dropping frame")


    def __empty_queue(self):

        while not self.q.empty():
            try:
                self.q.get_nowait()
            except queue.Empty:
                break