lavasnek_rs

lavasnek_rs

lavalink-rs bindings for Python

Cheat Sheet:

  • Functions that return a Result<T, E> mean that it can raise an exception. T is the type they return normally, and E is a list of possible exceptions that can raise.
  • Functions that return an Option<T> mean that the value returned can be None, where T would be the type of the returned value if not None.
  • If something returns a Future<T>, it means that it returns this, and that function should be awaited to work.
  • / on arguments means the end of positional arguments.
  • Self (with a capital S) means the type of self.
  • A type prefixed with impl means it's a Class that implements that Trait type.
 1# -*- coding: utf-8 -*-
 2"""
 3lavasnek_rs
 4-------------------
 5lavalink-rs bindings for Python
 6
 7Cheat Sheet:
 8
 9- Functions that return a `Result<T, E>` mean that it can raise an exception. `T` is the type they
10return normally, and `E` is a list of possible exceptions that can raise.
11- Functions that return an `Option<T>` mean that the value returned can be `None`, where `T` would be
12the type of the returned value if not `None`.
13- If something returns a `Future<T>`, it means that it returns
14[this](https://docs.python.org/3/library/asyncio-future.html?#asyncio.Future),
15and that function should be awaited to work.
16- / on arguments means the end of positional arguments.
17- Self (with a capital S) means the type of self.
18- A type prefixed with `impl` means it's a Class that implements that Trait type.
19"""
20
21from .lavasnek_rs import *
22
23__version__ = "0.1.0-alpha.5"
24__author__ = "vicky5124 <vickyf5124@gmail.com>"
25__license__ = "MPL-2.0"
26
27__all__ = [
28    "rust_sleep",
29    "log_something",
30    "Lavalink",
31    "LavalinkBuilder",
32    "PlayBuilder",
33    "LavalinkEventHandler",
34    "ConnectionInfo",
35    "Track",
36    "Tracks",
37    "TrackQueue",
38    "Info",
39    "PlaylistInfo",
40    "Node",
41    "Band",
42    "Stats",
43    "PlayerUpdate",
44    "TrackStart",
45    "TrackFinish",
46    "TrackException",
47    "TrackStuck",
48    "WebSocketClosed",
49    "PlayerDestroyed",
50    "NoSessionPresent",
51    "NetworkError",
52]
def rust_sleep(seconds, /)

Test function, do not use.

def log_something()

Test function, do not use.

class LavalinkBuilder:

__new__()

All of the methods that return Self also modify Self, so you can chain them, or call the individually.

Positional Arguments:

  • bot_id : Unsigned 64 bit integer
  • bot_token : String

Returns: Self

LavalinkBuilder()
async def build(cls, event_handler: object) -> lavasnek_rs.Lavalink:

Uses the Self data to build a Lavalink client and return it.

Can raise an exception if it's unable to connect to the lavalink server, discord server, or both.

Positional Arguments:

  • event_handler : impl LavalinkEventHandler

Returns: Future<Result<Lavalink, builtins.ConnectionError>>

def set_host(self, host: str) -> lavasnek_rs.LavalinkBuilder:

Sets the host. (Default to: 127.0.0.1)

Positional Arguments:

  • host : String

Returns: Self

def set_port(self, port: int) -> lavasnek_rs.LavalinkBuilder:

Sets the port. (Default to: 2333)

Positional Arguments:

  • port : Unsigned 16 bit integer

Returns: Self

def set_addr(self, address: str) -> lavasnek_rs.LavalinkBuilder:

Sets the host and port from an address.

Can raise an exception if the address is an invalid IPv4 or IPv6.

Positional Arguments:

  • address : String

Returns: Result<Self, ipaddress.AddressValueError>

def set_password(self, password: str) -> lavasnek_rs.LavalinkBuilder:

Sets the lavalink password. (Default to: "youshallnotpass")

Positional Arguments:

  • password : String

Returns: Self

def set_shard_count(self, shard_count: int) -> lavasnek_rs.LavalinkBuilder:

Sets the number of shards. (Default to: 1)

Positional Arguments:

  • shard_count : Unsigned 64 bit integer

Returns: Self

def set_bot_id(self, bot_id: int) -> lavasnek_rs.LavalinkBuilder:

Sets the ID of the bot.

Positional Arguments:

  • bot_id : Unsigned 64 bit integer

Returns: Self

def set_bot_token(self, bot_token: str) -> lavasnek_rs.LavalinkBuilder:

Sets the token of the bot.

Positional Arguments:

  • bot_token : String

Returns: Self

def set_is_ssl(self, is_ssl: bool) -> lavasnek_rs.LavalinkBuilder:

Sets if the lavalink server is behind SSL. (Default to: False)

Positional Arguments:

  • is_ssl : bool

Returns: Self

def set_start_gateway(self, start_gateway: bool) -> lavasnek_rs.LavalinkBuilder:

Sets if the discord gateway for voice connections should start or not. (Default to: True)

Positional Arguments:

  • start_gateway : bool

Returns: Self

def set_gateway_start_wait_time_secs(self, time: int) -> lavasnek_rs.LavalinkBuilder:

Sets the time to wait before starting the first discord gateway connection. (Default to: 6 seconds)

Positional Arguments:

  • time : Unsigned 64 bit integer

Returns: Self

def set_gateway_start_wait_time_millis(self, time: int) -> lavasnek_rs.LavalinkBuilder:

Sets the time to wait before starting the first discord gateway connection.

Positional Arguments:

  • time : Unsigned 64 bit integer

Returns: Self

class PlayBuilder:
PlayBuilder()
async def start(self) -> None:

Starts playing the track.

Returns: Future<Result<None, lavasnek_rs.NetworkError>>

async def queue(self) -> None:

Adds the track to the node queue.

If there's no queue loop running, this will start one up, and add it to the running loops on Lavalink.loops()

Needs for `Lavalink.create_session() to be called first.

Returns: Future<Result<None, [lavasnek_rs.NoSessionPresent, lavasnek_rs.NetworkError]>>

def to_track_queue(self) -> lavasnek_rs.TrackQueue:

Generates a TrackQueue from the builder.

Returns: TrackQueue

def requester(self, requester: int) -> lavasnek_rs.PlayBuilder:

Sets the person that requested the song

Positional Arguments:

  • requester : Unsigned 64 bit integer (User ID)

Returns: Self

def replace(self, replace: bool) -> lavasnek_rs.PlayBuilder:

Sets if the current playing track should be replaced with this new one.

Positional Arguments:

Returns: Self

def start_time_secs(self, start: int) -> lavasnek_rs.PlayBuilder:

Sets the time the track will start at in seconds.

Positional Arguments:

  • start : Unsigned 64 bit integer

Returns: Self

def finish_time_secs(self, finish: int) -> lavasnek_rs.PlayBuilder:

Sets the time the track will finish at in seconds.

Positional Arguments:

  • finish : Unsigned 64 bit integer

Returns: Self

def start_time_millis(self, start: int) -> lavasnek_rs.PlayBuilder:

Sets the time the track will start at in milliseconds.

Positional Arguments:

  • start : Unsigned 64 bit integer

Returns: Self

def finish_time_millis(self, finish: int) -> lavasnek_rs.PlayBuilder:

Sets the time the track will finish at in milliseconds.

Positional Arguments:

  • finish : Unsigned 64 bit integer

Returns: Self

class LavalinkEventHandler:

The lavalink event handler. This is a trait, so it defines the structure a class should have.

Make a class with the methods and signatures this class defines, and add that class to LavalinkBuilder.build()

If code inside any of the event raises an error, the traceback will be printed to stderr, and the variables sys.last_type, sys.last_value and sys.last_traceback will be set to the type, value and traceback of the printed exception respectively.

Some examples:

# Just a single event
class EventHandler:
    async def track_start(self, lava_client, event):
        print(event)

lavalink_client = await client_builder.build(EventHandler)
# No events
class EventHandler:
    pass

lavalink_client = await client_builder.build(EventHandler)
# Just a single event
class EventHandler:
    async def stats(self, lava_client, event):
        print(event)
    async def player_update(self, lava_client, event):
        print(event)
    async def track_start(self, lava_client, event):
        print(event)
    async def track_finish(self, lava_client, event):
        print(event)
    async def track_exception(self, lava_client, event):
        print(event)
    async def track_stuck(self, lava_client, event):
        print(event)
    async def websocket_closed(self, lava_client, event):
        print(event)
    async def player_destroyed(self, lava_client, event):
        print(event)

lavalink_client = await client_builder.build(EventHandler)
LavalinkEventHandler()
async def stats(self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.Stats) -> None:

Periodic event that returns the statistics of the server.

Positional Arguments:

Returns: Future<None>

async def player_update( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.PlayerUpdate) -> None:

Event that triggers when a player updates.

Positional Arguments:

Returns: Future<None>

async def track_start( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.TrackStart) -> None:

Event that triggers when a track starts playing.

Positional Arguments:

Returns: Future<None>

async def track_finish( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.TrackFinish) -> None:

Event that triggers when a track finishes playing.

Positional Arguments:

Returns: Future<None>

async def track_exception( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.TrackException) -> None:

Event that triggers when a track raises an exception on the Lavalink server.

Positional Arguments:

Returns: Future<None>

async def track_stuck( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.TrackStuck) -> None:

Event that triggers when a track gets stuck while playing.

Positional Arguments:

Returns: Future<None>

async def websocket_closed( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.WebsocketClosed) -> None:

Event that triggers when the websocket connection to the voice channel closes.

Positional Arguments:

Returns: Future<None>

async def player_destroyed( self, client: lavasnek_rs.Lavalink, event: lavasnek_rs.PlayerDestroyed) -> None:

Event that triggers when the player gets destroyed on a guild.

Positional Arguments:

Returns: Future<None>

class ConnectionInfo:

If you use a 3rd party method of joining a voice channel, you can get the values required for this from the VOICE_STATE_UPDATE and VOICE_SERVER_UPDATE events, and use raw_handle_event_voice_state_update() + raw_handle_event_voice_server_update() or manually build a dict with them, and use ConnectionInfo({ ... })

With hikari:

@bot.listen()
async def voice_state_update(event: hikari.VoiceStateUpdateEvent) -> None:
    await bot.data.lavalink.raw_handle_event_voice_state_update(
        event.state.guild_id,
        event.state.user_id,
        event.state.session_id,
        event.state.channel_id,
    )

@bot.listen()
async def voice_server_update(event: hikari.VoiceServerUpdateEvent) -> None:
    await bot.data.lavalink.raw_handle_event_voice_server_update(
        event.guild_id, event.endpoint, event.token
    )

Fields:

  • guild_id : Unsigned 64 bit integer
  • channel_id : Unsigned 64 bit integer
  • endpoint : String
  • token : String
  • session_id : String
ConnectionInfo()
class Track:

__new__()

Positional Arguments:

Returns: Self

Track()
track: str

The playable track.

Contains String

Information about the track.

Contains Option<Info>

class Tracks:
Tracks()
playlist_info: lavasnek_rs.PlaylistInfo

Information about the playlist.

Contains Option<PlaylistInfo>

tracks: List[lavasnek_rs.Track]

The tracks that can be played

Contains List<Track>

load_type: str

Contains String

class TrackQueue:
TrackQueue()
requester: int

The user id who requested the track if set by the PlayBuilder

Contains Option<Unsigned 64 bit integer>

The playable track.

Contains Track

start_time: int

The time the track will start at.

Contains Unsigned 64 bit integer

end_time: int

The time the track will finish at.

Contains Option<Unsigned 64 bit integer>

class Info:

__new__()

Arguments:

Returns: Self

Info()
is_seekable: bool

Contains bool

uri: str

Contains String

is_stream: bool

Contains bool

author: str

Contains String

identifier: str

Contains String

length: int

Contains Unsigned 64 bit integer

position: int

Contains Unsigned 64 bit integer

title: str

Contains String

class PlaylistInfo:
PlaylistInfo()
name: Optional[str]

The name of the playlist.

Contains Option<String>

selected_track: Optional[int]

Contains Option<Signed 64 bit integer>

class Node:
Node()
def get_data(self) -> Union[Any, Dict[Any, Any]]:

Use this to get the currently stored data on the Node.

T is whatever type you give to set_data's data parameter, but if you call this method before it, it will default to a Dict.

Returns T

def set_data(self, data: Union[Any, Dict[Any, Any]]) -> None:

Use this to set the tored data of the Node.

Returns None

volume: int

Contains Unsigned 16 bit integer

guild: int

Contains Unsigned 64 bit integer

is_paused: bool

Contains bool

is_on_loops: bool

Contains bool

now_playing: Optional[lavasnek_rs.TrackQueue]

Contains Option<TrackQueue>

queue: List[lavasnek_rs.TrackQueue]

Contains List<TrackQueue>

class Band:

See Lavalink.equalize_all for more info.

band_num: int = 14 # 0 to 14
bain: float = 0.125 # -0.25 to 1.0

band = Band(band_num, gain)
Band()
gain: float

Contains 64 bit float

band: int

Contains Unsigned 8 bit integer

class Stats:
Stats()
memory_free: int

Contains Signed 64 bit integer

op: str

Contains String

players: int

Contains Signed 64 bit integer

memory_reservable: int

Contains Signed 64 bit integer

memory_used: int

Contains Signed 64 bit integer

memory_allocated: int

Contains Signed 64 bit integer

playing_players: int

Contains Signed 64 bit integer

uptime: int

Contains Signed 64 bit integer

cpu_cores: int

Contains Signed 64 bit integer

frame_stats_nulled: Optional[int]

Contains Optional Signed 64 bit integer

frame_stats_deficit: Optional[int]

Contains Optional Signed 64 bit integer

frame_stats_sent: Optional[int]

Contains Optional Signed 64 bit integer

cpu_system_load: float

Contains 64 bit floating point

class PlayerUpdate:
PlayerUpdate()
state_position: int

Contains Signed 64 bit integer

state_time: int

Contains Signed 64 bit integer

guild_id: int

Contains Unsigned 64 bit integer

op: str

Contains String

class TrackStart:
TrackStart()
op: str

Contains String

track: str

Contains String

track_start_type: str

Contains String

guild_id: int

Contains Unsigned 64 bit integer

class TrackFinish:
TrackFinish()
op: str

Contains String

reason: str

Contains String

track: str

Contains String

guild_id: int

Contains Unsigned 64 bit integer

track_finish_type: str

Contains String

class TrackException:
TrackException()
exception_severity: str

Contains String

track_exception_type: str

Contains String

op: str

Contains String

guild_id: int

Contains Unsigned 64 bit integer

exception_cause: str

Contains String

error: str

Contains String

track: str

Contains String

exception_message: str

Contains String

class TrackStuck:
TrackStuck()
track_stuck_type: str

Contains String

op: str

Contains String

track: str

Contains String

threshold_ms: int

Contains Unsigned 64 bit integer

guild_id: int

Contains Unsigned 64 bit integer

class WebSocketClosed:
WebSocketClosed()
guild_id

Contains Unsigned 64 bit integer

by_remote

Contains bool

op

Contains String

websocket_closed_type

Contains String

user_id

Contains Unsigned 64 bit integer

code

Contains Unsigned 64 bit integer

class PlayerDestroyed:
PlayerDestroyed(*args, **kwargs)
guild_id

Contains Unsigned 64 bit integer

player_destroyed_type

Contains String

cleanup

Contains bool

op

Contains String

user_id

Contains Unsigned 64 bit integer

class NoSessionPresent(builtins.Exception):

Common base class for all non-exit exceptions.

Inherited Members
builtins.Exception
Exception
builtins.BaseException
with_traceback
args
class NetworkError(builtins.Exception):

Common base class for all non-exit exceptions.

Inherited Members
builtins.Exception
Exception
builtins.BaseException
with_traceback
args