interface.py

Go to the documentation of this file.

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

# Import ROS packages
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 SignalProcessingResponse
from rospeex_msgs.msg import SpeechSynthesisResponse
from rospeex_msgs.msg import SpeechSynthesisHeader
from rospeex_msgs.msg import SpeechSynthesisState
from rospeex_msgs.srv import SpeechRecognitionConfig


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'
    _SPI_CONFIG_SERVICE_NAME = 'spi_config'
    _NO_MESSAGE_WAV_FILENAME = 'nomessage.wav'
    _ACCEPT_MESSAGE_WAV_FILENAME = 'accept.wav'

    TEXT_RANGE = [0, 100]

    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

        # spi config
        self._spi_config_srv = None

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

        # 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:
            rospy.loginfo('enable speech recognition.')
            self._pub_sr = rospy.Publisher(
                self._SR_REQUEST_TOPIC_NAME,
                SpeechRecognitionRequest,
                queue_size=5
            )
            rospy.Subscriber(
                self._SR_RESPONSE_TOPIC_NAME,
                SpeechRecognitionResponse,
                self._sr_response_callback
            )

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

        if self._spi_enable:
            rospy.loginfo('enable signal processing interface.')
            self._pub_ss_state = rospy.Publisher(
                self._SPI_STATE_TOPIC_NAME,
                SpeechSynthesisState,
                queue_size=5
            )

            rospy.Subscriber(
                self._SPI_RESPONSE_TOPIC_NAME,
                SignalProcessingResponse,
                self._spi_response_callback
            )

            rospy.wait_for_service(self._SPI_CONFIG_SERVICE_NAME)
            self._spi_config_srv = rospy.ServiceProxy(
                self._SPI_CONFIG_SERVICE_NAME,
                SpeechRecognitionConfig
            )

    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:
            self._sr_req_id += 1
            self._play_package_sound(self._ACCEPT_MESSAGE_WAV_FILENAME)

    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
        """
        if self._sr_response:
            if response.header.user in (rospy.get_name(), 'spi'):
                self._sr_response(response.message)

        if not response.message:
            self._play_package_sound(self._NO_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, id, my, th, vi, fr, es)

            engine:  speech recognition engine setting (nict or google)

                    nict: supports ja, en, zh, ko, id, my, th, vi, fr, es

                    google: supports ja, en
        """
        self._spi_language = language
        self._spi_engine = engine
        if self._spi_config_srv:
            self._spi_config_srv(self._spi_engine, self._spi_language)

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

            language:  language setting (ja, en, zh, ko, id, my, th, vi)

            engine:  speech synthesis engine setting (nict or google)

                    nict: supports ja, en, zh, ko, id, my, th, vi

                    google: supports ja, en

            voice_font:  voice font setting

                    nict(ja): F128 or F117

                    nict(en): EF007

                    nict(zh): CJF101

                    nict(ko): KF001

                    nict(id/my/th/vi): None

                    google(ja/en): None

            limit: limit for message string length

                    (True:limit to 100 characters, False:no limit)
        """
        msg_check = True
        if limit:
            msg_check = self._check_text(message)

        if self._ss_enable and msg_check:
            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, id, my, th, vi)

            engine:  speech synthesis engine setting (nict or google)

                    nict: supports ja, en, zh, ko, id, my, th, vi

                    google: supports ja, en

            voice_font:  voice font setting

                    nict(ja): F128 or F117

                    nict(en): EF007

                    nict(zh): CJF101

                    nict(ko): KF001

                    nict(id/my/th/vi): None

                    google(ja/en): None

        """
        if self._ss_enable:
            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, id, my, th, vi, fr, es)

            engine:  speech recognition engine setting (nict or google)

                    nict: supports ja, en, zh, ko, id, my, th, vi, fr, es

                    google: supports ja, en
        """
        if self._sr_enable:
            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

        else:
            rospy.loginfo('sr interface is disabled.')

    def _check_text(self, text):
        """
        brief:
            check tts text
        param[in]:
            text: input mesage
        """
        res = True
        input_text = text
        if type(text) is str:
            input_text = text.decode('utf-8')

        if not len(input_text) in range(*self.TEXT_RANGE):
            msg = 'parameter failed. text length is not in range.'\
                  'Except: {range[0]} <= text_length:{value}'\
                  ' <= {range[1]}'.format(
                        range=self.TEXT_RANGE,
                        value=len(input_text)
                    )
            rospy.logerr(msg)
            res = False

        return res