Speech Recognition With Python

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

TESTCASES = [
  {
    'filename': 'audio/2830-3980-0043.wav',
    'text': 'experience proves this',
    'encoding': 'LINEAR16',
    'lang': 'en-US'
  },
  {
    'filename': 'audio/4507-16021-0012.wav',
    'text': 'why should one halt on the way',
    'encoding': 'LINEAR16',
    'lang': 'en-US'
  },
  {
    'filename': 'audio/8455-210777-0068.wav',
    'text': 'your power is sufficient i said',
    'encoding': 'LINEAR16',
    'lang': 'en-US'
  }
]

1
2
3
4
5
6
7

def read_wav_file(filename) -> Tuple[bytes, int]:
    with wave.open(filename, 'rb') as w:
        rate = w.getframerate()
        frames = w.getnframes()
        buffer = w.readframes(frames)

    return buffer, rate

1
2
3
4
5
6
7
8

def simulate_stream(buffer: bytes, batch_size: int = 4096):
    buffer_len = len(buffer)
    offset = 0
    while offset < buffer_len:
        end_offset = offset + batch_size
        buf = buffer[offset:end_offset]
        yield buf
        offset = end_offset

 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

from google.cloud import speech_v1
from google.cloud.speech_v1 import enums

def google_batch_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)
    client = speech_v1.SpeechClient()

    config = {
        'language_code': lang,
        'sample_rate_hertz': rate,
        'encoding': enums.RecognitionConfig.AudioEncoding[encoding]
    }

    audio = {
        'content': buffer
    }

    response = client.recognize(config, audio)
    # For bigger audio file, replace previous line with following:
    # operation = client.long_running_recognize(config, audio)
    # response = operation.result()

    for result in response.results:
        # First alternative is the most probable result
        alternative = result.alternatives[0]
        return alternative.transcript

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('google-cloud-batch-stt: "{}"'.format(
        google_batch_stt(t['filename'], t['lang'], t['encoding'])
    ))

 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

from google.cloud import speech
from google.cloud.speech import enums
from google.cloud.speech import types

def response_stream_processor(responses):
    print('interim results: ')

    transcript = ''
    num_chars_printed = 0
    for response in responses:
        if not response.results:
            continue

        result = response.results[0]
        if not result.alternatives:
            continue

        transcript = result.alternatives[0].transcript
        print('{0}final: {1}'.format(
            '' if result.is_final else 'not ',
            transcript
        ))

    return transcript

def google_streaming_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)

    client = speech.SpeechClient()

    config = types.RecognitionConfig(
        encoding=enums.RecognitionConfig.AudioEncoding[encoding],
        sample_rate_hertz=rate,
        language_code=lang
    )

    streaming_config = types.StreamingRecognitionConfig(
        config=config,
        interim_results=True
    )

    audio_generator = simulate_stream(buffer)  # chunk generator
    requests = (
        types.StreamingRecognizeRequest(audio_content=chunk)
        for chunk in audio_generator
    )
    responses = client.streaming_recognize(
        streaming_config, requests
    )
    # Now, put the transcription responses to use.
    return response_stream_processor(responses)

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('google-cloud-streaming-stt: "{}"'.format(
        google_streaming_stt(
            t['filename'], t['lang'], t['encoding']
        )
    ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import azure.cognitiveservices.speech as speechsdk

def azure_batch_stt(filename: str, lang: str, encoding: str) -> str:
    speech_config = speechsdk.SpeechConfig(
        subscription=AZURE_SPEECH_KEY,
        region=AZURE_SERVICE_REGION
    )
    audio_input = speechsdk.AudioConfig(filename=filename)
    speech_recognizer = speechsdk.SpeechRecognizer(
        speech_config=speech_config,
        audio_config=audio_input
    )
    result = speech_recognizer.recognize_once()

    return result.text if result.reason == speechsdk.ResultReason.RecognizedSpeech else None

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('azure-batch-stt: "{}"'.format(
        azure_batch_stt(t['filename'], t['lang'], t['encoding'])
    ))

 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

import time
import azure.cognitiveservices.speech as speechsdk

def azure_streaming_stt(filename: str, lang: str, encoding: str) -> str:
    speech_config = speechsdk.SpeechConfig(
        subscription=AZURE_SPEECH_KEY,
        region=AZURE_SERVICE_REGION
    )
    stream = speechsdk.audio.PushAudioInputStream()
    audio_config = speechsdk.audio.AudioConfig(stream=stream)
    speech_recognizer = speechsdk.SpeechRecognizer(
        speech_config=speech_config,
        audio_config=audio_config
    )

    # Connect callbacks to the events fired by the speech recognizer
    speech_recognizer.recognizing.connect(
        lambda evt: print('interim text: "{}"'.format(
            evt.result.text
        ))
    )
    speech_recognizer.recognized.connect(
        lambda evt:  print('azure-streaming-stt: "{}"'.format(
            evt.result.text
        ))
    )

    # start continuous speech recognition
    speech_recognizer.start_continuous_recognition()

    # push buffer chunks to stream
    buffer, rate = read_wav_file(filename)
    audio_generator = simulate_stream(buffer)
    for chunk in audio_generator:
      stream.write(chunk)
      time.sleep(0.1)  # to give callback a chance against fast loop

    # stop continuous speech recognition
    stream.close()
    time.sleep(0.5)  # give chance to VAD to kick in
    speech_recognizer.stop_continuous_recognition()
    time.sleep(0.5)  # Let all callback run

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    azure_streaming_stt(t['filename'], t['lang'], t['encoding'])

 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

import os

from ibm_watson import SpeechToTextV1
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator

def watson_batch_stt(filename: str, lang: str, encoding: str) -> str:
    authenticator = IAMAuthenticator(WATSON_API_KEY)
    speech_to_text = SpeechToTextV1(authenticator=authenticator)
    speech_to_text.set_service_url(WATSON_STT_URL)

    with open(filename, 'rb') as audio_file:
        response = speech_to_text.recognize(
            audio=audio_file,
            content_type='audio/{}'.format(
                os.path.splitext(filename)[1][1:]
            ),
            model=lang + '_BroadbandModel',
            max_alternatives=3,
        ).get_result()

    return response['results'][0]['alternatives'][0]['transcript']

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('watson-batch-stt: "{}"'.format(
        watson_batch_stt(t['filename'], t['lang'], t['encoding'])
    ))

  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

import json
import logging
import os
from queue import Queue
from threading import Thread
import time

from ibm_watson import SpeechToTextV1
from ibm_watson.websocket import RecognizeCallback, AudioSource
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator

# Watson websocket prints justs too many debug logs, so disable it
logging.disable(logging.CRITICAL)

# Chunk and buffer size
CHUNK_SIZE = 4096
BUFFER_MAX_ELEMENT = 10

# A callback class to process various streaming STT events
class MyRecognizeCallback(RecognizeCallback):
    def __init__(self):
        RecognizeCallback.__init__(self)
        self.transcript = None

    def on_transcription(self, transcript):
        # print('transcript: {}'.format(transcript))
        pass

    def on_connected(self):
        # print('Connection was successful')
        pass

    def on_error(self, error):
        # print('Error received: {}'.format(error))
        pass

    def on_inactivity_timeout(self, error):
        # print('Inactivity timeout: {}'.format(error))
        pass

    def on_listening(self):
        # print('Service is listening')
        pass

    def on_hypothesis(self, hypothesis):
        # print('hypothesis: {}'.format(hypothesis))
        pass

    def on_data(self, data):
        self.transcript = data['results'][0]['alternatives'][0]['transcript']
        print('{0}final: {1}'.format(
            '' if data['results'][0]['final'] else 'not ',
            self.transcript
        ))

    def on_close(self):
        # print("Connection closed")
        pass

def watson_streaming_stt(filename: str, lang: str, encoding: str) -> str:
    authenticator = IAMAuthenticator(WATSON_API_KEY)
    speech_to_text = SpeechToTextV1(authenticator=authenticator)
    speech_to_text.set_service_url(WATSON_STT_URL)

    # Make watson audio source fed by a buffer queue
    buffer_queue = Queue(maxsize=BUFFER_MAX_ELEMENT)
    audio_source = AudioSource(buffer_queue, True, True)

    # Callback object
    mycallback = MyRecognizeCallback()

    # Read the file
    buffer, rate = read_wav_file(filename)

    # Start Speech-to-Text recognition thread
    stt_stream_thread = Thread(
        target=speech_to_text.recognize_using_websocket,
        kwargs={
            'audio': audio_source,
            'content_type': 'audio/l16; rate={}'.format(rate),
            'recognize_callback': mycallback,
            'interim_results': True
        }
    )
    stt_stream_thread.start()

    # Simulation audio stream by breaking file into chunks and filling buffer queue
    audio_generator = simulate_stream(buffer, CHUNK_SIZE)
    for chunk in audio_generator:
        buffer_queue.put(chunk)
        time.sleep(0.5)  # give a chance to callback

    # Close the audio feed and wait for STTT thread to complete
    audio_source.completed_recording()
    stt_stream_thread.join()

    # send final result
    return mycallback.transcript

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('watson-cloud-streaming-stt: "{}"'.format(
        watson_streaming_stt(t['filename'], t['lang'], t['encoding'])
    ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pocketsphinx
import os

MODELDIR = os.path.join(os.path.dirname(pocketsphinx.__file__), 'model')

config = pocketsphinx.Decoder.default_config()
config.set_string('-hmm', os.path.join(MODELDIR, 'en-us'))
config.set_string('-lm', os.path.join(MODELDIR, 'en-us.lm.bin'))
config.set_string('-dict', os.path.join(MODELDIR, 'cmudict-en-us.dict'))

decoder = pocketsphinx.Decoder(config)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def sphinx_batch_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)
    decoder.start_utt()
    decoder.process_raw(buffer, False, False)
    decoder.end_utt()
    hypothesis = decoder.hyp()
    return hypothesis.hypstr

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text'])
    )
    print('sphinx-batch-stt: "{}"'.format(
        sphinx_batch_stt(t['filename'], t['lang'], t['encoding'])
    ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def sphinx_streaming_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)
    audio_generator = simulate_stream(buffer)

    decoder.start_utt()
    for chunk in audio_generator:
        decoder.process_raw(chunk, False, False)
    decoder.end_utt()

    hypothesis = decoder.hyp()
    return hypothesis.hypstr

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('sphinx-streaming-stt: "{}"'.format(
        sphinx_streaming_stt(t['filename'], t['lang'], t['encoding'])
    ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import deepspeech

model_file_path = 'deepspeech-0.6.0-models/output_graph.pbmm'
beam_width = 500
model = deepspeech.Model(model_file_path, beam_width)

# Add language model for better accuracy
lm_file_path = 'deepspeech-0.6.0-models/lm.binary'
trie_file_path = 'deepspeech-0.6.0-models/trie'
lm_alpha = 0.75
lm_beta = 1.85
model.enableDecoderWithLM(lm_file_path, trie_file_path, lm_alpha, lm_beta)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import numpy as np

def deepspeech_batch_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)
    data16 = np.frombuffer(buffer, dtype=np.int16)
    return model.stt(data16)

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('deepspeech-batch-stt: "{}"'.format(
        deepspeech_batch_stt(t['filename'], t['lang'], t['encoding'])
    ))

 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

def deepspeech_streaming_stt(filename: str, lang: str, encoding: str) -> str:
    buffer, rate = read_wav_file(filename)
    audio_generator = simulate_stream(buffer)

    # Create stream
    context = model.createStream()

    text = ''
    for chunk in audio_generator:
        data16 = np.frombuffer(chunk, dtype=np.int16)
        # feed stream of chunks
        model.feedAudioContent(context, data16)
        interim_text = model.intermediateDecode(context)
        if interim_text != text:
            text = interim_text
            print('inetrim text: {}'.format(text))

    # get final resut and close stream
    text = model.finishStream(context)
    return text

# Run tests
for t in TESTCASES:
    print('\naudio file="{0}"    expected text="{1}"'.format(
        t['filename'], t['text']
    ))
    print('deepspeech-streaming-stt: "{}"'.format(
        deepspeech_streaming_stt(t['filename'], t['lang'], t['encoding'])
    ))

 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

import speech_recognition as sr
from enum import Enum, unique

@unique
class ASREngine(Enum):
    sphinx = 0
    google = 1

def speech_to_text(filename: str, engine: ASREngine, language: str, show_all: bool = False) -> str:
    r = sr.Recognizer()

    with sr.AudioFile(filename) as source:
        audio = r.record(source)

    asr_functions = {
        ASREngine.sphinx: r.recognize_sphinx,
        ASREngine.google: r.recognize_google,
    }

    response = asr_functions[engine](audio, language=language, show_all=show_all)
    return response

# Run tests
for t in TESTCASES:
    filename = t['filename']
    text = t['text']
    lang = t['lang']

    print('\naudio file="{0}"    expected text="{1}"'.format(
        filename, text
    ))
    for asr_engine in ASREngine:
        try:
            response = speech_to_text(filename, asr_engine, language=lang)
            print('{0}: "{1}"'.format(asr_engine.name, response))
        except sr.UnknownValueError:
            print('{0} could not understand audio'.format(
                asr_engine.name
            ))
        except sr.RequestError as e:
            print('{0} error: {0}'.format(asr_engine.name, e))