signal_processing_interface.py

Go to the documentation of this file.

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

try:
    import roslib; roslib.load_manifest('rospeex')
except:
    pass

import os
import struct
import subprocess
import traceback
import threading

# get logger
import logging
logger = logging.getLogger(__name__)
logdebug = logger.debug
loginfo = logger.info
logwarn = logger.warn
logdebug = logger.error
logfatal = logger.critical

try:
    import rospy
    logdebug = rospy.logdebug
    loginfo = rospy.loginfo
    logwarn = rospy.logwarn
    logerr = rospy.logerr
    logfatal = rospy.logfatal
except:
    pass

from rospeex_core.validators import accepts
from rospeex_core.exceptions import SignalProcessingInterfaceException

class WaveFile(object):
    """ Wave file converter class """

    def __init__(self):
        """ initialize function """
        self._data = ''
        self._header = ''
        self._count = 0


    def build(self, data):
        """
        build wave file
        @param data: voice data
        @type  data: str
        """
        self._create_header(len(data))
        self._data = self._header
        fmt = '%dH' %len(data)
        pack_data = struct.pack(fmt, *data)
        self._data += pack_data
        return self._data


    def _create_header(self, data_size):
        """
        create wave file header
        @param data_size: wav data part size
        @type  data_size: int
        """
        self._header = []
        fmt_chunk = struct.pack('4s I 2H 2I 2H', *[
            'fmt ',         # chunk_id   = "fmt "
            16,             # chunk_byte = sizeof(fmt_chunk) - 8
            1,              # format_id  = 1
            1,              # channels   = 1
            16000,          # rate       = 16000
            32000,          # velocity   = 32000
            2,              # block_size = 2
            16])            # bits = 16

        data_chunk = struct.pack('4s I', *[
            'data',         # data chunk id
            data_size*2])     # data size

        file_size = len(data_chunk) + len(fmt_chunk) + 12 - 8 + data_size*2
        wav_header = struct.pack('4s I 4s', *[
            'RIFF',         # riff header
            file_size,      # filesize - 8
            'WAVE'])        # "WAVE"

        self._header = wav_header + fmt_chunk + data_chunk


class Connector(object):
    """ AppConnector class """

    def __init__(self, app_dir, controller, nict_mmcv, mmse_setting, vad_setting, ip_addr, recv_port, send_port, log_level):
        """
        initialize function
        @param app_dir: application directory
        @type  app_dir:
        @param controller:
        @type  controller:
        @param nict_mmcv:
        @type  nict_mmcv:
        @param mmse_setting:
        @type  mmse_setting:
        @param vad_setting:
        @type  vad_setting:
        @param ip_addr:
        @type  ip_addr:
        @param recv_port:
        @type  recv_port:
        @param send_port:
        @type  send_port:
        @param log_level:
        @type  log_level:
        """
        self._app_dir = app_dir
        self._controller = os.path.join(app_dir, controller)
        self._nict_mmcv = os.path.join(app_dir, nict_mmcv)
        self._mmse_setting = os.path.join(app_dir, mmse_setting)
        self._vad_setting = os.path.join(app_dir, vad_setting)
        self._cwd = app_dir
        self._ip_addr = ip_addr
        self._recv_port = recv_port
        self._send_port = send_port
        self._log_level = log_level
        self._process = None


    def launch(self):
        """
        launch application
        @raise SignalProcessingInterfaceException: NICTmmcvController is not installed.
        """
        # create arguments
        args = [self._controller,
                '-i', self._ip_addr,
                '-m', self._mmse_setting,
                '-v', self._vad_setting,
                '-b', self._nict_mmcv,
                '-l', self._log_level,
                '-s', str(self._send_port),
                '-c', str(self._recv_port)]

        options = {'stdin': subprocess.PIPE,
                   'stdout': subprocess.PIPE,
                   'cwd': self._cwd,
                   'close_fds': True}

        # launch process
        try:
            self._process = subprocess.Popen(args, **options)

        except (OSError, ValueError) as err:
            raise SignalProcessingInterfaceException('NICTmmcvController is not installed. Exception:%s' % str(err))

        except Exception as err:
            raise SignalProcessingInterfaceException('unknown exception. Traceback: %s' % traceback.format_exc())


    def read(self, data_size):
        """
        read data from stdout
        @param data_size: read data size
        @type  data_size: integer
        @return: read data
        @rtype: str
        @raize: SignalProcessingInterfaceException: NICTmmcvController is not launch
        """
        if self._process is None:
            raise SignalProcessingInterfaceException('NICTmmcvController is not launched.')

        return self._process.stdout.read(data_size)


    def write(self, data):
        """
        write data
        @param data: write data
        @type  data: str
        @raize: SignalProcessingInterfaceException: NICTmmcvController is not launch
        """
        if self._process is None:
            raise SignalProcessingInterfaceException('NICTmmcvController is not launched.')

        self._process.stdin.write(data)
        self._process.stdin.flush()


    def poll(self):
        """
        polling process
        @return errorcode
        @rtype errorcode
        @raize: SignalProcessingInterfaceException: NICTmmcvController is not launch
        """
        if self._process is None:
            raise SignalProcessingInterfaceException('NICTmmcvController is not launch.')

        return self._process.poll()


    def shutdown(self):
        """ shutdown application """
        if self._process is not None:
            self._process.kill()
            self._process.wait()
            self._process = None


class FrameSyncHeader:
    """
    Frame Sync Header class
    """
    DATA_MARK = 0
    START_MARK = 1
    END_MARK = 2
    TOF_MARK = 9
    EOF_MARK = 10


class ConnectorInterface(threading.Thread):
    """
    application controller interface class
    """

    @accepts(app_dir=basestring, controller=basestring, nict_mmcv=basestring, mmse_setting=basestring, vad_setting=basestring, ip_addr=basestring, recv_port=int, send_port=int)
    def __init__(self, app_dir, controller, nict_mmcv, mmse_setting, vad_setting, ip_addr, recv_port, send_port, log_level):
        """
        initialize function

        @param app_dir: application directory
        @type  app_dir: str
        @param controller: controller name
        @type  controller: str
        @param nict_mmcv: mmcv applicaton path
        @type  nict_mmcv: str
        @param mmse_setting: mmse setting file path
        @type  mmse_setting: str
        @param vad_setting: vad setting file path
        @type  vad_setting: str
        @param ip_addr: controller ip address
        @type  ip_addr: str
        @param recv_port: data receive port
        @type  recv_port: int
        @param send_port: data send port
        @type  send_port: int
        @param log_level: loglevel
        @type  log_level: str
        """
        threading.Thread.__init__(self)
        self._connector = Connector(app_dir, controller, nict_mmcv, mmse_setting, vad_setting, ip_addr, recv_port, send_port, log_level)
        self._voice_tmp = []
        self._voice_state = -1
        self._is_shutdown = False
        self._callback = None
        self.FRAME_SYNC_DATA_SIZE = 320
        self.FRAME_SYNC_HEADER_SIZE = 4
        self.FRAME_SYNC_SIZE = self.FRAME_SYNC_HEADER_SIZE + self.FRAME_SYNC_DATA_SIZE

    def shutdown(self):
        """
        shutdown application
        """
        self._is_shutdown = True

    def register_callback(self, callback):
        """
        get wav data
        @param callback: wav file callback
        @type  callback: function
        """
        self._callback = callback


    def set_play_sound_state(self, state):
        """
        set play sound state
        @param state: play sound state
        @type  state: True for playing sound.
                      False for NOT playing sound.
        """
        msg = 'play_sound_on\n\n'

        if state == False:
            msg = 'play_sound_off\n\n'

        self._connector.write(msg)


    def run(self):
        """
        thread main
        """
        # launch application
        self._connector.launch()

        # main loop
        while self._is_shutdown == False:
            # check application state
            if self._connector.poll() is not None:
                raise SignalProcessingInterfaceException('NICTmmcvController is terminated.')

            # read applcation data
            recv_data = self._connector.read(self.FRAME_SYNC_SIZE)
            if len(recv_data) == self.FRAME_SYNC_SIZE:
                header = struct.unpack('<I', recv_data[0:self.FRAME_SYNC_HEADER_SIZE])[0]
                # DATA Frame --------------------------------------------------------------->
                if header == FrameSyncHeader.DATA_MARK:
                    if self._voice_state == FrameSyncHeader.START_MARK:
                        fmt = '<%dH'%(self.FRAME_SYNC_DATA_SIZE/2)
                        voice_data = struct.unpack(fmt, recv_data[self.FRAME_SYNC_HEADER_SIZE:self.FRAME_SYNC_SIZE])
                        self._voice_tmp.extend(list(voice_data))
                # START Frame -------------------------------------------------------------->
                elif header == FrameSyncHeader.START_MARK:
                    loginfo('voice start ----------------------->')
                    self._voice_state = header
                    self._voice_tmp = []
                # END Frame ---------------------------------------------------------------->
                elif header == FrameSyncHeader.END_MARK:
                    if self._voice_state == FrameSyncHeader.START_MARK:
                        loginfo('voice end <-------------------------')
                        self._voice_state = header
                        wav = WaveFile()
                        wav_data = wav.build(self._voice_tmp)
                        self._voice_tmp = []
                        if self._callback != None:
                            self._callback(wav_data)
                # TOF/EOF Frame ------------------------------------------------------------->
                elif header == FrameSyncHeader.TOF_MARK or header == FrameSyncHeader.EOF_MARK:
                    self._voice_state = header
                    self._voice_state = []
                else:
                    raise SignalProcessingInterfaceException('Invalid header: %d' %header)

        # shutdown connector
        self._connector.shutdown()