play_pause.py

from enum import Enum

import numpy
import sounddevice as sd
from numpy._typing import NDArray
from PyQt6.QtCore import QObject, pyqtSignal
from scipy._lib._ccallback import CData
from sounddevice import CallbackFlags

from play_marker import PlayMarker


class StatusAudio(Enum):
    Playing = "playing"
    Pause = "paused"
    Stop = "stopped"


class PlayPauseAudio(QObject):
    """Класс для управления воспроизведением, паузой и остановкой аудио.

    Наследуется от QObject для использования сигналов Qt.
    Использует библиотеку sounddevice для потокового воспроизведения аудио.

    Сигналы:
        position_changed (pyqtSignal[int]): Сигнал с текущей позицией воспроизведения (в семплах)
    """

    position_changed = pyqtSignal(int)

    def __init__(self, song: NDArray[numpy.float32]) -> None:
        """Инициализирует объект воспроизведения аудио.

        Аргументы:
            song (NDArray[numpy.float32]): Аудиоданные в виде numpy массива
        """
        super().__init__()
        self.song = song
        self.sample_rate = 44100  # Стандартная частота дискретизации
        self.current_position = 0  # Текущая позиция в семплах
        self._state = 'stopped'  # Состояние: 'stopped', 'playing' или 'paused'
        self.stream: sd.OutputStream | None = None  # Аудиопоток sounddevice

    def play(self, play_marker_thread: PlayMarker) -> None:
        """Запускает или возобновляет воспроизведение аудио.

        Аргументы:
            play_marker_thread (PlayMarker): Поток для управления маркером воспроизведения
        """
        if self._state == StatusAudio.Playing.value:
            return

        if self._state == StatusAudio.Pause.value:
            self._state = StatusAudio.Playing.value
            self._start_stream()
            play_marker_thread.update_position(True)
        else:
            self.current_position = 0
            self._state = StatusAudio.Playing.value
            self._start_stream()
            play_marker_thread.update_position(True)

    def _start_stream(self) -> None:
        """Запускает аудиопоток для воспроизведения.

        Создает новый OutputStream с callback-функцией для потоковой передачи данных.
        """
        if self.stream:
            self.stream.close()

        def callback(outdata: NDArray[numpy.float64], frames: int, time: CData, status: CallbackFlags) -> None:
            """Callback-функция для передачи аудиоданных в звуковое устройство.

            Аргументы:
                outdata (NDArray[numpy.float64]): Буфер для выходных данных
                frames (int): Количество запрашиваемых кадров
                time: Временная метка
                status: Статус потока
            """
            if not self._state == 'playing':
                raise sd.CallbackStop
            available = len(self.song) - self.current_position
            if available <= 0:
                self.finished_stop()
                raise sd.CallbackStop

            chunk_size = min(frames, available)
            chunk = self.song[self.current_position : self.current_position + chunk_size]
            if len(outdata.shape) == 2 and len(chunk.shape) == 1:
                outdata[:chunk_size, 0] = chunk  # Для моно в стерео
            else:
                outdata[:chunk_size] = chunk
            self.position_changed.emit(self.current_position)
            self.current_position += chunk_size
            if chunk_size < frames:
                outdata[chunk_size:] = 0
                raise sd.CallbackStop

        self.stream = sd.OutputStream(
            samplerate=self.sample_rate,
            channels=1 if len(self.song.shape) == 1 else self.song.shape[1],
            callback=callback,
        )
        if self.stream:
            self.stream.start()

    def pause(self, play_marker_thread: PlayMarker) -> None:
        """Приостанавливает воспроизведение аудио.

        Аргументы:
            play_marker_thread (PlayMarker): Поток для управления маркером воспроизведения
        """
        if self._state == StatusAudio.Playing.value:
            self._state = StatusAudio.Pause.value
            if self.stream:
                self.stream.stop()
            play_marker_thread.pause_marker()

    def finished_stop(self) -> None:
        """Останавливает воспроизведение по завершении трека."""
        if self.stream:
            self.stream.stop()
            self.stream.close()
            self.stream = None
        self.current_position = 0
        self.position_changed.emit(0)

    def reset(self, play_marker_thread: PlayMarker) -> None:
        """Сбрасывает воспроизведение в начальное состояние.

        Аргументы:
            play_marker_thread (PlayMarker): Поток для управления маркером воспроизведения
        """
        self.current_position = 0
        self._state = StatusAudio.Pause.value
        if self.stream:
            self.stream.stop()
            self.stream.close()
            self.stream = None
        play_marker_thread.reset_marker()
        self.position_changed.emit(0)