feat: add (incomplete) nix flake

dave.py isn't packaged yet. Will try to do this myself but dealing with
vcpkg is a bit annoying.

errornocord/audio/__init__.py

@@ -0,0 +1,8 @@
+from . import discord, queue, utils, youtubedl
+
+__all__ = [
+    "discord",
+    "queue",
+    "utils",
+    "youtubedl",
+]

errornocord/audio/discord.py

@@ -0,0 +1,39 @@
+import audioop
+
+import disnake
+
+
+class TrackedAudioSource(disnake.AudioSource):
+    def __init__(self, source):
+        self._source = source
+        self.read_count = 0
+
+    def read(self) -> bytes:
+        data = self._source.read()
+        if data:
+            self.read_count += 1
+        return data
+
+    def fast_forward(self, seconds: int):
+        for _ in range(int(seconds / 0.02)):
+            self.read()
+
+    @property
+    def progress(self) -> float:
+        return self.read_count * 0.02
+
+
+class PCMVolumeTransformer(disnake.AudioSource):
+    def __init__(self, original: TrackedAudioSource, volume: float = 1.0) -> None:
+        if original.is_opus():
+            raise disnake.ClientException("AudioSource must not be Opus encoded")
+
+        self.original = original
+        self.volume = volume
+
+    def cleanup(self) -> None:
+        self.original.cleanup()
+
+    def read(self) -> bytes:
+        ret = self.original.read()
+        return audioop.mul(ret, 2, self.volume)

errornocord/audio/queue.py

@@ -0,0 +1,111 @@
+import collections
+from dataclasses import dataclass
+from typing import ClassVar, Optional
+
+import disnake
+
+from ..constants import BAR_LENGTH, EMBED_COLOR
+from .utils import format_duration
+from .youtubedl import YTDLSource
+
+
+@dataclass
+class Song:
+    player: YTDLSource
+    trigger_message: disnake.Message
+
+    def format(self, show_queuer=False, hide_preview=False, multiline=False) -> str:
+        title = f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''})"
+        duration = (
+            format_duration(self.player.duration) if self.player.duration else "stream"
+        )
+        if multiline:
+            queue_time = (
+                self.trigger_message.edited_at or self.trigger_message.created_at
+            )
+            return f"{title}\n**duration:** {duration}" + (
+                f", **queued by:** <@{self.trigger_message.author.id}> <t:{round(queue_time.timestamp())}:R>"
+                if show_queuer
+                else ""
+            )
+        return f"{title} [**{duration}**]" + (
+            f" (<@{self.trigger_message.author.id}>)" if show_queuer else ""
+        )
+
+    def embed(self, is_paused=False):
+        progress = 0
+        if self.player.duration:
+            progress = self.player.original.progress / self.player.duration
+
+        embed = disnake.Embed(
+            color=EMBED_COLOR,
+            title=self.player.title,
+            url=self.player.original_url,
+            description=(
+                f"{'⏸️ ' if is_paused else ''}"
+                f"`[{'#' * int(progress * BAR_LENGTH)}{'-' * int((1 - progress) * BAR_LENGTH)}]` "
+                + (
+                    f"**{format_duration(int(self.player.original.progress))}** / **{format_duration(self.player.duration)}** (**{round(progress * 100)}%**)"
+                    if self.player.duration
+                    else "[**stream**]"
+                )
+            ),
+            timestamp=self.trigger_message.edited_at or self.trigger_message.created_at,
+        )
+
+        uploader_value = None
+        if self.player.uploader_url:
+            if self.player.uploader:
+                uploader_value = f"[{self.player.uploader}]({self.player.uploader_url})"
+            else:
+                uploader_value = self.player.uploader_url
+        elif self.player.uploader:
+            uploader_value = self.player.uploader
+
+        if uploader_value:
+            embed.add_field(name="Uploader", value=uploader_value)
+        if self.player.like_count:
+            embed.add_field(name="Likes", value=f"{self.player.like_count:,}")
+        if self.player.view_count:
+            embed.add_field(name="Views", value=f"{self.player.view_count:,}")
+        if self.player.timestamp:
+            embed.add_field(name="Published", value=f"<t:{int(self.player.timestamp)}>")
+        if self.player.volume:
+            embed.add_field(name="Volume", value=f"{int(self.player.volume * 100)}%")
+
+        if self.player.thumbnail_url:
+            embed.set_image(self.player.thumbnail_url)
+
+        embed.set_footer(
+            text=f"Queued by {self.trigger_message.author.name}",
+            icon_url=(
+                self.trigger_message.author.avatar.url
+                if self.trigger_message.author.avatar
+                else None
+            ),
+        )
+
+        return embed
+
+    def __str__(self):
+        return self.__repr__()
+
+
+@dataclass
+class Player:
+    queue: ClassVar = collections.deque()
+    current: Optional[Song] = None
+
+    def queue_pop(self):
+        popped = self.queue.popleft()
+        self.current = popped
+        return popped
+
+    def queue_push(self, item):
+        self.queue.append(item)
+
+    def queue_push_front(self, item):
+        self.queue.appendleft(item)
+
+    def __str__(self):
+        return self.__repr__()

errornocord/audio/utils.py

@@ -0,0 +1,7 @@
+def format_duration(duration: int | float) -> str:
+    hours, duration = divmod(int(duration), 3600)
+    minutes, duration = divmod(duration, 60)
+    segments = [hours, minutes, duration]
+    if len(segments) == 3 and segments[0] == 0:
+        del segments[0]
+    return f"{':'.join(f'{s:0>2}' for s in segments)}"

errornocord/audio/youtubedl.py

@@ -0,0 +1,75 @@
+import asyncio
+from typing import Any, Optional
+
+import disnake
+import yt_dlp
+
+from ..constants import YTDL_OPTIONS
+from .discord import PCMVolumeTransformer, TrackedAudioSource
+
+ytdl = yt_dlp.YoutubeDL(YTDL_OPTIONS)
+
+
+class YTDLSource(PCMVolumeTransformer):
+    def __init__(
+        self,
+        source: TrackedAudioSource,
+        *,
+        data: dict[str, Any],
+        volume: float = 0.5,
+    ):
+        super().__init__(source, volume)
+
+        self.description = data.get("description")
+        self.duration = data.get("duration")
+        self.id = data.get("id")
+        self.like_count = data.get("like_count")
+        self.original_url = data.get("original_url")
+        self.thumbnail_url = data.get("thumbnail")
+        self.timestamp = data.get("timestamp")
+        self.title = data.get("title")
+        self.uploader = data.get("uploader")
+        self.uploader_url = data.get("uploader_url")
+        self.view_count = data.get("view_count")
+
+    @classmethod
+    async def from_url(
+        cls,
+        url,
+        *,
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+        stream: bool = False,
+    ):
+        loop = loop or asyncio.get_event_loop()
+        data: Any = await loop.run_in_executor(
+            None,
+            lambda: ytdl.extract_info(url, download=not stream),
+        )
+
+        if "entries" in data:
+            if not data["entries"]:
+                raise Exception("no results found!")
+            data = data["entries"][0]
+            if "url" not in data:
+                raise Exception("no url returned!")
+
+        return cls(
+            TrackedAudioSource(
+                disnake.FFmpegPCMAudio(
+                    data["url"] if stream else ytdl.prepare_filename(data),
+                    before_options="-vn -reconnect 1",
+                ),
+            ),
+            data=data,
+        )
+
+    def __repr__(self):
+        return f"<YTDLSource title={self.title} original_url={self.original_url} duration={self.duration}>"
+
+    def __str__(self):
+        return self.__repr__()
+
+
+def __reload_module__():
+    global ytdl
+    ytdl = yt_dlp.YoutubeDL(YTDL_OPTIONS)