human/Human.py

#encoding = utf8
import logging
import queue
from queue import Queue

import numpy as np

from tts.Chunk2Mal import Chunk2Mal

logger = logging.getLogger(__name__)

class Human:
    def __init__(self):
        self._tts = None
        self._fps = 25  # 20 ms per frame
        self._sample_rate = 16000
        self._chunk = self._sample_rate // self._fps  # 320 samples per chunk (20ms * 16000 / 1000)
        self._chunk_2_mal = Chunk2Mal()

    def on_destroy(self):
        if self._tts is not None:
            self._tts.stop()
        logger.info('human destroy')

    def set_tts(self, tts):
        if self._tts == tts:
            return

        self._tts = tts
        self._tts.start()

    def read(self, txt):
        if self._tts is None:
            logger.warning('tts is none')
            return

        self._tts.push_txt(txt)

    def push_audio_chunk(self, chunk):
        self._chunk_2_mal.push_chunk(chunk)
        self._audio_chunk_queue.put(chunk)

    def pull_audio_chunk(self):
        try:
            chunk = self._audio_chunk_queue.get(block=True, timeout=1.0)
            type = 1
        except queue.Empty:
            chunk = np.zeros(self._chunk, dtype=np.float32)
            type = 0
        return chunk, type