test_rospeex_ss_say.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
import codecs


class TestRospeexSS(unittest.TestCase):
    """ Speech Synthesis Test class """
    def __init__(self, *args, **kwargs):
        super(TestRospeexSS, self).__init__(*args, **kwargs)
        self._ready = 1
        self._wait_flag = True

    def _ss_response(self, message):
        """
        speech synthesis response callback
        @param message: message class
        @type  message: str
        """
        self._ready = 1
        print 'received response'

    def read_input_data(self, target_file):
        """
        read input data
        @param target_file: target test data
        @type  target_file: str
        """
        data = ""
        try:
            print target_file
            f_log = codecs.open(target_file, 'r', 'utf_8')
            data = f_log.read()
            f_log.close()
        except:
            print 'no file'
        return data


    def execute_case(self, message, target_lang, target_engine, target_voice=""):
        """
        execute test case
        @param message:
        @type  message:
        @param target_lang:
        @type  target_lang:
        @param target_engine:
        @type  target_engine:
        @param target_voice:
        @type  target_voice:
        """
        self._ready = 0
        # rospy.loginfo('Message: %s  VoiceFont: %s' % (message, target_voice))
        self._interface.say(message, target_lang, target_engine, target_voice)
        while not self._ready and self._wait_flag:
            rospy.sleep(0.1)  # wait until callback response


    def test_run(self):
        """
        @param target_engine:
        @type  target_engine:
        @param target_dir:
        @type  target_dir:
        @param target_files_ja:
        @type  target_files_ja:
        @param  target_files_en:
        @type  target_files_en:
        """
        self._ready = 1
        rospy.init_node('test_rospeex_ss')
        self._interface = ROSpeexInterface()
        self._interface.init()
        self._interface.register_ss_response(self._ss_response)
        rospy.sleep(0.5)

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

        if not rospy.is_shutdown():
            for target_file in target_files:
                filename = os.path.join( target_dir, target_file )
                message = self.read_input_data(filename)
                for target_voicefont in target_voicefonts:
                    self.execute_case(message, target_language, target_engine, target_voicefont)


if __name__ == '__main__':
    rostest.rosrun('rospeex_if', 'rospeex_ss_test', TestRospeexSS, sys.argv)