79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import discord
|
|
|
|
FRAME_LENGTH_SECONDS = 0.02
|
|
|
|
|
|
from zyrarbot.audio.effects.pipeline import AudioEffectsPipeline
|
|
from zyrarbot.audio.effects.registry import create_default_effects
|
|
|
|
try:
|
|
import audioop
|
|
except ImportError: # pragma: no cover
|
|
try:
|
|
import audioop_lts as audioop
|
|
except ImportError:
|
|
audioop = None
|
|
|
|
|
|
class YTDLSource(discord.PCMVolumeTransformer):
|
|
"""Źródło audio (strumień YouTube lub plik lokalny) z metadanymi utworu."""
|
|
|
|
def __init__(
|
|
self,
|
|
source: discord.AudioSource,
|
|
*,
|
|
data: dict[str, Any],
|
|
volume: float = 0.5,
|
|
effects_pipeline: AudioEffectsPipeline | None = None,
|
|
):
|
|
super().__init__(source, volume)
|
|
self.data = data
|
|
self.title: str | None = data.get("title")
|
|
self.url: str | None = data.get("url")
|
|
self.id: str = uuid.uuid4().hex
|
|
self.duration: float | None = data.get("duration")
|
|
self.elapsed: float = 0.0
|
|
self.plays_left: int | None = data.get("plays_left")
|
|
self.played: bool = False
|
|
self.effects_pipeline: AudioEffectsPipeline = (
|
|
effects_pipeline if effects_pipeline is not None else AudioEffectsPipeline(create_default_effects())
|
|
)
|
|
|
|
@property
|
|
def is_local(self) -> bool:
|
|
return bool(self.data.get("is_local", False))
|
|
|
|
@property
|
|
def filepath(self) -> str | None:
|
|
return self.data.get("filepath")
|
|
|
|
@property
|
|
def display_url(self) -> str | None:
|
|
return self.data.get("webpage_url") or self.url
|
|
|
|
@property
|
|
def has_web_link(self) -> bool:
|
|
url = self.display_url
|
|
return bool(url and str(url).startswith("http"))
|
|
|
|
def set_active_effects(self, effect_ids: Any) -> None:
|
|
if self.effects_pipeline:
|
|
self.effects_pipeline.active_ids = effect_ids
|
|
|
|
def read(self) -> bytes:
|
|
data = self.original.read()
|
|
if not data:
|
|
return b""
|
|
self.elapsed += FRAME_LENGTH_SECONDS
|
|
if self.effects_pipeline is not None:
|
|
return self.effects_pipeline.process(data, volume=self._volume)
|
|
if audioop is not None and self._volume != 1.0:
|
|
return audioop.mul(data, 2, self._volume)
|
|
return data
|
|
|