API Reference

These pages were automatically generated from docstrings in code.

They might be outdated, or incomplete.

Audio

Video and audio downloading related classes, mainly used for Discord voice bots.

class royalnet.audio.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 youtube_dl.YoutubeDL.extract_info() function.

Warning

Does not download the info, for that use royalnet.audio.YtdlInfo.retrieve_for_url().

_default_ytdl_args = {'ignoreerrors': True, 'no_warnings': True, 'noplaylist': True, 'outtmpl': '%(title)s-%(id)s.%(ext)s', 'quiet': True}

classmethod retrieve_for_url(url, **ytdl_args) → List[royalnet.audio.ytdlinfo.YtdlInfo]

Fetch the info for an url through YoutubeDL.

Returns

A list containing the infos for the requested videos.

to_discord_embed() → discord.embeds.Embed

Return this info as a discord.Embed.

class royalnet.audio.YtdlFile(url: str, info: Optional[royalnet.audio.ytdlinfo.YtdlInfo] = None, filename: Optional[str] = None)

Information about a youtube-dl downloaded file.

_default_ytdl_args = {'ignoreerrors': True, 'no_warnings': True, 'noplaylist': True, 'outtmpl': '%(epoch)s-%(title)s-%(id)s.%(ext)s', 'quiet': True}

delete()

download_file(**ytdl_args) → None

classmethod download_from_url(url: str, **ytdl_args) → List[royalnet.audio.ytdlfile.YtdlFile]

has_info() → bool

is_downloaded() → bool

open()

update_info(**ytdl_args) → None

class royalnet.audio.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.

is_opus()

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

Returns

False.

read()

Reads 20ms worth of audio.

If the audio is complete, then returning an empty bytes-like object to signal this is the way to do so.

class royalnet.audio.YtdlDiscord(ytdl_file: royalnet.audio.ytdlfile.YtdlFile)

convert_to_pcm() → None

classmethod create_and_ready_from_url(url, **ytdl_args) → List[royalnet.audio.ytdldiscord.YtdlDiscord]

classmethod create_from_url(url, **ytdl_args) → List[royalnet.audio.ytdldiscord.YtdlDiscord]

delete() → None

property info

pcm_available()

ready_up()

spawn_audiosource() → royalnet.audio.fileaudiosource.FileAudioSource

class royalnet.audio.YtdlMp3(ytdl_file: royalnet.audio.ytdlfile.YtdlFile)

convert_to_mp3() → None

classmethod create_and_ready_from_url(url, **ytdl_args) → List[royalnet.audio.ytdlmp3.YtdlMp3]

classmethod create_from_url(url, **ytdl_args) → List[royalnet.audio.ytdlmp3.YtdlMp3]

delete() → None

property info

pcm_available()

ready_up()

Bots

Various bot interfaces, and a generic class to create new ones.

class royalnet.bots.TelegramBot(*, telegram_config: royalnet.bots.telegram.TelegramConfig, royalnet_config: Optional[royalnet.network.royalnetconfig.RoyalnetConfig] = None, database_config: Optional[royalnet.database.databaseconfig.DatabaseConfig] = None, sentry_dsn: Optional[str] = None, commands: List[Type[royalnet.commands.command.Command]] = None)

A bot that connects to Telegram.

_data_factory() → Type[royalnet.commands.commanddata.CommandData]

async _handle_callback_query(update: telegram.update.Update)

async _handle_message(update: telegram.update.Update)

async _handle_update(update: telegram.update.Update)

_init_client()

Create the telegram.Bot, and set the starting offset.

_interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]

interface_name = 'telegram'

async run()

A blocking coroutine that should make the bot start listening to commands and requests.

class royalnet.bots.TelegramConfig(token: str)

The specific configuration to be used for royalnet.database.TelegramBot.

class royalnet.bots.DiscordBot(*, discord_config: royalnet.bots.discord.DiscordConfig, royalnet_config: Optional[royalnet.network.royalnetconfig.RoyalnetConfig] = None, database_config: Optional[royalnet.database.databaseconfig.DatabaseConfig] = None, sentry_dsn: Optional[str] = None, commands: List[Type[royalnet.commands.command.Command]] = None)

A bot that connects to Discord.

_bot_factory() → Type[discord.client.Client]

Create a custom DiscordClient class inheriting from discord.Client.

_data_factory() → Type[royalnet.commands.commanddata.CommandData]

_init_client()

Create an instance of the DiscordClient class created in royalnet.bots.DiscordBot._bot_factory().

_init_voice()

Initialize the variables needed for the connection to voice chat.

_interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]

async add_to_music_data(dfiles: List[royalnet.audio.ytdldiscord.YtdlDiscord], guild: discord.guild.Guild)

Add a list of royalnet.audio.YtdlDiscord to the corresponding music_data object.

async advance_music_data(guild: discord.guild.Guild)

Try to play the next song, while it exists. Otherwise, just return.

interface_name = 'discord'

async run()

Login to Discord, then run the bot.

async update_activity_with_source_title()

Change the bot’s presence (using discord.Client.change_presence()) to match the current listening status.

If multiple guilds are using the bot, the bot will always have an empty presence.

class royalnet.bots.DiscordConfig(token: str)

The specific configuration to be used for royalnet.bots.DiscordBot.

class royalnet.bots.GenericBot(*, royalnet_config: Optional[royalnet.network.royalnetconfig.RoyalnetConfig] = None, database_config: Optional[royalnet.database.databaseconfig.DatabaseConfig] = None, commands: List[Type[royalnet.commands.command.Command]] = None, sentry_dsn: Optional[str] = None, loop: asyncio.events.AbstractEventLoop = None)

A generic bot class, to be used as base for the other more specific classes, such as royalnet.bots.TelegramBot and royalnet.bots.DiscordBot.

_data_factory() → Type[royalnet.commands.commanddata.CommandData]

_init_commands(commands: List[Type[royalnet.commands.command.Command]]) → None

Generate the commands dictionary required to handle incoming messages, and the network_handlers dictionary required to handle incoming requests.

_init_database(commands: List[Type[royalnet.commands.command.Command]], database_config: royalnet.database.databaseconfig.DatabaseConfig)

Create an royalnet.database.Alchemy with the tables required by the commands. Then, find the chain that links the master_table to the identity_table.

_init_royalnet(royalnet_config: royalnet.network.royalnetconfig.RoyalnetConfig)

Create a royalnet.network.RoyalnetLink, and run it as a asyncio.Task.

_interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]

async _network_handler(request_dict: dict) → dict

Handle a single dict received from the royalnet.network.RoyalnetLink.

Returns

Another dict, formatted as a royalnet.network.Response.

interface_name = NotImplemented

async run()

A blocking coroutine that should make the bot start listening to commands and requests.

Commands

class royalnet.commands.CommandInterface

alchemy = NotImplemented

bot = NotImplemented

loop = NotImplemented

name = NotImplemented

async net_request(message, destination: str) → dict

Send data through a royalnet.network.RoyalnetLink and wait for a royalnet.network.Reply.

Parameters

• message – The data to be sent. Must be pickle-able.

• destination – The destination of the request, either in UUID format or node name.

prefix = NotImplemented

register_keyboard_key(key_name: str, callback: Callable)

register_net_handler(message_type: str, network_handler: Callable)

Register a new handler for messages received through Royalnet.

unregister_keyboard_key(key_name: str)

unregister_net_handler(message_type: str)

Remove a Royalnet handler.

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

description = NotImplemented

A small description of the command, to be displayed when the command is being autocompleted.

name = NotImplemented

The main name of the command. To have /example on Telegram, the name should be example.

require_alchemy_tables = {}

A set of royalnet.database tables that must exist for this command to work.

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

syntax = ''

The syntax of the command, to be displayed when a royalnet.error.InvalidInputError is raised, in the format (required_arg) [optional_arg].

class royalnet.commands.CommandData

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 a royalnet.error.UnregisteredError if this is True and the call has no author.

Raises

royalnet.error.UnregisteredError –

async keyboard(text: str, keyboard: Dict[str, Callable]) → None

Send a keyboard having the keys of the dict as keys and calling the correspondent values on a press.

The function should be passed the CommandData instance as a argument.

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.

class royalnet.commands.CommandArgs

An interface to access the arguments of a command with ease.

__getitem__(item)

Arguments can be accessed with an array notation, such as args[0].

Raises

royalnet.error.InvalidInputError – if the requested argument does not exist.

joined(*, require_at_least=0) → str

Get the arguments as a space-joined string.

Parameters

require_at_least – the minimum amount of arguments required, will raise royalnet.error.InvalidInputError if the requirement is not fullfilled.

Raises

royalnet.error.InvalidInputError – if there are less than require_at_least arguments.

Returns

The space-joined string.

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

Match the royalnet.utils.commandargs.joined() to a regex pattern.

Parameters

pattern – The regex pattern to be passed to re.match().

Raises

royalnet.error.InvalidInputError – if the pattern doesn’t match.

Returns

The matched groups, as returned by re.Match.groups().

optional(index: int, default=None)

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.

Database

Relational database classes and methods.

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

A wrapper around SQLAlchemy declarative that allows to use multiple databases at once while maintaining a single table-class for both of them.

__init__(database_uri: str, tables: Set)

Create a new Alchemy object.

Parameters

• database_uri – The uri of the database, as described at https://docs.sqlalchemy.org/en/13/core/engines.html .

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

_create_tables(tables: Set)

session_acm()

Use Alchemy as a asyncronous context manager (to be used in async with statements).

session_cm()

Use Alchemy as a context manager (to be used in with statements).

royalnet.database.relationshiplinkchain(starting_class, ending_class) → Optional[tuple]

Find the path to follow to get from the starting table to the ending table.

class royalnet.database.DatabaseConfig(database_uri: str, master_table: Type, identity_table: Type, identity_column_name: str)

The configuration to be used for the royalnet.database.Alchemy component of royalnet.bots.GenericBot.

Network

Royalnet (websocket) related classes.

class royalnet.network.RoyalnetLink(master_uri: str, secret: str, link_type: str, request_handler, *, loop: asyncio.events.AbstractEventLoop = None)

async connect()

Connect to the royalnet.network.RoyalnetServer at self.master_uri.

async identify() → None

async receive() → royalnet.network.package.Package

Recieve a Package from the royalnet.network.RoyalnetServer.

Raises

royalnet.network.royalnetlink.ConnectionClosedError –

async request(message, destination)

async run(loops: numbers.Real = inf)

Blockingly run the Link.

async send(package: royalnet.network.package.Package)

exception royalnet.network.NetworkError(error_data: dict, *args)

exception royalnet.network.NotConnectedError

The royalnet.network.RoyalnetLink is not connected to a royalnet.network.RoyalnetServer.

exception royalnet.network.NotIdentifiedError

The royalnet.network.RoyalnetLink has not identified yet to a royalnet.network.RoyalnetServer.

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

A Royalnet package, the data type with which a royalnet.network.RoyalnetLink communicates with a royalnet.network.RoyalnetServer 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. Usually a royalnet.network.Message.

• 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.network.package.Package

Create a Package from a dictionary.

static from_json_bytes(b: bytes) → royalnet.network.package.Package

Create a Package from UTF8-encoded JSON bytes.

static from_json_string(string: str) → royalnet.network.package.Package

Create a Package from a JSON string.

reply(data) → royalnet.network.package.Package

Reply to this Package with another Package.

Parameters

data – The data that should be sent. Usually a royalnet.network.Message.

Returns

The reply Package.

to_dict() → dict

Convert the Package into a dictionary.

to_json_bytes() → bytes

Convert the Package into UTF8-encoded JSON bytes.

to_json_string() → str

Convert the Package into a JSON string.

class royalnet.network.RoyalnetServer(address: str, port: int, required_secret: str, *, loop: asyncio.events.AbstractEventLoop = None)

find_client(*, nid: str = None, link_type: str = None) → List[royalnet.network.royalnetserver.ConnectedClient]

find_destination(package: royalnet.network.package.Package) → List[royalnet.network.royalnetserver.ConnectedClient]

Find a list of destinations for the package.

Parameters

package – The package to find the destination of.

Returns

A list of ConnectedClients to send the package to.

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

async route_package(package: royalnet.network.package.Package) → None

Executed every time a package is received and must be routed somewhere.

async serve()

async start()

class royalnet.network.RoyalnetConfig(master_uri: str, master_secret: str)

exception royalnet.network.ConnectionClosedError

The royalnet.network.RoyalnetLink’s connection was closed unexpectedly. The link can’t be used anymore.

class royalnet.network.Request(handler: str, data: dict)

A request sent from a royalnet.network.RoyalnetLink to another.

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

static from_dict(d: dict)

to_dict()

class royalnet.network.Response

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

classmethod from_dict(d: dict) → royalnet.network.response.Response

Recreate the response from a received dict.

raise_on_error()

Raise an Exception if the Response is an error, do nothing otherwise.

to_dict() → dict

Prepare the Response to be sent by converting it to a JSONable dict.

class royalnet.network.ResponseSuccess(data: Optional[dict] = None)

A response to a successful royalnet.network.Request.

raise_on_error()

Raise an Exception if the Response is an error, do nothing otherwise.

class royalnet.network.ResponseError(name: str, description: str, extra_info: Optional[dict] = None)

A response to a invalid royalnet.network.Request.

raise_on_error()

Raise an Exception if the Response is an error, do nothing otherwise.

Utils

Miscellaneous useful functions and classes.

async royalnet.utils.asyncify(function: Callable, *args, **kwargs)

Convert a function into a coroutine.

Warning

The coroutine cannot be cancelled, and any attempts to do so will result in unexpected outputs.

royalnet.utils.safeformat(string: str, **words: str) → str

str.format() something, but ignore missing keys instead of raising an error.

Parameters

• string – The base string to be formatted.

• words – The words to format the string with.

Returns

The formatted string.

royalnet.utils.cdj(class_: Any) → dict

Return a dict of the class attributes without the __module__, __dict__, __weakref__ and __doc__ keys, to be used while generating dynamically SQLAlchemy declarative table classes.

Parameters

class_ – The object that you want to dict-ify.

Returns

The class dict.

Warning

You can’t dict-ify classes with __slots__!

async royalnet.utils.sleep_until(dt: datetime.datetime) → None

Block the call until the specified datetime.

Warning

Accurate only to seconds.

royalnet.utils.plusformat(i: int) → str

Convert an int to a str, prepending a + if it’s greater than 0.

Parameters

i – the int to convert.

Returns

The resulting str.

class royalnet.utils.NetworkHandler

The NetworkHandler functions are called when a specific Message type is received.

message_type = NotImplemented

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

Convert a list to a str by adding final between the last two elements and middle between the others.

Parameters

• l – the input list.

• 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.

royalnet.utils.plusformat(i: int) → str

Convert an int to a str, prepending a + if it’s greater than 0.

Parameters

i – the int to convert.

Returns

The resulting str.

royalnet.utils.fileformat(string: str) → str

Ensure a string can be used as a filename by replacing all non-word characters with underscores.

Parameters

string – the input string.

Returns

A valid filename string.

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

Convert the weird date string returned by youtube-dl into the YYYY-MM-DD format.

Parameters

• string – the input string, in the YYYYMMDD format.

• separator – the string to add between the years, the months and the days. Defaults to -.

Returns

The resulting string, in the format YYYY-MM-DD format.

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

royalnet.utils.telegram_escape(string: str) → str

Escape a string to be sent through Telegram, and format it using RoyalCode.

Warning

Currently escapes everything, even items in code blocks.

royalnet.utils.discord_escape(string: str) → str

Escape a string to be sent through Discord, and format it using RoyalCode.

Warning

Currently escapes everything, even items in code blocks.

royalnet.utils.splitstring(s: str, max: int) → List[str]

royalnet.utils.parse_5etools_entry(entry) → str

royalnet.utils.ordinalformat(number: int)

Web

royalnet.web.create_app(config_obj: Type, blueprints: List[royalnet.web.royalprint.Royalprint])

Create a flask.Flask application object.

Gets the app.secret_key from the SECRET_KEY envvar.

Also requires a DB_PATH key in config_obj to initialize the database connection.

Warning

The code for this class was written at 1 AM, and I have no clue of how and why it works or even if it really does work. Use with caution?

Parameters

• config_obj – The object to be passed to flask.Flask.config.from_object().

• blueprints – A list of blueprints to be registered to the application.

Returns

The created flask.Flask.

class royalnet.web.Royalprint(name, import_name, static_folder=None, static_url_path=None, template_folder=None, url_prefix=None, subdomain=None, url_defaults=None, root_path=None, required_tables: Optional[set] = None)

An edited flask.Blueprint containing an additional required_tables parameter.

Error

exception royalnet.error.CurrentlyDisabledError

This feature is temporarely disabled and is not available right now.

exception royalnet.error.ExternalError

Something went wrong in a non-Royalnet component and the command execution cannot be completed.

exception royalnet.error.FileTooBigError

The file to be downloaded would be too big to store; therefore, it has been skipped.

exception royalnet.error.InvalidConfigError

The bot has not been configured correctly, therefore the command can not function.

exception royalnet.error.InvalidInputError

The command has received invalid input and cannot complete.

exception royalnet.error.KeyboardExpiredError

A special type of exception that can be raised in keyboard handlers to mark a specific keyboard as expired.

exception royalnet.error.NoneFoundError

The element that was being looked for was not found.

exception royalnet.error.RoyalnetRequestError(error: ResponseError)

An error was raised while handling the Royalnet request.

This exception contains the royalnet.network.ResponseError that was returned by the other Link.

exception royalnet.error.RoyalnetResponseError

The royalnet.network.Response that was received is invalid.

exception royalnet.error.TooManyFoundError

Multiple elements matching the request were found, and only one was expected.

exception royalnet.error.UnregisteredError

The command required a registered user, and the user was not registered.

exception royalnet.error.UnsupportedError

The command is not supported for the specified interface.