Creation of ASR api

.gitignore

@@ -1,3 +1,5 @@
 version.py
 *.egg-info/
 *.so
+
+.idea

setup.py

@@ -9,7 +9,7 @@
 import pkgconfig
 
 
-VERSION = "0.1.0"
+VERSION = "0.2.0"
 PACKAGE = "yapykaldi"
 PACKAGE_DIR = os.path.join('src', 'python')
 

src/python/yapykaldi/asr.py

@@ -0,0 +1,120 @@
+"""
+Yapykaldi ASR: Class definition for ASR component. It connects to a source and an optional sink
+"""
+from __future__ import (print_function, division, absolute_import, unicode_literals)
+from builtins import *
+import struct
+from threading import Event
+import numpy as np
+from .logger import logger
+from .nnet3 import KaldiNNet3OnlineDecoder, KaldiNNet3OnlineModel
+from .gmm import KaldiGmmOnlineDecoder, KaldiGmmOnlineModel
+from .io import AudioSourceBase
+from .utils import volume_indicator
+
+
+ONLINE_MODELS = {'nnet3': KaldiNNet3OnlineModel, 'gmm': KaldiGmmOnlineModel}
+ONLINE_DECODERS = {'nnet3': KaldiNNet3OnlineDecoder, 'gmm': KaldiGmmOnlineDecoder}
+
+
+class Asr(object):
+    """API for ASR"""
+    # pylint: disable=too-many-instance-attributes, useless-object-inheritance
+
+    def __init__(self, model_dir, model_type, stream, timeout=2, debug=False):
+        """
+        :param model_dir: Path to model directory
+        :param model_type: Type of ASR model 'nnet3' or 'hmm'
+        :param stream: Audio source object
+        :param timeout: (default 2) Time to wait for a new data buffer before stopping recognition due to unavailability
+        of data
+        :param debug: (default False) Flag to set logger to log audio chunk volume and partially decoded string and
+        likelihood
+        """
+        self.model_dir = model_dir
+        self.model_type = model_type
+
+        self.stream = stream  # type: AudioSourceBase
+
+        logger.info("Trying to initialize %s model from %s", self.model_type, self.model_dir)
+        self.model = ONLINE_MODELS[self.model_type](self.model_dir)
+        logger.info("Successfully initialized %s model from %s", self.model_type, self.model_dir)
+
+        self.timeout = timeout
+
+        self._finalize = Event()
+
+        self._string_partially_recognized_callbacks = []
+        self._string_fully_recognized_callbacks = []
+
+        self._debug = debug
+
+    def recognize(self):
+        """Method to start the recognition process on audio stream added to process queue"""
+
+        if self._finalize.is_set():
+            raise Exception("Asr object not initialized for recognition")
+
+        logger.info("Trying to initialize %s model decoder", self.model_type)
+        decoder = ONLINE_DECODERS[self.model_type](self.model)
+        logger.info("Successfully initialized %s model decoder", self.model_type)
+
+        decoded_string = ""
+        while not self._finalize.is_set():
+            try:
+                chunk = self.stream.get_next_chunk(self.timeout)
+                data = np.array(struct.unpack_from('<%dh' % self.stream.chunksize, chunk), dtype=np.float32)
+            except StopIteration as e:  # pylint: disable=invalid-name
+                logger.info("Stream reached it end")
+                logger.error(e)
+                self.stop()
+            except Exception as e:  # pylint: disable=invalid-name, broad-except
+                logger.error("Other exception happened: %s", e)
+                break
+            else:
+                if decoder.decode(self.stream.rate, data, self._finalize.is_set()):
+                    decoded_string, likelihood = decoder.get_decoded_string()
+
+                    if self._debug:
+                        chunk_volume_level = volume_indicator(data)
+                        logger.info("Chunk volume level: %s", chunk_volume_level)
+                        logger.info("Partially decoded (%s): %s", likelihood, decoded_string)
+
+                    for callback in self._string_partially_recognized_callbacks:
+                        callback(decoded_string)
+                else:
+                    raise RuntimeError("Decoding failed")
+
+        logger.info("Decoding of input stream is complete")
+        logger.info("Final result (%s): %s", likelihood, decoded_string)
+
+        for callback in self._string_fully_recognized_callbacks:
+            callback(decoded_string)
+
+    def stop(self):
+        """Stop ASR process"""
+        logger.info("Stop ASR")
+        self._finalize.set()
+        self.stream.stop()
+
+    def start(self):
+        """Begin ASR process"""
+        logger.info("Starting speech recognition")
+        # Reset internal states at the start of a new call
+
+        self._finalize.clear()
+
+        self.stream.start()
+
+    def register_callback(self, callback, partial=False):
+        """
+        Register a callback to receive the decoded string both partial and complete.
+
+        :param callback: a function taking a single string as it's parameter
+        :param partial: (default False) flag to set callback for partial recognitions
+        :return: None
+        """
+        if partial:
+            self._string_partially_recognized_callbacks += [callback]
+        else:
+            self._string_fully_recognized_callbacks += [callback]

src/python/yapykaldi/io/__init__.py

@@ -0,0 +1,14 @@
+"""
+Yapykaldi I/O: Classes and functions for I/O operations with all the wrappers
+"""
+
+__all__ = [
+    # From .sources
+    "AudioSourceBase", "PyAudioMicrophoneSource", "WaveFileSource",
+
+    # From .sinks
+    "WaveFileSink"
+]
+
+from .sources import AudioSourceBase, PyAudioMicrophoneSource, WaveFileSource
+from .sinks import WaveFileSink

src/python/yapykaldi/io/sinks.py

@@ -0,0 +1,46 @@
+"""Audio sinks supported by Yapykaldi"""
+import wave
+import pyaudio
+
+
+class WaveFileSink(object):
+    """WaveFileSink class"""
+
+    def __init__(self, wavpath, fmt=pyaudio.paInt16, channels=1, rate=16000, chunk=1024):
+        """
+
+        :param wavpath: location where to save audio to
+        :param fmt: (default pyaudio.paInt16) Data type of the audio stream
+        :param channels: (default 1) Number of channels of the audio stream
+        :param rate: (default 16000) Sampling frequency of the audio stream
+        :param chunk: (default 1024) Size of the audio stream buffer
+        """
+        self._pyaudio = pyaudio.PyAudio()
+        self.wavpath = wavpath
+        self.format = fmt
+        self.channels = channels
+        self.rate = rate
+        self.chunk = chunk
+
+        self.frames = []
+
+    def add_chunk(self, frames):
+        """Add frame chunk to the WaveFileSink object
+
+        :param frames: audio frames to be added to the sink object
+        """
+        # Only append method works for both python 2 and 3
+        # List concatenation does not work as it converts byte strings to int
+        self.frames.append(frames)
+
+    def write_frames(self, frames=None):
+        """Write audio frames into a file
+
+        :param frames: (default None) Frames to write to a file. This bypasses the frames stored in the sink object.
+        """
+        wav_out = wave.open(self.wavpath, 'wb')
+        wav_out.setnchannels(self.channels)
+        wav_out.setsampwidth(self._pyaudio.get_sample_size(self.format))
+        wav_out.setframerate(self.rate)
+        wav_out.writeframes(b''.join(frames if frames else self.frames))
+        wav_out.close()

src/python/yapykaldi/io/sources.py

@@ -0,0 +1,187 @@
+"""Audio sources supported by Yapykaldi"""
+from __future__ import print_function, division, absolute_import, unicode_literals
+from builtins import *
+import math
+import wave
+from threading import Event, Thread
+from queue import Empty, Queue
+import pyaudio
+
+from .sinks import WaveFileSink
+from ..logger import logger
+
+try:
+    from typing import Optional
+except ImportError:
+    pass
+
+
+class AudioSourceBase(object):
+    """The AudioSource
+    It requires some setup before we can get audio bytes from it and
+    requires some teardown afterwards
+
+    The right order is:
+    1. source = AudioSourceBase()
+    2. source.open()                # to open the file, connect the mic etc.
+    3. source.start()               # actually start getting audio data
+    4. source.get_next_chunk()      # use the audio data
+    5. source.stop()                # stop getting audio data
+    6. source.close()               # close the file
+
+    Some sources only support opening them once but
+        they should all support going through start, get.., stop
+        several times
+
+    """
+    # pylint: disable=useless-object-inheritance
+
+    def __init__(self, rate=16000, chunksize=1024):
+        self.rate = rate
+        self.chunksize = chunksize
+
+    def open(self):
+        raise NotImplementedError()
+
+    def start(self):
+        raise NotImplementedError()
+
+    def stop(self):
+        raise NotImplementedError()
+
+    def close(self):
+        raise NotImplementedError()
+
+    def get_next_chunk(self, timeout):
+        raise NotImplementedError()
+
+
+class PyAudioMicrophoneSource(AudioSourceBase):
+    def __init__(self, fmt=pyaudio.paInt16, channels=1, rate=16000, chunksize=1024, saver=None):
+        """
+        :param fmt: (default pyaudio.paInt16) format of the audio data
+        :param channels: (default 1) number of channels in audio data
+        :param rate: (default 16000) sampling frequency of audio data
+        :param chunksize: (default 1024) size of audio data buffer
+        :param saver: (default None) audio sink object
+        """
+        super().__init__(rate=rate, chunksize=chunksize)
+
+        self._pyaudio = pyaudio.PyAudio()
+        self.format = fmt
+        self.channels = channels
+
+        self.stream = None  # type: Optional[pyaudio.PyAudio]
+
+        self.saver = saver  # type: WaveFileSink
+
+        self._queue = Queue()
+        self._worker = None  # type: Optional[Thread]
+
+        self._stop = Event()
+
+    def open(self):
+        # This function is needed to maintain generality in api of stream sources
+        pass
+
+    def start(self):
+        # Start async process to put audio chunks in a queue
+        self._stop.clear()
+        self._worker = Thread(target=self._listen, args=(self._stop,))
+        logger.info("Starting audio stream in a separate thread")
+        self._worker.start()
+
+    def _listen(self, stop_event):
+        stream = self._pyaudio.open(format=self.format,
+                                    channels=self.channels,
+                                    rate=self.rate,
+                                    input=True,
+                                    frames_per_buffer=self.chunksize)
+
+        while not stop_event.wait(0):
+            chunk = stream.read(self.chunksize)
+            # logger.debug("{}\t+1 chunks in the queue".format(self._queue.qsize()))
+            self._queue.put(chunk)
+
+        stream.stop_stream()
+        stream.close()
+        logger.info("Stopped streaming audio")
+
+    def get_next_chunk(self, timeout=1):
+        try:
+            # logger.debug("{}\t-1 chunks in the queue".format(self._queue.qsize()))
+            chunk = self._queue.get(block=True, timeout=timeout)
+            if self.saver:
+                self.saver.add_chunk(chunk)
+            return chunk
+        except Empty:
+            raise StopIteration()
+
+    def stop(self):
+        if not self._stop.is_set():
+            self._stop.set()
+
+            logger.info("Waiting for audio stream to stop")
+            self._worker.join()
+            logger.info("Exited audio stream thread")
+        else:
+            logger.info("No running audio stream to stop")
+
+    def close(self):
+        self._pyaudio.terminate()
+
+        if self.saver:
+            self.saver.write_frames()
+
+
+class WaveFileSource(AudioSourceBase):
+    def __init__(self, filename, rate=16000, chunksize=1024):
+        """
+        :param filename: path to the wave file
+        :type filename: str
+        :param rate: (default 16000) sampling frequency of audio data
+        :param chunksize: (default 1024) size of audio data buffer
+        """
+        super().__init__(rate=rate, chunksize=chunksize)
+        self.filename = filename
+        self.wavf = None
+        self.total_num_frames = None
+        self.total_chunks = None
+        self.read_chunks = None
+
+    def open(self):
+        if not self.wavf:
+            self.wavf = wave.open(self.filename, 'rb')
+            assert self.wavf.getnchannels() == 1
+            assert self.wavf.getsampwidth() == 2
+            assert self.wavf.getnframes() > 0
+            assert self.wavf.getframerate() == self.rate
+            logger.info("Stream opened from %s", self.filename)
+        else:
+            logger.error("Stream already open from %s. Call the close() method first", self.filename)
+
+    def start(self):
+        self.total_num_frames = self.wavf.getnframes()
+        self.total_chunks = math.floor(self.total_num_frames / self.chunksize)
+        self.read_chunks = 0
+
+    def get_next_chunk(self, timeout):
+        if self.read_chunks < self.total_chunks:
+            frames = self.wavf.readframes(self.chunksize)
+            self.read_chunks += 1
+            return frames
+
+        raise StopIteration()
+
+    def stop(self):
+        # This function is needed to maintain generality in api of stream sources
+        pass
+
+    def close(self):
+        self.wavf.close()
+        logger.info("Stream closed from %s", self.filename)
+
+        self.wavf = None
+        self.total_num_frames = None
+        self.total_chunks = None
+        self.read_chunks = None

src/python/yapykaldi/logger.py

@@ -0,0 +1,4 @@
+import logging
+
+LOGGER_NAME = "yapykaldi"
+logger = logging.getLogger(LOGGER_NAME)