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

[pre-commit.ci] auto fixes from pre-commit.com hooks 

for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2026-06-11 17:52:44 +00:00
parent 0bdca1975c
commit 5c5d68d1f2
1 changed files with 18 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
pomice/pool.py
View File

@ -18,7 +18,8 @@ from urllib.parse import quote
import aiohttp import aiohttp
import orjson as json import orjson as json
from discord import Client, VoiceChannel from discord import Client
from discord import VoiceChannel
from discord.ext import commands from discord.ext import commands
try: try:
@ -31,13 +32,7 @@ from websockets import exceptions
from . import __version__ from . import __version__
from . import applemusic from . import applemusic
from . import spotify from . import spotify
from .enums import ( from .enums import TrackType, NodeAlgorithm, PlaylistType, SearchType, URLRegex
TrackType,
NodeAlgorithm,
PlaylistType,
SearchType,
URLRegex
)
from .exceptions import InvalidSpotifyClientAuthorization from .exceptions import InvalidSpotifyClientAuthorization
from .exceptions import LavalinkVersionIncompatible from .exceptions import LavalinkVersionIncompatible
from .exceptions import NodeConnectionFailure from .exceptions import NodeConnectionFailure
@ -145,7 +140,7 @@ class Node:
self._resume_timeout: int = resume_timeout self._resume_timeout: int = resume_timeout
self._secure: bool = secure self._secure: bool = secure
self._fallback: bool = fallback self._fallback: bool = fallback
self._location = location self._location = location
self._websocket_uri: str = f"{'wss' if self._secure else 'ws'}://{self._host}:{self._port}" self._websocket_uri: str = f"{'wss' if self._secure else 'ws'}://{self._host}:{self._port}"
@ -218,13 +213,13 @@ class Node:
def bot(self) -> Client: def bot(self) -> Client:
"""Property which returns the discord.py client linked to this node""" """Property which returns the discord.py client linked to this node"""
return self._bot return self._bot
@property @property
def location(self) -> str: def location(self) -> str:
""" """
Property which returns the default region unless set specifically Property which returns the default region unless set specifically
""" """
return self._location return self._location
@property @property
@ -984,7 +979,9 @@ class NodePool:
return len(self._nodes.values()) return len(self._nodes.values())
@classmethod @classmethod
def get_best_node(cls, *, algorithm: NodeAlgorithm, channel: Optional[VoiceChannel] = None) -> Node: def get_best_node(
cls, *, algorithm: NodeAlgorithm, channel: Optional[VoiceChannel] = None,
) -> Node:
"""Fetches the best node based on an NodeAlgorithm. """Fetches the best node based on an NodeAlgorithm.
This option is preferred if you want to choose the best node This option is preferred if you want to choose the best node
from a multi-node setup using either the node's latency from a multi-node setup using either the node's latency
@ -998,7 +995,7 @@ class NodePool:
based on how players it has. This method will return a node with based on how players it has. This method will return a node with
the least amount of players the least amount of players
""" """
available_nodes: List[Node] = [node for node in cls._nodes.values() if node._available] available_nodes: List[Node] = [node for node in cls._nodes.values() if node._available]
if not available_nodes: if not available_nodes:
@ -1011,20 +1008,19 @@ class NodePool:
elif algorithm == NodeAlgorithm.by_players: elif algorithm == NodeAlgorithm.by_players:
tested_nodes = {node: len(node.players.keys()) for node in available_nodes} tested_nodes = {node: len(node.players.keys()) for node in available_nodes}
return min(tested_nodes, key=tested_nodes.get) # type: ignore return min(tested_nodes, key=tested_nodes.get) # type: ignore
elif algorithm == NodeAlgorithm.by_location and isinstance(channel, VoiceChannel): elif algorithm == NodeAlgorithm.by_location and isinstance(channel, VoiceChannel):
tested_nodes = {} tested_nodes = {}
chosen_region = channel.rtc_region chosen_region = channel.rtc_region
if not chosen_region: if not chosen_region:
return cls.get_best_node(algorithm=NodeAlgorithm.by_ping) return cls.get_best_node(algorithm=NodeAlgorithm.by_ping)
if (node := next( if node := next(
(node for node in available_nodes if node.location == chosen_region), (node for node in available_nodes if node.location == chosen_region), None,
None ):
)):
return node return node
return random.choice(available_nodes) return random.choice(available_nodes)
else: else:
raise ValueError( raise ValueError(