human/tts/tts_base.py

#encoding = utf8

import logging
import time

from eventbus import EventBus
from nlp import NLPCallback
from utils import AsyncTaskQueue

logger = logging.getLogger(__name__)


class TTSBase(NLPCallback):
    def __init__(self, handle):
        self._handle = handle
        self._message_queue = AsyncTaskQueue('TTSBaseQueue', 5)
        self._is_running = True
        EventBus().register('stop', self.on_stop)
        EventBus().register('clear_cache', self.on_clear_cache)

    def __del__(self):
        EventBus().unregister('stop', self.on_stop)
        EventBus().unregister('clear_cache', self.on_clear_cache)

    def on_stop(self, *args, **kwargs):
        self.stop()

    def on_clear_cache(self, *args, **kwargs):
        logger.info('TTSBase clear_cache')
        self._message_queue.clear()

    @property
    def handle(self):
        return self._handle

    @handle.setter
    def handle(self, value):
        self._handle = value

    async def _request(self, txt: str, index):
        # print('_request:', txt)
        if not self._is_running:
            logger.info('TTSBase::_request is not running')
            return

        t = time.time()
        stream = await self._on_request(txt)
        if stream is None or self._is_running is False:
            logger.warn(f'-------stream is None or is_running {self._is_running}')
            return
        logger.info(f'-------tts time:{time.time() - t:.4f}s, txt:{txt}')
        if self._handle is not None and self._is_running:
            await self._on_handle(stream, index)
        else:
            logger.info(f'handle is None, running:{self._is_running}')

    async def _on_request(self, text: str):
        pass

    async def _on_handle(self, stream, index):
        pass

    async def _on_close(self):
        pass

    def on_message(self, txt: str):
        self.message(txt)

    def message(self, txt):
        if not self._is_running:
            logger.info('TTSBase::message is not running')
            return
        txt = txt.strip()
        if len(txt) == 0:
            # logger.info(f'message is empty')
            return
        logger.info(f'message:{txt}')
        index = 0
        if self._handle is not None:
            index = self._handle.get_index()
            # print(f'message txt-index:{txt}, index {index}')
        self._message_queue.add_task(self._request, txt, index)

    def stop(self):
        self._is_running = False
        self._message_queue.add_task(self._on_close)
        self._message_queue.stop()

    def pause_talk(self):
        logger.info(f'TTSBase pause_talk')
        self._message_queue.clear()