test_sr_nict.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# import python libraries
import os
import logging
import unittest
import ConfigParser
import time

from nose.tools import raises
from rospeex_core.sr.base.session import PacketType
from rospeex_core.sr.nict.sync_client import SyncClient
from rospeex_core.sr.nict.mime_message import MIMEMessage
from rospeex_core.sr.nict.streaming_client import StreamingClient
from rospeex_core import exceptions as ext


# setup logging
FORMAT = '[%(levelname)s] %(message)s @%(filename)s:%(funcName)s:%(lineno)d'
logging.basicConfig(format=FORMAT, level=logging.DEBUG)
logger = logging.getLogger(__name__)


class TestSyncClient(unittest.TestCase):
    def setUp(self):
        base_dir = os.path.dirname(__file__)
        filename = os.path.join(base_dir, 'config.cfg')
        settings = ConfigParser.ConfigParser()
        settings.read(filename)
        self.flac_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'flac_file')
        )
        self.broken_wav_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'broken_wav_file')
        )
        self.wav_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'wav_file')
        )

        multi_lang_wav = settings.get('SpeechRecognition', 'wav_hello')
        self.wav_hello = self.get_opt_subsection(multi_lang_wav.splitlines())

        for key in self.wav_hello.keys():
            f = self.wav_hello[key]
            self.wav_hello[key] = os.path.join(
                    base_dir,
                    f)

        multi_lang_text = settings.get('SpeechRecognition', 'text_hello')
        self.text_hello = self.get_opt_subsection(multi_lang_text.splitlines())
        for key in self.text_hello.keys():
            t = self.text_hello[key]
            self.text_hello[key] = unicode(t, 'utf-8')

        self._streaming_result = None

    def get_opt_subsection(self, lines):
        conf = {}
        for item in lines:
            pair = item.split(':')
            if pair[0]:
                conf[pair[0]] = pair[1].strip()
        return conf

    @raises(ext.ParameterException)
    def test_request_invalid_audio_format_flac(self):
        request_data = open(self.flac_file, 'rb').read()
        client = SyncClient()
        client.request(request_data)

    @raises(ext.ParameterException)
    def test_request_invalid_audio_format_broken_wav(self):
        language = 'ja'
        request_data = open(self.broken_wav_file, 'rb').read()
        client = SyncClient()
        client.request(data=request_data, language=language)

    @raises(ext.ParameterException)
    def test_request_invalid_language(self):
        language = 'hoge'
        request_data = open(self.wav_file, 'rb').read()
        client = SyncClient()
        client.request(data=request_data, language=language)

    @raises(ext.RequestTimeoutException)
    def test_request_invalid_request_timeout(self):
        language = 'ja'
        request_data = open(self.wav_file, 'rb').read()
        client = SyncClient()
        client.request(data=request_data, language=language, timeout=1)

    def test_request_valid_language(self):
        for language in SyncClient.LANGUAGES:
            request_data = open(self.wav_hello[language], 'rb').read()
            client = SyncClient()
            msg = client.request(data=request_data, language=language)
            logger.info(msg)
            assert msg == self.text_hello[language]

    def test_support_streaming(self):
        client = SyncClient()
        assert not client.support_streaming()


class TestMIMEMessage(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_valid_mime_message(self):
        MIMEMessage.create_http_header()
        MIMEMessage.create_http_header('HOGE')
        MIMEMessage.create_header_request('ja', 1)
        MIMEMessage.create_header_request('en', 100)
        MIMEMessage.create_finish_request()
        MIMEMessage.create_data_request('hoge')
        MIMEMessage.create_bundle_request('hoge', 'hoge')

    @raises(Exception)
    def test_invalid_create_data_request(self):
        MIMEMessage.create_data_request(None)

    @raises(Exception)
    def test_invalid_create_bundle_request(self):
        MIMEMessage.create_bundle_request('hoge', None)


class TestStreamingClient(unittest.TestCase):
    def setUp(self):
        base_dir = os.path.dirname(__file__)
        filename = os.path.join(base_dir, 'config.cfg')
        settings = ConfigParser.ConfigParser()
        settings.read(filename)
        self.flac_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'flac_file')
        )
        self.broken_wav_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'broken_wav_file')
        )
        self.wav_file = os.path.join(
            base_dir,
            settings.get('SpeechRecognition', 'wav_file')
        )

        multi_lang_wav = settings.get('SpeechRecognition', 'wav_hello')
        self.wav_hello = self.get_opt_subsection(multi_lang_wav.splitlines())

        for key in self.wav_hello.keys():
            f = self.wav_hello[key]
            self.wav_hello[key] = os.path.join(
                    base_dir,
                    f)

        multi_lang_text = settings.get('SpeechRecognition', 'text_hello')
        self.text_hello = self.get_opt_subsection(multi_lang_text.splitlines())
        for key in self.text_hello.keys():
            t = self.text_hello[key]
            self.text_hello[key] = unicode(t, 'utf-8')

    def get_opt_subsection(self, lines):
        conf = {}
        for item in lines:
            pair = item.split(':')
            if pair[0]:
                conf[pair[0]] = pair[1].strip()
        return conf

    @raises(ext.ParameterException)
    def test_request_invalid_audio_format_flac(self):
        request_data = open(self.flac_file, 'rb').read()
        client = StreamingClient()
        client.request(request_data)

    @raises(ext.ParameterException)
    def test_request_invalid_audio_format_broken_wav(self):
        language = 'ja'
        request_data = open(self.broken_wav_file, 'rb').read()
        client = StreamingClient()
        client.request(data=request_data, language=language)

    @raises(ext.ParameterException)
    def test_request_invalid_language(self):
        language = 'hoge'
        request_data = open(self.wav_file, 'rb').read()
        client = StreamingClient()
        client.request(data=request_data, language=language)

    @raises(ext.RequestTimeoutException)
    def test_request_invalid_request_timeout(self):
        language = 'ja'
        request_data = open(self.wav_file, 'rb').read()
        client = StreamingClient()
        client.request(data=request_data, language=language, timeout=1)

    def test_request_valid_language(self):
        for language in StreamingClient.LANGUAGES:
            request_data = open(self.wav_file, 'rb').read()
            client = StreamingClient()
            msg = client.request(data=request_data, language=language)
            logger.info(msg)

    def test_support_streaming(self):
        client = StreamingClient()
        assert client.support_streaming()

    def _streaming_cb(self, msg):
        self._streaming_result = msg

    def test_add_streaming_packet(self):
        for language in StreamingClient.LANGUAGES:
            request_data = open(self.wav_hello[language], 'rb').read()
            client = StreamingClient(language)
            client.set_streaming_config(language)
            client.register_streaming_cb(self._streaming_cb)

            client.add_streaming_packet(PacketType.START, None)
            client.add_streaming_packet(PacketType.DATA, request_data)
            client.add_streaming_packet(PacketType.END, None)
            time.sleep(10)

            logger.info(self._streaming_result)
            assert self._streaming_result == self.text_hello[language]

if __name__ == '__main__':
    import rosunit
    rosunit.unitrun(
        'rospeex_core',
        'sr_nict_sync_client',
        TestSyncClient,
        None,
        coverage_packages=['rospeex_core.sr']
    )

    rosunit.unitrun(
        'rospeex_core',
        'sr_nict_streaming_client',
        TestStreamingClient,
        None,
        coverage_packages=['rospeex_core.sr']
    )

    rosunit.unitrun(
        'rospeex_core',
        'sr_nict_mime_message',
        TestMIMEMessage,
        None,
        coverage_packages=['rospeex_core.sr']
    )