client3/libs/lib_player.py

"""
新版播放器,包含划词朗读/AI朗读的TTS实时播放和语音合成的音频文件播放
sudo apt-get install python3-pyqt5.qtmultimedia
sudo apt-get install libqt5multimedia5-plugins
"""
import threading
from queue import Queue
import sounddevice as sd
import time
import os
import sys
import config

from PyQt5.QtWidgets import QApplication

from libs import lib_tts, lib_tts_dipai
import lib_tts_grpc
from PyQt5.QtMultimedia import QMediaPlayer, QMediaContent
from PyQt5.QtCore import QUrl, QObject, pyqtSignal


# 获取家目录
def get_home_env():
    if sys.platform == "win32":
        return 'APPDATA'
    return 'HOME'


# 生成临时音频文件保存路径
def gen_audio_save_path():
    _dir = os.path.join(os.environ[get_home_env()], "xAssistant", "tts_cache")
    if not os.path.exists(_dir):
        os.makedirs(_dir)
    _fp = f"{int(time.time())}.pcm"
    _audio_fp = os.path.join(_dir, _fp)
    print(_audio_fp)
    return _audio_fp


# 计算待合成文本的音频时长（预估）
def calculate_duration(text: str, speed_mode: int):
    """
    :param text: 待合成文本
    :param speed_mode: 语速模式 1:较快 2:较慢  其他为正常
    """
    if speed_mode == 1:
        _speed = 8.145
    elif speed_mode == 2:
        _speed = 4.064
    else:
        _speed = 5.358
    _duration = len(text) / _speed
    return int(_duration)


# TTS临时文件播放线程
class PlayTmpFile(threading.Thread):
    def __init__(self, fp, queue):
        super().__init__()
        self.daemon = True
        self._tmp_file = fp
        self._run_flag = True
        self._audio_queue = queue

    def run(self):
        if self._tmp_file is None or not os.path.exists(self._tmp_file):
            return
        self._audio_queue.queue.clear()
        with open(self._tmp_file, "rb") as f:
            while self._run_flag:
                data = f.read(4096)
                if not data:
                    self._audio_queue.put("--end--")
                    break
                self._audio_queue.put(data)
                time.sleep(0.02)

    def stop(self):
        self._run_flag = False


# TTS播放器（支持播放、暂停、继续、停止）
class TTSSteamPlayer(QObject):
    playing_signal = pyqtSignal()
    paused_signal = pyqtSignal()
    stopped_signal = pyqtSignal()

    def __init__(self, cb_fn):
        super().__init__()
        self._audio_queue = Queue()
        self._tmp_file = None
        self._is_paused = False
        # self._select_default_output_device()
        self._player = sd.RawOutputStream(samplerate=16000, channels=1, dtype='int16')
        self._player.start()
        self._tts_thread = None
        self._run_flag = True
        self._callback = cb_fn
        self.engine_type = config.config["ENGINE_TYPE"]

    # 设置默认输出设备
    def _select_default_output_device(self):
        try:
            system_output_device = sd.query_devices(kind='output')
            sd.default.device = system_output_device['index']
        except Exception as e:
            pass

    # 播放
    def play(self, tts_info: dict):
        self.playing_signal.emit()
        if self._is_paused:
            self._is_paused = False
            return
        self._audio_queue.queue.clear()
        if tts_info:  # 有文本时，调用语音合成来播放实时音频
            self._tmp_file = gen_audio_save_path()
            text = tts_info.get("text")
            vcn = tts_info.get("vcn")
            speed = tts_info.get("speed")
            # self._tts_thread = lib_tts.TTSManagerNew(text, vcn, speed, self._audio_queue)
            # self._tts_thread = lib_tts_dipai.TTSManagerNew(text, vcn, speed, self._audio_queue)
            # self._tts_thread = lib_tts_grpc.TTSManagerNew(text, vcn, speed, self._audio_queue)
            if self.engine_type == "1":
                self._tts_thread = lib_tts.TTSManagerNew(text, vcn, speed, self._audio_queue)
            elif self.engine_type == "2":
                self._tts_thread = lib_tts_dipai.TTSManagerNew(text, vcn, speed, self._audio_queue)
            else:
                self._tts_thread = lib_tts_grpc.TTSManagerNew(text, vcn, speed, self._audio_queue)

            self._tts_thread.start()
            threading.Thread(target=self._write_stream, name="_write_stream", args=(False,), daemon=True).start()
        else:  # 无文本时，播放临时文件
            self._tts_thread = PlayTmpFile(self._tmp_file, self._audio_queue)
            self._tts_thread.start()
            threading.Thread(target=self._write_stream, name="_write_stream", args=(True,), daemon=True).start()

    # 暂停
    def pause(self):
        self.paused_signal.emit()
        self._is_paused = True

    def resume(self):
        self._is_paused = False

    def is_paused(self):
        return self._is_paused

    # 停止
    def stop(self):
        self._is_paused = False
        self._audio_queue.queue.clear()
        self._audio_queue.put("--stop--")

    # 销毁播放器
    def quit(self):
        self._is_paused = False
        self._run_flag = False
        self._audio_queue.queue.clear()
        self._player.stop()

    # 向播放器写入音频流
    def _write_stream(self, is_file=False):
        while self._run_flag:
            if self._is_paused:
                time.sleep(0.02)
                continue
            if not self._audio_queue.empty():
                buf = self._audio_queue.get()
                if isinstance(buf, str):
                    if buf == "--end--" or buf == "--stop--":
                        if self._tts_thread:
                            self._tts_thread.stop()
                        self._callback(buf)
                    else:  # 异常
                        self._callback(buf)
                    break
                if not is_file:
                    self._write_tmp(buf)
                try:
                    self._player.write(buf)
                except:
                    continue
        self.stopped_signal.emit()
        self._audio_queue.queue.clear()

    # 写入临时文件用于重新播放
    def _write_tmp(self, data):
        with open(self._tmp_file, "ab") as f:
            f.write(data)


# 音频文件播放器
class AudioPlayer(QObject):
    playing_signal = pyqtSignal()
    paused_signal = pyqtSignal()
    stopped_signal = pyqtSignal()
    rate_changed_signal = pyqtSignal(dict)

    def __init__(self):
        super().__init__()
        self.duration = 1
        self.position = 0
        self.media_player = QMediaPlayer()
        self.play_state = None
        self.media_player.stateChanged.connect(self.state_changed)  # 获取播放状态
        self.media_player.positionChanged.connect(self.position_changed)  # 获取播放进度
        self.media_player.durationChanged.connect(self.duration_changed)  # 获取音频总时长

    def set_media(self, audio_path: str):
        self.media_player.setMedia(QMediaContent(QUrl.fromLocalFile(audio_path.replace('\\', '/'))))

    # 播放状态改变
    def state_changed(self, state):
        if state == QMediaPlayer.State.PlayingState:
            self.play_state = QMediaPlayer.State.PlayingState
            self.playing_signal.emit()
        elif state == QMediaPlayer.State.PausedState:
            self.play_state = QMediaPlayer.State.PausedState
            self.paused_signal.emit()
        elif state == QMediaPlayer.State.StoppedState:
            self.play_state = QMediaPlayer.State.StoppedState
            self.stopped_signal.emit()

    # 播放进度改变
    def position_changed(self, position):
        try:
            self.position = position
            rate = position / self.duration
            # 格式化时间
            # duration = self.format_time(self.duration)
            # position = self.format_time(position)
            info = {"rate": rate, "position": position, "duration": self.duration}
            self.rate_changed_signal.emit(info)
        except ZeroDivisionError:
            pass

    @staticmethod
    def format_time(input_time):
        minute = int(input_time / 60000)
        second = int((input_time - minute * 60000) / 1000)
        # 单数前面加0
        if minute < 10:
            minute = f"0{minute}"
        if second < 10:
            second = f"0{second}"
        output_time = f"{minute}:{second}"
        return output_time

    # 音频总时长改变
    def duration_changed(self, duration):
        self.duration = duration

    # 设置播放进度
    def set_position(self, float_val: float):
        val = float_val * self.duration
        self.media_player.setPosition(val)

    def set_time(self, val):
        self.media_player.setPosition(val)

    # 设置播放倍速
    def set_playback_speed(self, speed: float):
        if speed == 0.5:
            speed = 0.8
        self.media_player.setPlaybackRate(speed)

    # 播放
    def play(self, *args):
        self.media_player.play()

    # 暂停
    def pause(self):
        self.media_player.pause()

    # 继续
    def resume(self):
        self.media_player.play()

    # 停止
    def stop(self):
        try:
            self.media_player.stop()
        except:
            pass

    # 销毁播放器
    def quit(self):
        pass

    def is_playing(self):
        return self.play_state == QMediaPlayer.State.PlayingState


if __name__ == '__main__':
    app = QApplication([])
    _play_audio_vlc_thread = AudioPlayer()
    _play_audio_vlc_thread.set_media(r"/media/llh/data/test/19.wav")
    _play_audio_vlc_thread.play()
    app.exec_()