162 lines
5.5 KiB
Python
162 lines
5.5 KiB
Python
import unittest
|
|
from Queue import Queue
|
|
|
|
from os.path import dirname, join
|
|
from speech_recognition import WavFile, AudioData
|
|
|
|
from mycroft.client.speech.listener import AudioConsumer, RecognizerLoop
|
|
from mycroft.client.speech.recognizer_wrapper import (
|
|
RemoteRecognizerWrapperFactory
|
|
)
|
|
|
|
__author__ = 'seanfitz'
|
|
|
|
|
|
class MockRecognizer(object):
|
|
def __init__(self, transcription=None):
|
|
self.transcriptions = []
|
|
|
|
def recognize_google(self, audio, key=None, language=None, show_all=False):
|
|
return self.transcriptions.pop(0)
|
|
|
|
def set_transcriptions(self, transcriptions):
|
|
self.transcriptions = transcriptions
|
|
|
|
|
|
class AudioConsumerTest(unittest.TestCase):
|
|
"""
|
|
AudioConsumerTest
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.loop = RecognizerLoop()
|
|
self.queue = Queue()
|
|
self.recognizer = MockRecognizer()
|
|
|
|
self.consumer = AudioConsumer(
|
|
self.loop.state,
|
|
self.queue,
|
|
self.loop,
|
|
self.loop.wakeup_recognizer,
|
|
self.loop.mycroft_recognizer,
|
|
RemoteRecognizerWrapperFactory.wrap_recognizer(
|
|
self.recognizer, 'google'))
|
|
|
|
def __create_sample_from_test_file(self, sample_name):
|
|
root_dir = dirname(dirname(dirname(__file__)))
|
|
filename = join(
|
|
root_dir, 'test', 'client', 'data', sample_name + '.wav')
|
|
wavfile = WavFile(filename)
|
|
with wavfile as source:
|
|
return AudioData(
|
|
source.stream.read(), wavfile.SAMPLE_RATE,
|
|
wavfile.SAMPLE_WIDTH)
|
|
|
|
def test_audio_pos_front_back(self):
|
|
audio = self.__create_sample_from_test_file('mycroft_in_utterance')
|
|
self.queue.put(audio)
|
|
tolerance = 4000
|
|
begin = 70000
|
|
end = 92000
|
|
|
|
monitor = {}
|
|
self.recognizer.set_transcriptions(
|
|
["what's the weather next week", ""])
|
|
|
|
def wakeword_callback(message):
|
|
monitor['pos_begin'] = message.get('pos_begin')
|
|
monitor['pos_end'] = message.get('pos_end')
|
|
|
|
self.loop.once('recognizer_loop:wakeword', wakeword_callback)
|
|
self.consumer.read_audio()
|
|
|
|
pos_begin = monitor.get('pos_begin')
|
|
self.assertIsNotNone(pos_begin)
|
|
diff = abs(pos_begin - begin)
|
|
self.assertTrue(
|
|
diff <= tolerance,
|
|
str(diff) + " is not less than " + str(tolerance))
|
|
|
|
pos_end = monitor.get('pos_end')
|
|
self.assertIsNotNone(pos_end)
|
|
diff = abs(pos_end - end)
|
|
self.assertTrue(
|
|
diff <= tolerance,
|
|
str(diff) + " is not less than " + str(tolerance))
|
|
|
|
def test_wakeword_in_beginning(self):
|
|
self.queue.put(self.__create_sample_from_test_file('mycroft'))
|
|
monitor = {}
|
|
self.recognizer.set_transcriptions([
|
|
"what's the weather next week", ""])
|
|
|
|
def callback(message):
|
|
monitor['utterances'] = message.get('utterances')
|
|
|
|
self.loop.once('recognizer_loop:utterance', callback)
|
|
self.consumer.read_audio()
|
|
utterances = monitor.get('utterances')
|
|
self.assertIsNotNone(utterances)
|
|
self.assertTrue(len(utterances) == 1)
|
|
self.assertEquals("what's the weather next week", utterances[0])
|
|
|
|
def test_wakeword_in_phrase(self):
|
|
self.queue.put(self.__create_sample_from_test_file('mycroft'))
|
|
monitor = {}
|
|
self.recognizer.set_transcriptions([
|
|
"he can do other stuff too", "what's the weather in cincinnati"])
|
|
|
|
def callback(message):
|
|
monitor['utterances'] = message.get('utterances')
|
|
|
|
self.loop.once('recognizer_loop:utterance', callback)
|
|
self.consumer.read_audio()
|
|
utterances = monitor.get('utterances')
|
|
self.assertIsNotNone(utterances)
|
|
self.assertTrue(len(utterances) == 2)
|
|
self.assertEquals("he can do other stuff too", utterances[0])
|
|
self.assertEquals("what's the weather in cincinnati", utterances[1])
|
|
|
|
def test_call_and_response(self):
|
|
audio = self.__create_sample_from_test_file('mycroft')
|
|
self.queue.put(audio)
|
|
monitor = {}
|
|
self.recognizer.set_transcriptions(["mycroft", ""])
|
|
|
|
def wakeword_callback(message):
|
|
monitor['wakeword'] = message.get('utterance')
|
|
|
|
def utterance_callback(message):
|
|
monitor['utterances'] = message.get('utterances')
|
|
|
|
self.loop.once('recognizer_loop:wakeword', wakeword_callback)
|
|
self.consumer.read_audio()
|
|
|
|
self.assertIsNotNone(monitor.get('wakeword'))
|
|
|
|
self.queue.put(audio)
|
|
self.recognizer.set_transcriptions(
|
|
["what's the weather next week", ""])
|
|
self.loop.once('recognizer_loop:utterance', utterance_callback)
|
|
self.consumer.read_audio()
|
|
|
|
utterances = monitor.get('utterances')
|
|
self.assertIsNotNone(utterances)
|
|
self.assertTrue(len(utterances) == 1)
|
|
self.assertEquals("what's the weather next week", utterances[0])
|
|
|
|
def test_ignore_wakeword_when_sleeping(self):
|
|
self.queue.put(self.__create_sample_from_test_file('mycroft'))
|
|
self.loop.sleep()
|
|
monitor = {}
|
|
self.recognizer.set_transcriptions(["", ""])
|
|
|
|
def wakeword_callback(message):
|
|
monitor['wakeword'] = message.get('utterance')
|
|
|
|
self.loop.once('recognizer_loop:wakeword', wakeword_callback)
|
|
self.consumer.read_audio()
|
|
|
|
self.assertIsNone(monitor.get('wakeword'))
|
|
self.assertTrue(self.loop.state.sleeping)
|