test_rospeex_sr_stream.py

Go to the documentation of this file.

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

# import ros packages
import rospy
import unittest
import rostest

# Import other libraries
from rospeex_if import ROSpeexInterface
from rospeex_core.sr.base.session import PacketType

# import python libraries
import os
import sys

# import ROS messages
import std_msgs.msg
from rospeex_msgs.msg import SignalProcessingStream
from rospeex_msgs.msg import SignalProcessingResponse

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

        self._sr_req_id = 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.')
        rospy.logdebug(self._message.decode('utf-8').encode('utf-8'))
        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 voice data: voice data (binary wav)
        @type  voice data: str
        @param target_lang: recognition ranguage
        @type  target_lang: str
        @param target_engine: target speech recognition engine
        @type  target_engine: str (nict | google | microsoft)
        """
        rospy.sleep(5) # wait for rospeex sr node
        self._ready = 0

        self._send_spi_stream('', PacketType.START)
        self._send_spi_stream(voice_data, PacketType.DATA)
        self._send_spi_stream('', PacketType.END)

        self._send_spi_response(voice_data)
        self._wait_count += 1

        while not self._ready and self._wait_flag:
            rospy.sleep(0.1)  # wait until callback response

    def _send_spi_stream(self, voice_data, packet_type):
        """
        publish /spi_stream topic

        @param voice_data: voice data (binary wav)
        @type  voice_data: str
        @param packet_type: send packet type (PacketType)
        @type  packet_type: integer
        """
        msg = SignalProcessingStream()
        msg.header = std_msgs.msg.Header()
        msg.header.stamp = rospy.Time.now()
        msg.header.seq = self._sr_req_id
        msg.packet_type = packet_type
        msg.packet_data = voice_data
        self._pub_spi_stream.publish(msg)
        self._sr_req_id += 1

    def _send_spi_response(self, voice_data):
        """
        publish /spi_response topic

        @param voice_data: voice data (binary wav)
        @type  voice_data: str
        """
        msg = SignalProcessingResponse()
        msg.header.user = rospy.get_name()
        msg.header.request_id = str(self._sr_req_id)
        msg.data = voice_data
        self._pub_spi.publish(msg)
        self._sr_req_id += 1

    def _dummy(self):
        return 1

    def test_run(self):
        self._ready = 1
        rospy.init_node('test_rospeex_sr')
        self._interface = ROSpeexInterface()
        self._interface.init(ss=False, sr=True, spi=True)
        self._interface.register_sr_response(self._sr_response)
        rospy.sleep(1)

        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', ''))
        expected_results = list(rospy.get_param('~expected_results', ''))
        self._wait_flag = rospy.get_param('~wait_flag', True)

        self._pub_spi = rospy.Publisher(
            self._SPI_RESPONSE_TOPIC_NAME,
            SignalProcessingResponse,
            queue_size=10
        )

        self._pub_spi_stream = rospy.Publisher(
            self._SPI_STREAM_TOPIC_NAME,
            SignalProcessingStream,
            queue_size=100
        )

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

        for engine in target_engine_list:
            cnt = 0
            if not rospy.is_shutdown():
                for target_file in target_files:
                    self._interface.set_spi_config(language=target_language, engine=engine)
                    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)

                        res = self._message.decode('utf-8').encode('utf-8')
                        msg = expected_results[cnt].encode('utf-8')
                        self.assertEqual(res, msg, 'Speech recognition failed.')
                    else:
                        rospy.logerror('invalid voice data')
                    
                    cnt += 1

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


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