feat: initial commit, as im not gonna stop if i dont commit

mafic/__init__.py

@@ -15,7 +15,10 @@
 
 import logging
 
-from . import libraries as __libraries
+from . import __libraries
+from .errors import *
+from .node import *
+from .player import *
 
 del __libraries
 

mafic/libraries.py → mafic/__libraries.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from os import getenv
+from typing import Any
 
 from pkg_resources import DistributionNotFound, get_distribution
 
@@ -11,9 +12,12 @@
 __all__ = (
     "Client",
     "Connectable",
+    "dumps",
+    "ExponentialBackoff",
+    "GuildVoiceStatePayload",
+    "loads",
     "VoiceProtocol",
     "VoiceServerUpdatePayload",
-    "GuildVoiceStatePayload",
 )
 
 libraries = ("nextcord", "disnake", "py-cord", "discord.py", "discord")
@@ -53,21 +57,34 @@
 if library == "nextcord":
     from nextcord import Client, VoiceProtocol
     from nextcord.abc import Connectable
+    from nextcord.backoff import ExponentialBackoff
     from nextcord.types.voice import (
         GuildVoiceState as GuildVoiceStatePayload,
         VoiceServerUpdate as VoiceServerUpdatePayload,
     )
 elif library == "disnake":
     from disnake import Client, VoiceProtocol
     from disnake.abc import Connectable
+    from disnake.backoff import ExponentialBackoff
     from disnake.types.voice import (
         GuildVoiceState as GuildVoiceStatePayload,
         VoiceServerUpdate as VoiceServerUpdatePayload,
     )
 else:
     from discord import Client, VoiceProtocol
     from discord.abc import Connectable
+    from discord.backoff import ExponentialBackoff
     from discord.types.voice import (
         GuildVoiceState as GuildVoiceStatePayload,
         VoiceServerUpdate as VoiceServerUpdatePayload,
     )
+
+
+try:
+    from orjson import dumps as _dumps, loads
+
+    def dumps(obj: Any) -> str:
+        return _dumps(obj).decode()
+
+except ImportError:
+    from json import dumps, loads

mafic/node.py

@@ -0,0 +1,153 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+from asyncio import create_task, sleep
+from logging import getLogger
+from typing import TYPE_CHECKING
+
+from aiohttp import ClientSession, WSMsgType
+
+from .__libraries import ExponentialBackoff, dumps, loads
+
+if TYPE_CHECKING:
+    from asyncio import Task
+    from typing import Any
+
+    from aiohttp import ClientWebSocketResponse
+
+    from .__libraries import Client, VoiceServerUpdatePayload
+    from .typings import Coro, OutgoingMessage
+
+_log = getLogger(__name__)
+
+
+class Node:
+    def __init__(
+        self,
+        *,
+        host: str,
+        port: int,
+        label: str,
+        password: str,
+        client: Client,
+        secure: bool = False,
+        heartbeat: int = 30,
+        timeout: float = 10,
+        session: ClientSession | None = None,
+    ) -> None:
+        self._host = host
+        self._port = port
+        self._label = label
+        self.__password = password
+        self._secure = secure
+        self._heartbeat = heartbeat
+        self._timeout = timeout
+        self._client = client
+        self.__session = session
+
+        self._rest_uri = f"http{'s' if secure else ''}://{host}:{port}"
+        self._ws_uri = f"ws{'s' if secure else ''}://{host}:{port}"
+
+        self._ws: ClientWebSocketResponse | None = None
+        self._ws_task: Task[None] | None = None
+
+        self._available = False
+        self._backoff = ExponentialBackoff()
+
+    @property
+    def host(self) -> str:
+        return self._host
+
+    @property
+    def port(self) -> int:
+        return self._port
+
+    @property
+    def label(self) -> str:
+        return self._label
+
+    @property
+    def client(self) -> Client:
+        return self._client
+
+    @property
+    def secure(self) -> bool:
+        return self._secure
+
+    async def _connect(self) -> None:
+        await self._client.wait_until_ready()
+        assert self._client.user is not None
+
+        if self.__session is None:
+            self.__session = ClientSession()
+
+        session = self.__session
+
+        headers: dict[str, str] = {
+            "Authorization": self.__password,
+            "User-Id": str(self._client.user.id),
+            "Client-Name": f"Mafic/{__import__('mafic').__version__}",
+        }
+
+        self._ws = await session.ws_connect(  # pyright: ignore[reportUnknownMemberType]
+            self._ws_uri,
+            timeout=self._timeout,
+            heartbeat=self._heartbeat,
+            headers=headers,
+        )
+        self._ws_task = create_task(
+            self._ws_listener(), name=f"mafic node {self._label}"
+        )
+
+        self._available = True
+
+        await sleep(1)
+        if self._available:
+            self._backoff = ExponentialBackoff()
+
+    async def _ws_listener(self) -> None:
+        if self._ws is None:
+            raise RuntimeError(
+                "Websocket is not connected but attempted to listen, report this."
+            )
+
+        async for message in self._ws:
+            # Please aiohttp, fix your typehints.
+            _type: WSMsgType = message.type  # pyright: ignore[reportUnknownMemberType]
+
+            if _type is WSMsgType.CLOSED:
+                self._available = False
+                self._ws = None
+
+                wait_time = self._backoff.delay()
+                _log.warning("Websocket closed, reconnecting in %.2f...", wait_time)
+
+                await sleep(wait_time)
+                create_task(self._connect())
+                return
+            else:
+                create_task(self._handle_msg(message.json(loads=loads)))
+
+    async def __send(self, data: OutgoingMessage) -> None:
+        if self._ws is None:
+            raise RuntimeError(
+                "Websocket is not connected but attempted to send, report this."
+            )
+
+        await self._ws.send_json(data, dumps=dumps)
+
+    def send_voice_server_update(
+        self, guild_id: int, session_id: str, data: VoiceServerUpdatePayload
+    ) -> Coro[None]:
+        return self.__send(
+            {
+                "op": "voiceUpdate",
+                "guildId": str(guild_id),
+                "sessionId": session_id,
+                "event": data,
+            }
+        )
+
+    async def _handle_msg(self, data: dict[str, Any]) -> None:
+        raise NotImplementedError

mafic/player.py

@@ -4,27 +4,52 @@
 
 from typing import TYPE_CHECKING
 
-from .libraries import VoiceProtocol
+from .__libraries import VoiceProtocol
+from .pool import NodePool
 
 if TYPE_CHECKING:
-    from .libraries import (
+    from .__libraries import (
         Client,
         Connectable,
         GuildVoiceStatePayload,
         VoiceServerUpdatePayload,
     )
+    from .node import Node
 
 
 class Player(VoiceProtocol):
-    def __init__(self, client: Client, channel: Connectable) -> None:
+    def __init__(
+        self,
+        client: Client,
+        channel: Connectable,
+        *,
+        node: Node | None = None,
+    ) -> None:
         self.client: Client = client
         self.channel: Connectable = channel
+        self._node = node
+
+        self._guild_id: int | None = None
+        self._session_id: str | None = None
+        self._server_state: VoiceServerUpdatePayload | None = None
 
     async def on_voice_state_update(self, data: GuildVoiceStatePayload) -> None:
         raise NotImplementedError
 
     async def on_voice_server_update(self, data: VoiceServerUpdatePayload) -> None:
-        raise NotImplementedError
+        if self._node is None:
+            self._node = NodePool.get_node(
+                guild_id=data["guild_id"], endpoint=data["endpoint"]
+            )
+
+        self._server_state = data
+
+        if self._guild_id is None or self._session_id is None:
+            return
+
+        await self._node.send_voice_server_update(
+            guild_id=self._guild_id, session_id=self._session_id, data=data
+        )
 
     async def connect(self, *, timeout: float, reconnect: bool) -> None:
         raise NotImplementedError

mafic/pool.py

@@ -0,0 +1,59 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+from random import choice
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import ClassVar, Optional
+
+    from aiohttp import ClientSession
+
+    from .__libraries import Client
+    from .node import Node
+
+
+class NodePool:
+    _nodes: ClassVar[dict[str, Node]] = {}
+
+    @property
+    def nodes(self) -> dict[str, Node]:
+        return self._nodes
+
+    @classmethod
+    def create_node(
+        cls,
+        *,
+        host: str,
+        port: int,
+        label: str,
+        password: str,
+        client: Client,
+        secure: bool = False,
+        heartbeat: int = 30,
+        timeout: float = 10,
+        session: ClientSession | None = None,
+    ) -> Node:
+        node = Node(
+            host=host,
+            port=port,
+            label=label,
+            password=password,
+            client=client,
+            secure=secure,
+            heartbeat=heartbeat,
+            timeout=timeout,
+            session=session,
+        )
+
+        # TODO: assign dicts for regions and such
+        cls._nodes[label] = node
+
+        return node
+
+    @classmethod
+    def get_node(cls, *, guild_id: str | int, endpoint: Optional[str]) -> Node:
+        # TODO: use guild id, endpoint and other stuff like usage to determine node
+
+        return choice(list(cls._nodes.values()))

mafic/typings/__init__.py

@@ -0,0 +1,4 @@
+# SPDX-License-Identifier: MIT
+
+from .misc import *
+from .outgoing import *

mafic/typings/misc.py

@@ -0,0 +1,10 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+from typing import Any, Coroutine, TypeVar
+
+__all__ = ("Coro",)
+T = TypeVar("T")
+
+Coro = Coroutine[Any, Any, T]