test_rospeex_sr.py

Go to the documentation of this file.

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

# import ros packages
try:
    import roslib; roslib.load_manifest('rospeex')
except:
    pass

import rospy
import unittest
import rostest

# Import other libraries
from rospeex_if import ROSpeexInterface

# import python libraries
import os
import sys

class TestRospeexSR(unittest.TestCase):
    """ Speech Synthesis Test class """
    def setUp(self):
        self._message = ''
        self._ready = 1
        self._interface = None
        self._wait_flag = True
        self._wait_count = 0

    def _sr_response(self, message):
        """
        speech synthesis response callback

        @param message: message class
        @type  message: str
        """
        self._message = message
        self._ready = 1
        rospy.loginfo('Message Received :' + message)
        self._wait_count -= 1

    def _read_input_data(self, target_file):
        """
        read input data

        @param target_file: target test data
        @type  target_file: str
        """
        data = None
        with open(target_file, 'rb') as f_log:
            data = f_log.read()

        if data is None:
            rospy.logwarn('[%s] is not found.' % target_file)
        return data


    def _execute_case(self, voice_data, target_lang, target_engine):
        """
        execute test case

        @param message: voice data (binary wav)
        @type  message: str
        @param target_lang: recognition ranguage
        @type  target_lang: str
        @param target_engine: target speech recognition engine
        @type  target_engine: str (nict | google)
        """
        self._ready = 0
        self._interface.recognize(voice_data, target_lang, target_engine)
        self._wait_count += 1
        while not self._ready and self._wait_flag:
            rospy.sleep(0.1)  # wait until callback response

    def test_run(self):
        self._ready = 1
        rospy.init_node('test_rospeex_sr')
        self._interface = ROSpeexInterface()
        self._interface.init()
        self._interface.register_sr_response(self._sr_response)
        rospy.sleep(0.5)

        target_dir = rospy.get_param('~target_dir', '../testdata')
        target_engine = rospy.get_param('~target_engine', 'nict')
        target_engine_list = list(rospy.get_param('~target_engine_list', ''))
        target_language = rospy.get_param('~target_language', 'ja')
        target_files = list(rospy.get_param('~target_files', ''))
        self._wait_flag = rospy.get_param('~wait_flag', True)

        if len(target_engine_list) == 0:
            target_engine_list.append(target_engine)

        for engine in target_engine_list:
            if not rospy.is_shutdown():
                for target_file in target_files:
                    rospy.loginfo(target_file)
                    filename = os.path.join(target_dir, target_file)
                    voice_data = self._read_input_data(filename)
                    if voice_data:
                        self._execute_case(voice_data, target_language, engine)
                    else:
                        rospy.logerror('invalid voice data')

        while self._wait_count > 0:
                rospy.sleep(0.1)

if __name__ == '__main__':
    rostest.rosrun('rospeex_if', 'rospeex_sr_test', TestRospeexSR, sys.argv)