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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
#!/usr/bin/env python2
# -*- coding: utf-8-*-
import os
import traceback
import json
import tempfile
import logging
from abc import ABCMeta, abstractmethod
import requests
import yaml
"""
The default Speech-to-Text implementation which relies on PocketSphinx.
"""
class TranscriptionMode:
NORMAL, KEYWORD, MUSIC = range(3)
class AbstractSTTEngine(object):
"""
Generic parent class for all STT engines
"""
__metaclass__ = ABCMeta
@classmethod
@abstractmethod
def is_available(cls):
return True
@abstractmethod
def transcribe(self, audio_file_path, PERSONA_ONLY=False, MUSIC=False):
pass
class PocketSphinxSTT(AbstractSTTEngine):
SLUG = 'sphinx'
def __init__(self, lmd="languagemodel.lm", dictd="dictionary.dic",
lmd_persona="languagemodel_persona.lm", dictd_persona="dictionary_persona.dic",
lmd_music=None, dictd_music=None, **kwargs): #FIXME: get init args from config
"""
Initiates the pocketsphinx instance.
Arguments:
speaker -- handles platform-independent audio output
lmd -- filename of the full language model
dictd -- filename of the full dictionary (.dic)
lmd_persona -- filename of the 'Persona' language model (containing, e.g., 'Jasper')
dictd_persona -- filename of the 'Persona' dictionary (.dic)
"""
self._logger = logging.getLogger(__name__)
# quirky bug where first import doesn't work
try:
import pocketsphinx as ps
except:
import pocketsphinx as ps
hmm_dir = self._get_hmm_dir()
self._logfiles = {}
with tempfile.NamedTemporaryFile(prefix='psdecoder_music_', suffix='.log', delete=False) as f:
self._logfiles[TranscriptionMode.MUSIC] = f.name
with tempfile.NamedTemporaryFile(prefix='psdecoder_keyword_', suffix='.log', delete=False) as f:
self._logfiles[TranscriptionMode.KEYWORD] = f.name
with tempfile.NamedTemporaryFile(prefix='psdecoder_normal_', suffix='.log', delete=False) as f:
self._logfiles[TranscriptionMode.NORMAL] = f.name
self._decoders = {}
if lmd_music and dictd_music:
self._decoders[TranscriptionMode.MUSIC] = ps.Decoder(hmm=hmm_dir, lm=lmd_music, dict=dictd_music, logfn=self._logfiles[TranscriptionMode.MUSIC])
self._decoders[TranscriptionMode.KEYWORD] = ps.Decoder(hmm=hmm_dir, lm=lmd_persona, dict=dictd_persona, logfn=self._logfiles[TranscriptionMode.KEYWORD])
self._decoders[TranscriptionMode.NORMAL] = ps.Decoder(hmm=hmm_dir, lm=lmd, dict=dictd, logfn=self._logfiles[TranscriptionMode.NORMAL])
def __del__(self):
for filename in self._logfiles.values():
os.remove(filename)
@classmethod
def _get_hmm_dir(cls): #FIXME: Replace this as soon as we have a config module
hmm_dir = None
# Try to get hmm_dir from config
profile_path = os.path.join(os.path.dirname(__file__), 'profile.yml')
if os.path.exists(profile_path):
with open(profile_path, 'r') as f:
profile = yaml.safe_load(f)
if 'pocketsphinx' in profile and 'hmm_dir' in profile['pocketsphinx']:
hmm_dir = profile['pocketsphinx']['hmm_dir']
if not hmm_dir:
hmm_dir = "/usr/local/share/pocketsphinx/model/hmm/en_US/hub4wsj_sc_8k"
return hmm_dir
def transcribe(self, audio_file_path, mode=TranscriptionMode.NORMAL):
"""
Performs STT, transcribing an audio file and returning the result.
Arguments:
audio_file_path -- the path to the audio file to-be transcribed
PERSONA_ONLY -- if True, uses the 'Persona' language model and dictionary
MUSIC -- if True, uses the 'Music' language model and dictionary
"""
wavFile = file(audio_file_path, 'rb')
wavFile.seek(44)
decoder = self._decoders[TranscriptionMode.NORMAL]
decoder.decode_raw(wavFile)
result = decoder.get_hyp()
with open(self._logfiles[mode], 'r+') as f:
if mode == TranscriptionMode.KEYWORD:
modename = "[KEYWORD]"
elif mode == TranscriptionMode.MUSIC:
modename = "[MUSIC]"
else:
modename = "[NORMAL]"
for line in f:
self._logger.debug("%s %s", modename, line.strip())
f.truncate()
print "==================="
print "JASPER: " + result[0]
print "==================="
return result[0]
@classmethod
def is_available(cls):
return (pkgutil.get_loader('pocketsphinx') is not None and os.path.exits(cls._get_hmm_dir()))
"""
Speech-To-Text implementation which relies on the Google Speech API.
This implementation requires a Google API key to be present in profile.yml
To obtain an API key:
1. Join the Chromium Dev group: https://groups.google.com/a/chromium.org/forum/?fromgroups#!forum/chromium-dev
2. Create a project through the Google Developers console: https://console.developers.google.com/project
3. Select your project. In the sidebar, navigate to "APIs & Auth." Activate the Speech API.
4. Under "APIs & Auth," navigate to "Credentials." Create a new key for public API access.
5. Add your credentials to your profile.yml. Add an entry to the 'keys' section using the key name 'GOOGLE_SPEECH.' Sample configuration:
6. Set the value of the 'stt_engine' key in your profile.yml to 'google'
Excerpt from sample profile.yml:
...
timezone: US/Pacific
stt_engine: google
keys:
GOOGLE_SPEECH: $YOUR_KEY_HERE
"""
class GoogleSTT(AbstractSTTEngine):
SLUG = 'google'
RATE = 16000
def __init__(self, api_key=None, **kwargs): #FIXME: get init args from config
"""
Arguments:
api_key - the public api key which allows access to Google APIs
"""
self.api_key = api_key
self.http = requests.Session()
def transcribe(self, audio_fp, mode=TranscriptionMode.NORMAL):
"""
Performs STT via the Google Speech API, transcribing an audio file and returning an English
string.
Arguments:
audio_file_path -- the path to the .wav file to be transcribed
"""
url = "https://www.google.com/speech-api/v2/recognize?output=json&client=chromium&key=%s&lang=%s&maxresults=6&pfilter=2" % (
self.api_key, "en-us")
with open(audio_file_path, 'rb') as f:
data = f.read()
try:
headers = {'Content-type': 'audio/l16; rate=%s' % GoogleSTT.RATE}
response = self.http.post(url, data=data, headers=headers)
response.encoding = 'utf-8'
response_read = response.text
decoded = json.loads(response_read.split("\n")[1])
text = decoded['result'][0]['alternative'][0]['transcript']
if text:
print "==================="
print "JASPER: " + text
print "==================="
return text
except Exception:
traceback.print_exc()
@classmethod
def is_available(cls):
return True
"""
Returns a Speech-To-Text engine.
Currently, the supported implementations are the default Pocket Sphinx and
the Google Speech API
Arguments:
engine_type - one of "sphinx" or "google"
kwargs - keyword arguments passed to the constructor of the STT engine
"""
def get_engines():
return [stt_engine for stt_engine in AbstractSTTEngine.__subclasses__() if hasattr(stt_engine, 'SLUG') and stt_engine.SLUG]
def newSTTEngine(stt_engine, **kwargs):
selected_engines = filter(lambda engine: hasattr(engine, "SLUG") and engine.SLUG == stt_engine, get_engines())
if len(selected_engines) == 0:
raise ValueError("No STT engine found for slug '%s'" % stt_engine)
else:
if len(selected_engines) > 1:
print("WARNING: Multiple STT engines found for slug '%s'. This is most certainly a bug." % stt_engine)
engine = selected_engines[0]
if not engine.is_available():
raise ValueError("STT engine '%s' is not available (due to missing dependencies, missing dependencies, etc.)" % stt_engine)
return engine(**kwargs)
|
