ws_server_tornado.py

Go to the documentation of this file.

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

# python libraries
import os
import sys
import threading
import struct
import logging

# pypi libraries
import tornado
import tornado.ioloop
import tornado.web
import tornado.websocket
from tornado.ioloop import PeriodicCallback
from tornado.web import StaticFileHandler

# ros package
import rospkg

# local libraries
import logging_util

# import tornado
version_list = [0, 0, 0]
for idx, value in enumerate(tornado.version.split('.')):
    version_list[idx] = int(value)
major, minor, revision = version_list
version = major * 1000**2 + minor * 1000 + revision

if version < 2002000:
    from ws_tornado_210 import ModWebSocketHandler as WebSocketHandler
else:
    from tornado.websocket import WebSocketHandler


__all__ = [
    'WebAudioMonotorServer',
]


# create logger
logger = logging_util.get_logger(__file__)


class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.render('index.html')


class MyWebSocketHandler(WebSocketHandler):
    """ tornado websocket handler
    """
    def initialize(self, factory):
        logger.info('initialize handler')
        self.session_id = id(self)
        self.factory = factory
        self.ping_callback = None
        self.timeout_callback = None
        self.is_open = False
        self.ping_callback = PeriodicCallback(self._send_ping, 2000)
        self.timeout_callback = PeriodicCallback(self._check_timeout, 1000)
        self.timeout_count = 0

    def check_origin(self, origin):
        return True

    def open(self):
        logger.info('websocket handler opened')
        self.is_open = True
        self.ping_callback.start()
        self.factory.register_session(self)

    def write_message(self, message, binary=False):
        if self.is_open and not self.ws_connection.client_terminated:
            self.ws_connection.write_message(message, binary)

    def on_message(self, payload):
        logger.debug('receive message')
        is_binary = isinstance(payload, str)
        self.factory.add_recv_packet(payload, is_binary, self.session_id)

    def on_pong(self, data):
        logger.debug('on pong [%s]', self.session_id)
        self.timeout_callback.stop()
        self.timeout_count = 0

    def _send_ping(self):
        logger.debug('send ping [%s]', self.session_id)
        ping_str = '{}'.format(self.session_id).encode('utf-8')
        self.timeout_callback.stop()
        self.timeout_callback.start()
        self.ping(ping_str)

    def _check_timeout(self):
        logger.info('time out occured [%s]', self.session_id)
        self.timeout_count += 1
        if self.timeout_count > 100:
            self.close()
            self.timeout_callback.stop()
            self.factory.unregister_session(self)

    def on_close(self):
        logger.info('websocket handler closed [%s]', self.session_id)
        self.ping_callback.stop()
        self.timeout_callback.stop()
        self.is_open = False
        self.factory.unregister_session(self)

    def close(self):
        if self.is_open:
            self.ping_callback.stop()
            self.timeout_callback.stop()
            self.is_open = False
            super(WebSocketHandler, self).close()

    def on_finish(self):
        logger.info('finish connection [%s]', self.session_id)

    def onVadConnected(self):
        payload = '[VAD SERVER Connected]'.encode('utf-8')
        self.write_message(payload, False)

    def onVadDisconnected(self):
        payload = '[VAD SERVER Disconnected]'.encode('utf-8')
        self.write_message(payload, False)

    def onEpdMessage(self, text):
        payload = '[SS] {text}'.format(text=text).encode('utf-8')
        self.write_message(payload, False)

    def onSRReceived(self, text):
        payload = u'[SR_TEXT] ' + text.decode('utf-8')
        self.write_message(payload.encode('utf-8'), False)

    def onSSReceived(self, text):
        payload = u'[SS_TEXT] ' + text.decode('utf-8')
        self.write_message(payload.encode('utf-8'), False)

    def sendMessage(self, data, binary):
        self.write_message(data, binary)


class WebAudioMonitorServerFactory(object):
    """ AudioMonitorServerFactory class """
    _VAD_HEADER = 22
    _SS_TEXT = 30
    _SR_TEXT = 31

    def __init__(self, uri, recv_audio_packet_cb, recv_command_packet_cb):
        self._recv_audio_packet_cb = recv_audio_packet_cb
        self._recv_command_packet_cb = recv_command_packet_cb
        self._clients = {}
        self._vad_connected = False
        self._session_id_list = []
        self._mutex = threading.Lock()

    def add_recv_packet(self, packet, is_binary, session_id):
        """ add data to data que

        :param packet:
        :param is_binary:
        :param session_id:
        """
        if session_id in self._session_id_list:
            # check primary session
            if session_id == self._session_id_list[0]:
                self._recv_packet(packet, is_binary)
        else:
            logger.debug('target session [%s] is not found.', session_id)

    def _recv_packet(self, packet, is_binary):
        """ reveive packet process

        :param packet:
        :param is_binary:
        """
        if is_binary:
            logger.debug('receive binary data. length [%d]', len(packet))
            conv_packet = bytearray(packet)
            header = conv_packet[0]
            if header == self._VAD_HEADER:
                self._recv_audio_packet_cb(conv_packet[1:])
            else:
                logger.warn('invalid header packet. header [%d]', header)

        else:
            logger.debug('receive string data [%s]', packet)
            self._recv_command_packet_cb(packet)

    def register_session(self, session):
        """ register new session

        :param session_id:
        """
        session_id = session.session_id
        logger.info('add session [%s]', session_id)

        # add data que for this session
        with self._mutex:
            self._clients[session_id] = session
            self._session_id_list.append(session_id)

        # if vad connected, emmit vad connected event.
        if self._vad_connected:
            self.notify_vad_connected()

    def unregister_session(self, session):
        """ delete session id

        :session_id:
        """
        with self._mutex:
            session_id = session.session_id
            logger.info('delete session [%s]', session_id)
            self._session_id_list = [id for id in self._session_id_list if id != session_id]
            if session_id in self._clients:
                self._clients[session_id].close()
                del self._clients[session_id]

        # if vad connected, emmit vad connected event.
        if self._vad_connected:
            self.notify_vad_connected()

    def send_packet(self, packet):
        """ add send packet by protocol class

        :packet:
        """
        with self._mutex:
            send_packet = struct.pack('B', self._VAD_HEADER)
            send_packet += packet
            for client in self._clients.values():
                logger.debug('send packet len: %d', len(send_packet))
                client.sendMessage(send_packet, True)

    def notify_vad_connected(self):
        """ notify vad connected
        """
        with self._mutex:
            self._vad_connected = True
            for key, client in self._clients.items():
                logger.debug('notify to valid client [%s]', key)
                client.onVadConnected()

    def notify_vad_disconnected(self):
        """ notify vad disconnected

        """
        with self._mutex:
            self._vad_connected = False
            for client in self._clients.values():
                client.onVadDisconnected()

    def notify_epd_message(self, message):
        """ notify epd message
        """
        if len(message) == 0:
            return None

        with self._mutex:
            for client in self._clients.values():
                client.onEpdMessage(message)

    def _create_send_text_packet(self, header, text):
        utf8_text = text.decode('utf-8')
        encoded_text = utf8_text.encode('utf-8')
        send_packet = struct.pack('B', header)
        send_packet += encoded_text
        return send_packet

    def notify_ss_text(self, text):
        """ notify speech synthesis result
        """
        if len(text) == 0:
            return None

        send_packet = self._create_send_text_packet(self._SS_TEXT, text)
        with self._mutex:
            for client in self._clients.values():
                client.sendMessage(send_packet, True)

    def notify_sr_text(self, text):
        """ notify speech recognition result
        """
        if len(text) == 0:
            return None

        send_packet = self._create_send_text_packet(self._SR_TEXT, text)
        with self._mutex:
            for client in self._clients.values():
                client.sendMessage(send_packet, True)


class WebAudioMonotorServer(object):
    """ AudioPacketServer class """
    def __init__(self, host, port, debug):
        """ init function """
        self._thread = None
        self._factory = None
        self._instance = None
        self._host = host
        self._port = port
        self._url = 'ws://{host}:{port}'.format(
            host=host,
            port=port
        )
        self._recv_auido_packet_cb = None
        self._recv_command_packet_cb = None

    def start(self):
        """ start websocket server """
        logger.info('start websocket server at %s', self._url)
        self._factory = WebAudioMonitorServerFactory(
            self._url,
            self._recv_auido_packet,
            self._recv_command_packet
        )

        # both under one Twisted Web Site
        rp = rospkg.RosPack()
        audiomonitor_path = rp.get_path('rospeex_webaudiomonitor')
        path = os.path.join(audiomonitor_path, 'www')
        static_path = os.path.join(path, 'static')

        settings = {
            'static_path': static_path,
            'template_path': path
        }
        handlers = [
            (r'/', MainHandler),
            (r'/static/(.*)', StaticFileHandler, {'path': static_path}),
            (r'/websocket', MyWebSocketHandler, dict(factory=self._factory))
        ]
        app = tornado.web.Application(handlers, **settings)
        app.listen(self._port)

        self._instance = tornado.ioloop.IOLoop.instance()
        self._thread = threading.Thread(
            target=self._instance.start
        )
        self._thread.start()

    def stop(self):
        """ stop websocket server
        """
        self._instance.stop()
        self._thread.join(0.1)

    def _recv_auido_packet(self, packet):
        """ recv packet
        """
        if self._recv_auido_packet_cb:
            self._recv_auido_packet_cb(packet)

    def _recv_command_packet(self, packet):
        """ recv audio packet
        """
        if self._recv_command_packet_cb:
            self._recv_command_packet_cb(packet)

    def set_websocket_callback(self, recv_audio_cb, recv_command_cb):
        """ set websocket serer callback

        :param recv_packet_cb:
        """
        self._recv_auido_packet_cb = recv_audio_cb
        self._recv_command_packet_cb = recv_command_cb

    def send_packet(self, packet):
        """ notify vad connected
        """
        self._factory.send_packet(packet)

    def notify_vad_connected(self):
        """ notify vad connected
        """
        self._factory.notify_vad_connected()

    def notify_vad_disconnected(self):
        """ notify vad disconnected
        """
        self._factory.notify_vad_disconnected()

    def notify_epd_message(self, message):
        """ notify vad disconnected
        """
        self._factory.notify_epd_message(message)

    def notify_sr_text(self, text):
        """ notify speech recognition text

        :param text:
        """
        self._factory.notify_sr_text(text)

    def notify_ss_text(self, text):
        """ notify speech synthesis text

        :param text:
        """
        self._factory.notify_ss_text(text)


def main():
    # setup loggger
    logging_util.setup_logging(default_level=logging.DEBUG)
    server = WebAudioMonotorServer(
        host='127.0.0.1',
        port=9000,
        debug=True
    )
    server.start()

    logger.info('wait for enter')
    raw_input()


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