Trixy/PicoVoice/Porcupine.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
import os
import struct
import wave
import configparser
from datetime import datetime
from threading import Thread
import pvporcupine
from pvrecorder import PvRecorder
import platform
from VoicePlay import Voice
from DynLoader import DynLoaderMod as dyn


class Porcupine(Thread):

    def __init__(self):
        super(Porcupine, self).__init__()
        config = configparser.ConfigParser()
        config.read('config/pico.ini')
        Keywords=[]
        Sensitivity=[]
        isAdmin=[]
        bHasKeywords = False

        self._model_path=""
        self._keyword_paths=""
        self._sensitivities=0.5

        self._lang = "de"
        if "lang" in config["global"]:
            self._lang = config["global"]["lang"]
        self._os = platform.system().lower()

        if "ApiKey" in config["PicoVoice"]:
            self._access_key = config["PicoVoice"]["apikey"]
        self._library_path = pvporcupine.LIBRARY_PATH


        if "model" in config["Porcupine"]:
            self._model_path = 'ressources/models/porcupine/' + config["Porcupine"]["model"]

        for k,v in config.items():
            if "weakword" in v and v["weakword"]!="":
                print("  - Weakword '"+k+"': " + v["weakword"].strip())
                Keywords.append('ressources/models/porcupine/' + v["weakword"].strip())
                bHasKeywords = True
                if "sensitivities" in v:
                    Sensitivity.append(float(v["sensitivities"]))
                else:
                    Sensitivity.append(float(0.5))
                if "admin" in v and (v["admin"]=="1" or v["admin"]=="true" or v["admin"]=="True" or v["admin"]=="TRUE"):
                    isAdmin.append(True)
                else:
                    isAdmin.append(False)


        if bHasKeywords == True:
            self._keyword_paths = Keywords
        self._sensitivities = Sensitivity
        self.isAdmin = isAdmin

        if "InputDeviceIndex" in config["Hardware"]:
            self._input_device_index = int(config["Hardware"]["InputDeviceIndex"])
        else:
            self._input_device_index = None
        self._output_path = None

    instance = None
    @staticmethod
    def getInstance():
        if Porcupine.instance==None:
            Porcupine.instance = Porcupine()
        return Porcupine.instance

    def run(self):
        print("Porcupine.run()")
        print("  - Generate Keyword List")

        _access_key = Porcupine.getInstance()._access_key
        _library_path = Porcupine.getInstance()._library_path
        _model_path = Porcupine.getInstance()._model_path
        _keyword_paths = Porcupine.getInstance()._keyword_paths
        _sensitivities = Porcupine.getInstance()._sensitivities
        _output_path = Porcupine.getInstance()._output_path
        __func = Porcupine.getInstance().__func
        _isAdmin = Porcupine.getInstance().isAdmin
        dynloader:dyn = dyn.getInstance()

        keywords = list()
        for x in self._keyword_paths:
            keyword_phrase_part = os.path.basename(x).replace('.ppn', '').split('_')
            if len(keyword_phrase_part) > 6:
                keywords.append(' '.join(keyword_phrase_part[0:-6]))
            else:
                keywords.append(keyword_phrase_part[0])
                
        porcupine = None
        recorder = None
        wav_file = None
        try:
            '''porcupine = pvporcupine.create(
                access_key=_access_key,
                library_path=_library_path,
                model_path=_model_path,
                keyword_paths=_keyword_paths,
                sensitivities=_sensitivities
            )'''
            porcupine = pvporcupine.create(
                access_key=_access_key,
                model_path=_model_path,
                keyword_paths=_keyword_paths,
                sensitivities=_sensitivities
            )
            recorder = PvRecorder(device_index=self._input_device_index, frame_length=porcupine.frame_length)
            print("Frame Length: ",porcupine.frame_length)
            recorder.start()
            if _output_path is not None:
                wav_file = wave.open(self._output_path, "w")
                wav_file.setparams((1, 2, 16000, 512, "NONE", "NONE"))

            print('Listening {')
            for keyword, sensitivity, admin in zip(keywords, _sensitivities, _isAdmin):
                print(keyword,str(sensitivity), admin)
            print('}')

            bRunning = True
            while bRunning:
                pcm = recorder.read()

                if wav_file is not None:
                    wav_file.writeframes(struct.pack("h" * len(pcm), *pcm))

                result = porcupine.process(pcm)
                if result >= 0:
                    #print('[%s] Detected %s' % (str(datetime.now()), keywords[result]))
                    if(self.__func):
                        #bRunning = False
                        recorder.stop()
                        dynloader.execute("WeakWord",result,keywords[result],_isAdmin[result])
                        __func(result,keywords[result],_isAdmin[result])
                        recorder.start()
        except pvporcupine.PorcupineInvalidArgumentError as e:
            print("One or more arguments provided to Porcupine is invalid: {\n" +
                  "\tAcccess Key: "+_access_key+"\n" +
                  "\tLibrary Path: "+_library_path+"\n" +
                  "\tModel Path: "+_model_path+"\n" +
                  "\tKeyword Paths: ['"+', '.join(_keyword_paths)+"']\n" +
                  "\tSensitivities: ['"+', '.join([str(int) for int in _sensitivities])+"']\n" +
                  "}")
            print("If all other arguments seem valid, ensure that '"+self._access_key+"' is a valid AccessKey")
            print("Error: "+','.join(e.args))
            Voice.getInstance().Say("Ausnahmefehler in Porcubine Argument")
            raise e
        except pvporcupine.PorcupineActivationError as e:
            print("AccessKey activation error")
            Voice.getInstance().Say("Ausnahmefehler in Porcubine Aktivierung")
            raise e
        except pvporcupine.PorcupineActivationLimitError as e:
            print("AccessKey '"+self._access_key+"' has reached it's temporary device limit")
            Voice.getInstance().Say("Ausnahmefehler in Porcubine Aktivierungs-Limit")
            raise e
        except pvporcupine.PorcupineActivationRefusedError as e:
            print("AccessKey '"+self._access_key+"' refused")
            Voice.getInstance().Say("Ausnahmefehler in Porcubine Aktivierung zurückgewiesen")
            raise e
        except pvporcupine.PorcupineActivationThrottledError as e:
            print("AccessKey '"+self._access_key+"' has been throttled")
            Voice.getInstance().Say("Ausnahmefehler in Porcubine Zugriff wurde gedrosselt")
            raise e
        except pvporcupine.PorcupineError as e:
            print("Failed to initialize Porcupine")
            Voice.getInstance().Say("Ausnahmefehler bei der Porcubine Initialisierung")
            raise e
        except KeyboardInterrupt:
            print('Stopping ...')
        finally:
            if porcupine is not None:
                porcupine.delete()
            if recorder is not None:
                recorder.delete()
            if wav_file is not None:
                wav_file.close()

    def RegisterFunc(self, func):
        self.__func = func

    @staticmethod
    def show_audio_devices():
        devices = PvRecorder.get_audio_devices()
        for i in range(len(devices)):
            print("index: "+str(i)+", device name: '"+devices[i]+"'")