node.py

Go to the documentation of this file.

# -*- coding: utf-8 -*-

# Import Python libraries
import rospy
import threading
import traceback

# Import local libraries
from rospeex_core import logging_util

# Import Messages
from rospeex_msgs.msg import SpeechRecognitionResponse
from rospeex_msgs.msg import SpeechRecognitionRequest
from rospeex_msgs.msg import SignalProcessingStream
from rospeex_msgs.srv import SpeechRecognitionConfig

# Import speech synthesis client
from rospeex_core import exceptions as ext
from rospeex_core.sr import Factory
from rospeex_core.sr import Engine


class SpeechRecognition(object):
    """ SpeechRecognition class """

    _MY_NODE_NAME = 'rospeex_sr'
    _REQUEST_TOPIC_NAME = 'sr_req'
    _RESPONSE_TOPIC_NAME = 'sr_res'
    _SPI_CONFIG_SERVICE_NAME = 'spi_config'
    _STREAM_TOPIC_NAME = 'spi_stream'

    def __init__(self):
        """ init """
        self._publisher = None
        self._user_request_dict = {}

        # streaming settings
        self._stream_client = None
        self._stream_id = 0
        self._stream_engine = Engine.NICT
        self._stream_language = 'ja'
        self._sr_config_service = None
        self._stream_client_lock = threading.Lock()

        # paramters
        self._google_api_key = None
        self._microsoft_api_key = None

    def _load_parameters(self):
        """ load paramters from roscore """
        self._google_api_key = rospy.get_param('~google_api_key', None)
        self._microsoft_api_key = rospy.get_param('~microsoft_api_key', None)

    def _publish_streaming_result(self, text):
        """ publish streaming result data
        @param text: send text
        @type  text: str
        """
        rospy.logdebug('SR result received')
        response = SpeechRecognitionResponse()
        response.header.engine = self._stream_engine
        response.header.language = self._stream_language
        response.header.user = 'spi'
        response.header.request_id = str(self._stream_id)
        response.message = text
        self._stream_id += 1
        self._publisher.publish(response)

    def _update_user_dictionary(self, header):
        """ renew user request dictionary
        @param header: request header
        @type  header: str
        """
        if header.user in self._user_request_dict:
            self._user_request_dict[header.user] += 1
            if int(header.request_id) != self._user_request_dict[header.user]:
                # check request id
                msg = 'invalid request id. user:{} '\
                      'current id:{}'\
                      'previous id:{}'.format(
                            header.user,
                            int(header.request_id),
                            self._user_request_dict[header.user]
                      )
                rospy.logwarn(msg)

            # update request id
            self._user_request_dict[header.user] = int(header.request_id)

        else:
            update_dict = {header.user: int(header.request_id)}
            self._user_request_dict.update(update_dict)
            rospy.logdebug('add new user:%s' % header.user)

    def _streaming_callback(self, request):
        """
        """
        with self._stream_client_lock:
            self._stream_client.add_streaming_packet(
                request.packet_type,
                request.packet_data
            )

    def _streaming_config(self, request):
        """
        """
        client = None
        try:
            client = self._create_client(
                engine=request.engine,
                language=request.language,
            )

        except Exception as err:
            rospy.logwarn(err)
            return False

        with self._stream_client_lock:
            self._stream_client = client

        return True

    def _sync_callback(self, request):
        """ callback function for sr_req topic
        @param request: topic data
        @type  request: str
        """
        rospy.loginfo(request)
        response = SpeechRecognitionResponse()
        response.header = request.header

        # update user dictionary
        self._update_user_dictionary(request.header)

        # Create speeech synthesis client
        try:
            # execute speech synthtesis
            client = self._create_client(
                engine=request.header.engine,
                language=request.header.language
            )

            # request client
            response.message = client.request(
                request.data,
                request.header.language
            )

            client.join()

        except ext.SpeechRecognitionException as err:
            rospy.logerr('[SpeechRecognitionException] ' + str(err))
            response.memo = str(err)

        except ext.ParameterException as err:
            rospy.logerr('[ParameterException] ' + str(err))
            response.memo = str(err)

        except ext.ServerException as err:
            rospy.logerr('[ServerException] ' + str(err))
            response.memo = str(err)

        # send speech synthesis response to subscribe node
        # sr result contains NOT ascii format.
        rospy.loginfo(response.header)
        self._publisher.publish(response)

    def _create_client(self, engine, language):
        """ create client
        @param engine:
        @param language:
        """
        client = Factory.create(
            engine=engine,
            language=language,
            google_api_key=self._google_api_key,
            microsoft_api_key=self._microsoft_api_key
        )
        client.register_streaming_cb(
            self._publish_streaming_result
        )
        return client

    def run(self):
        """ run rospeex sr node """
        rospy.init_node(self._MY_NODE_NAME)

        # load parameters
        self._load_parameters()

        # create stream client
        self._stream_client = self._create_client(
            engine=self._stream_engine,
            language=self._stream_language
        )

        self._sr_config_service = rospy.Service(
            self._SPI_CONFIG_SERVICE_NAME,
            SpeechRecognitionConfig,
            self._streaming_config
        )

        # create subscriber
        rospy.Subscriber(
            self._REQUEST_TOPIC_NAME,
            SpeechRecognitionRequest,
            self._sync_callback
        )

        rospy.Subscriber(
            self._STREAM_TOPIC_NAME,
            SignalProcessingStream,
            self._streaming_callback
        )

        # create publisher
        self._publisher = rospy.Publisher(
            self._RESPONSE_TOPIC_NAME,
            SpeechRecognitionResponse,
            queue_size=10
        )

        rospy.loginfo("start speech recognition node")
        rospy.spin()

    def join(self, timeout=None):
        self._stream_client.join(timeout)


def main():
    try:
        logging_util.setup_logging(use_ros_logging=True)
        node = SpeechRecognition()
        node.run()

    except rospy.ROSInterruptException:
        pass

    except Exception:
        rospy.logfatal(traceback.format_exc())
        return 2

    finally:
        node.join()

    return 1