interface.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_if')
except:
    pass

import os
import rospy
import rospkg
import subprocess

# Import ROS messages
from rospeex_msgs.msg import SpeechRecognitionRequest
from rospeex_msgs.msg import SpeechSynthesisRequest
from rospeex_msgs.msg import SpeechRecognitionResponse
from rospeex_msgs.msg import SpeechSynthesisResponse
from rospeex_msgs.msg import SignalProcessingResponse
from rospeex_msgs.msg import SpeechSynthesisHeader
from rospeex_msgs.msg import SpeechSynthesisState

class ROSpeexInterface(object):
    """
    class:
        ROSpeexInterface class
    brief:
        Provides rospeex interface for python.
    """
    _SR_REQUEST_TOPIC_NAME = 'sr_req'
    _SR_RESPONSE_TOPIC_NAME = 'sr_res'
    _SS_REQUEST_TOPIC_NAME = 'ss_req'
    _SS_RESPONSE_TOPIC_NAME = 'ss_res'
    _SPI_RESPONSE_TOPIC_NAME = 'spi_res'
    _SPI_STATE_TOPIC_NAME = 'ss_state'
    _NO_MESSAGE_WAV_FILENAME = 'nomessage.wav'
    _ACCEPT_MESSAGE_WAV_FILENAME = 'accept.wav'

    def __init__(self):
        # define topic names
        # callback lists
        self._sr_response = None
        self._ss_response = None

        # publisher lists
        self._pub_sr = None
        self._pub_ss = None
        self._pub_ss_state = None

        # request id count
        self._ss_req_id = 0
        self._sr_req_id = 0

        self._sr_queue_num = 0
        self._sr_queue_disp = False

        # spi settings
        self._spi_language = 'ja'
        self._spi_engine = 'nict'

    def init(self, ss=True, sr=True, spi=True):
        """
        brief:
            Initializing rospeex.
        param[in]:
            ss:  set true to enable speech synthesis

            sr: set true to enable speech recognition

            spi: set true to enable signal processing interface (waveform monitor)
        """
        # enable flags
        self._ss_enable = ss
        self._sr_enable = sr
        self._spi_enable = spi

        # publish topic for ss/sr request
        if self._sr_enable == True:
            rospy.loginfo('enable speech recognition.')
            self._pub_sr = rospy.Publisher(self._SR_REQUEST_TOPIC_NAME, SpeechRecognitionRequest)
            rospy.Subscriber(self._SR_RESPONSE_TOPIC_NAME, SpeechRecognitionResponse, self._sr_response_callback)

        if self._ss_enable == True:
            rospy.loginfo('enable speech synthesis.')
            self._pub_ss = rospy.Publisher(self._SS_REQUEST_TOPIC_NAME, SpeechSynthesisRequest)
            rospy.Subscriber(self._SS_RESPONSE_TOPIC_NAME, SpeechSynthesisResponse, self._ss_response_callback)

        if self._spi_enable == True:
            rospy.loginfo('enable signal processing interface.')
            self._pub_ss_state = rospy.Publisher(self._SPI_STATE_TOPIC_NAME, SpeechSynthesisState)
            rospy.Subscriber(self._SPI_RESPONSE_TOPIC_NAME, SignalProcessingResponse, self._spi_response_callback)


    def play_sound(self, sound_path):
        """
        brief:
            playing audio file.
        param[in]:
            sound_path: sound file path
        """
        # disable mic input
        self._publish_ss_state(True)

        cmd = ['aplay', '-q', sound_path]
        try:
            subprocess.check_call(cmd)
        except subprocess.CalledProcessError as err:
            rospy.logwarn(str(err))
        finally:
            self._publish_ss_state(False)


    def _publish_ss_state(self, state):
        """
        brief:
            send ss state to spi node.
        param[in]:
            state: ss state flag.
        """
        if self._pub_ss_state:
            msg = SpeechSynthesisState()
            msg.header.request_type = SpeechSynthesisHeader.REQUEST_TYPE_SAY
            msg.header.engine = ''
            msg.header.voice_font = ''
            msg.header.language = ''
            msg.header.user = rospy.get_name()
            msg.header.request_id = ''
            msg.play_state = state
            self._pub_ss_state.publish(msg)


    def _play_package_sound(self, file_name):
        """
        brief:
            playing audio file.
        param[in]:
            file_name: sound file name (file location: rospeex/sound)
        """
        rp = rospkg.RosPack()
        rospeex_dir = rp.get_path('rospeex_if')
        sound_path  = os.path.join(rospeex_dir, 'sound', file_name)
        self.play_sound(sound_path)


    def _spi_response_callback(self, response):
        """
        brief:
            Response from signal processing interface (wave monitor).
        param[in]:
            response:  response from ros node
        """

        if self._spi_enable == True and self._pub_sr != None:
            # process messsage data
            msg = SpeechRecognitionRequest()
            msg.header.user = rospy.get_name()
            msg.header.request_id = str(self._sr_req_id)
            msg.header.language = self._spi_language
            msg.header.engine = self._spi_engine
            msg.data = response.data
            self._pub_sr.publish(msg)
            self._sr_req_id += 1
            self._sr_queue_num += 1
            if self._sr_queue_num > 1:
                self._sr_queue_disp = True
            if self._sr_queue_disp == True:
                rospy.loginfo('speech recognition in progress: %d' %self._sr_queue_num)

            # notice to user
            self._play_package_sound("accept.wav")

    def _ss_response_callback(self, response):
        """
        brief:
            Response from speech synthesis node.
        param[in]:
            response:  speech synthesis result audio file (wave)
        """
        if self._ss_response and response.header.user == rospy.get_name():
            self._ss_response(response.data)


    def register_ss_response(self, func):
        """
        brief:
            Registering a speech synthesis callback function.
        param[in]:
            func:  form: func(data)

                data: speech synthesis result audio file (wave)
        """
        self._ss_response = func


    def _sr_response_callback(self, response):
        """
        brief:
            Response from speech recognition node.
        param[in]:
            response:  speech recognition result surface string
        """

        self._sr_queue_num -= 1
        if self._sr_queue_disp == True:
            rospy.loginfo('speech recognition in progress: %d' %self._sr_queue_num)
        if self._sr_queue_num == 0:
            self._sr_queue_disp = False

        if self._sr_response and response.header.user == rospy.get_name():
            self._sr_response(response.message)

        if not response.message:
            self._play_package_sound(self._NO_MESSAGE_WAV_FILENAME)
        elif self._spi_enable != True:
            self._play_package_sound(self._ACCEPT_MESSAGE_WAV_FILENAME)


    def register_sr_response(self, func):
        """
        brief:
            Registering a speech recognition callback function.
        param[in]:
            func:  form: func(message)

                message: speech recognition result surface string
        """
        self._sr_response = func


    def set_spi_config(self, language='ja', engine='nict'):
        """
        brief:
            Setting signal processing interface (wave monitor) parameters.
        param[in]:
            language:  language setting (ja, en, zh, ko)

            engine:  speech recognition engine setting (nict or google)

                    nict: supports ja, en, zh, ko

                    google: supports ja, en
        """
        self._spi_language = language
        self._spi_engine = engine


    def say(self, message, language='ja', engine='nict', voice_font='*'):
        """
        brief:
            Performs speech synthesis, and outputs to the speaker.
        param[in]:
            message:  text for performing speech synthesis

            language:  language setting (ja, en, zh, ko)

            engine:  speech synthesis engine setting (nict or google or voicetext)

                    nict: supports ja, en, zh, ko

                    google: supports ja, en

                    voicetext: supports ja

            voice_font:  voice font setting

                    nict(ja): F128 or F117

                    nict(en): EF007

                    nict(zh): CJF101

                    nict(ko): KF001

                    google(ja/en): None

                    voicetext(ja): show, haruka, hikari, takeru, santa, bear
        """
        if self._ss_enable == True:
            msg = SpeechSynthesisRequest()
            msg.header.request_type = SpeechSynthesisHeader.REQUEST_TYPE_SAY
            msg.header.engine = engine
            msg.header.voice_font = voice_font
            msg.header.language = language
            msg.header.user = rospy.get_name()
            msg.header.request_id = str(self._ss_req_id)
            msg.message = message
            msg.memo = ""
            self._pub_ss.publish(msg)
            self._ss_req_id += 1
        else:
            rospy.loginfo('ss interface is disabled.')


    def tts(self, message, file_name, language='ja', engine='nict', voice_font='*'):
        """
        brief:
            Performs speech synthesis, and outputs to the wave file.
        param[in]:
            message:  text to perform speech synthesis

            file_name:  filepath to output

            language:  language setting (ja, en, zh, ko)

            engine:  speech synthesis engine setting (nict or google)

                    nict: supports ja, en, zh, ko

                    google: supports ja, en

            voice_font:  voice font setting

                    nict(ja): F128 or F117

                    nict(en): EF007

                    nict(zh): CJF101

                    nict(ko): KF001

                    google(ja/en): None

        """
        if self._ss_enable == True:
            msg = SpeechSynthesisRequest()
            msg.header.request_type = SpeechSynthesisHeader.REQUEST_TYPE_TTS
            msg.header.engine = engine
            msg.header.voice_font = voice_font
            msg.header.language = language
            msg.header.user = rospy.get_name()
            msg.header.request_id = str(self._ss_req_id)
            msg.message = message
            msg.memo = file_name
            self._pub_ss.publish(msg)
            self._ss_req_id += 1
        else:
            rospy.loginfo('ss interface is disabled.')


    def recognize(self, data, language='ja', engine='nict'):
        """
        brief:
            Recognizing voice data.
        param[in]:
            data:  voice file (wave format: 16kHz, 16bit, mono, LE)

            language:  language setting (ja, en, zh, ko)

            engine:  speech recognition engine setting (nict or google)

                    nict: supports ja, en, zh, ko

                    google: supports ja, en
        """
        if self._sr_enable == True:
            msg = SpeechRecognitionRequest()
            msg.header.user = rospy.get_name()
            msg.header.request_id = str(self._sr_req_id)
            msg.header.language = language
            msg.header.engine = engine
            msg.data = data
            self._pub_sr.publish(msg)
            self._sr_req_id += 1
            self._sr_queue_num += 1
            if self._sr_queue_num > 1:
                self._sr_queue_disp = True
            if self._sr_queue_disp == True:
                rospy.loginfo('speech recognition in progress: %d' %self._sr_queue_num)
        else:
            rospy.loginfo('sr interface is disabled.')