docomo_hoya.py

Go to the documentation of this file.

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

import rospy
import json
import traceback
import requests

# import library
from rospeex_core import logging_util
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
from rospeex_core.ss import nict

logger = logging_util.get_logger(__name__)

class Client(IClient):
    """ SpeechSynthesisCient_VoiceText class """
    LANGUAGES = ['ja']

    # docomo_hoya parameters
    VERSION = "v1"
    URL = "https://api.apigw.smt.docomo.ne.jp/voiceText/%s/textToSpeech?" % VERSION
    SPEAKER_LIST = ['show', 'haruka', 'hikari', 'takeru', 'santa', 'bear', '*']
    EMOTION_LIST = ['heppiness', 'anger', 'sadness']
    EMOTION_SPEAKER = ['haruka', 'hikari', 'takeru', 'santa', 'bear']
    EMOTION_RANGE = [1, 2]
    TEXT_RANGE = [0, 200]
    PITCH_RANGE = [50, 200]
    SPEED_RANGE = [50, 400]
    VOLUME_RANGE = [50, 200]
    DEFAULT_VOICE_FONT = 'show'

    def __init__(self):
        """ init function
        @param args: parameter list for initialize
        @type args: list
        @param kwargs: parameter dictionary for initialize
        """
        self._key = ''
        self._pitch = 100
        self._speed = 100
        self._volume = 100
        self._emotion = None
        self._emotion_level = 1
        self._load_parameter()

    def _load_parameter(self):
        """ load parameter from rospy """
        try:
            self._key = rospy.get_param('~docomo_api_key', '')
            self._pitch = rospy.get_param('~docomo_hoya_pitch', 100)
            self._speed = rospy.get_param('~docomo_hoya_speed', 100)
            self._volume = rospy.get_param('~docomo_hoya_volume', 100)
            self._emotion = rospy.get_param('~docomo_hoya_emotion', None)
            self._emotion_level = rospy.get_param(
                '~docomo_hoya_emotion_level',
                1
            )

        except Exception:
            pass

    @accepts(message=basestring, language=str, voice_font=str, timeout=int)
    def request(self, message, language='ja', voice_font='show', timeout=10):
        """
        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 SpeechSynthesisException:
        """
        # check language
        check_language(language, self.LANGUAGES)

        # check api key
        self._check_api_key()

        # check speaker
        self._check_voice_font(voice_font)
        if voice_font == '*':
            voice_font = self.DEFAULT_VOICE_FONT

        # check emotion
        if self._emotion != None:
            self._check_emotion(voice_font)

        # check text / pitch / speed / volume
        self._check_text(message)
        self._check_pitch()
        self._check_speed()
        self._check_volume()

        # send data
        try:
            client = nict.Client()
            client.request(message, language, '*', 10)
        except Exception:
            pass

        # create parameters
        params = {
            'APIKEY': self._key,
            'Content-Type': 'application/x-www-form-urlencoded',
            'text': message,
            'speaker': voice_font,
            'pitch': self._pitch,
            'speed': self._speed,
            'volume': self._volume
        }

        try:
            self.URL = self.URL
            response = requests.post(
                self.URL,
                params=params,
                timeout=timeout
            )

        except requests.exceptions.Timeout as err:
            msg = 'request time out. Exception: {err}'.format(
                err=str(err)
            )
            raise ext.RequestTimeoutException(msg)

        except requests.exceptions.ConnectionError as err:
            msg = 'network connection error. Exception: {err}'.format(
                err=str(err)
            )
            raise ext.InvalidRequestException(msg)

        except requests.exceptions.RequestException as err:
            msg = 'invalid request error. Exception: {err}'.format(
                err=str(err)
            )
            raise ext.InvalidRequestException(msg)

        # check response
        if response.status_code == 200:
            return response.content

        else:
            content = json.loads(response.content)
            self._check_errors(response.status_code, content)

    def _check_api_key(self):
        """ check api key
        @raises ParameterException
        """
        if self._key is None or self._key == '':
            msg = 'parameter failed. if you want to use docomo_hoya engine'\
                   'you MUST set api key for docomo_hoya api.'
            raise ext.ParameterException(msg)

    def _check_text(self, text):
        """ check tts text
        @param text: text
        @type str
        @raises ParameterException
        """
        if not len(text) in range(*self.TEXT_RANGE):
            msg = 'parameter failed. text length is not in range.'\
                  'Except: {range[0]} <= text_length:{value}'\
                  ' <= {range[1]}'.format(
                        range=self.TEXT_RANGE,
                        value=len(text)
                    )
            raise ext.ParameterException(msg)

    def _check_voice_font(self, voice_font):
        """ check voice font
        @param voice_font: voice font
        @type voice_font: str
        @raises ParameterException
        """
        if voice_font not in self.SPEAKER_LIST:
            msg = 'parameter failed [{voice_font}].'\
                  'you choose voice fonts following parametrs '\
                  '{speaker_list}'.format(
                        voice_font=voice_font,
                        speaker_list=str(self.SPEAKER_LIST)
                    )
            raise ext.ParameterException(msg)

    def _check_emotion(self, voice_font):
        """ check emotion and emotion_level
        @param voice_font: voice font
        @type voice_font: str
        @raises ParameterException
        """
        if self._emotion is None:
            return

        if voice_font not in self.EMOTION_SPEAKER:
            msg = 'parameter failed. if you want to use emotion option,'\
                  'you choose following speakers %s' % (self.EMOTION_SPEAKER)
            raise ext.ParameterException(msg)

        if self._emotion not in self.EMOTION_LIST:
            msg = 'parameter failed. if you want to use emotion option,'\
                  'you choose following emotions %s' % (self.EMOTION_LIST)
            raise ext.ParameterException(msg)

        if self._emotion_level not in range(*self.EMOTION_RANGE):
            msg = 'parameter failed. emotion level is not in range.'\
                  'Except: {range[0]} <= emotion_level:{value}' \
                  ' <= {range[1]}'.format(
                        range=self.EMOTION_RANGE,
                        value=self._emotion_level
                    )
            raise ext.ParameterException(msg)

    def _check_volume(self):
        """ check volume level
        @raises ParameterException
        """
        if self._volume not in range(*self.VOLUME_RANGE):
            msg = 'parameter failed. volume level is not in range.'\
                  'Except: {range[0]} <= volume:{value} <= {range[1]}'.format(
                        range=self.VOLUME_RANGE,
                        value=self._volume,
                    )
            raise ext.ParameterException(msg)

    def _check_speed(self):
        """ check speed level
        @raises ParameterException
        """
        if self._speed not in range(*self.SPEED_RANGE):
            msg = 'parameter failed. speed level is not in range.'\
                  'Except: {range[0]} <= speed:{value} <= {range[1]}'.format(
                        range=self.SPEED_RANGE,
                        value=self._speed,
                    )
            raise ext.ParameterException(msg)

    def _check_pitch(self):
        """ check speed level
        @raises ParameterException
        """
        if self._pitch not in range(*self.PITCH_RANGE):
            msg = 'parameter failed. pitch level is not in range.'\
                  'Except: {range[0]} <= pitch:{value} <= {range[1]}'.format(
                        range=self.PITCH_RANGE,
                        value=self._pitch,
                    )
            raise ext.ParameterException(msg)

    def _check_errors(self, status_code, content):
        """ check server errors
        @param status_code: status code
        @type  status_code: int
        @param content: response message
        @type  content: str
        @raises InvalidRequestException
        @raises InvalidResponseException
        """
        try:
            message = content['error']['message']
        except KeyError as err:
            msg = 'invalid response. Message: {message}'.format(
                message=str(err)
            )
            raise ext.InvalidResponseException(msg)

        if status_code == 400:
            msg = 'invalid request. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidRequestException(msg)

        elif status_code == 401:
            msg = 'auth failed. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidRequestException(msg)

        elif status_code == 404:
            msg = 'target url is not avariable. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidRequestException(msg)

        elif status_code == 405:
            msg = 'invalid request. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidRequestException(msg)

        elif status_code == 500:
            msg = 'server internal error. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidResponseException(msg)

        elif status_code == 503:
            msg = 'service unavailable error. Message: {message}'.format(
                message=message
            )
            raise ext.InvalidResponseException(msg)

        else:
            msg = 'undefined error. Message: {message} '\
                  'Traceback:{trace}'.format(
                        message=message,
                        trace=traceback.format_exc()
                    )
            raise ext.InvalidResponseException(msg)