human/Human.py

#encoding = utf8
import logging
import multiprocessing as mp

from tts.Chunk2Mal import Chunk2Mal

logger = logging.getLogger(__name__)


class Human:
    def __init__(self):
        self._tts = None
        self._fps = 50  # 20 ms per frame
        self._batch_size = 16
        self._sample_rate = 16000
        self._chunk = self._sample_rate // self._fps  # 320 samples per chunk (20ms * 16000 / 1000)
        self._chunk_2_mal = Chunk2Mal(self)
        self._stride_left_size = 10
        self._stride_right_size = 10
        self._feat_queue = mp.Queue(2)

    def get_fps(self):
        return self._fps

    def get_batch_size(self):
        return self._batch_size

    def get_chunk(self):
        return self._chunk

    def get_stride_left_size(self):
        return self._stride_left_size

    def get_stride_right_size(self):
        return self._stride_right_size

    def on_destroy(self):
        self._chunk_2_mal.stop()

        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()
        self._chunk_2_mal.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)

    def push_feat_queue(self, mel_chunks):
        self._feat_queue.put(mel_chunks)