session.py

Go to the documentation of this file.

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

# python libraries
import abc
import threading
from Queue import Queue
from Queue import Empty

# local libraries
from rospeex_core import logging_util
from rospeex_core import exceptions as ext


__all__ = [
    'ISession',
    'Session',
    'IState',
    'SessionState',
    'PacketType',
]


# get logger
logger = logging_util.get_logger(__name__)


class ISession(object):
    """ Session Interface class
    """
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def set_next_state(self, state):
        pass    # pragma: no cover

    @abc.abstractmethod
    def add_packet(self, type, data):
        pass    # pragma: no cover

    @abc.abstractmethod
    def check_completion(self):
        pass    # pragma: no cover

    @abc.abstractmethod
    def result(self):
        pass    # pragma: no cover

    @abc.abstractmethod
    def register_result_cb(self):
        pass    # pragma: no cover

    @abc.abstractmethod
    def unregister_result_cb(self):
        pass    # pragma: no cover


class IState(object):
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def run(self, packet_data):
        pass    # pragma: no cover

    @abc.abstractmethod
    def next(self, packet_type):
        pass    # pragma: no cover

    @abc.abstractmethod
    def result(self):
        pass    # pragma: no cover

    @abc.abstractmethod
    def state(self):
        pass    # pragma: no cover


class PacketType(object):
    """ Packet Type
    """
    START = 0
    DATA = 1
    END = 2
    CANCEL = 3

    __TYPE_TO_STR = {
        START: 'START',
        END: 'END',
        DATA: 'DATA',
        CANCEL: 'CANCEL',
    }

    @classmethod
    def to_str(cls, packet_type):
        if packet_type in cls.__TYPE_TO_STR.keys():
            return cls.__TYPE_TO_STR[packet_type]
        return 'UNKNOWN TYPE'

    @classmethod
    def check_packet_type(cls, packet_type):
        if packet_type not in cls.__TYPE_TO_STR.keys():
            raise ext.InvalidPacketTypeException()


class SessionState():
    ERROR = -1
    INIT = 0
    START = 1
    DATA = 2
    END = 3

    __STATE_TO_STR = {
        ERROR: 'ERROR',
        INIT: 'INIT',
        START: 'START',
        DATA: 'DATA',
        END: 'END'
    }

    @classmethod
    def to_str(cls, state):
        if state in cls.__STATE_TO_STR.keys():
            return cls.__STATE_TO_STR[state]
        return 'UNKNOWN STATE'

    @classmethod
    def check_state(cls, state):
        if state not in cls.__STATE_TO_STR.keys():
            raise ext.InvalidSessionStateException()


class Session(ISession, threading.Thread):

    def __init__(self, state):
        """ init nict speech recognition session

        @param num_workers: number of worker thread
        """
        threading.Thread.__init__(self)

        # data settings
        self._data_que = Queue()

        # que settings
        self._state = state
        self._next_state = None
        self._next_stata_lock = threading.Lock()

        # callback settings
        self._result_cb_lock = threading.Lock()
        self._result_cb_list = []

        # thread settings
        self._stop_request = threading.Event()
        # self.daemon = True

    def set_next_state(self, state):
        """ add packet

        @param state: next state
        @return: None
        """
        with self._next_stata_lock:
            self._next_state = state

    def _get_next_state(self):
        """ get next state
        @return: next state
        """
        ret = self._state
        with self._next_stata_lock:
            if self._next_state:
                ret = self._next_state
                self._next_state = None
        return ret

    def add_packet(self, packet_type, data):
        """ add send packet
        @param packet_type: packet type
        @type  packet_type: PakcetType (0, 1, 2, 3)
        @param data: input data
        @raises InvalidPacketTypeException:
        @return: None
        """
        # check packet type
        PacketType.check_packet_type(packet_type)

        # check current state
        self._data_que.put_nowait([packet_type, data])

    def run(self):
        """ run thread
        @return: None
        """
        while not self._stop_request.isSet():
            # process send packet
            try:
                # get data from queue
                data_type, data = self._data_que.get(timeout=0.05)

                self._process(data_type, data)

                # finish process data
                self._data_que.task_done()

            except Empty:
                pass

    def _process(self, data_type, data):
        """ process input data
        @param data_type: packet data type
        @param data: data
        """
        # get next state
        self._state = self._state.next(data_type)
        self._state.run(data)

        if self._state.state() == SessionState.ERROR:
            logger.debug('Error Occured. [%s]', self._state.result())
            self._state = self._get_next_state()

        elif self._state.state() == SessionState.END:
            self._result_text = self._state.result()
            logger.info('End Session.')

            # call all callback functions
            with self._result_cb_lock:
                for callback in self._result_cb_list:
                    callback(self._result_text)

            # renew state
            self._state = self._get_next_state()

    def join(self, timeout=None):
        """ end thread
        @param timeout: time out time [s]
        @return: None
        """
        self._stop_request.set()
        self._data_que.join()
        super(Session, self).join(timeout)

    def check_completion(self):
        """ check completion tasks
        @return: True for finish tasks / False for executing tasks
        """
        return self._data_que.qsize() == 0

    def wait_completion(self, timeout=None):
        """ wait completions seesion task
        @return: None
        """
        self._data_que.join(timeout)

    def result(self):
        """ get result text
        @return: result text
        """
        return self._result_text

    def state(self):
        """ get current session state
        @return: session state
        """
        return self._state.state()

    def register_result_cb(self, cb):
        """ register result callback function
        @param cb: callback function
        @return: None
        """
        with self._result_cb_lock:
            self._result_cb_list.append(cb)
            self._result_cb_list = set(self._result_cb_list)

    def unregister_result_cb(self, cb):
        """ unregister result callback function
        @param cb: callback function
        @return: None
        """
        with self._result_cb_lock:
            new_list = [c for c in self._result_cb_list if c is not cb]
            self._result_cb_list = new_list