voice-gateway-service/alibabacloud-nls-python-sdk-dev/nls/stream_input_tts.py

# Copyright (c) Alibaba, Inc. and its affiliates.

import logging
import uuid
import json
import threading
from enum import IntEnum

from nls.core import NlsCore
from . import logging
from .exception import StartTimeoutException, WrongStateException, InvalidParameter

__STREAM_INPUT_TTS_NAMESPACE__ = "FlowingSpeechSynthesizer"

__STREAM_INPUT_TTS_REQUEST_CMD__ = {
    "start": "StartSynthesis",
    "send": "RunSynthesis",
    "stop": "StopSynthesis",
}
__STREAM_INPUT_TTS_REQUEST_NAME__ = {
    "started": "SynthesisStarted",
    "sentence_begin": "SentenceBegin",
    "sentence_synthesis": "SentenceSynthesis",
    "sentence_end": "SentenceEnd",
    "completed": "SynthesisCompleted",
    "task_failed": "TaskFailed",
}

__URL__ = "wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1"

__all__ = ["NlsStreamInputTtsSynthesizer"]


class NlsStreamInputTtsRequest:
    def __init__(self, task_id, session_id, appkey):
        self.task_id = task_id
        self.appkey = appkey
        self.session_id = session_id

    def getStartCMD(self, voice, format, sample_rate, volumn, speech_rate, pitch_rate, ex):
        self.voice = voice
        self.format = format
        self.sample_rate = sample_rate
        self.volumn = volumn
        self.speech_rate = speech_rate
        self.pitch_rate = pitch_rate
        cmd = {
            "header": {
                "message_id": uuid.uuid4().hex,
                "task_id": self.task_id,
                "name": __STREAM_INPUT_TTS_REQUEST_CMD__["start"],
                "namespace": __STREAM_INPUT_TTS_NAMESPACE__,
                "appkey": self.appkey,
            },
            "payload": {
                "session_id": self.session_id,
                "voice": self.voice,
                "format": self.format,
                "sample_rate": self.sample_rate,
                "volumn": self.volumn,
                "speech_rate": self.speech_rate,
                "pitch_rate": self.pitch_rate,
            },
        }
        if ex:
            cmd["payload"].update(ex)        
        return json.dumps(cmd)

    def getSendCMD(self, text):
        cmd = {
            "header": {
                "message_id": uuid.uuid4().hex,
                "task_id": self.task_id,
                "name": __STREAM_INPUT_TTS_REQUEST_CMD__["send"],
                "namespace": __STREAM_INPUT_TTS_NAMESPACE__,
                "appkey": self.appkey,
            },
            "payload": {"text": text},
        }
        return json.dumps(cmd)

    def getStopCMD(self):
        cmd = {
            "header": {
                "message_id": uuid.uuid4().hex,
                "task_id": self.task_id,
                "name": __STREAM_INPUT_TTS_REQUEST_CMD__["stop"],
                "namespace": __STREAM_INPUT_TTS_NAMESPACE__,
                "appkey": self.appkey,
            },
        }
        return json.dumps(cmd)


class NlsStreamInputTtsStatus(IntEnum):
    Begin = 1
    Start = 2
    Started = 3
    WaitingComplete = 3
    Completed = 4
    Failed = 5
    Closed = 6

class ThreadSafeStatus:
    def __init__(self, state: NlsStreamInputTtsStatus):
        self._state = state
        self._lock = threading.Lock()
    
    def get(self) -> NlsStreamInputTtsStatus:
        with self._lock:
            return self._state
    
    def set(self, state: NlsStreamInputTtsStatus):
        with self._lock:
            self._state = state


class NlsStreamInputTtsSynthesizer:
    """
    Api for text-to-speech
    """

    def __init__(
        self,
        url=__URL__,
        token=None,
        appkey=None,
        session_id=None,
        on_data=None,
        on_sentence_begin=None,
        on_sentence_synthesis=None,
        on_sentence_end=None,
        on_completed=None,
        on_error=None,
        on_close=None,
        callback_args=[],
    ):
        """
        NlsSpeechSynthesizer initialization

        Parameters:
        -----------
        url: str
            websocket url.
        akid: str
            access id from aliyun. if you provide a token, ignore this argument.
        appkey: str
            appkey from aliyun
        session_id: str
            32-character string, if empty, sdk will generate a random string.
        on_data: function
            Callback object which is called when partial synthesis result arrived
            arrived.
            on_result_changed has two arguments.
            The 1st argument is binary data corresponding to aformat in start
            method.
            The 2nd argument is *args which is callback_args.
        on_sentence_begin: function
            Callback object which is called when detected sentence start.
            on_start has two arguments.
            The 1st argument is message which is a json format string.
            The 2nd argument is *args which is callback_args.
        on_sentence_synthesis: function
            Callback object which is called when detected sentence synthesis.
            The incremental timestamp is returned within payload.
            on_start has two arguments.
            The 1st argument is message which is a json format string.
            The 2nd argument is *args which is callback_args.
        on_sentence_end: function
            Callback object which is called when detected sentence end.
            The timestamp of the whole sentence is returned within payload.
            on_start has two arguments.
            The 1st argument is message which is a json format string.
            The 2nd argument is *args which is callback_args.
        on_completed: function
            Callback object which is called when recognition is completed.
            on_completed has two arguments.
            The 1st argument is message which is a json format string.
            The 2nd argument is *args which is callback_args.
        on_error: function
            Callback object which is called when any error occurs.
            on_error has two arguments.
            The 1st argument is message which is a json format string.
            The 2nd argument is *args which is callback_args.
        on_close: function
            Callback object which is called when connection closed.
            on_close has one arguments.
            The 1st argument is *args which is callback_args.
        callback_args: list
            callback_args will return in callbacks above for *args.
        """
        if not token or not appkey:
            raise InvalidParameter("Must provide token and appkey")
        self.__response_handler__ = {
            __STREAM_INPUT_TTS_REQUEST_NAME__["started"]: self.__synthesis_started,
            __STREAM_INPUT_TTS_REQUEST_NAME__["sentence_begin"]: self.__sentence_begin,
            __STREAM_INPUT_TTS_REQUEST_NAME__[
                "sentence_synthesis"
            ]: self.__sentence_synthesis,
            __STREAM_INPUT_TTS_REQUEST_NAME__["sentence_end"]: self.__sentence_end,
            __STREAM_INPUT_TTS_REQUEST_NAME__["completed"]: self.__synthesis_completed,
            __STREAM_INPUT_TTS_REQUEST_NAME__["task_failed"]: self.__task_failed,
        }
        self.__callback_args = callback_args
        self.__url = url
        self.__appkey = appkey
        self.__token = token
        self.__session_id = session_id
        self.start_sended = threading.Event()
        self.started_event = threading.Event()
        self.complete_event = threading.Event()
        self.__on_sentence_begin = on_sentence_begin
        self.__on_sentence_synthesis = on_sentence_synthesis
        self.__on_sentence_end = on_sentence_end
        self.__on_data = on_data
        self.__on_completed = on_completed
        self.__on_error = on_error
        self.__on_close = on_close
        self.__allow_aformat = ("pcm", "wav", "mp3")
        self.__allow_sample_rate = (
            8000,
            11025,
            16000,
            22050,
            24000,
            32000,
            44100,
            48000,
        )
        self.state = ThreadSafeStatus(NlsStreamInputTtsStatus.Begin)     
        if not self.__session_id:
            self.__session_id = uuid.uuid4().hex   
        self.request = NlsStreamInputTtsRequest(
            uuid.uuid4().hex, self.__session_id, self.__appkey
        )

    def __handle_message(self, message):
        logging.debug("__handle_message")
        try:
            __result = json.loads(message)
            if __result["header"]["name"] in self.__response_handler__:
                __handler = self.__response_handler__[__result["header"]["name"]]
                __handler(message)
            else:
                logging.error("cannot handle cmd{}".format(__result["header"]["name"]))
                return
        except json.JSONDecodeError:
            logging.error("cannot parse message:{}".format(message))
            return

    def __syn_core_on_open(self):
        logging.debug("__syn_core_on_open")
        self.start_sended.set()

    def __syn_core_on_data(self, data, opcode, flag):
        logging.debug("__syn_core_on_data")
        if self.__on_data:
            self.__on_data(data, *self.__callback_args)

    def __syn_core_on_msg(self, msg, *args):
        logging.debug("__syn_core_on_msg:msg={} args={}".format(msg, args))
        self.__handle_message(msg)

    def __syn_core_on_error(self, msg, *args):
        logging.debug("__sr_core_on_error:msg={} args={}".format(msg, args))

    def __syn_core_on_close(self):
        logging.debug("__sr_core_on_close")
        if self.__on_close:
            self.__on_close(*self.__callback_args)
        self.state.set(NlsStreamInputTtsStatus.Closed)
        self.start_sended.set()
        self.started_event.set()
        self.complete_event.set()

    def __synthesis_started(self, message):
        logging.debug("__synthesis_started")
        self.started_event.set()

    def __sentence_begin(self, message):
        logging.debug("__sentence_begin")
        if self.__on_sentence_begin:
            self.__on_sentence_begin(message, *self.__callback_args)

    def __sentence_synthesis(self, message):
        logging.debug("__sentence_synthesis")
        if self.__on_sentence_synthesis:
            self.__on_sentence_synthesis(message, *self.__callback_args)

    def __sentence_end(self, message):
        logging.debug("__sentence_end")
        if self.__on_sentence_end:
            self.__on_sentence_end(message, *self.__callback_args)

    def __synthesis_completed(self, message):
        logging.debug("__synthesis_completed")
        if self.__on_completed:
            self.__on_completed(message, *self.__callback_args)
        self.__nls.shutdown()
        logging.debug("__synthesis_completed shutdown done")
        self.complete_event.set()


    def __task_failed(self, message):
        logging.debug("__task_failed")
        self.start_sended.set()
        self.started_event.set()
        self.complete_event.set()
        if self.__on_error:
            self.__on_error(message, *self.__callback_args)
        self.state.set(NlsStreamInputTtsStatus.Failed)

    def startStreamInputTts(
        self,
        voice="longxiaochun",
        aformat="pcm",
        sample_rate=24000,
        volume=50,
        speech_rate=0,
        pitch_rate=0,
        ex:dict=None,
    ):
        """
        Synthesis start

        Parameters:
        -----------
        voice: str
            voice for text-to-speech, default is xiaoyun
        aformat: str
            audio binary format, support: 'pcm', 'wav', 'mp3', default is 'pcm'
        sample_rate: int
            audio sample rate, default is 24000, support:8000, 11025, 16000, 22050,
            24000, 32000, 44100, 48000
        volume: int
            audio volume, from 0~100, default is 50
        speech_rate: int
            speech rate from -500~500, default is 0
        pitch_rate: int
            pitch for voice from -500~500, default is 0
        ex: dict
            dict which will merge into 'payload' field in request
        """

        self.__nls = NlsCore(
            url=self.__url,
            token=self.__token,
            on_open=self.__syn_core_on_open,
            on_message=self.__syn_core_on_msg,
            on_data=self.__syn_core_on_data,
            on_close=self.__syn_core_on_close,
            on_error=self.__syn_core_on_error,
            callback_args=[],
        )

        if aformat not in self.__allow_aformat:
            raise InvalidParameter("format {} not support".format(aformat))
        if sample_rate not in self.__allow_sample_rate:
            raise InvalidParameter("samplerate {} not support".format(sample_rate))
        if volume < 0 or volume > 100:
            raise InvalidParameter("volume {} not support".format(volume))
        if speech_rate < -500 or speech_rate > 500:
            raise InvalidParameter("speech_rate {} not support".format(speech_rate))
        if pitch_rate < -500 or pitch_rate > 500:
            raise InvalidParameter("pitch rate {} not support".format(pitch_rate))

        request = self.request.getStartCMD(
            voice, aformat, sample_rate, volume, speech_rate, pitch_rate, ex
        )
        
        last_state = self.state.get()
        if last_state != NlsStreamInputTtsStatus.Begin:
            logging.debug("start with wrong state {}".format(last_state))
            self.state.set(NlsStreamInputTtsStatus.Failed)
            raise WrongStateException("start with wrong state {}".format(last_state))

        logging.debug("start with request: {}".format(request))
        self.__nls.start(request, ping_interval=0, ping_timeout=None)
        self.state.set(NlsStreamInputTtsStatus.Start)
        if not self.start_sended.wait(timeout=10):
            logging.debug("syn start timeout")
            raise StartTimeoutException(f"Waiting Connection before Start over 10s")

        if last_state != NlsStreamInputTtsStatus.Begin:
            logging.debug("start with wrong state {}".format(last_state))
            self.state.set(NlsStreamInputTtsStatus.Failed)
            raise WrongStateException("start with wrong state {}".format(last_state))

        if not self.started_event.wait(timeout=10):
            logging.debug("syn started timeout")
            self.state.set(NlsStreamInputTtsStatus.Failed)
            raise StartTimeoutException(f"Waiting Started over 10s")
        self.state.set(NlsStreamInputTtsStatus.Started)

    def sendStreamInputTts(self, text):
        """
        send text to server

        Parameters:
        -----------
        text: str
            utf-8 text
        """
        last_state = self.state.get()
        if last_state != NlsStreamInputTtsStatus.Started:
            logging.debug("send with wrong state {}".format(last_state))
            self.state.set(NlsStreamInputTtsStatus.Failed)
            raise WrongStateException("send with wrong state {}".format(last_state))

        request = self.request.getSendCMD(text)
        logging.debug("send with request: {}".format(request))
        self.__nls.send(request, None)

    def stopStreamInputTts(self):
        """
        Synthesis end
        """

        last_state = self.state.get()
        if last_state != NlsStreamInputTtsStatus.Started:
            logging.debug("send with wrong state {}".format(last_state))
            self.state.set(NlsStreamInputTtsStatus.Failed)
            raise WrongStateException("stop with wrong state {}".format(last_state))


        request = self.request.getStopCMD()
        logging.debug("stop with request: {}".format(request))
        self.__nls.send(request, None)
        self.state.set(NlsStreamInputTtsStatus.WaitingComplete)
        self.complete_event.wait()
        self.state.set(NlsStreamInputTtsStatus.Completed)
        self.shutdown()

    def shutdown(self):
        """
        Shutdown connection immediately
        """

        self.__nls.shutdown()