Wake Word Plugins

WakeWord plugins classify audio and report if a certain word or sound is present or not

These plugins usually correspond to the name of the voice assistant, "hey mycroft", but can also be used for other purposes

ww

List of Wake Word plugins

Plugin Type
ovos-ww-plugin-pocketsphinx phonemes
ovos-ww-plugin-vosk text samples
ovos-ww-plugin-snowboy model
ovos-ww-plugin-precise model
ovos-ww-plugin-precise-lite model
ovos-ww-plugin-nyumaya model
ovos-ww-plugin-nyumaya-legacy model
neon_ww_plugin_efficientwordnet model
mycroft-porcupine-plugin model
ovos-ww-plugin-hotkeys keyboard

Standalone Usage

first lets get some boilerplate ouf of the way for the microphone handling logic

import pyaudio

# helper class
class CyclicAudioBuffer:
    def __init__(self, duration=0.98, initial_data=None,
                 sample_rate=16000, sample_width=2):
        self.size = self.duration_to_bytes(duration, sample_rate, sample_width)
        initial_data = initial_data or self.get_silence(self.size)
        # Get at most size bytes from the end of the initial data
        self._buffer = initial_data[-self.size:]

    @staticmethod
    def duration_to_bytes(duration, sample_rate=16000, sample_width=2):
        return int(duration * sample_rate) * sample_width

    @staticmethod
    def get_silence(num_bytes):
        return b'\0' * num_bytes

    def append(self, data):
        """Add new data to the buffer, and slide out data if the buffer is full
        Arguments:
            data (bytes): binary data to append to the buffer. If buffer size
                          is exceeded the oldest data will be dropped.
        """
        buff = self._buffer + data
        if len(buff) > self.size:
            buff = buff[-self.size:]
        self._buffer = buff

    def get(self):
        """Get the binary data."""
        return self._buffer

# pyaudio params
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK = 1024
MAX_RECORD_SECONDS = 20
SAMPLE_WIDTH = pyaudio.get_sample_size(FORMAT)
audio = pyaudio.PyAudio()

# start Recording
stream = audio.open(channels=CHANNELS, format=FORMAT,
    rate=RATE, frames_per_buffer=CHUNK, input=True)


def load_plugin():
    # Wake word initialization
    config = {"model": "path/to/hey_computer.model"}
    return MyHotWord("hey computer", config=config)


def listen_for_ww(plug):
    # TODO - see examples below
    return False

plug = load_plugin()

print(f"Waiting for wake word {MAX_RECORD_SECONDS} seconds")
found = listen_for_ww(plug)

if found:
    print("Found wake word!")
else:
    print("No wake word found")

# stop everything
plug.stop()
stream.stop_stream()
stream.close()
audio.terminate()

new style plugins

New style plugins expect to receive live audio, they may keep their own cyclic buffers internally


def listen_for_ww(plug):
    for i in range(0, int(RATE / CHUNK * MAX_RECORD_SECONDS)):
        data = stream.read(CHUNK)
        # feed data directly to streaming prediction engines
        plug.update(data)
        # streaming engines return result here
        found = plug.found_wake_word(data)
        if found:
            return True

old style plugins (DEPRECATED)

Old style plugins expect to receive ~3 seconds of audio data at once

def listen_for_ww(plug):
    # used for old style non-streaming wakeword (deprecated)
    audio_buffer = CyclicAudioBuffer(plug.expected_duration,
                                     sample_rate=RATE, sample_width=SAMPLE_WIDTH)
    for i in range(0, int(RATE / CHUNK * MAX_RECORD_SECONDS)):
        data = stream.read(CHUNK)
        # add data to rolling buffer, used by non-streaming engines
        audio_buffer.append(data)
        # non streaming engines check the byte_data in audio_buffer
        audio_data = audio_buffer.get()
        found = plug.found_wake_word(audio_data)
        if found:
            return True

new + old style plugins (backwards compatibility)

if you are unsure what kind of plugin you will be using you can be compatible with both approaches like ovos-core

def listen_for_ww(plug):
    # used for old style non-streaming wakeword (deprecated)
    audio_buffer = CyclicAudioBuffer(plug.expected_duration,
                                     sample_rate=RATE, sample_width=SAMPLE_WIDTH)
    for i in range(0, int(RATE / CHUNK * MAX_RECORD_SECONDS)):
        data = stream.read(CHUNK)
        # old style engines will ignore the update
        plug.update(data)
        # streaming engines will ignore the byte_data
        audio_buffer.append(data)
        audio_data = audio_buffer.get()
        found = plug.found_wake_word(audio_data)
        if found:
            return True

Plugin Template

from ovos_plugin_manager.templates.hotwords import HotWordEngine
from threading import Event


class MyWWPlugin(HotWordEngine):
    def __init__(self, key_phrase="hey mycroft", config=None, lang="en-us"):
        super().__init__(key_phrase, config, lang)
        self.detection = Event()
        # read config settings for your plugin
        self.sensitivity = self.config.get("sensitivity", 0.5)
        # TODO - plugin stuff
        # how does your plugin work? phonemes? text? models?
        self.engine = MyWW(key_phrase) 

    def found_wake_word(self, frame_data):
        """Check if wake word has been found.

        Checks if the wake word has been found. Should reset any internal
        tracking of the wake word state.

        Arguments:
            frame_data (binary data): Deprecated. Audio data for large chunk
                                      of audio to be processed. This should not
                                      be used to detect audio data instead
                                      use update() to incrementally update audio
        Returns:
            bool: True if a wake word was detected, else False
        """
        detected = self.detection.is_set()
        if detected:
            self.detection.clear()
        return detected

    def update(self, chunk):
        """Updates the hotword engine with new audio data.

        The engine should process the data and update internal trigger state.

        Arguments:
            chunk (bytes): Chunk of audio data to process
        """
        if self.engine.found_it(chunk): # TODO - check for wake word
            self.detection.set()

    def stop(self):
        """Perform any actions needed to shut down the wake word engine.

        This may include things such as unloading data or shutdown
        external processess.
        """
        self.engine.bye()  # TODO - plugin specific shutdown