nambrogi
/
pomice
mirror of https://github.com/cloudwithax/pomice.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: add event models

This commit is contained in:
NiceAesth 2024-02-23 13:43:37 +02:00
parent 3657c28495
commit 861071cf19
6 changed files with 188 additions and 205 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/advanced.py
View File

@ -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

3
pomice/__init__.py
View File

@ -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 *

197
pomice/events.py
View File

@ -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"<Pomice.TrackStartEvent player={self.player!r} track={self.track!r}>"
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"<Pomice.TrackEndEvent player={self.player!r} track_id={self.track!r} "
f"reason={self.reason!r}>"
)
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"<Pomice.TrackStuckEvent player={self.player!r} track={self.track!r} "
f"threshold={self.threshold!r}>"
)
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"<Pomice.TrackExceptionEvent player={self.player!r} exception={self.exception!r}>"
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"<Pomice.WebSocketClosedPayload guild={self.guild!r} code={self.code!r} "
f"reason={self.reason!r} by_remote={self.by_remote!r}>"
)
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"<Pomice.WebsocketClosedEvent payload={self.payload!r}>"
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"<Pomice.WebsocketOpenEvent target={self.target!r} ssrc={self.ssrc!r}>"

5
pomice/models/__init__.py
View File

@ -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)

176
pomice/models/events.py Normal file
View File

@ -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"<Pomice.TrackStartEvent player={self.player!r} track={self.track!r}>"
@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"<Pomice.TrackEndEvent player={self.player!r} track={self.track!r} reason={self.reason!r}>"
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"<Pomice.TrackStuckEvent player={self.player!r} track={self.track!r} threshold={self.threshold!r}>"
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"<Pomice.TrackExceptionEvent player={self.player!r} track={self.track!r} exception={self.exception!r}>"
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"<Pomice.WebSocketClosedPayload guild_id={self.guild_id!r} code={self.code!r} "
f"reason={self.reason!r} by_remote={self.by_remote!r}>"
)
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"<Pomice.WebSocketClosedEvent payload={self.payload!r}>"
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"<Pomice.WebSocketOpenEvent target={self.target!r} ssrc={self.ssrc!r}>"

10
pomice/player.py
View File

@ -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)