signal_processing_interface.py

Go to the documentation of this file.

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

# import python libraries
import os
import struct
import subprocess
import traceback
import threading

# import local libraries
from rospeex_core.validators import accepts
from rospeex_core.exceptions import SignalProcessingInterfaceException
from rospeex_core import logging_util
from rospeex_msgs.msg import SignalProcessingStream

# get logger
logger = logging_util.get_logger(__name__)


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)
        ]

        devnull = open(os.devnull, 'w')
        options = {
            'stdin': subprocess.PIPE,
            'stdout': subprocess.PIPE,
            'stderr': devnull,
            '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 not self._process:
            msg = 'NICTmmcvController is not launched.'
            raise SignalProcessingInterfaceException(msg)
        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 not self._process:
            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 not self._process:
            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
    """
    INIT = -1
    DATA_MARK = 0
    START_MARK = 1
    END_MARK = 2
    TOF_MARK = 9
    EOF_MARK = 10

    __HEADER_TO_STR = {
        INIT: 'INIT',
        DATA_MARK: 'DATA_MARK',
        START_MARK: 'START_MARK',
        END_MARK: 'END_MARK',
        TOF_MARK: 'TOF_MARK',
        EOF_MARK: 'EOF_MARK',
    }

    @classmethod
    def is_valid(cls, header):
        if header in cls.__HEADER_TO_STR.keys():
            return True
        return False

    @classmethod
    def to_str(cls, header):
        if header in cls.__HEADER_TO_STR:
            return cls.__HEADER_TO_STR[header]
        return 'UNKNOWN'


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._stop_request = threading.Event()
        self._callback = None
        self._voice_data = []
        self._stream_callback = None
        self.FRAME_SYNC_DATA_SIZE = 320
        self.FRAME_SYNC_HEADER_SIZE = 4
        self.STREAMING_SEND_NUM = 5
        self.STREAMING_DATA_SIZE = self.STREAMING_SEND_NUM * self.FRAME_SYNC_DATA_SIZE
        self.FRAME_SYNC_SIZE = self.FRAME_SYNC_HEADER_SIZE + self.FRAME_SYNC_DATA_SIZE

    def join(self, timeout=None):
        """
        shutdown application
        """
        self._stop_request.set()
        super(ConnectorInterface, self).join(timeout)

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

    def register_stream_callback(self, callback):
        self._stream_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 not state:
            msg = 'play_sound_off\n\n'

        self._connector.write(msg)

    def _send_streaming_data(self, packet_type, packet_data=''):
        """
        @param packet_type:
        @type  packet_type:
        @param packet_data:
        @type  packet_data:
        """
        if self._stream_callback:
            self._stream_callback(packet_type, packet_data)

    def _split_framesync_packet(self, recv_data):
        """
        @param recv_data:
        @type  recv_data:
        """
        header = None
        data = None
        if len(recv_data) == self.FRAME_SYNC_SIZE:
            raw_header = recv_data[0:self.FRAME_SYNC_HEADER_SIZE]
            header = struct.unpack('<I', raw_header)[0]

        if header == FrameSyncHeader.DATA_MARK:
            fmt = '<%dH' % (self.FRAME_SYNC_DATA_SIZE/2)
            data = struct.unpack(
                fmt,
                recv_data[self.FRAME_SYNC_HEADER_SIZE:self.FRAME_SYNC_SIZE]
            )

        return header, data

    def _send_streaming_data_packet(self, streaming_data):
        """
        @param streaming_data:
        @type  streaming_data:
        """
        if len(streaming_data) > 0:
            fmt = '{}H'.format(len(streaming_data))
            pack_data = struct.pack(fmt, *streaming_data)
            self._send_streaming_data(
                SignalProcessingStream.DATA,
                pack_data
            )

    def run(self):
        """ thread main """
        streaming_data = []
        audio_state = FrameSyncHeader.INIT

        # launch application
        self._connector.launch()

        # main loop
        while not self._stop_request.isSet():
            # check application state
            if self._connector.poll():
                msg = 'NICTmmcvController is terminated.'
                raise SignalProcessingInterfaceException(msg)

            # read applcation data
            recv_data = self._connector.read(self.FRAME_SYNC_SIZE)

            # split framesync data to header and data
            header, voice_data = self._split_framesync_packet(recv_data)

            # DATA Frame
            if header == FrameSyncHeader.DATA_MARK:
                # logger.debug('DATA_MARK received')
                data_accept_state = [
                    FrameSyncHeader.START_MARK,
                    FrameSyncHeader.DATA_MARK
                ]

                if audio_state in data_accept_state:
                    streaming_data.extend(list(voice_data))
                    self._voice_data.extend(list(voice_data))
                    if len(streaming_data) > self.STREAMING_DATA_SIZE:
                        self._send_streaming_data_packet(streaming_data)
                        streaming_data = []
                    audio_state = FrameSyncHeader.DATA_MARK

            # START Frame
            elif header == FrameSyncHeader.START_MARK:
                # send start packet
                logger.debug('START_MARK received')
                self._send_streaming_data(SignalProcessingStream.START)
                audio_state = FrameSyncHeader.START_MARK

            # END Frame
            elif header == FrameSyncHeader.END_MARK:
                logger.debug('END_MARK received')
                send_end_headers = [
                    FrameSyncHeader.START_MARK,
                    FrameSyncHeader.DATA_MARK
                ]

                if audio_state in send_end_headers:
                    # send data packet
                    self._send_streaming_data_packet(streaming_data)
                    streaming_data = []

                    # send end packet
                    self._send_streaming_data(SignalProcessingStream.END)

                    if self._callback:
                        wav = WaveFile()
                        wav_data = wav.build(self._voice_data)
                        self._callback(wav_data)

                    self._voice_data = []
                audio_state = FrameSyncHeader.END_MARK

            # TOF Frame
            elif header == FrameSyncHeader.TOF_MARK:
                logger.debug('TOF_MARK received')
                audio_state = FrameSyncHeader.END_MARK

            # EOF Frame
            elif header == FrameSyncHeader.EOF_MARK:
                logger.debug('EOF_MARK received')
                audio_state = FrameSyncHeader.END_MARK

            else:
                msg = 'Invalid header: {}'.format(header)
                raise SignalProcessingInterfaceException(msg)

        # shutdown connector
        self._connector.shutdown()