#!/usr/bin/env python2 # -*- coding: utf-8-*- import os import wave import json import tempfile import logging import urllib import urlparse from abc import ABCMeta, abstractmethod import requests import yaml import jasperpath import diagnose import vocabcompiler class AbstractSTTEngine(object): """ Generic parent class for all STT engines """ __metaclass__ = ABCMeta VOCABULARY_TYPE = None @classmethod def get_config(cls): return {} @classmethod def get_instance(cls, vocabulary_name, phrases): config = cls.get_config() if cls.VOCABULARY_TYPE: vocabulary = cls.VOCABULARY_TYPE(vocabulary_name, path=jasperpath.config( 'vocabularies')) if not vocabulary.matches_phrases(phrases): vocabulary.compile(phrases) config['vocabulary'] = vocabulary instance = cls(**config) return instance @classmethod def get_passive_instance(cls): phrases = vocabcompiler.get_keyword_phrases() return cls.get_instance('keyword', phrases) @classmethod def get_active_instance(cls): phrases = vocabcompiler.get_all_phrases() return cls.get_instance('default', phrases) @classmethod @abstractmethod def is_available(cls): return True @abstractmethod def transcribe(self, fp): pass class PocketSphinxSTT(AbstractSTTEngine): """ The default Speech-to-Text implementation which relies on PocketSphinx. """ SLUG = 'sphinx' VOCABULARY_TYPE = vocabcompiler.PocketsphinxVocabulary def __init__(self, vocabulary, hmm_dir="/usr/local/share/" + "pocketsphinx/model/hmm/en_US/hub4wsj_sc_8k"): """ Initiates the pocketsphinx instance. Arguments: vocabulary -- a PocketsphinxVocabulary instance hmm_dir -- the path of the Hidden Markov Model (HMM) """ self._logger = logging.getLogger(__name__) # quirky bug where first import doesn't work try: import pocketsphinx as ps except: import pocketsphinx as ps with tempfile.NamedTemporaryFile(prefix='psdecoder_', suffix='.log', delete=False) as f: self._logfile = f.name self._decoder = ps.Decoder(hmm=hmm_dir, logfn=self._logfile, **vocabulary.decoder_kwargs) def __del__(self): os.remove(self._logfile) @classmethod def get_config(cls): # FIXME: Replace this as soon as we have a config module config = {} # HMM dir # Try to get hmm_dir from config profile_path = jasperpath.config('profile.yml') if os.path.exists(profile_path): with open(profile_path, 'r') as f: profile = yaml.safe_load(f) try: config['hmm_dir'] = profile['pocketsphinx']['hmm_dir'] except KeyError: pass return config def transcribe(self, fp): """ Performs STT, transcribing an audio file and returning the result. Arguments: fp -- a file object containing audio data """ fp.seek(44) # FIXME: Can't use the Decoder.decode_raw() here, because # pocketsphinx segfaults with tempfile.SpooledTemporaryFile() data = fp.read() self._decoder.start_utt() self._decoder.process_raw(data, False, True) self._decoder.end_utt() result = self._decoder.get_hyp() with open(self._logfile, 'r+') as f: for line in f: self._logger.debug(line.strip()) f.truncate() print "===================" print "JASPER: " + result[0] print "===================" return [result[0]] @classmethod def is_available(cls): return diagnose.check_python_import('pocketsphinx') class GoogleSTT(AbstractSTTEngine): """ 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 """ SLUG = 'google' def __init__(self, api_key=None, language='en-us'): # FIXME: get init args from config """ Arguments: api_key - the public api key which allows access to Google APIs """ self._logger = logging.getLogger(__name__) self._request_url = None self._language = None self._api_key = None self._http = requests.Session() self.language = language self.api_key = api_key @property def request_url(self): return self._request_url @property def language(self): return self._language @language.setter def language(self, value): self._language = value self._regenerate_request_url() @property def api_key(self): return self._api_key @api_key.setter def api_key(self, value): self._api_key = value self._regenerate_request_url() def _regenerate_request_url(self): if self.api_key and self.language: query = urllib.urlencode({'output': 'json', 'client': 'chromium', 'key': self.api_key, 'lang': self.language, 'maxresults': 6, 'pfilter': 2}) self._request_url = urlparse.urlunparse( ('https', 'www.google.com', '/speech-api/v2/recognize', '', query, '')) else: self._request_url = None @classmethod def get_config(cls): # FIXME: Replace this as soon as we have a config module config = {} # HMM dir # Try to get hmm_dir from config profile_path = jasperpath.config('profile.yml') if os.path.exists(profile_path): with open(profile_path, 'r') as f: profile = yaml.safe_load(f) if 'keys' in profile and 'GOOGLE_SPEECH' in profile['keys']: config['api_key'] = profile['keys']['GOOGLE_SPEECH'] return config def transcribe(self, fp): """ 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 """ if not self.api_key: self._logger.critical('API key missing, transcription request ' + 'aborted.') return [] elif not self.language: self._logger.critical('Language info missing, transcription ' + 'request aborted.') return [] wav = wave.open(fp, 'rb') frame_rate = wav.getframerate() wav.close() data = fp.read() headers = {'content-type': 'audio/l16; rate=%s' % frame_rate} r = self._http.post(self.request_url, data=data, headers=headers) try: r.raise_for_status() except requests.exceptions.HTTPError: self._logger.critical('Request failed with http status %d', r.status_code) if r.status_code == requests.codes['forbidden']: self._logger.warning('Status 403 is probably caused by an ' + 'invalid Google API key.') return [] r.encoding = 'utf-8' try: # We cannot simply use r.json() because Google sends invalid json # (i.e. multiple json objects, seperated by newlines. We only want # the last one). response = json.loads(list(r.text.strip().split('\n', 1))[-1]) if len(response['result']) == 0: # Response result is empty raise ValueError('Nothing has been transcribed.') results = [alt['transcript'] for alt in response['result'][0]['alternative']] except ValueError as e: self._logger.warning('Empty response: %s', e.args[0]) results = [] except (KeyError, IndexError): self._logger.warning('Cannot parse response.', exc_info=True) results = [] else: # Convert all results to uppercase results = tuple(result.upper() for result in results) self._logger.info('Transcribed: %r', results) return results @classmethod def is_available(cls): return diagnose.check_network_connection() def get_engine_by_slug(slug=None): """ Returns: An STT Engine implementation available on the current platform Raises: ValueError if no speaker implementation is supported on this platform """ if not slug or type(slug) is not str: raise TypeError("Invalid slug '%s'", slug) selected_engines = filter(lambda engine: hasattr(engine, "SLUG") and engine.SLUG == slug, get_engines()) if len(selected_engines) == 0: raise ValueError("No STT engine found for slug '%s'" % slug) else: if len(selected_engines) > 1: print(("WARNING: Multiple STT engines found for slug '%s'. " + "This is most certainly a bug.") % slug) engine = selected_engines[0] if not engine.is_available(): raise ValueError(("STT engine '%s' is not available (due to " + "missing dependencies, missing " + "dependencies, etc.)") % slug) return engine def get_engines(): def get_subclasses(cls): subclasses = set() for subclass in cls.__subclasses__(): subclasses.add(subclass) subclasses.update(get_subclasses(subclass)) return subclasses return [tts_engine for tts_engine in list(get_subclasses(AbstractSTTEngine)) if hasattr(tts_engine, 'SLUG') and tts_engine.SLUG]