spi_connector.py

Go to the documentation of this file.

# -*- coding: utf-8 -*-

# python libraries
import time
import socket
import struct
import threading
import copy
import select

# pypi libraries

# local libraries
import logging_util
import framesync


__all__ = [
    'SPIConnector'
]


logger = logging_util.get_logger(__file__)


class ConnectorBase(threading.Thread):
    """ ConnectorBase class
    """
    RECONNECT_TIME = 2.0

    def __init__(self):
        """ init class """
        super(ConnectorBase, self).__init__()
        self._recv_callback = None
        self._connection = None
        self._is_connected = False
        self._keep_alive = True
        self._state = True
        self._ext_msg = None

    def _create_connection(self):
        """ create connection
        """
        pass

    def connect(self):
        """ connect to audio server
        """
        if self._connection:
            self.disconnect()

        try:
            self._connection = self._create_connection()
            self._is_connected = True
            logger.info('[%s] connected.', self.__class__.__name__)

        except Exception as err:
            logger.warn('[%s] %s', self.__class__.__name__, err)
            self.disconnect()
            time.sleep(self.RECONNECT_TIME)

    def disconnect(self):
        """ disconnect from server / client """
        if self._is_connected:
            if self._connection is not None:
                self._connection.close()
                self._connection = None
            self._is_connected = False

    def set_recv_callback(self, recv_cb):
        """ set receive callback

        :param recv_cb:
        """
        self._recv_callback = recv_cb

    def reset_recv_callback(self):
        """ reset receive callback
        """
        self._recv_callback = None

    def send(self, packet):
        """ send packet

        :param packet:
        """
        if self.is_connected():
            self._connection.sendall(packet)
        else:
            logger.warn('[%s] is not connected.', self.__class__.__name__)

    def is_connected(self):
        """ check connection

        :returns: True for connected / False for disconnected.
        """
        return self._is_connected

    def get_state(self):
        """ check connector status

        :returns: True for running / False for some error occured.
        """
        return self._state

    def set_state(self, state):
        """ set connector status

        :param state:
        """
        self._state = state

    def set_ext_msg(self, msg):
        """ set exception message

        :param msg:
        """
        self._ext_msg = msg

    def run(self):
        while self._keep_alive:
            try:
                # process
                self.connect()
                if self.is_connected():
                    self._process()

            except Exception as err:
                logger.warn(str(err))
                logger.info('[%s] retry to connect.', self.__class__.__name__)
                self.set_state(False)

            finally:
                self.disconnect()
                time.sleep(self.RECONNECT_TIME)
                logger.info('[%s] disconnect.', self.__class__.__name__)
                self.set_ext_msg(None)

        logger.info('[%s] exit success.', self.__class__.__name__)

    def end(self):
        self._keep_alive = False
        self.disconnect()

    def _process(self):
        pass


class VADConnector(ConnectorBase):
    """ VADConnector class
        本接続クラスは、音声情報の通信に使用される
    """
    def __init__(self, host, port, received_cb):
        """ init class """
        super(VADConnector, self).__init__()
        self._host = host
        self._port = port
        self._received_cb = received_cb
        self._received_packet = ''
        self._send_packet_que = []
        self._send_packet_lock = threading.Lock()

    def _create_connection(self):
        """ create connection
        """
        msg = '[{name}] create connection. {host}:{port}'.format(
            host=self._host,
            port=self._port,
            name=self.__class__.__name__
        )
        logger.info(msg)
        connection = socket.socket(
            socket.AF_INET,
            socket.SOCK_STREAM
        )
        connection.settimeout(10.0)
        connection.connect((self._host, self._port))
        return connection

    def _process(self):
        """ process data
        """
        while self._keep_alive:
            # read from socket
            rready, wready, xready = select.select(
                [self._connection],
                [],
                [],
                0.01
            )

            for sock in rready:
                recv_packet = sock.recv(framesync.VAD_FRAME_SIZE)
                self._process_packet(recv_packet)

            # write socket
            send_packet = self._pop_packet()
            for packet in send_packet:
                super(VADConnector, self).send(packet)

            # if error occured outside
            if self._ext_msg:
                logger.warn('[%s] %s', self.__class__.__name__, self._ext_msg)
                break

    def _process_packet(self, packet):
        """ proces packet data

        :param packet:
        """
        # TODO test
        logger.debug('receive packet [%d] from vad server.', len(packet))
        self._received_packet += packet
        while len(self._received_packet) >= framesync.VAD_FRAME_SIZE:
            # get new packet
            slice_packet = self._received_packet[:framesync.VAD_FRAME_SIZE]
            self._received_packet = self._received_packet[framesync.VAD_FRAME_SIZE:]

            # get header data
            header = struct.unpack('>1I', slice_packet[:framesync.VAD_HEADER_SIZE])[0]
            if header == framesync.HEADER_DATA_MARK:
                pass
            elif header == framesync.HEADER_START_MARK:
                logger.debug('receive start mark')
            elif header == framesync.HEADER_END_MARK:
                logger.debug('receive end mark')
            elif header == framesync.HEADER_TOF_MARK:
                logger.debug('receive tof mark')
            elif header == framesync.HEADER_EOF_MARK:
                logger.debug('receive eof mark')
            else:
                logger.debug('invalid mark %d', header)
            self._received_cb(slice_packet)

    def send(self, packet):
        """ send packet

        :param packet:
        """
        # なぜ、ブラウザから送られてくるヘッダ情報がリトルエンディアンになっているのか？
        header = struct.unpack('<1I', bytes(packet[0:4]))[0]
        logger.debug('receive frame sync header [%d] from browser.', header)
        if header in framesync.VALID_HEADER_MARK_LIST:
        # if header in [framesync.HEADER_TOF_MARK, framesync.HEADER_DATA_MARK]:
            self._push_packet(packet)
            # super(VADConnector, self).send(packet)
        else:
            logger.warn('invald frame header [%d]', header)

    def _push_packet(self, packet):
        with self._send_packet_lock:
            self._send_packet_que.append(packet)
            logger.debug('push vad packet. len [%d].', len(self._send_packet_que))

    def _pop_packet(self):
        ret_packet = []
        with self._send_packet_lock:
            ret_packet = copy.copy(self._send_packet_que)
            self._send_packet_que = []
        return ret_packet


class EPDConnector(ConnectorBase):
    """ EPDConnector class
        本サーバクラスは、EPD命令を受信するために使用される
    """
    def __init__(self, host, port, received_cb):
        """ init class """
        super(EPDConnector, self).__init__()
        self._host = host
        self._port = port
        self._sock = socket.socket(
            socket.AF_INET,
            socket.SOCK_STREAM
        )
        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._sock.bind((self._host, self._port))
        self._sock.settimeout(10.0)
        self._received_cb = received_cb
        self._send_packet_que = []
        self._send_packet_lock = threading.Lock()

    def _create_connection(self):
        """ create connection
        """
        msg = '[{name}] create connection. {host}:{port}'.format(
            host=self._host,
            port=self._port,
            name=self.__class__.__name__
        )
        logger.info(msg)
        self._sock.listen(10)
        connection, addr = self._sock.accept()
        return connection

    def _process(self):
        """ process data
        """
        while self._keep_alive:
            # read from socket
            rready, wready, xready = select.select(
                [self._connection],
                [],
                [],
                0.1
            )

            for sock in rready:
                recv_packet = sock.recv(1024)
                self._received_cb(recv_packet)

            # write socket
            send_packet = self._pop_packet()
            for packet in send_packet:
                super(EPDConnector, self).send(packet)

            # if error occured outside
            if self._ext_msg:
                logger.warn('[%s] %s', self.__class__.__name__, self._ext_msg)
                break

    def send(self, packet):
        """ send packet

        :param packet:
        """
        self._push_packet(packet)

    def _push_packet(self, packet):
        with self._send_packet_lock:
            self._send_packet_que.append(packet)

    def _pop_packet(self):
        ret_packet = []
        with self._send_packet_lock:
            ret_packet = copy.copy(self._send_packet_que)
            self._send_packet_que = []
        return ret_packet


class SPIConnector(threading.Thread):
    """ SPIConnector class
    """
    def __init__(self, vad_host, vad_port, epd_host, epd_port):
        super(SPIConnector, self).__init__()
        self._vad_connected_cb = None
        self._vad_disconnected_cb = None
        self._vad_received_cb = None
        self._epd_received_cb = None

        # init audio connector
        self._vad_connector = VADConnector(
            host=vad_host,
            port=vad_port,
            received_cb=self._vad_received
        )

        # init epd connector
        self._epd_connector = EPDConnector(
            host=epd_host,
            port=epd_port,
            received_cb=self._epd_received
        )

        self._is_connected = False
        self._keep_alive = True

    def run(self):
        """ run thread """
        # start client / server
        self._vad_connector.start()
        self._epd_connector.start()

        while self._keep_alive:
            self._check_connection()
            time.sleep(1)

            if not self._vad_connector.get_state():
                self._epd_connector.set_ext_msg('vad connector error occured')
            elif not self._epd_connector.get_state():
                self._vad_connector.set_ext_msg('epd connector error occured')
            self._vad_connector.set_state(True)
            self._epd_connector.set_state(True)

    def end(self):
        # end audio connector
        self._vad_connector.end()
        self._vad_connector.join(0.1)

        # end epd connector
        self._epd_connector.end()
        self._epd_connector.join(0.1)

        # end spi connector
        self._keep_alive = False

    def _vad_received(self, packet):
        if self._vad_received_cb:
            self._vad_received_cb(packet)

    def _epd_received(self, packet):
        if self._epd_received_cb:
            self._epd_received_cb(packet)

    def _check_connection(self):
        """ check connection

        :returns:
        """
        # get connection state
        auido_con = self._vad_connector.is_connected()
        epd_con = self._epd_connector.is_connected()
        connected = auido_con and epd_con

        # check state changed
        if connected and not self._is_connected:
            # state [disconnected -> connected]
            if self._vad_connected_cb:
                self._vad_connected_cb()
            logger.info('vad connection state changed. [Disconnected->Connected]')
            self._is_connected = True

        elif not connected and self._is_connected:
            # state [connected -> disconnected]
            if self._vad_disconnected_cb:
                self._vad_disconnected_cb()
            logger.info('vad connection state changed. [Connected->Disconnected]')
            self._is_connected = False
        return connected

    def set_vad_callback(self, vad_connected, vad_disconnected):
        """ set vad callback

        :param vad_connected:
        :param vad_disconnected:
        """
        self._vad_connected_cb = vad_connected
        self._vad_disconnected_cb = vad_disconnected

    def reset_vad_callback(self):
        """ reset vad callback
        """
        self._vad_connected_cb = None
        self._vad_disconnected_cb = None

    def set_recv_callback(self, vad_received, epd_received):
        self._vad_received_cb = vad_received
        self._epd_received_cb = epd_received

    def reset_recv_callback(self):
        self._vad_received_cb = None
        self._epd_received_cb = None

    def send_vad_packet(self, packet):
        """ send audio packet

        :param packet:
        """
        self._vad_connector.send(packet)

    def send_epd_packet(self, packet):
        """ send epd packet

        :param packet:
        """
        self._epd_connector.send(packet)


def main():
    logging_util.setup_logging()

    vad_host = '127.0.0.1'
    vad_port = 16001

    epd_host = '127.0.0.1'
    epd_port = 5002

    connector = SPIConnector(
        vad_host=vad_host,
        vad_port=vad_port,
        epd_host=epd_host,
        epd_port=epd_port
    )

    try:
        connector.start()
        logger.info('wait for [enter]')
        raw_input()

    except KeyboardInterrupt:
        pass

    except Exception as err:
        logger.error(str(err))

    finally:
        logger.info('exit connector')
        connector.end()
        logger.info('wait end threads')
        connector.join(0.1)
        logger.info('finish process')


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