node.py

Go to the documentation of this file.

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

# Import Python packages
import os
import sys
import rospy
import rospkg
import datetime
import platform
from datetime import datetime as dt

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

# Import other libraries
from rospeex_core import logging_util
from rospeex_core.spi.signal_processing_interface import ConnectorInterface
from rospeex_core import exceptions as ext

logging_util.setup_logging(use_ros_logging=True)


class SignalProcessingInterface(object):
    """ SignalProcessingInterface class """

    _SPI_RESPONSE_TOPIC_NAME = 'spi_res'
    _SS_STATE_TOPIC_NAME = 'ss_state'
    _SPI_STREAM_TOPIC_NAME = 'spi_stream'

    def __init__(self):
        """ init function """
        package = rospkg.RosPack()
        rospeex_dir = package.get_path('rospeex_core')
        self._connector = None
        self._ip_addr = '127.0.0.1'
        self._app_dir = os.path.join(rospeex_dir, 'bin')
        self._controller = 'NICTmmcvController'
        self._nict_mmcv = 'NICTmmcv'
        self._mmse_setting = 'NICTmmcv.mmse.ini'
        self._vad_setting = 'NICTmmcv.vad.ini'
        self._audio_dir = 'rospeex'
        self._recv_port = 5002
        self._send_port = 16001
        self._websocket_port = 9000
        self._vadoff_during_play = True
        self._request_id = 0
        self._stream_seq = 0
        self._log_level = 'warn'
        self._audio_monitor = None
        self._pub_spi = None
        self._pub_spi_stream = None
        self._ss_state_dict = {}

    def _ss_state_response(self, response):
        """ receive speech synthesis node state
        @param response: response data
        @type  response: str
        """
        rospy.loginfo(response)

        # add dict
        user = response.header.user
        if response.header.user not in self._ss_state_dict:
            self._ss_state_dict[user] = [0, dt.now()]

        # update state
        if response.play_state is True:
            self._ss_state_dict[user][0] += 1
        else:
            self._ss_state_dict[user][0] -= 1
        self._ss_state_dict[user][1] = dt.now()

        # update state
        state_list = [state[0] for state in self._ss_state_dict.values()]
        if self._vadoff_during_play:
            play_state = True if sum(state_list) > 0 else False
        else:
            play_state = False
        self._connector.set_play_sound_state(play_state)

    def _ui_response(self, data):
        """ receive user interface response
        @param data: voice data
        @type  data: str
        """
        msg = SignalProcessingResponse()
        msg.header.user = rospy.get_name()
        msg.header.request_id = str(self._request_id)
        msg.data = data
        self._pub_spi.publish(msg)
        self._request_id += 1

        now = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
        filename = '%s.wav' % now
        filepath = os.path.join(self._audio_dir, filename)
        with open(filepath, 'wb') as wavefile:
            wavefile.write(data)

    def _stream_response(self, packet_type, packet_data):
        """ receive user interface response

        @param data: voice data
        @type  data: str
        """
        msg = SignalProcessingStream()
        msg.header = std_msgs.msg.Header()
        msg.header.stamp = rospy.Time.now()
        msg.packet_type = packet_type
        msg.packet_data = packet_data
        self._pub_spi_stream.publish(msg)

    def _update_connection_parameter(self):
        """ update connection parameter """
        self._ip_addr = rospy.get_param(
            '~ip_addr',
            self._ip_addr
        )
        self._recv_port = rospy.get_param(
            '~recv_port',
            self._recv_port
        )
        self._send_port = rospy.get_param(
            '~send_port',
            self._send_port
        )
        self._log_level = rospy.get_param(
            '~log_level',
            self._log_level
        )
        self._websocket_port = rospy.get_param(
            '~websocket_port',
            self._websocket_port
        )
        self._audio_dir = rospy.get_param(
            '~audio_dir',
            self._audio_dir
        )
        self._vadoff_during_play = rospy.get_param(
            '~vadoff_during_play',
            self._vadoff_during_play
        )
        self._audio_monitor = None

        # show parameter list
        msg = 'parameter:\n'\
              '\tip_addr           :{}\n'\
              '\trecv_port         :{}\n'\
              '\tsend_port         :{}\n'\
              '\twebsocket_port    :{}\n'\
              '\tvadoff_during_play:{}'.format(
                    self._ip_addr,
                    self._recv_port,
                    self._send_port,
                    self._websocket_port,
                    self._vadoff_during_play
                )
        rospy.loginfo(msg)

    def _update_application_dir(self):
        """ update application directory. (32bit or 64bit) """
        arch = platform.architecture()
        try:
            if arch[0] == '32bit':
                self._app_dir = os.path.join(self._app_dir, 'x86')
            else:
                self._app_dir = os.path.join(self._app_dir, 'x64')

        except IndexError:
            msg = 'platform.architecture() returns unexcepted value. '\
                  'Value: {}'.format(str(arch))
            rospy.info(msg)
            self._app_dir = os.path.join(self._app_dir, 'x64')

    def shutdown(self):
        """ shutdown speech processing interface """
        if self._connector:
            self._connector.join()

        if self._audio_monitor:
            self._audio_monitor.exit()

    def run(self):
        """ run application """
        rospy.init_node('rospeex_spi')

        # create publisher / sucscriber
        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
        )
        rospy.Subscriber(
            self._SS_STATE_TOPIC_NAME,
            SpeechSynthesisState,
            self._ss_state_response
        )

        # update parameters
        self._update_connection_parameter()

        # update application directory
        self._update_application_dir()

        # create path
        if not os.path.exists(self._audio_dir):
            os.makedirs(self._audio_dir)

        # create connection
        self._connector = ConnectorInterface(
            app_dir=self._app_dir,
            controller=self._controller,
            nict_mmcv=self._nict_mmcv,
            mmse_setting=self._mmse_setting,
            vad_setting=self._vad_setting,
            ip_addr=self._ip_addr,
            recv_port=self._recv_port,
            send_port=self._send_port,
            log_level=self._log_level
        )

        self._connector.register_callback(self._ui_response)
        self._connector.register_stream_callback(self._stream_response)
        self._connector.start()

        # main loop
        rospy.loginfo('start signal processing interface node.')
        rate = rospy.Rate(100)
        while not rospy.is_shutdown():
            rate.sleep()


def main():
    """ mian function """
    spi_node = None

    try:
        spi_node = SignalProcessingInterface()
        spi_node.run()

    except ext.SignalProcessingInterfaceException as err:
        rospy.logfatal(str(err))

    except rospy.ROSInterruptException as err:
        rospy.loginfo(str(err))

    except Exception as err:
        rospy.logfatal(str(err))

    finally:
        if spi_node is not None:
            spi_node.shutdown()
    return 0


if __name__ == '__main__':
    sys.exit(main())