sync_client.py

Go to the documentation of this file.

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

# import python libraries
import urllib2
import json
import base64
import socket
import traceback
import ssl

# import local libraries
from rospeex_core.validators import accepts
from rospeex_core.validators import check_wave_data
from rospeex_core.validators import check_language
from rospeex_core import exceptions as ext
from rospeex_core.sr.base import IClient


class SyncClient(IClient):
    """ SpeechRecognitionClient_NICT class """

    AUDIO_LENGTH = 16000
    FRAMERATE = 16000
    CHANNELS = 1
    SAMPWIDTH = 2
    URL = 'http://rospeex.nict.go.jp/nauth_json/jsServices/VoiceTraSR'
    LANGUAGES = ['ja', 'en', 'zh', 'ko']

    @accepts(data=str, language=str, timeout=int)
    def request(
        self,
        data,
        language='ja',
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT
    ):
        """ send speech recognition request to server

        @param data: speech binary data
        @type  data: str
        @param language: speech data language
        @type  language: str
        @param timeout: time out time[ms]
        @type  timeout: int
        """
        check_wave_data(
            data,
            self.FRAMERATE,
            self.CHANNELS,
            self.SAMPWIDTH,
            self.AUDIO_LENGTH
        )
        check_language(language, self.LANGUAGES)

        # load voice data and encord to base64
        voice_send = base64.b64encode(data)

        data = {
            'method': 'recognize',
            'params': [
                '1.1',
                {
                    'audio': voice_send,
                    'audioType': 'audio/x-wav',
                    'voiceType': '*',
                    'language': language,
                    'applicationType': 'rospeex'
                }
            ]
        }

        # send request to server
        res_data = None
        try:
            req = urllib2.Request(self.URL)
            req.add_header('Content-Type', 'application/json')
            res = urllib2.urlopen(req, json.dumps(data), timeout=timeout)
            res_data = json.loads(res.read())

        except urllib2.URLError as err:
            if isinstance(err.reason, socket.timeout):
                msg = 'request time out. Exception: {}'.format(str(err))
                raise ext.RequestTimeoutException(msg)

            msg = 'request url error. Exception: {}'.format(str(err))
            raise ext.InvalidRequestException(msg)

        except urllib2.HTTPError as err:
            msg = 'http error. {} Exception: {}'.format(err.code, err.msg)
            raise ext.InvalidResponseException(msg)

        except (ssl.SSLError, socket.timeout) as err:
            raise ext.RequestTimeoutException(str(err))

        except:
            msg = 'unknown exception. Traceback: {}'.format(
                traceback.format_exc()
            )
            raise ext.SpeechRecognitionException(msg)

        # get result
        if res_data['error'] is not None:
            if 'faultCode' in res_data['error']:
                if res_data['error']['faultCode'] == 'Server.userException':
                    raise ext.InvalidResponseException(
                        'the audio format is not supported.'
                    )
            msg = 'server response error. msg:{}'.format(res_data['error'])
            raise ext.InvalidResponseException(msg)

        return res_data['result']

    def support_streaming(self):
        """
        check support streaming
        @return: True for support streaming / False for NOT support streaming
        """
        return False

    def add_streaming_packet(self, packet_type, packet_data):
        """ add streaming packet
        @param packet_type:
        @type  packet_type: int
        @param packet_data:
        @param packet_data: str
        """
        pass

    def register_streaming_cb(self, cb):
        """
        register streaming result callback
        @param cb:
        @type cb:
        """
        pass

    def unregister_streaming_cb(self, cb):
        """
        unregister streaming result callback
        @param cb:
        @type cb:
        """
        pass

    def set_streaming_config(self, language):
        """ set streaming config
        @param language:
        """
        pass

    def join(self, timeout=None):
        """
        join streaming client
        @param timeout:
        @type timeout:
        """
        pass