nict.py

Go to the documentation of this file.

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

import urllib2
import json
import base64
import socket
import traceback
import logging


# import library
from rospeex_core.validators import accepts
from rospeex_core.validators import check_language
from rospeex_core import exceptions as ext
from rospeex_core.ss.base import IClient


logger = logging.getLogger(__name__)


class Client(IClient):
    """ SpeechSynthesisCient_NICT class """
    URL = 'http://rospeex.nict.go.jp/nauth_json/jsServices/VoiceTraSS'
    FORMAT = 'x-wav'
    LANGUAGES = ['ja', 'en', 'zh', 'ko']
    VOICEFONT_DICT = {
        'ja': ['F128', 'F117', '*'],
        'en': ['EF007', '*'],
        'zh': ['CJF101', '*'],
        'ko': ['KF001', '*']
    }

    def __init__(self):
        """ initialize function """
        pass

    @accepts(message=basestring, language=str, voice_font=str, timeout=int)
    def request(
        self,
        message,
        language='ja',
        voice_font='*',
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT
    ):
        """
        Send speech synthesis request to server,
        and get speech synthesis result.
        @param message: message
        @type  message: str
        @param language: speech synthesis language
        @type  language: str
        @param voice_font: taraget voice font
        @type  voice_font: str
        @param timeout: request timeout time (second)
        @type  timeout: float
        @return: voice data (wav format binary)
        @rtype: str
        @raise InvalidResponseException:
        @raise InvalidRequestException: send invalid request.
        @raise InvalidResponseException: server error.
        @raise RequestTimeoutException: timeout error.
        """
        check_language(language, self.LANGUAGES)
        self._check_voice_font(voice_font, language)

        # get tts data
        decorded_data = self._request_tts(
            message,
            language,
            voice_font,
            timeout
        )

        # check decorded data format
        ret_code = self._check_decorded_data(decorded_data)

        # check fault code
        voice = None
        if not ret_code:
            if decorded_data['error']['faultCode'] == 'Server.userException':
                msg = 'the format is not supported.'
                raise ext.InvalidResponseException(msg)
            msg = 'server response error. msg:%s' % decorded_data['error']
            raise ext.InvalidResponseException(msg)
        else:
            voice = base64.b64decode(decorded_data['result']['audio'])
        return voice

    def _check_voice_font(self, voice_font, language):
        """ check voice font
        @param voice_font:
        @type  voice_font:
        @param language:
        @type  language:
        @raise ParameterException
        """
        if voice_font not in self.VOICEFONT_DICT[language]:
            msg = 'invalid voice font [{voice_font}].'\
                  ' Expect: {voice_font_list}'.format(
                        voice_font=voice_font,
                        voice_font_list=str(self.VOICEFONT_DICT[language])
                    )
            raise ext.ParameterException(msg)

    def _check_decorded_data(self, decorded_data):
        """
        @param decorded_data:
        @type  decorded_data:
        @return: True to success / False to failture
        @rtype: bool
        @raise InvalidResponseException:
        """
        # result pattern
        if not decorded_data:
            msg = 'invalid server response. response has no data.'
            raise ext.InvalidResponseException(msg)

        ret_code = False
        if 'result' in decorded_data:
            if decorded_data['result'] is not None:
                if 'audio' in decorded_data['result']:
                    ret_code = True
                else:
                    msg = 'invalid server response.'\
                          ' response has no audio data.'
                    raise ext.InvalidResponseException(msg)

        # check error pattern
        if 'error' in decorded_data:
            if decorded_data['error'] is not None:
                if 'faultCode' in decorded_data['error']:
                    ret_code = False
                else:
                    msg = 'invalid server response. response has no faultcode.'
                    raise ext.InvalidResponseException(msg)

        return ret_code

    def _request_tts(self, message, language, voice_font, timeout):
        """
        request tts to NICT server
        @param message: message
        @type  message: str
        @param language: speech synthesis language
        @type  language: str
        @param voice_font: taraget voice font
        @type  voice_font: str
        @param timeout: request timeout time (second)
        @type  timeout: float
        @return: voice data (wav format binary)
        @rtype: str
        @raise InvalidRequestException: send invalid request.
        @raise InvalidResponseException: server error.
        @raise RequestTimeoutException: timeout error.
        """
        # create request data
        data = {
            'method': 'speak',
            'params': [
                '1.1',
                {
                    'language': language,
                    'text': message,
                    'voiceType': voice_font,
                    'audioType': 'audio/%s' % self.FORMAT,
                    'applicationType': 'rospeex'
                }
            ]
        }

        # create request
        decorded_data = None
        request = urllib2.Request(self.URL)
        request.add_header('Content-Type', 'application/json')

        # get response
        try:
            response = urllib2.urlopen(
                request,
                json.dumps(data),
                timeout=timeout
            )
            decorded_data = json.loads(response.read())

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

            msg = 'request url error. Exception: %s' % str(err)
            raise ext.InvalidRequestException(msg)

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

        except socket.timeout as err:
            msg = 'request time out. Exception: %s' % str(err)
            raise ext.RequestTimeoutException(msg)

        except:
            msg = 'unknown exception. Traceback: %s' % traceback.format_exc()
            raise ext.SpeechSynthesisException(msg)

        return decorded_data