audio_editor.py

from collections import defaultdict
from enum import Enum
from pathlib import Path
from threading import Lock, Thread
from typing import DefaultDict, Union

from pydub import AudioSegment
from PyQt6.QtCore import QObject, pyqtSignal

from class_of_changes import (ChangeOfOctave, ChangeOfSampleRate,
                              ChangeOfSpeed, ChangeOfVolume, CutOutFragment,
                              GlueFragments, Reverse)
from stack import Stack

ten = 1000


class WorkerSignals(QObject):
    """Класс сигналов для обработки событий в фоновых потоках.

    Атрибуты:
        finished: Сигнал, испускаемый при успешном завершении операции.
        error: Сигнал, испускаемый при возникновении ошибки."""

    finished = pyqtSignal()
    error = pyqtSignal(str)


class Extension(Enum):
    Wav = ".wav"
    Mp3 = ".mp3"


class AudioEditor:
    """Класс аудиоредактора, который выполняет все манипуляции со звуком."""

    _history_of_actions: Stack[
        Union[ChangeOfSpeed, ChangeOfVolume, CutOutFragment, GlueFragments, Reverse, ChangeOfSampleRate, ChangeOfOctave]
    ]

    def __init__(self, path: str):
        """Инициализирует аудиоредактор.

        Args:
            path (str): Путь к аудиофайлу для редактирования.

        Raises:
            TypeException: Если файл не в формате WAV или MP3.
        """
        self._signals = WorkerSignals()
        self._history_of_actions = Stack()
        self._history_of_actions_str: list[str] = []
        self._changes: DefaultDict[str, list[ChangeOfVolume | ChangeOfSpeed | ChangeOfSampleRate | ChangeOfOctave]] = (
            defaultdict(list)
        )
        self._reversed = False
        self._song = None
        self._lock = Lock()
        self._song_path = Path(path)
        song_name = self._song_path.name
        extension = self._song_path.suffix
        if extension == Extension.Wav.value:
            self._song = AudioSegment.from_wav(song_name)
        elif extension == Extension.Mp3.value:
            self._song = AudioSegment.from_mp3(song_name)
        else:
            raise TypeException("Wrong type of audio-extension")
        self._current = self._song

    def change_volume(self, x: float) -> None:
        """Изменяет громкость аудио.

        Создает объект Change_Of_Volume, выполняет execute и добавляет действие в историю.

        Аргументы:
            x (float): Значение изменения громкости.
        """
        action = ChangeOfVolume(x)
        self.apply_change(action, "volume", x)

    def change_speed(self, x: float) -> None:
        """Изменяет скорость воспроизведения аудио.

        Создает объект Change_Of_Speed, выполняет execute и добавляет действие в историю.

        Аргументы:
            x (float): Значение изменения скорости.
        """
        action = ChangeOfSpeed(x)
        self.apply_change(action, "speed", x)

    def change_sample_rate(self, x: int) -> None:
        """Изменяет частоту аудио.

        Создает объект ChangeOfSampleRate, выполняет execute и добавляет действие в историю.

        Аргументы:
            x (int): Значение изменения частоты.
        """
        action = ChangeOfSampleRate(x)
        self.apply_change(action, "sample_rate", x)

    def change_octave(self, x: float) -> None:
        """Изменяет высоту аудио.

        Создает объект ChangeOfOctave, выполняет execute и добавляет действие в историю.

        Аргументы:
            x (float): Значение изменения высоты.
        """
        action = ChangeOfOctave(x)
        self.apply_change(action, "octave", x)

    def apply_change(
        self,
        action: Union[ChangeOfVolume, ChangeOfSpeed, ChangeOfSampleRate, ChangeOfOctave],
        change_type: str,
        value: float,
    ) -> None:
        self._history_of_actions.push(action)
        self._changes[action.__class__.__name__].append(action)
        self._history_of_actions_str.append(f"changed {change_type} on {value}")

    def cut_fragment(self, first_index: int, second_index: int, path: str) -> None:
        """Вырезает фрагмент аудио в отдельном потоке.

        Args:
            first_index (int): Начальный индекс фрагмента (в миллисекундах).
            second_index (int): Конечный индекс фрагмента (в миллисекундах).
            path (str): Путь для сохранения вырезанного фрагмента."""

        def worker() -> None:
            try:
                with self._lock:
                    current_song = self.update_original_song()
                    action = CutOutFragment(first_index, second_index)
                    self._song = action.execute(self._song)
                    cut_out = current_song[first_index:second_index]
                    cut_out.export(path)
                    self._history_of_actions.push(action)
                    self._history_of_actions_str.append(f"cut out fragment from {first_index} to {second_index}")
                    self._signals.finished.emit()
            except Exception as e:
                self._signals.error.emit(f"Error in audio processing: {e}")

        Thread(target=worker, daemon=True).start()

    def glue_fragments(self, fragment_one: AudioSegment, fragment_two: AudioSegment, song_path: str) -> None:
        """Склеивает два аудиофрагмента и сохраняет результат.

        Args:
            fragment_one (AudioSegment): Первый аудиофрагмент.
            fragment_two (AudioSegment): Второй аудиофрагмент.
            song_path (str): Путь для сохранения склеенного аудио."""
        current = self.update_original_song()
        if not fragment_one:
            fragment_one = current
        if not fragment_two:
            fragment_two = current
        action = GlueFragments(fragment_one, fragment_two)
        song = action.execute()
        if song_path:
            song.export(song_path)
        self._history_of_actions_str.append("glued two fragments")

    def save(self, path: str) -> None:
        """Сохраняет текущее состояние аудиофайла.

        Аргументы:
            path (str): Путь для сохранения файла.
        """
        self._song = self.update_original_song()
        self._song.export(path)

    def history_changes(self) -> list[str]:
        """Возвращает историю изменений в виде списка строк.

        Возвращает:
            list[str]: Список описаний выполненных действий.
        """
        if self._history_of_actions_str:
            return self._history_of_actions_str
        return []

    def backwards(self) -> None:
        """Переворачивает аудиофайл (воспроизведение в обратном порядке).

        Создает объект Reverse, выполняет execute и добавляет действие в историю.
        """
        if self._current:
            action = Reverse()
            self._history_of_actions.push(action)
            self._history_of_actions_str.append("backwards")
            self._reversed = not self._reversed

    def undo(
        self,
    ) -> (
        Union[ChangeOfVolume, ChangeOfSpeed, ChangeOfSampleRate, CutOutFragment, GlueFragments, Reverse, ChangeOfOctave]
        | None
    ):
        """Отменяет последнее действие.

        Возвращает:
            Any | None: Возвращает объект последнего действия (для Change_Of_Speed/Volume),
                       либо None для других действий.
        """
        with self._lock:
            if not self._history_of_actions.is_empty():
                action = self._history_of_actions.pop()
                if isinstance(action, CutOutFragment):
                    self._song = action.song
                if isinstance(action, Reverse):
                    self._reversed = not self._reversed
                if self._changes[action.__class__.__name__]:
                    self._changes[action.__class__.__name__].pop()
                self._history_of_actions_str.append("undo the last action")
                if action.change_slider():
                    return action
            return None

    def update_original_song(self) -> AudioSegment:
        """Обновляет оригинальный аудиофайл с учетом всех изменений.

        Возвращает:
            AudioSegment: Обновленный аудиофайл.
        """
        result_song = self._song
        for change in self._changes:
            if len(self._changes[change]) >= 1:
                result_song = self._changes[change][-1].execute(result_song)
        if self._reversed and result_song:
            result_song = result_song.reverse()
        return result_song


class TypeException(Exception):
    """Исключение для неверного типа аудиофайла.

    Аргументы:
        message (str): Сообщение об ошибке.
    """

    def __init__(self, message: str) -> None:
        super().__init__(message)