From 861071cf19102c0a9e193d8a32f71e3a7e36c3c9 Mon Sep 17 00:00:00 2001 From: NiceAesth Date: Fri, 23 Feb 2024 13:43:37 +0200 Subject: [PATCH] feat: add event models --- examples/advanced.py | 2 +- pomice/__init__.py | 3 +- pomice/events.py | 197 -------------------------------------- pomice/models/__init__.py | 5 +- pomice/models/events.py | 176 ++++++++++++++++++++++++++++++++++ pomice/player.py | 10 +- 6 files changed, 188 insertions(+), 205 deletions(-) delete mode 100644 pomice/events.py create mode 100644 pomice/models/events.py diff --git a/examples/advanced.py b/examples/advanced.py index b8d2d3a..ef14061 100644 --- a/examples/advanced.py +++ b/examples/advanced.py @@ -125,7 +125,7 @@ class Music(commands.Cog): return player.dj == ctx.author or ctx.author.guild_permissions.kick_members - # The following are events from pomice.events + # The following are events from pomice.models.events # We are using these so that if the track either stops or errors, # we can just skip to the next track diff --git a/pomice/__init__.py b/pomice/__init__.py index 84b4718..e4d5456 100644 --- a/pomice/__init__.py +++ b/pomice/__init__.py @@ -7,6 +7,7 @@ Copyright (c) 2023, cloudwithax Licensed under GPL-3.0 """ + import discord if not discord.version_info.major >= 2: @@ -27,7 +28,7 @@ __license__ = "GPL-3.0" __copyright__ = "Copyright (c) 2023, cloudwithax" from .enums import * -from .events import * +from .models import * from .exceptions import * from .filters import * from .objects import * diff --git a/pomice/events.py b/pomice/events.py deleted file mode 100644 index 062c2e4..0000000 --- a/pomice/events.py +++ /dev/null @@ -1,197 +0,0 @@ -from __future__ import annotations - -from abc import ABC -from typing import Any -from typing import Optional -from typing import Tuple -from typing import TYPE_CHECKING - -from discord import Client -from discord import Guild -from discord.ext import commands - -from .objects import Track -from .pool import NodePool - -if TYPE_CHECKING: - from .player import Player - -__all__ = ( - "PomiceEvent", - "TrackStartEvent", - "TrackEndEvent", - "TrackStuckEvent", - "TrackExceptionEvent", - "WebSocketClosedPayload", - "WebSocketClosedEvent", - "WebSocketOpenEvent", -) - - -class PomiceEvent(ABC): - """The base class for all events dispatched by a node. - Every event must be formatted within your bot's code as a listener. - i.e: If you want to listen for when a track starts, the event would be: - ```py - @bot.listen - async def on_pomice_track_start(self, event): - ``` - """ - - name = "event" - handler_args: Tuple - - def dispatch(self, bot: Client) -> None: - bot.dispatch(f"pomice_{self.name}", *self.handler_args) - - -class TrackStartEvent(PomiceEvent): - """Fired when a track has successfully started. - Returns the player associated with the event and the pomice.Track object. - """ - - name = "track_start" - - __slots__ = ( - "player", - "track", - ) - - def __init__(self, data: dict, player: Player): - self.player: Player = player - self.track: Optional[Track] = self.player._current - - # on_pomice_track_start(player, track) - self.handler_args = self.player, self.track - - def __repr__(self) -> str: - return f"" - - -class TrackEndEvent(PomiceEvent): - """Fired when a track has successfully ended. - Returns the player associated with the event along with the pomice.Track object and reason. - """ - - name = "track_end" - - __slots__ = ("player", "track", "reason") - - def __init__(self, data: dict, player: Player): - self.player: Player = player - self.track: Optional[Track] = self.player._ending_track - self.reason: str = data["reason"] - - # on_pomice_track_end(player, track, reason) - self.handler_args = self.player, self.track, self.reason - - def __repr__(self) -> str: - return ( - f"" - ) - - -class TrackStuckEvent(PomiceEvent): - """Fired when a track is stuck and cannot be played. Returns the player - associated with the event along with the pomice.Track object - to be further parsed by the end user. - """ - - name = "track_stuck" - - __slots__ = ("player", "track", "threshold") - - def __init__(self, data: dict, player: Player): - self.player: Player = player - self.track: Optional[Track] = self.player._ending_track - self.threshold: float = data["thresholdMs"] - - # on_pomice_track_stuck(player, track, threshold) - self.handler_args = self.player, self.track, self.threshold - - def __repr__(self) -> str: - return ( - f"" - ) - - -class TrackExceptionEvent(PomiceEvent): - """Fired when a track error has occured. - Returns the player associated with the event along with the error code and exception. - """ - - name = "track_exception" - - __slots__ = ("player", "track", "exception") - - def __init__(self, data: dict, player: Player): - self.player: Player = player - self.track: Optional[Track] = self.player._ending_track - # Error is for Lavalink <= 3.3 - self.exception: str = data.get( - "error", - "", - ) or data.get("exception", "") - - # on_pomice_track_exception(player, track, error) - self.handler_args = self.player, self.track, self.exception - - def __repr__(self) -> str: - return f"" - - -class WebSocketClosedPayload: - __slots__ = ("guild", "code", "reason", "by_remote") - - def __init__(self, data: dict): - self.guild: Optional[Guild] = NodePool.get_node().bot.get_guild(int(data["guildId"])) - self.code: int = data["code"] - self.reason: str = data["code"] - self.by_remote: bool = data["byRemote"] - - def __repr__(self) -> str: - return ( - f"" - ) - - -class WebSocketClosedEvent(PomiceEvent): - """Fired when a websocket connection to a node has been closed. - Returns the reason and the error code. - """ - - name = "websocket_closed" - - __slots__ = ("payload",) - - def __init__(self, data: dict, _: Any) -> None: - self.payload: WebSocketClosedPayload = WebSocketClosedPayload(data) - - # on_pomice_websocket_closed(payload) - self.handler_args = (self.payload,) - - def __repr__(self) -> str: - return f"" - - -class WebSocketOpenEvent(PomiceEvent): - """Fired when a websocket connection to a node has been initiated. - Returns the target and the session SSRC. - """ - - name = "websocket_open" - - __slots__ = ("target", "ssrc") - - def __init__(self, data: dict, _: Any) -> None: - self.target: str = data["target"] - self.ssrc: int = data["ssrc"] - - # on_pomice_websocket_open(target, ssrc) - self.handler_args = self.target, self.ssrc - - def __repr__(self) -> str: - return f"" diff --git a/pomice/models/__init__.py b/pomice/models/__init__.py index 7653113..0535fbc 100644 --- a/pomice/models/__init__.py +++ b/pomice/models/__init__.py @@ -1,6 +1,9 @@ import pydantic from pydantic import ConfigDict +from .events import * +from .version import * + class BaseModel(pydantic.BaseModel): - model_config = ConfigDict(populate_by_name=True) + model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True) diff --git a/pomice/models/events.py b/pomice/models/events.py new file mode 100644 index 0000000..625e2d0 --- /dev/null +++ b/pomice/models/events.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import abc +from enum import Enum +from enum import unique +from typing import TYPE_CHECKING +from typing import Literal +from discord import Guild +from pydantic import computed_field +from pydantic import Field +from pomice.models import BaseModel +from pomice.player import Player +from pomice.objects import Track +from pomice.pool import NodePool + +if TYPE_CHECKING: + from discord import Client + +__all__ = ( + "PomiceEvent", + "TrackStartEvent", + "TrackEndEvent", + "TrackStuckEvent", + "TrackExceptionEvent", + "WebSocketClosedPayload", + "WebSocketClosedEvent", + "WebSocketOpenEvent", +) + + +class PomiceEvent(BaseModel, abc.ABC): + """The base class for all events dispatched by a node. + Every event must be formatted within your bot's code as a listener. + i.e: If you want to listen for when a track starts, the event would be: + ```py + @bot.listen + async def on_pomice_track_start(self, event): + ``` + """ + + name: str + + @abc.abstractmethod + def dispatch(self, bot: Client) -> None: ... + + +class TrackStartEvent(PomiceEvent): + """Fired when a track has successfully started. + Returns the player associated with the event and the pomice.Track object. + """ + + name: Literal["track_start"] + player: Player + track: Track + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.player, self.track) + + def __repr__(self) -> str: + return f"" + + +@unique +class TrackEndEventReason(str, Enum): + FINISHED = "finished" + LOAD_FAILED = "loadfailed" + STOPPED = "stopped" + REPLACED = "replaced" + CLEANUP = "cleanup" + + @classmethod + def _missing_(cls, value: object) -> TrackEndEventReason: + if isinstance(value, str): + return TrackEndEventReason(value.casefold()) + + +class TrackEndEvent(PomiceEvent): + """Fired when a track has successfully ended. + Returns the player associated with the event along with the pomice.Track object and reason. + """ + + name: Literal["track_end"] + player: Player + track: Track + reason: TrackEndEventReason + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.player, self.track, self.reason) + + def __repr__(self) -> str: + return f"" + + +class TrackStuckEvent(PomiceEvent): + """Fired when a track has been stuck for a while. + Returns the player associated with the event along with the pomice.Track object and threshold. + """ + + name: Literal["track_stuck"] + player: Player + track: Track + threshold: float = Field(alias="thresholdMs") + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.player, self.track, self.threshold) + + def __repr__(self) -> str: + return f"" + + +class TrackExceptionEvent(PomiceEvent): + """Fired when there is an exception while playing a track. + Returns the player associated with the event along with the pomice.Track object and exception. + """ + + name: Literal["track_exception"] + player: Player + track: Track + exception: str = Field(alias="error") + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.player, self.track, self.exception) + + def __repr__(self) -> str: + return f"" + + +class WebSocketClosedPayload(BaseModel): + """The payload for the WebSocketClosedEvent.""" + + guild_id: int = Field(alias="guildId") + code: int + reason: str + by_remote: bool = Field(alias="byRemote") + + @computed_field + @property + def guild(self) -> Guild: + return NodePool.get_node().bot.get_guild(self.guild_id) + + def __repr__(self) -> str: + return ( + f"" + ) + + +class WebSocketClosedEvent(PomiceEvent): + """Fired when the websocket connection to the node is closed. + Returns the player associated with the event and the code and reason for the closure. + """ + + name: Literal["websocket_closed"] + payload: WebSocketClosedPayload + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.payload) + + def __repr__(self) -> str: + return f"" + + +class WebSocketOpenEvent(PomiceEvent): + """Fired when the websocket connection to the node is opened. + Returns the player associated with the event. + """ + + name: Literal["websocket_open"] + target: str + ssrc: str + + def dispatch(self, bot: Client) -> None: + bot.dispatch(f"pomice_{self.name}", self.target, self.ssrc) + + def __repr__(self) -> str: + return f"" diff --git a/pomice/player.py b/pomice/player.py index 73a5ec0..8ce87af 100644 --- a/pomice/player.py +++ b/pomice/player.py @@ -16,9 +16,9 @@ from discord.ext import commands from . import events from .enums import SearchType -from .events import PomiceEvent -from .events import TrackEndEvent -from .events import TrackStartEvent +from pomice.models.events import PomiceEvent +from pomice.models.events import TrackEndEvent +from pomice.models.events import TrackStartEvent from .exceptions import FilterInvalidArgument from .exceptions import FilterTagAlreadyInUse from .exceptions import FilterTagInvalid @@ -355,9 +355,9 @@ class Player(VoiceProtocol): async def _dispatch_event(self, data: dict) -> None: event_type: str = data["type"] - event: PomiceEvent = getattr(events, event_type)(data, self) + event: PomiceEvent = getattr(events, event_type)(player=self, **data) - if isinstance(event, TrackEndEvent) and event.reason not in ("REPLACED", "replaced"): + if isinstance(event, TrackEndEvent) and event.reason != "replaced": self._current = None event.dispatch(self._bot)