g lk - 2024-01-29

entry file:

    from pocketsphinx import Endpointer, Decoder, set_loglevel, Config, get_model_path
    import os
    import sounddevice
    import asyncio
    from util.string import get_absolute_path

    class Porcupine():
        model_path =  get_absolute_path("offline_wakeup\\cmusphinx-zh-cn-5.2_back\\zh_cn.cd_cont_5000")
        lm = get_absolute_path('offline_wakeup\\Sphinx_Keyword_zh_cn\\', '1225.lm')
        dic =  get_absolute_path("offline_wakeup\\Sphinx_Keyword_zh_cn\\", "1225.dic")

        print("model_path", model_path)
        print("lm", lm)
        print("dic", dic)

        sampleRate = 16000

        def __init__(self, on_status_change=None, config=None):
            self.on_status_change = on_status_change

            # 是否在听
            self.is_listening = False

            self.ep = Endpointer(vad_mode=3)

            self.block_size = self.ep.frame_bytes // 2

            self.initialize()

        def initialize(self):
            """
            """
            """
                pocketsphinx.Config Doc: https://pocketsphinx.readthedocs.io/en/latest/pocketsphinx.html#pocketsphinx.Config
                Config Doc: https://pocketsphinx.readthedocs.io/en/latest/config_params.html#Config
            """

            try:
                config = Config(
                    hmm=self.model_path, 
                    lm=self.lm, 
                    dict=self.dic, 
                    samprate=self.sampleRate, 
                    # remove_noise=True
                )
            except Exception as e:
              print('Config error', e)

            try:
                self.stream = sounddevice.RawInputStream(
                    samplerate=self.sampleRate,
                    blocksize=self.block_size,
                    dtype="int16",
                    channels=1,
                )
            except Exception as e:
              print('RawInputStream error', e)

            self.decoder = Decoder(config)

        def _status_change(self, data=None):
            self.on_status_change and asyncio.run(self.on_status_change(data))

        def start(self,):
            with self.stream:
                print("请说...")
                self.is_listening = True
                self._status_change({ 
                    "t": "is_listening",
                })
                try:
                    while True:
                        frame, _ = self.stream.read(self.block_size)
                        prev_in_speech = self.ep.in_speech
                        speech = self.ep.process(frame)
                        if speech is not None:
                            if not prev_in_speech:
                                print("开始时间:", self.ep.speech_start)
                                self.decoder.start_utt()
                            self.decoder.process_raw(speech)

                            # 获取当前的识别假设。
                            hyp = self.decoder.hyp()

                            # 如果有识别结果
                            if hyp is not None:
                                print("PARTIAL RESULT: ", hyp.hypstr)

                            # 如果不在说话了
                            if not self.ep.in_speech:
                                print("结束时间:", self.ep.speech_end)
                                self.decoder.end_utt()
                                result = self.decoder.hyp().hypstr
                                print("识别结果:", result)
                                print("评分", self.decoder.hyp().score)
                                if any(value in result for value in ["小薇", "小微"]):
                                    print("😀")
                                    self._status_change({ 
                                        "t": "keyword",
                                        "val": result
                                    })
                except Exception as e:
                    self._status_change({"t": "error", "val": f"endpointer error: {str(e)}" })
                    print('出现异常', e)

        def stop(self,):
            if self.stream:
                self.stream.abort()


    if __name__ == "__main__":
        porcupine_instance = Porcupine()
        porcupine_instance.start()

        # while True:
        #     pass

After the program starts, I don't speak, but it recognizes the other keywords I set. At the same time, if I accurately say the keywords I set, it can also trigger keywords. However, I don't want other confusing keywords to be easily triggered. What should I do? Thank you to everyone.

Running results

 

Last edit: g lk 2024-01-29