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_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(*, network_config: Optional[royalnet.network.networkconfig.NetworkConfig] = 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, secrets_name: str = '__default__')

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.

_initialize()
_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.

static safe_api_call(f: Callable, *args, **kwargs) → Optional
class royalnet.bots.DiscordBot(*, network_config: Optional[royalnet.network.networkconfig.NetworkConfig] = 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, secrets_name: str = '__default__')

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.

_initialize()
_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.GenericBot(*, network_config: Optional[royalnet.network.networkconfig.NetworkConfig] = 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, secrets_name: str = '__default__')

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() → None

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

_init_database()

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_loop()
_init_network()

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

_init_sentry()
_initialize()
_interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]
async _network_handler(request_dict: dict) → dict

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

Returns

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

get_secret(username: str)
interface_name = NotImplemented
run()

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

run_blocking(verbose=False)
set_secret(username: str, password: str)

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.NetworkLink 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)
aliases = NotImplemented

A list of possible aliases for a command. To have /e as alias for /example, one should set aliases to ["e"].

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 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 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], *flags) → 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.

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.

Display an error message to the user, along with the correct syntax for the command.

exception royalnet.commands.UnsupportedError(message='')

A requested feature is not available on this interface.

Display an error message to the user, telling them to use another interface.

exception royalnet.commands.KeyboardExpiredError(message='')

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

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

async connect()

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

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

Recieve a Package from the royalnet.network.NetworkServer.

Raises

royalnet.network.royalnetlink.ConnectionClosedError –

async request(message, destination)
async run()

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.NetworkLink is not connected to a royalnet.network.NetworkServer.

exception royalnet.network.NotIdentifiedError

The royalnet.network.NetworkLink has not identified yet to a royalnet.network.NetworkServer.

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.NetworkLink communicates with a royalnet.network.NetworkServer 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.NetworkServer(address: str, port: int, required_secret: str, *, loop: asyncio.events.AbstractEventLoop = None)
find_client(*, nid: str = None, link_type: str = None) → List[royalnet.network.networkserver.ConnectedClient]
find_destination(package: royalnet.network.package.Package) → List[royalnet.network.networkserver.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.

run_blocking(verbose=False)
serve()
class royalnet.network.NetworkConfig(master_uri: str, master_secret: str)
exception royalnet.network.ConnectionClosedError

The royalnet.network.NetworkLink’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.NetworkLink 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.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.

property args
exception royalnet.error.RoyalnetResponseError

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