API Reference¶

This page is autogenerated from the docstrings inside the code.

Alchemy¶

Relational database classes and methods.

class royalnet.alchemy.Alchemy(database_uri: str, tables: Set)¶

A wrapper around sqlalchemy.orm that allows the instantiation of multiple engines at once while maintaining a single declarative class for all of them.

__init__(database_uri: str, tables: Set)¶

Create a new Alchemy object.

Parameters

database_uri – The database URI .
tables – The set of tables to be created and used in the selected database. Check the tables submodule for more details.

get(table: Union[str, type]) → sqlalchemy.sql.schema.Table¶

Get the table with a specified name or class.

Parameters: table – The table name or table class you want to get.
Raises: TableNotFoundError – if the requested table was not found.

session_acm()¶

Create a Session as a async context manager (that can be used in async with statements).

The Session will be closed safely when the context manager exits (even in case of error).

Example

You can use the async context manager like this:

async with alchemy.session_acm() as session:
    # Do some stuff
    ...
    # Commit the session
    await asyncify(session.commit)

session_cm()¶

Create a Session as a context manager (that can be used in with statements).

The Session will be closed safely when the context manager exits (even in case of error).

Example

You can use the context manager like this:

with alchemy.session_cm() as session:
    # Do some stuff
    ...
    # Commit the session
    session.commit()

royalnet.alchemy.table_dfs(starting_table: sqlalchemy.sql.schema.Table, ending_table: sqlalchemy.sql.schema.Table) → tuple¶

Depth-first-search for the path from the starting table to the ending table.

Returns: A tuple containing the path, starting from the starting table and ending at the ending table.

exception royalnet.alchemy.AlchemyException¶: Base class for Alchemy exceptions.

exception royalnet.alchemy.TableNotFoundError¶: The requested table was not found.

Backpack¶

A Pack that is imported by default by all Royalnet instances.

Keep things here to a minimum!

Bard¶

class royalnet.bard.YtdlInfo(info: Dict[str, Any])¶

A wrapper around youtube_dl extracted info.

__init__(info: Dict[str, Any])¶: Create a YtdlInfo from the dict returned by the YoutubeDL.extract_info() function.

Warning

Does not download the info, to do that use retrieve_for_url().

classmethod from_url(url, loop: Optional[asyncio.events.AbstractEventLoop] = None, **ytdl_args) → List[royalnet.bard.ytdlinfo.YtdlInfo]¶

Fetch the info for an url through YoutubeDL.

Returns: A list containing the infos for the requested videos.

class royalnet.bard.YtdlFile(url: str, info: Optional[royalnet.bard.ytdlinfo.YtdlInfo] = None, filename: Optional[str] = None, ytdl_args: Optional[Dict[str, Any]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None)¶

A representation of a file download with youtube_dl.

__init__(url: str, info: Optional[royalnet.bard.ytdlinfo.YtdlInfo] = None, filename: Optional[str] = None, ytdl_args: Optional[Dict[str, Any]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None)¶: Create a YtdlFile instance.

Warning

Please avoid using directly __init__(), use from_url() instead!

aopen()¶

Open the downloaded file as an async context manager (and download it if it isn’t available yet).

Example

You can use the async context manager like this:

async with ytdlfile.aopen() as file:
    b: bytes = file.read()

default_ytdl_args = {'ignoreerrors': True, 'no_warnings': False, 'noplaylist': True, 'outtmpl': './downloads/%(epoch)s-%(title)s-%(id)s.%(ext)s', 'quiet': False}¶

async delete_asap()¶: As soon as nothing is using the file, delete it.

async download_file() → None¶: Download the file.

classmethod from_url(url: str, **ytdl_args) → List[royalnet.bard.ytdlfile.YtdlFile]¶: Create a list of YtdlFile from a URL.

property has_info¶: Does the YtdlFile have info available?

property is_downloaded¶: Has the file been downloaded yet?

async retrieve_info() → None¶: Retrieve info about the YtdlFile through YoutubeDL.

class royalnet.bard.YtdlMp3(ytdl_file: royalnet.bard.ytdlfile.YtdlFile)¶

A representation of a YtdlFile conversion to mp3.

async convert_to_mp3() → None¶: Convert the file to mp3 with ffmpeg.

async delete_asap() → None¶: Delete the mp3 file.

classmethod from_url(url, **ytdl_args) → List[royalnet.bard.ytdlmp3.YtdlMp3]¶: Create a list of YtdlMp3 from a URL.

property info¶: Shortcut to get the YtdlInfo of the object.

property is_converted¶: Has the file been converted?

class royalnet.bard.YtdlDiscord(ytdl_file: royalnet.bard.ytdlfile.YtdlFile)¶

A representation of a YtdlFile conversion to the discord PCM format.

async convert_to_pcm() → None¶: Convert the file to pcm with ffmpeg.

async delete_asap() → None¶: Delete the mp3 file.

embed() → discord.embeds.Embed¶: Return this info as a discord.Embed.

classmethod from_url(url, **ytdl_args) → List[royalnet.bard.ytdldiscord.YtdlDiscord]¶: Create a list of YtdlMp3 from a URL.

property info¶: Shortcut to get the YtdlInfo of the object.

property is_converted¶: Has the file been converted?

spawn_audiosource()¶

Commands¶

class royalnet.commands.CommandInterface(config: Dict[str, Any])¶

property alchemy¶: A shortcut for serf.alchemy.

async call_herald_event(destination: str, event_name: str, **kwargs) → dict¶

Call an event function on a different Serf.

Example

You can run a function on a DiscordSerf from a TelegramSerf.

config = None¶: The config section for the pack of the command.

constellation = None¶

A reference to the Constellation that is implementing this CommandInterface.

Example

A reference to a Constellation.

property loop¶: A shortcut for serf.loop.

name = NotImplemented¶

The name of the CommandInterface that’s being implemented.

Examples

telegram, discord, console…

prefix = NotImplemented¶

The prefix used by commands on the interface.

Examples

/ on Telegram, ! on Discord.

serf = None¶

A reference to the Serf that is implementing this CommandInterface.

Example

A reference to a TelegramSerf.

class royalnet.commands.Command(interface: royalnet.commands.commandinterface.CommandInterface)¶

property alchemy¶: A shortcut for interface.alchemy.

aliases = []¶

A list of possible aliases for a command.

Example

To be able to call /e as an alias for /example, one should set aliases to ["e"].

property config¶: A shortcut for interface.config.

description = NotImplemented¶: A small description of the command, to be displayed when the command is being autocompleted.

property loop¶: A shortcut for interface.loop.

name = NotImplemented¶

The main name of the command.

Example

To be able to call /example on Telegram, the name should be "example".

async run(args: royalnet.commands.commandargs.CommandArgs, data: royalnet.commands.commanddata.CommandData) → None¶

property serf¶: A shortcut for interface.serf.

syntax = ''¶: The syntax of the command, to be displayed when a InvalidInputError is raised, in the format (required_arg) [optional_arg].

class royalnet.commands.CommandData(interface: royalnet.commands.commandinterface.CommandInterface, session: Optional[sqlalchemy.orm.session.Session], loop: asyncio.events.AbstractEventLoop)¶

async delete_invoking(error_if_unavailable=False) → None¶

Delete the invoking message, if supported by the interface.

The invoking message is the message send by the user that contains the command.

Parameters: error_if_unavailable – if True, raise an exception if the message cannot been deleted.

async get_author(error_if_none: bool = False)¶

Try to find the identifier of the user that sent the message. That probably means, the database row identifying the user.

Parameters: error_if_none – Raise an exception if this is True and the call has no author.

async reply(text: str) → None¶

Send a text message to the channel where the call was made.

Parameters: text – The text to be sent, possibly formatted in the weird undescribed markup that I’m using.

property session¶

Get the Alchemy Session, if it is available.

Raises: UnsupportedError – if no session is available.

async session_commit()¶: Commit the changes to the session.

class royalnet.commands.CommandArgs¶

An interface to easily access the arguments of a command.

Inherits from list.

__getitem__(item)¶

Access arguments as if they were a list.

Raises: InvalidInputError – if the requested argument does not exist.

Examples

# /pasta spaghetti aldente
>>> self[0]
"spaghetti"
>>> self[1]
"aldente"
>>> self[2]
# InvalidInputError: Missing argument #3.
>>> self[0:2]
["spaghetti", "aldente"]

joined(*, require_at_least=0) → str¶

Get the arguments as a space-joined string.

Parameters: require_at_least – the minimum amount of arguments required.
Raises: InvalidInputError – if there are less than require_at_least arguments.
Returns: The space-joined string.

Examples

# /pasta spaghetti aldente
>>> self.joined()
"spaghetti aldente"
>>> self.joined(require_at_least=3)
# InvalidInputError: Not enough arguments specified (minimum is 3).

match(pattern: Union[str, Pattern], *flags) → Sequence[AnyStr]¶

Match the joined() string to a re.Pattern-like object.

Parameters: pattern – The regex pattern to be passed to re.match().
Raises: InvalidInputError – if the pattern doesn’t match.
Returns: The matched groups, as returned by re.Match.groups().

optional(index: int, default=None) → Optional[str]¶

Get the argument at a specific index, but don’t raise an error if nothing is found, instead returning the default value.

Parameters

index – The index of the argument you want to retrieve.
default – The value returned if the argument is missing.

Returns

Either the argument or the default value, defaulting to None.

Examples

# /pasta spaghetti aldente
>>> self.optional(0)
"spaghetti"
>>> self.optional(2)
None
>>> self.optional(2, default="carbonara")
"carbonara"

exception royalnet.commands.CommandError(message='')¶

Something went wrong during the execution of this command.

Display an error message to the user, explaining what went wrong.

exception royalnet.commands.InvalidInputError(message='')¶: The command has received invalid input and cannot complete.

exception royalnet.commands.UnsupportedError(message='')¶: A requested feature is not available on this interface.

exception royalnet.commands.ConfigurationError(message='')¶: The command cannot work because of a wrong configuration by part of the Royalnet admin.

exception royalnet.commands.ExternalError(message='')¶: The command failed to execute, but the problem was because of an external factor (such as an external API going down).

exception royalnet.commands.UserError(message='')¶: The command failed to execute, and the error is because of something that the user did.

exception royalnet.commands.ProgramError(message='')¶: The command encountered an error in the program.

class royalnet.commands.Event(interface: royalnet.commands.commandinterface.CommandInterface)¶

A remote procedure call triggered by a royalnet.herald request.

__init__(interface: royalnet.commands.commandinterface.CommandInterface)¶: Bind the event to a Serf.

property alchemy¶: A shortcut for interface.alchemy.

property config¶: A shortcut for interface.config.

interface = None¶: The CommandInterface available to this Event.

property loop¶: A shortcut for interface.loop.

name = NotImplemented¶: The event_name that will trigger this event.

async run(**kwargs)¶

property serf¶: A shortcut for interface.serf.

Constellation¶

The part of royalnet that handles the webserver and webpages.

It uses many features of starlette.

class royalnet.constellation.Constellation(alchemy_cfg: Dict[str, Any], herald_cfg: Dict[str, Any], packs_cfg: Dict[str, Any], constellation_cfg: Dict[str, Any], **_)¶

The class that represents the webserver.

It runs multiple Star, which represent the routes of the website.

It also handles the Alchemy connection, and Herald connections too.

Interface = None¶: The CommandInterface class of this Constellation.

address = None¶: The address that the Constellation will bind to when run.

alchemy = None¶: The Alchemy of this Constellation.

events = None¶: A dictionary containing all Event that can be handled by this Constellation.

herald = None¶: The Link object connecting the Constellation to the rest of the herald network.

herald_task = None¶: A reference to the aio.Task that runs the rh.Link.

init_herald(herald_cfg: Dict[str, Any])¶: Create a rh.Link.

interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]¶: Create the rc.CommandInterface class for the Constellation.

async network_handler(message: Union[royalnet.herald.request.Request, royalnet.herald.broadcast.Broadcast]) → royalnet.herald.response.Response¶

port = None¶: The port on which the Constellation will listen for connection on.

register_events(events: List[Type[royalnet.commands.event.Event]], pack_cfg: Dict[str, Any])¶

register_exc_stars(exc_stars: List[Type[royalnet.constellation.star.ExceptionStar]], pack_cfg: Dict[str, Any])¶

register_page_stars(page_stars: List[Type[royalnet.constellation.star.PageStar]], pack_cfg: Dict[str, Any])¶

run_blocking()¶

classmethod run_process(alchemy_cfg: Dict[str, Any], herald_cfg: Dict[str, Any], sentry_cfg: Dict[str, Any], packs_cfg: Dict[str, Any], constellation_cfg: Dict[str, Any], logging_cfg: Dict[str, Any])¶

Blockingly create and run the Constellation.

This should be used as the target of a multiprocessing.Process.

running = None¶: Is the Constellation server currently running?

starlette = None¶: The Starlette app.

class royalnet.constellation.Star(config: Dict[str, Any], constellation: Constellation)¶

A Star is a class representing a part of the website.

It shouldn’t be used directly: please use PageStar and ExceptionStar instead!

property Session¶: A shortcut for the Alchemy Session of the Constellation.

property alchemy¶: A shortcut for the Alchemy of the Constellation.

async page(request: starlette.requests.Request) → starlette.responses.Response¶

The function generating the Response to a web Request.

If it raises an error, the corresponding ExceptionStar will be used to handle the request instead.

property session_acm¶: A shortcut for alchemy.session_acm() of the Constellation.

class royalnet.constellation.PageStar(config: Dict[str, Any], constellation: Constellation)¶

A PageStar is a class representing a single route of the website (for example, /api/user/get).

To create a new website route you should create a new class inheriting from this class with a function overriding page() and changing the values of path and optionally methods.

methods = ['GET']¶

The HTTP methods supported by the Star, in form of a list.

By default, a Star only supports the GET method, but more can be added.

Example

methods: List[str] = ["GET", "POST", "PUT", "DELETE"]

path = NotImplemented¶

The route of the star.

Example

path: str = '/api/user/get'

class royalnet.constellation.ExceptionStar(config: Dict[str, Any], constellation: Constellation)¶

An ExceptionStar is a class that handles an Exception raised by another star by returning a different response than the one originally intended.

The handled exception type is specified in the error.

It can also handle standard webserver errors, such as 404 Not Found: to handle them, set error to an int of the corresponding error code.

To create a new exception handler you should create a new class inheriting from this class with a function overriding page() and changing the value of error.

error = None¶

The error that should be handled by this star. It should be either a subclass of Exception, or the int of an HTTP error code.

Examples

error: int = 404

error: Type[Exception] = ValueError

royalnet.constellation.shoot(code: int, description: str) → starlette.responses.JSONResponse¶: Create a error JSONResponse with the passed error code and description.

Herald¶

class royalnet.herald.Config(name: str, address: str, port: int, secret: str, secure: bool = False, path: str = '/')¶

copy(name: Optional[str] = None, address: Optional[str] = None, port: Optional[int] = None, secret: Optional[str] = None, secure: Optional[bool] = None, path: Optional[str] = None)¶: Create an exact copy of this configuration, but with different parameters.

classmethod from_config(*, name: str, address: str, port: int, secret: str, secure: bool = False, path: str = '/', enabled: ... = Ellipsis)¶

property url¶

exception royalnet.herald.HeraldError¶: A generic royalnet.herald error.

exception royalnet.herald.ConnectionClosedError¶: The Link’s connection was closed unexpectedly. The link can’t be used anymore.

exception royalnet.herald.LinkError¶: An error for something that happened in a Link.

exception royalnet.herald.InvalidServerResponseError¶: The Server sent invalid data to the Link.

exception royalnet.herald.ServerError¶: An error for something that happened in a Server.

class royalnet.herald.Link(config: royalnet.herald.config.Config, request_handler, *, loop: asyncio.events.AbstractEventLoop = None)¶

async broadcast(destination: str, broadcast: royalnet.herald.broadcast.Broadcast) → None¶

async connect()¶: Connect to the Server at config.url.

async identify() → None¶

async receive() → royalnet.herald.package.Package¶

Recieve a Package from the Server.

Raises: ConnectionClosedError –

async request(destination: str, request: royalnet.herald.request.Request) → royalnet.herald.response.Response¶

async run()¶: Blockingly run the Link.

async send(package: royalnet.herald.package.Package)¶: Send a package to the Server.

class royalnet.herald.Package(data: dict, *, source: str, destination: str, source_conv_id: Optional[str] = None, destination_conv_id: Optional[str] = None)¶

A data type with which a Link communicates with a Server or another Link.

Contains info about the source and the destination.

__init__(data: dict, *, source: str, destination: str, source_conv_id: Optional[str] = None, destination_conv_id: Optional[str] = None)¶

Create a Package.

Parameters

data – The data that should be sent.
source – The nid of the node that created this Package.
destination – The link_type of the destination node, or alternatively, the nid of the node. Can also be the NULL value to send the message to nobody.
source_conv_id – The conversation id of the node that created this package. Akin to the sequence number on IP packets.
destination_conv_id – The conversation id of the node that this Package is a reply to.

static from_dict(d) → royalnet.herald.package.Package¶: Create a Package from a dictionary.

static from_json_bytes(b: bytes) → royalnet.herald.package.Package¶: Create a Package from UTF-8-encoded JSON bytes.

static from_json_string(string: str) → royalnet.herald.package.Package¶: Create a Package from a JSON string.

reply(data) → royalnet.herald.package.Package¶

Reply to this Package with another Package.

Parameters: data – The data that should be sent. Usually a Request.
Returns: The reply Package.

to_dict() → dict¶: Convert the Package into a dictionary.

to_json_bytes() → bytes¶: Convert the Package into UTF-8-encoded JSON bytes.

to_json_string() → str¶: Convert the Package into a JSON string.

class royalnet.herald.Request(handler: str, data: dict, msg_type: Optional[str] = None)¶

A request sent from a Link to another.

It contains the name of the requested handler, in addition to the data.

classmethod from_dict(d: dict)¶

to_dict()¶

class royalnet.herald.Response¶

A base class to be inherited by all other response types.

classmethod from_dict(d: dict) → royalnet.herald.response.Response¶: Recreate the response from a received dict.

to_dict() → dict¶: Prepare the Response to be sent by converting it to a JSONable dict.

class royalnet.herald.ResponseSuccess(data: Optional[dict] = None)¶: A response to a successful Request.

class royalnet.herald.ResponseFailure(name: str, description: str, extra_info: Optional[dict] = None)¶: A response to a invalid Request.

class royalnet.herald.Server(config: royalnet.herald.config.Config, *, loop: asyncio.events.AbstractEventLoop = None)¶

find_client(*, nid: str = None, link_type: str = None) → List[royalnet.herald.server.ConnectedClient]¶

find_destination(package: royalnet.herald.package.Package) → List[royalnet.herald.server.ConnectedClient]¶

Find a list of destinations for the package.

Parameters: package – The package to find the destination of.
Returns: A list of ConnectedClient to send the package to.

async listener(websocket: websockets.server.WebSocketServerProtocol, path)¶

async route_package(package: royalnet.herald.package.Package) → None¶: Executed every time a Package is received and must be routed somewhere.

async run()¶

run_blocking(logging_cfg: Dict[str, Any])¶

serve()¶

class royalnet.herald.Broadcast(handler: str, data: dict, msg_type: Optional[str] = None)¶

classmethod from_dict(d: dict)¶

to_dict()¶

Serf¶

class royalnet.serf.Serf(loop: asyncio.events.AbstractEventLoop, alchemy_cfg: Dict[str, Any], herald_cfg: Dict[str, Any], packs_cfg: Dict[str, Any], **_)¶

An abstract class, to be used as base to implement Royalnet bots on multiple interfaces (such as Telegram or Discord).

Data = None¶: The CommandData class of this Serf.

Interface = None¶: The CommandInterface class of this Serf.

alchemy = None¶: The Alchemy object connecting this Serf to a database.

async call(command: royalnet.commands.command.Command, data: royalnet.commands.commanddata.CommandData, parameters: List[str])¶

commands = None¶: The dict connecting each command name to its Command object.

data_factory() → Type[royalnet.commands.commanddata.CommandData]¶: Create the CommandData for the Serf.

events = None¶: A dictionary containing all Event that can be handled by this Serf.

herald = None¶: The Link object connecting the Serf to the rest of the Herald network.

herald_task = None¶: A reference to the asyncio.Task that runs the Link.

property identity_chain¶: Find a relationship path starting from the master table and ending at the identity table, and return it.

identity_table = None¶: The identity table containing the interface data (such as the Telegram user data) and that is in a many-to-one relationship with the master table.

init_alchemy(alchemy_cfg: Dict[str, Any], tables: Set[type]) → None¶: Create and initialize the Alchemy with the required tables, and find the link between the master table and the identity table.

init_herald(herald_cfg: Dict[str, Any])¶: Create a Link and bind Event.

interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]¶: Create the CommandInterface class for the Serf.

interface_name = NotImplemented¶

loop = None¶: The event loop this Serf is running on.

master_table = None¶: The central table listing all users. It usually is User.

async network_handler(message: Union[royalnet.herald.request.Request, royalnet.herald.broadcast.Broadcast]) → royalnet.herald.response.Response¶

register_commands(commands: List[Type[royalnet.commands.command.Command]], pack_cfg: Dict[str, Any]) → None¶: Initialize and register all commands passed as argument.

register_events(events: List[Type[royalnet.commands.event.Event]], pack_cfg: Dict[str, Any])¶

async run()¶: A coroutine that starts the event loop and handles command calls.

classmethod run_process(**kwargs)¶

Blockingly create and run the Serf.

This should be used as the target of a multiprocessing.Process.

exception royalnet.serf.SerfError¶: Base class for all royalnet.serf errors.

Utils¶

async royalnet.utils.asyncify(function: Callable, *args, loop: Optional[asyncio.events.AbstractEventLoop] = None, **kwargs)¶: Asyncronously run the function in a different thread or process, preventing it from blocking the event loop.

Warning

If the function has side effects, it may behave strangely.

async royalnet.utils.sleep_until(dt: datetime.datetime) → None¶: Sleep until the specified datetime.

Warning

Accurate only to seconds.

royalnet.utils.andformat(l: Collection[str], middle=', ', final=' and ') → str¶

Convert a iterable (such as a list) to a str by adding final between the last two elements and middle between the others.

Parameters

l – the input iterable.
middle – the str to be added between the middle elements.
final – the str to be added between the last two elements.

Returns

The resulting str.

Examples

>>> andformat(["Steffo", "Kappa", "Proto"])
"Steffo, Kappa and Proto"

>>> andformat(["Viktya", "Sensei", "Cate"], final=" e ")
"Viktya, Sensei e Cate"

>>> andformat(["Paltri", "Spaggia", "Gesù", "Mallllco"], middle="+", final="+")
"Paltri+Spaggia+Gesù+Mallllco"

royalnet.utils.underscorize(string: str) → str¶

Replace all non-word characters in a str with underscores.

It is particularly useful when you want to use random strings from the Internet as filenames.

Parameters: string – the input string.
Returns: The resulting string.

Example

>>> underscorize("LE EPIC PRANK [GONE WRONG!?!?]")
"LE EPIC PRANK _GONE WRONG_____"

royalnet.utils.ytdldateformat(string: Optional[str], separator: str = '-') → str¶

Convert the date str returned by youtube_dl into: the YYYY-MM-DD format.

Parameters

string – the input str, in the YYYYMMDD format used by youtube_dl.
separator – the str to add between the years, the months and the days. Defaults to "-".

Returns

The resulting str in the new format.

Example

>>> ytdldateformat("20111111")
"2011-11-11"

>>> ytdldateformat("20200202", separator=".")
"2020.02.02"

royalnet.utils.numberemojiformat(l: List[str]) → str¶

Convert a list to a Unicode string with one item on every line numbered with emojis.

Parameters: l – the list to convert.
Returns: The resulting Unicode string.

Examples

Cannot be displayed, as Sphinx does not render emojis properly.

royalnet.utils.ordinalformat(number: int) → str¶

Convert a int to the corresponding English ordinal str.

Parameters: number – the number to convert.
Returns: //en.wikipedia.org/wiki/Ordinal_numeral>`_.
Return type: The corresponding English `ordinal numeral <https

Examples

>>> ordinalformat(1)
"1st"
>>> ordinalformat(2)
"2nd"
>>> ordinalformat(11)
"11th"
>>> ordinalformat(101)
"101st"
>>> ordinalformat(112)
"112th"
>>> ordinalformat(0)
"0th"

royalnet.utils.to_urluuid(uuid: uuid.UUID) → str¶: Return a base64 url-friendly short UUID.

royalnet.utils.from_urluuid(b: str) → uuid.UUID¶

class royalnet.utils.MultiLock¶

A lock that can allow both simultaneous access and exclusive access to a resource.

exclusive()¶: Acquire the lock for exclusive access.

normal()¶: Acquire the lock for simultaneous access.

class royalnet.utils.FileAudioSource(file)¶

A discord.AudioSource that uses a io.BufferedIOBase as an input instead of memory.

The stream should be in the usual PCM encoding.

Warning

This AudioSource will consume (and close) the passed stream.

__init__(file)¶

Create a FileAudioSource.

Parameters: file – the file to be played back.

is_opus()¶

This audio file isn’t Opus-encoded, but PCM-encoded.

Returns: False.

read()¶

Reads 20ms of audio.

If the stream has ended, then return an empty bytes-like object.

stop()¶: Stop the FileAudioSource. Once stopped, a FileAudioSource will immediatly stop reading more bytes from the file.

royalnet.utils.init_sentry(sentry_cfg: Dict[str, Any])¶

royalnet.utils.sentry_exc(exc: Exception, level: str = 'ERROR')¶

royalnet.utils.init_logging(logging_cfg: Dict[str, Any])¶