API Reference

Base classes

class clu.base.BaseClient(name, version=None, loop=None, log_dir=None, log=None, verbose=False)

Bases: object

A base client that can be used for listening or for an actor.

This class defines a new client. Clients differ from actors in that they do not receive commands or issue replies, but do send commands to other actors and listen to the keyword-value flow. All actors are also clients and any actor should subclass from BaseClient.

Normally a new instance of a client or actor is created by passing a configuration file path to from_config which defines how the client must be started.

Parameters

• name (str) – The name of the client.

• version (Optional[str]) – The version of the client.

• loop (Optional[asyncio.AbstractEventLoop]) – The event loop. If None, the current event loop will be used.

• log_dir (Optional[Union[pathlib.Path, str]]) – The directory where to store the file logs.

• log (Optional[SDSSLogger]) – A Logger instance to be used for logging instead of creating a new one.

• verbose (Union[bool, int]) – Whether to log to stdout. Can be an integer logging level.

classmethod from_config(config, *args, **kwargs)

Parses a configuration file.

Parameters

config (Union[Dict[str, Any], pathlib.Path, str]) – A configuration dictionary or the path to a YAML configuration file. If the file contains a section called 'actor' or 'client', that section will be used instead of the whole file.

send_command(actor, *args, **kwargs)

Sends a command to an actor and returns a Command instance.

Parameters

actor (str) –

setup_logger(log, log_dir, verbose=False)

Starts the file logger.

Parameters

• log (Any) –

• log_dir (Optional[Union[pathlib.Path, str]]) –

• verbose (Union[bool, int]) –

abstract async start()

Runs the client.

Parameters

self (clu.base.T) –

Return type

clu.base.T

async stop()

Shuts down all the remaining tasks.

class clu.base.BaseActor(*args, schema=None, **kwargs)

Bases: clu.base.BaseClient

An actor based on asyncio.

This class expands BaseClient with a parsing system for new commands and placeholders for methods for handling new commands and writing replies, which should be overridden by the specific actors.

Parameters

schema (SchemaType) –

load_schema(schema, is_file=True)

Loads and validates the actor schema.

Parameters

schema (Optional[Union[Dict[str, Any], pathlib.Path, str]]) –

Return type

Optional[clu.model.Model]

abstract new_command()

Handles a new command.

Must be overridden by the subclass and call parse_command with a Command object.

abstract parse_command(command)

Parses and executes a Command. Must be overridden.

Parameters

command (clu.command.BaseCommand) –

send_command()

Sends a command to another actor.

write(message_code='i', message=None, command=None, broadcast=False, validate=True, call_internal=True, **kwargs)

Writes a message to user(s).

The reply to the users will be formatted by the actor class into message specific to the communication channel. Additional keywords passed to this method will be used to complete the message (as long as they don’t overlap with named parameters). For example

actor.write('i', message={'text': 'Hi!', 'value1': 1})

and

actor.write('i', text='Hi!', value1=1)

are equivalent and

actor.write('i', message={'text': 'Hi!'}, value1=1)

is equivalent to

actor.write('i', message={'text': 'Hi!', 'value1': 1})

This method generates a Reply object that is passed to the _write_internal method in the class, which processes it and outputs the message to the users using the appropriate transport. If command is passed, the reply is added to command.replies.

Parameters

• message_code (str) – The message code (e.g., 'i' or ':').

• message (Optional[Dict[str, Any] | str]) – The keywords to be output. Must be a dictionary of pairs {keyword: value}.

• command (Optional[BaseCommand]) – The command to which we are replying. If not set, it is assumed that this is a broadcast.

• broadcast (bool) – Whether to broadcast the message to all the users or only to the commander.

• validate (bool) – Validate the reply against the actor schema. This is ignored if the actor was not started with knowledge of its own schema.

• call_internal (bool) – Whether to call the actor internal write method. Should be True but it’s sometimes useful to call write with call_internal=False when one is overriding the method and wants to control when to call the internal method.

• kwargs – Keyword arguments that will used to update the message.

Return type

Reply

class clu.base.Reply(message_code, message, command=None, broadcast=False, use_validation=False, validated=False)

Bases: object

A reply from a command or actor to be sent to the users.

Parameters

• message_code (str) –

• message (Dict[str, Any]) –

• command (Optional[BaseCommand]) –

• broadcast (bool) –

• use_validation (bool) –

• validated (bool) –

Client

class clu.client.AMQPClient(name, url=None, user='guest', password='guest', host='localhost', port=5672, virtualhost='/', ssl=False, version=None, loop=None, log_dir=None, log=None, models=[])

Bases: clu.base.BaseClient

Defines a new client based on the AMQP standard.

To start a new client first instantiate the class and then run start as a coroutine. Note that start does not block so you will need to use asyncio’s run_forever or a similar system

>>> loop = asyncio.get_event_loop()
>>> client = await AMQPClient('my_client', host='localhost').start()
>>> loop.run_forever()

Parameters

• name (str) – The name of the client.

• url (Optional[str]) – RFC3986 formatted broker address. When used, the other connection keyword arguments are ignored.

• user (str) – The user to connect to the AMQP broker. Defaults to guest.

• password (str) – The password for the user. Defaults to guest.

• host (str) – The host where the AMQP message broker runs. Defaults to localhost.

• virtualhost (str) – Virtualhost parameter. '/' by default.

• port (int) – The port on which the AMQP broker is running. Defaults to 5672.

• ssl (bool) – Whether to use TLS/SSL connection.

• version (Optional[str]) – The version of the client.

• loop (Optional[asyncio.AbstractEventLoop]) – The event loop. If None, the current event loop will be used.

• log_dir (Optional[PathLike]) – The directory where to store the logs. Defaults to $HOME/logs/<name> where <name> is the name of the actor.

• log (Optional[SDSSLogger]) – A Logger instance to be used for logging instead of creating a new one.

• parser – A click command parser that is a subclass of CluGroup. If None, the active parser will be used.

• models (List[str]) – A list of actor models whose schemas will be monitored.

async handle_reply(message)

Handles a reply received from the exchange.

Creates a new instance of AMQPReply from the message. If the reply is valid it updates any running command.

Parameters

message (aio_pika.message.IncomingMessage) – The message received.

Returns

reply – The AMQPReply object created from the message.

Return type

clu.client.AMQPReply

async run_forever()

Runs the event loop forever.

async send_command(consumer, command_string, command_id=None)

Commands another actor over its RCP queue.

Parameters

• consumer (str) – The actor we are commanding.

• command_string (str) – The command string that will be parsed by the remote actor.

• command_id (Optional[str]) – The command ID associated with this command. If empty, an unique identifier will be attached.

async start(exchange_name='sdss_exchange')

Starts the connection to the AMQP broker.

Parameters

exchange_name (str) –

async stop()

Cancels queues and closes the connection.

running_commands

External commands currently running.

Type

dict

class clu.client.AMQPReply(message, log=None)

Bases: object

Wrapper for an IncomingMessage that expands and decodes it.

Parameters

• message (apika.IncomingMessage) – The message that contains the reply.

• log (Optional[logging.Logger]) – A message logger.

Variables

• is_valid – Whether the message is valid and correctly parsed.

• body – The body of the message, as a JSON dictionary.

• info – The info dictionary.

• headers – The headers of the message, decoded if they are bytes.

• message_code – The message code.

• sender – The name of the actor that sends the reply.

• command_id – The command ID.

Actor

class clu.actor.AMQPBaseActor(*args, schema=None, **kwargs)

Bases: clu.client.AMQPClient, clu.base.BaseActor

An actor class that uses AMQP message brokering.

This class differs from LegacyActor in that it uses an AMQP messaging broker (typically RabbitMQ) to communicate with other actors in the system, instead of standard TCP sockets. Although the internals and protocols are different the entry points and behaviour for both classes should be almost identical.

This class needs to be subclassed with a command parser.

See the documentation for AMQPActor and AMQPClient for additional parameter information.

Parameters

schema (Optional[SchemaType]) – The path to the datamodel schema for the actor, in JSON Schema format. If the schema is provided all replies will be validated against it. An invalid reply will fail and not be emitted. The schema can also be set when subclassing by setting the class schema attribute.

async new_command(message, ack=True)

Handles a new command received by the actor.

Parameters

message (aio_pika.message.IncomingMessage) –

async start(**kwargs)

Starts the connection to the AMQP broker.

class clu.actor.AMQPActor(*args, schema=None, **kwargs)

Bases: clu.parsers.click.ClickParser, clu.actor.AMQPBaseActor

An AMQP actor that uses a click parser.

Parameters

schema (Optional[SchemaType]) –

class clu.actor.TCPBaseActor(name, host=None, port=None, *args, **kwargs)

Bases: clu.base.BaseActor

A TCP base actor that replies using JSON.

This implementation of BaseActor uses TCP as command/reply channel and replies to the user by sending a JSON-valid string. This makes it useful as a “device” actor that is not connected to the central message parsing system but that we still want to accept commands and reply with easily parseable messages.

Commands received by this actor must be in the format [<uid>] <command string>, where <uid> is any integer unique identifier that will be used as command_id and appended to any reply.

This is a base actor that does not include a parser. It must be subclassed with a concrete parser that overrides parse_command.

Parameters

• name (str) – The actor name.

• host (Optional[str]) – The host where the TCP server will run.

• port (Optional[int]) – The port of the TCP server.

• args – Arguments to be passed to BaseActor.

• kwargs – Arguments to be passed to BaseActor.

new_command(transport, command_str)

Handles a new command received by the actor.

Parameters

• transport (clu.actor.CustomTransportType) –

• command_str (bytes) –

new_user(transport)

Assigns userID to new client connection.

Parameters

transport (clu.actor.CustomTransportType) –

async run_forever()

Runs the actor forever, keeping the loop alive.

async start()

Starts the TCP server.

Parameters

self (clu.actor.TCPBaseActor_co) –

Return type

clu.actor.TCPBaseActor_co

async stop()

Stops the client connection and running tasks.

write(*args, **kwargs)

Writes a message to user(s) as a JSON.

A header keyword with the commander_id (i.e., the user id of the transport that sent the command), command_id, message_code, and sender is added to each message. The payload of the message is written to data. An example of a valid message is

{
    "header": {
        "command_id": 0,
        "commander_id": 1,
        "message_code": "i",
        "sender": "test_camera"
    },
    "message": {
        "camera": {
            "name": "test_camera",
            "uid": "DEV_12345"
        },
        "status": {
            "temperature": 25.0,
            "cooler": 10.0
        }
    }
}

Although the messsage is displayed here in multiple lines, it is written as a single line to the TCP clients to facilitate parsing. For a multiline output, which is more human-readable, use the multiline command.

See write for details on the allowed parameters.

server

The server to talk to this actor.

Type

TCPStreamServer

transports

Mapping of commander_id to transport

class clu.actor.JSONActor(*args, **kwargs)

Bases: clu.parsers.click.ClickParser, clu.actor.TCPBaseActor

An implementation of TCPBaseActor that uses a Click command parser.

send_command(*args, **kwargs)

Not implemented for JSONActor.

Command

class clu.command.BaseCommand(commander_id=0, command_id=0, consumer_id=0, actor=None, parent=None, status_callback=None, call_now=False, default_keyword='text', loop=None)

Bases: _asyncio.Future, clu.tools.StatusMixIn[clu.tools.CommandStatus], Generic[clu.command.Actor_co]

Base class for commands of all types (user and device).

A BaseCommand instance is a Future whose result gets set when the status is done (either successfully or not).

Parameters

• commander_id (Union[int, str]) – The ID of the commander issuing this command. Can be a string or an integer. Normally the former is used for new-style actor and the latter for legacy actors.

• command_id (Union[int, str]) – The ID associated to this command. As with the commander_id, it can be a string or an integer

• consumer_id (Union[int, str]) – The actor that is consuming this command. Normally this is our own actor but if we are commanding another actor consumer_id will be the destination actor.

• actor (Optional[Actor_co]) – The actor instance associated to this command.

• parent (Optional[BaseCommand[Actor_co]]) – Another BaseCommand object that is issuing this subcommand. Messages emitted by the command will use the parent command_id.

• status_callback (Optional[Callable[[CommandStatus], Any]]) – A function to call when the status changes.

• call_now (bool) – Whether to call status_callback when initialising the command.

• default_keyword (str) – The keyword to use when writing a message that does not specify a keyword.

• loop (Optional[asyncio.AbstractEventLoop]) – The event loop.

debug(*args, **kwargs)

Writes a debug-level message.

error(*args, **kwargs)

Writes an error-level message (does not fail the command).

fail(*args, **kwargs)

Convenience method to mark a command FAILED.

finish(*args, **kwargs)

Convenience method to mark a command DONE.

info(*args, **kwargs)

Writes an info-level message.

set_status(status, message=None, silent=False, **kwargs)

Same as status but allows to specify a message to the users.

Parameters

• status (Union[clu.tools.CommandStatus, str]) –

• message (Optional[Dict[str, Any]]) –

• silent (bool) –

Return type

clu.command.BaseCommand

warning(*args, **kwargs)

Writes a warning-level message.

write(message_code='i', message=None, broadcast=False, **kwargs)

Writes to the user(s).

Parameters

• message_code (str) – The message code (e.g., 'i' or ':').

• message (Optional[Union[Dict[str, Any], str]]) – The text to be output. If None, only the code will be written.

• broadcast (bool) –

replies: List[clu.base.Reply]

A list of replies this command has received.

Type

Reply

property status: Optional[clu.tools.CommandStatus]

Returns the status.

class clu.command.Command(command_string='', transport=None, **kwargs)

Bases: clu.command.BaseCommand[clu.command.Actor_co]

A command from a user.

Parameters

• command_string (str) – The string that defines the body of the command.

• transport (Optional[Any]) – The TCP transport associated with this command (only relevant for LegacyActor commands).

parse()

Parses the command.

Return type

clu.command.Command

body

The body of the command.

ctx

The Context running this command. Only relevant if using the built-in click-based parser.

raw_command_string

The raw command string.

class clu.command.TimedCommand(command_string, delay=1)

Bases: object

A command to be executed on a loop.

Parameters

• command_string (str) – The command string to run.

• delay (float) – How many seconds to wait between repeated calls.

done(task)

Marks the execution of a command.

async run(actor)

Run the command.

Parameters

actor (clu.base.BaseActor) –

class clu.command.TimedCommandList(actor, resolution=0.5, loop=None)

Bases: list

A list of TimedCommand objects that will be executed on a loop.

Parameters

• actor (clu.base.BaseActor) – The actor in which the commands are to be run.

• resolution – In seconds, how frequently to check if any of the TimedCommand must be executed.

• loop (Optional[asyncio.AbstractEventLoop]) –

add_command(command_string, **kwargs)

Adds a new TimedCommand.

Parameters

command_string (str) –

async poller()

The polling loop.

start()

Starts the loop.

Return type

clu.command.TimedCommandList

async stop()

Cancel the poller.

property running: bool

Returns True if the poller is running.

clu.command.parse_legacy_command(command_string)

Parses a command received by a legacy actor.

Parameters

command_string (str) – The command to parse, including an optional header.

Returns

command_id, command_body – The command ID, and the command body parsed from the command string.

Return type

Tuple[int, str]

Legacy

class clu.legacy.actor.BaseLegacyActor(name, host, port, tron_host=None, tron_port=None, models=[], version=None, loop=None, log_dir=None, log=None, verbose=False, schema=None)

Bases: clu.base.BaseActor

An actor that provides compatibility with the SDSS opscore protocol.

The TCP servers need to be started by awaiting the coroutine start. Note that start does not block so you will need to use asyncio’s run_forever or a similar system

>>> loop = asyncio.get_event_loop()
>>> my_actor = await LegacyActor('my_actor', '127.0.0.1', 9999, loop=loop)
>>> my_actor.start()
>>> loop.run_forever()

Parameters

• name (str) – The name of the actor.

• host (str) – The host where the TCP server will run.

• port (int) – The port of the TCP server.

• tron_host (Optional[str]) – The host on which Tron is running.

• tron_port (Optional[int]) – The port on which Tron is running.

• models (List[str]) – A list of strings with the actors whose models will be tracked.

• version (Optional[str]) – The version of the actor.

• loop (Optional[asyncio.AbstractEventLoop]) – The event loop. If None, the current event loop will be used.

• log_dir (Optional[PathLike]) – The directory where to store the logs. Defaults to $HOME/logs/<name> where <name> is the name of the actor.

• log (Optional[logging.Logger]) – A Logger instance to be used for logging instead of creating a new one.

• schema (Optional[PathLike]) – The path to the datamodel schema for the actor, in JSON Schema format. If the schema is provided all replies will be validated against it. An invalid reply will fail and not be emitted. The schema can also be set when subclassing by setting the class schema attribute.

• verbose (bool) –

static format_user_output(user_id, command_id, message_code, msg_str=None)

Formats a string to send to users.

Parameters

• user_id (int) –

• command_id (int) –

• message_code (str) –

• msg_str (Optional[str]) –

Return type

str

static get_user_command_id(command=None, user_id=0, command_id=0)

Returns commander_id, command_id based on user-supplied information.

Parameters

• command (Optional[clu.command.Command]) – User command; used as a default for user_id and command_id.

• user_id (Optional[int]) – If None then use command.user_id.

• command_id (Optional[int]) – If None then use command.command_id.

Returns

user_id, command_id – The commander ID and the command ID, parsed from the inputs. If they cannot be determined, returns zeros.

Return type

Tuple[int, int]

new_command(transport, command_str)

Handles a new command received by the actor.

Parameters

• transport (clu.actor.CustomTransportType) –

• command_str (bytes) –

new_user(transport)

Assigns userID to new client connection.

Parameters

transport (clu.actor.CustomTransportType) –

async run_forever()

Runs the actor forever, keeping the loop alive.

send_command(target, command_string, command_id=None)

Sends a command through the hub.

Parameters

• target (str) – The actor to command.

• command_string (str) – The command to send.

• command_id (Optional[int]) – The command id. If None, a sequentially increasing value will be used. You should not specify a command_id unless you really know what you’re doing.

show_new_user_info(user_id)

Shows information for new users. Called when a new user connects.

Parameters

user_id (int) –

show_user_info(user_id)

Shows user information including your user_id.

Parameters

user_id (int) –

show_user_list()

Shows a list of connected users. Broadcast to all users.

show_version(user_id=None)

Shows actor version.

Parameters

user_id (Optional[int]) –

async start()

Starts the server and the Tron client connection.

Parameters

self (clu.legacy.actor.T) –

Return type

clu.legacy.actor.T

async stop()

Stops the client connection and running tasks.

write(message_code='i', message=None, command=None, user_id=None, command_id=None, concatenate=True, broadcast=False, validate=True, **kwargs)

Writes a message to user(s).

Parameters

• message_code (str) – The message code (e.g., 'i' or ':').

• message (Optional[Dict[str, Any]]) – The keywords to be output. Must be a dictionary of pairs {keyword: value}. If value is a list it will be converted into a comma-separated string. To prevent unexpected casting, it is recommended for value to always be a string.

• command (Optional[clu.command.Command]) – User command; used as a default for user_id and command_id. If the command is done, it is ignored.

• user_id (Optional[int]) – The user (transport) to which to write. None defaults to 0.

• command_id (Optional[int]) – If None then use command.command_id.

• concatenate (bool) – Concatenates all the keywords to be output in a single reply with the keyword-values joined with semicolons. Otherwise each keyword will be output as a different message.

• broadcast (bool) – Whether to broadcast the reply. Equivalent to user_id=0.

• validate (bool) – Validate the reply against the actor model. This is ignored if the actor was not started with knowledge of its own schema.

• kwargs – Keyword arguments that will be added to the message. If a keyword is both in message and in kwargs, the value in kwargs supersedes message.

models

Actor models.

Type

dict

transports

Mapping of user_id to transport

tron

The client connection to Tron.

Type

TronConnection

class clu.legacy.actor.LegacyActor(name, host, port, tron_host=None, tron_port=None, models=[], version=None, loop=None, log_dir=None, log=None, verbose=False, schema=None)

Bases: clu.parsers.click.ClickParser, clu.legacy.actor.BaseLegacyActor

A legacy actor that uses the ClickParser.

Parameters

• name (str) –

• host (str) –

• port (int) –

• tron_host (Optional[str]) –

• tron_port (Optional[int]) –

• models (List[str]) –

• version (Optional[str]) –

• loop (Optional[asyncio.AbstractEventLoop]) –

• log_dir (Optional[PathLike]) –

• log (Optional[logging.Logger]) –

• verbose (bool) –

• schema (Optional[PathLike]) –

class clu.legacy.tron.TronConnection(host, port=6093, name='tron', models=[], **kwargs)

Bases: clu.base.BaseClient

Allows to send commands to Tron and manages the feed of replies.

Parameters

• name (str) – The name of the client.

• host (str) – The host on which Tron is running.

• port (int) – The port on which Tron is running.

• models (List[str]) – A list of strings with the actors whose models will be tracked.

• kwargs – Arguments to be passed to BaseClient.

async get_keys()

Gets all the keys for the models being tracked.

send_command(target, command_string, commander='tron.tron', mid=None)

Sends a command through the hub.

Parameters

• target – The actor to command.

• command_string – The command to send.

• commander – The actor or client sending the command. The format for Tron is “commander message_id target command” where commander needs to start with a letter and have a program and a user joined by a dot. Otherwise the command will be accepted but the reply will fail to parse.

• mid – The message id. If None, a sequentially increasing value will be used. You should not specify a mid unless you really know what you’re doing.

async start(get_keys=True)

Starts the connection to Tron.

Parameters

get_keys (bool) – If True, gets all the keys in the models.

stop()

Closes the connection.

keyword_dicts

The KeysDictionary associated with each actor to track.

Type

dict

models

The model and values of each actor being tracked.

Type

dict

class clu.legacy.tron.TronKey(name, key, keyword=None, model=None, callback=None)

Bases: clu.model.Property

A Tron model key with callbacks.

Similar to Property but stores the original key with the keyword datamodel.

Parameters

• name (str) –

• key (Key) –

• keyword (Optional[Keyword]) –

• model (Optional[TronModel]) –

• callback (Optional[Callable[[TronKey], Any]]) –

update_keyword(keyword)

Updates the keyword and value.

Parameters

keyword (Optional[clu.legacy.types.messages.Keyword]) –

class clu.legacy.tron.TronModel(keydict, callback=None, log=None)

Bases: clu.model.BaseModel[clu.legacy.tron.TronKey]

A JSON-compliant model for actor keywords.

Parameters

• keydict (KeysDictionary) – A dictionary of keys that define the datamodel.

• callback (Callable[[TronModel], Any]) – A function or coroutine to call when the datamodel changes. The function is called with the instance of TronModel and the modified keyword. If the callback is a coroutine, it is scheduled as a task.

• log (Optional[logging.Logger]) – Where to log messages.

parse_reply(reply)

Parses a reply and updates the datamodel.

reload()

Reloads the model. Clears callbacks.

Maskbits

class clu.tools.Maskbit(value)

Bases: enum.Flag

A maskbit enumeration. Intended for subclassing.

property active_bits: List[clu.tools.Maskbit]

Returns a list of non-combination flags that match the value.

class clu.tools.CommandStatus(value)

Bases: clu.tools.Maskbit

An enumeration.

DONE = 1

CANCELLED = 2

FAILED = 4

TIMEDOUT = 8

READY = 16

RUNNING = 32

CANCELLING = 64

FAILING = 128

DEBUG = 256

ACTIVE_STATES = 224

FAILED_STATES = 14

FAILING_STATES = 192

DONE_STATES = 15

ALL_STATES = 255

property is_combination: bool

Returns True if a flag is a combination.

property did_fail: bool

Command failed or was cancelled.

property did_succeed: bool

Command finished with DONE status.

property is_active: bool

Command is running, cancelling or failing.

property is_done: bool

Command is done (whether successfully or not).

property is_failing: bool

Command is being cancelled or is failing.

static code_to_status(code, default=None)

Returns the status associated with a code.

If the code doesn’t have an associated status, returns default. default defaults to CommandStatus.RUNNING.

Parameters

default (Optional[clu.tools.CommandStatus]) –

Return type

clu.tools.CommandStatus

MixIns

class clu.tools.StatusMixIn(maskbit_flags, initial_status=None, callback_func=None, call_now=False)

Bases: Generic[clu.tools.MaskbitType]

A mixin that provides status tracking with callbacks.

Provides a status property that executes a list of callbacks when the status changes.

Parameters

• maskbit_flags – A class containing the available statuses as a series of maskbit flags. Usually as subclass of enum.Flag.

• initial_status – The initial status.

• callback_func – The function to call if the status changes.

• call_now – Whether the callback function should be called when initialising.

Variables

callbacks – A list of the callback functions to call.

do_callbacks()

Calls functions in callbacks.

async wait_for_status(value)

Awaits until the status matches value.

property status

Returns the status.

class clu.tools.CaseInsensitiveDict(values)

Bases: Dict[str, clu.tools.T]

A dictionary that performs case-insensitive operations.

Parameters

values (Any) –

Model

class clu.model.BaseModel(name, callback=None, **kwargs)

Bases: clu.tools.CaseInsensitiveDict[clu.model.T], clu.tools.CallbackMixIn

A JSON-compliant model.

Parameters

• name (str) – The name of the model.

• callback (Optional[Callable[[Any], Any]]) – A function or coroutine to call when the datamodel changes. The function is called with the instance of BaseModel and the key that changed.

flatten()

Returns a dictionary of values.

Return a dictionary in which the Property instances are replaced with their values.

Return type

Dict[str, Any]

jsonify()

Returns a JSON string with the model.

Return type

str

class clu.model.Model(name, schema, is_file=False, **kwargs)

Bases: clu.model.BaseModel[clu.model.Property]

A model with JSON validation.

In addition to the parameters in BaseModel, the following parameters are accepted:

Parameters

• schema (SchemaType) – A valid JSON schema, to be used for validation.

• is_file (bool) – Whether the input schema is a filepath or a dictionary.

• name (str) –

static check_schema(schema)

Checks whether a JSON schema is valid.

Parameters

schema (Dict[str, Any]) – The schema to check as a dictionary.

Returns

result – Returns True if the schema is a valid JSON schema, False otherwise.

Return type

bool

update_model(instance)

Validates a new instance and updates the model.

Parameters

instance (Dict[str, Any]) –

class clu.model.ModelSet(client, actors=[], get_schema_command='get_schema', raise_exception=True, **kwargs)

Bases: dict

A dictionary of Model instances.

Given a list of actors, queries each of the actors to return their own schemas, which are then parsed and loaded as Model instances. Since obtaining the schema require sending a command to the actor, that process happens when the coroutine load_schemas is awaited, which should usually occur when the client is started.

Parameters

• client (clu.base.BaseClient) – A client with a connection to the actors to monitor.

• actors (List[str]) – A list of actor models whose schemas will be loaded.

• get_schema_command (str) – The command to send to the actor to get it to return its own schema.

• raise_exception (bool) – Whether to raise an exception if any of the models cannot be loaded.

• kwargs – Keyword arguments to be passed to Model.

Example

>>> model_set = ModelSet(client, actors=['sop', 'guider'])
>>> model_set['sop']
<Model (name='sop')>

async load_schemas(actors=None)

Loads the actor schames.

Parameters

actors (Optional[List[str]]) –

class clu.model.Property(name, value=None, model=None, callback=None)

Bases: clu.tools.CallbackMixIn

A model property with callbacks.

Parameters

• name (str) – The name of the property.

• value (Optional[Any]) – The value of the property.

• model (Optional[Any]) – The parent model.

• callback (Optional[Callable[[Any], Any]]) – The function or coroutine that will be called if the value of the key if updated. The callback is called with the instance of Property as the only argument. Note that the callback will be scheduled even if the value does not change.

flatten()

Returns a dictionary with the name and value of the property.

Return type

Dict[str, Any]

property value: Any

The value associated to the key.

Parser

class clu.parsers.click.ClickParser

Bases: object

A command parser that uses Click at its base.

parse_command(command)

Parses a user command using the Click internals.

Parameters

command (clu.parsers.click.T) –

Return type

clu.parsers.click.T

context_obj = {}

Parameters to be set in the context object.

Type

dict

parser_args: List[Any] = []

Arguments to be passed to each command in the parser. Note that the command is always passed first.

Type

list

class clu.parsers.click.CluCommand(*args, context_settings=None, **kwargs)

Bases: click.core.Command

Override click.Command to pass the actor and command.

done_callback(task, exception_handler=None)

Checks if the command task has been successfully done.

invoke(ctx)

As click.Command.invoke but passes the actor and command.

class clu.parsers.click.CluGroup(name=None, commands=None, **attrs)

Bases: click.core.Group

Override click.Group.

Makes all child commands instances of CluCommand.

command(*args, **kwargs)

Override click.Group to use CluCommand by default.

group(*args, **kwargs)

Creates a new group inheriting from this class.

parse_args(ctx, args)

Given a context and a list of arguments this creates the parser and parses the arguments, then modifies the context as necessary. This is automatically invoked by make_context().

class clu.parsers.json.JSONParser

Bases: object

A parser that receives commands and arguments as a JSON string.

See JSON parser for details on implementation and use-cases.

parse_command(command)

Parses a user command.

The command string must be a serialised JSON-like string that contains at least a keyword command with the name of the callback, and any number of additional arguments which will be passed to it.

Parameters

command (clu.parsers.json.T) –

Return type

clu.parsers.json.T

callbacks: Dict[str, Callable[[...], Coroutine]] = {}

Mapping of command verb to callback coroutine.

Type

dict

parser_args: List[Any] = []

Additional arguments to be passed to each command in the parser. Note that the command is always passed first.

Type

list

clu.parsers.click.timeout(seconds)

A decorator to timeout the command after a number of seconds.

Parameters

seconds (float) –

Sockets

class clu.protocol.TCPProtocol(loop=None, connection_callback=None, data_received_callback=None, max_connections=None)

Bases: asyncio.protocols.Protocol

A TCP server/client based on asyncio protocols.

This is a high-level implementation of the client and server asyncio protocols. See asyncio protocol for details.

Parameters

• loop (asyncio.AbstractEventLoop) – The event loop. The current event loop is used by default.

• connection_callback (Optional[ConnectionCallbackType]) – Callback to call when a new client connects.

• data_received_callback (Optional[Callable[[str], Any]]) – Callback to call when a new data is received.

• max_connections (Optional[int]) – How many clients the server accepts. If None, unlimited connections are allowed.

async classmethod create_server(host, port, **kwargs)

Returns a Server connection.

Parameters

• host (str) –

• port (int) –

async classmethod create_client(host, port, **kwargs)

Returns a Transport and Protocol.

Parameters

• host (str) –

• port (int) –

connection_made(transport)

Receives a connection and calls the connection callback.

Parameters

transport (asyncio.transports.Transport) –

data_received(data)

Decodes the received data.

Parameters

data (bytes) –

connection_lost(exc)

Called when connection is lost.

class clu.protocol.PeriodicTCPServer(periodic_callback=None, sleep_time=1, **kwargs)

Bases: clu.protocol.TCPProtocol

A TCP server that runs a callback periodically.

Parameters

• period_callback – Callback to run every iteration.

• sleep_time (float) – The delay between two calls to periodic_callback.

• kwargs – Parameters to pass to TCPProtocol

• periodic_callback (Optional[ConnectionCallbackType]) –

async classmethod create_client(*args, **kwargs)

Returns a Transport and Protocol.

async classmethod create_server(host, port, *args, **kwargs)

Returns a Server connection.

Parameters

• host (str) –

• port (int) –

property periodic_callback: Optional[Callable[[Any], Any]]

Returns the periodic callback.

class clu.protocol.TCPStreamServer(host, port, connection_callback=None, data_received_callback=None, loop=None, max_connections=None)

Bases: object

A TCP server based on asyncio streams.

This is a high-level implementation of the asyncio server using streams. See asyncio streams for details.

Parameters

• host (str) – The server host.

• port (int) – The server port.

• connection_callback (Optional[ConnectionCallbackType]) – Callback to call when a new client connects or disconnects.

• data_received_callback (Optional[DataReceivedCallbackType]) – Callback to call when a new data is received.

• loop (Optional[asyncio.AbstractEventLoop]) – The event loop. The current event loop is used by default.

• max_connections (Optional[int]) – How many clients the server accepts. If None, unlimited connections are allowed.

async start()

Starts the server and returns a Server connection.

Return type

asyncio.events.AbstractServer

stop()

Stops the server.

serve_forever()

Exposes TCPStreamServer.server.serve_forever.

async connection_made(reader, writer)

Called when a new client connects to the server.

Stores the writer protocol in transports, calls the connection callback, if any, and starts a loop to read any incoming data.

Parameters

• reader (asyncio.streams.StreamReader) –

• writer (asyncio.streams.StreamWriter) –

class clu.protocol.TCPStreamPeriodicServer(host, port, periodic_callback=None, sleep_time=1, **kwargs)

Bases: clu.protocol.TCPStreamServer

A TCP server that calls a function periodically.

Parameters

• host (str) – The server host.

• port (int) – The server port.

• period_callback – Callback to run every iteration. It is called for each transport that is connected to the server and receives the transport object.

• sleep_time (float) – The delay between two calls to periodic_callback.

• kwargs – Parameters to pass to TCPStreamServer

• periodic_callback (Optional[Callable[[asyncio.Transport], Any]]) –

async start()

Starts the server and returns a Server connection.

Return type

asyncio.events.AbstractServer

stop()

Stops the server.

property periodic_callback

Returns the periodic callback.

class clu.protocol.TCPStreamClient(host, port)

Bases: object

An object containing a writer and reader stream to a TCP server.

Parameters

• host (str) –

• port (int) –

async open_connection()

Creates the connection.

close()

Closes the stream.

async clu.protocol.open_connection(host, port)

Returns a TCP stream connection with a writer and reader.

This function is equivalent to doing

>>> client = TCPStreamClient('127.0.0.1', 5555)
>>> await client.open_connection()

Instead just do

>>> client = await TCPStreamClient('127.0.0.1', 5555)
>>> client.writer.write('Hi!\n'.encode())

Parameters

• host (str) – The host of the TCP server.

• port (int) – The port of the TCP server.

Returns

client (TCPStreamClient) – A container for the stream reader and writer.

Return type

clu.protocol.TCPStreamClient

class clu.protocol.TopicListener(url=None, user='guest', password='guest', host='localhost', virtualhost='/', port=5672, ssl=False)

Bases: object

A class to declare and listen to AMQP queues with topic conditions.

Parameters

• url (str) – RFC3986 formatted broker address. When used, the other keyword arguments are ignored.

• user (str) – The user to connect to the RabbitMQ broker.

• password (str) – The password for the user.

• host (str) – The host where the RabbitMQ message broker runs.

• virtualhost (str) – Virtualhost parameter. '/' by default.

• port (int) – The port on which the RabbitMQ message broker is running.

• ssl (bool) – Whether to use TLS/SSL connection.

async connect(exchange_name, exchange_type=<ExchangeType.TOPIC: 'topic'>, on_return_raises=True)

Initialise the connection.

Parameters

• exchange_name (str) – The name of the exchange to create.

• exchange_type (aio_pika.exchange.ExchangeType) – The type of exchange to create.

Return type

clu.protocol.TopicListener

async add_queue(queue_name, callback=None, bindings='*')

Adds a queue with bindings.

Parameters

• queue_name (str) – The name of the queue to create.

• callback (Optional[Callable[[aio_pika.message.IncomingMessage], Any]]) – A callable that will be called when a new message is received in the queue. Can be a coroutine.

• bindings (Union[str, List[str]]) – The list of bindings for the queue. Can be a list of string or a single string in which the bindings are comma-separated.

Return type

aio_pika.queue.Queue

async stop()

Cancels queues and closes the connection.

class clu.device.Device(host, port, callback=None)

Bases: clu.tools.CallbackMixIn

A class that handles the TCP connection to a device.

There are two ways to create a new device. You can create a subclass from Device and override the process_message method which handles how you react to a new line being received

class MyDevice(Device):

    async def process_message(self, line):
        print(line)

my_device = MyDevice('192.168.1.10', 4444)
await my_device.start()

Note that process_message must be a coroutine. Alternatively you can pass a callback that will be called instead of process_message when a new message arrives. The callback must also be a coroutine

async def printer(line):
    print(line)

my_device = MyDevice('192.168.1.10', 4444, callback=printer)
await my_device.start()

Parameters

• host (str) – The host of the device.

• port (int) – The port on which the device is serving.

• callback (Optional[Callable[[str], Any]]) – The callback to call with each new message received from the client (after decoding into a string). If no callback is specified, process_message is called. If the callback is not a coroutine, it will be converted to one.

is_connected()

Returns True if the connection is open.

Return type

bool

async process_message(line)

Processes a newly received message.

Parameters

line (str) –

async start()

Opens the connection and starts the listener.

async stop()

Closes the connection and stops the listener.

write(message, newline='\n')

Write to the device. The message is encoded and a new line added.

Parameters

message (str) –

Tools

clu.escape(value)

Escapes a text using json.dumps.

Parameters

value (Any) –

class clu.tools.ActorHandler(actor, level=40, keyword='text', code_mapping=None, filter_warnings=None)

Bases: logging.Handler

A handler that outputs log messages as actor keywords.

Parameters

• actor – The actor instance.

• level (int) – The level above which records will be output in the actor.

• keyword (str) – The keyword around which the messages will be output.

• code_mapping (Optional[Dict[int, str]]) – A mapping of logging levels to actor codes. The values provided override the default mapping. For example, to make input log messages with info level be output as debug, code_mapping={logging.INFO: 'd'}.

• filter_warnings (Optional[List[Type[Warning]]]) – A list of warning classes that will be issued to the actor. Subclasses of the filter warning are accepted, any other warnings will be ignored.

emit(record)

Emits the record.

Parameters

record (logging.LogRecord) –

class clu.tools.CallbackMixIn(callbacks=[], loop=None)

Bases: object

A mixin for executing callbacks.

Parameters

• callbacks (List[Callable[[Any], Any]]) – A list of functions or coroutines to be called.

• loop (Optional[asyncio.AbstractEventLoop]) –

notify(*args)

Calls the callback functions with some arguments.

Coroutine callbacks are scheduled as a task. Synchronous callbacks are scheduled with call_soon.

register_callback(callback_func)

Adds a callback function or coroutine function.

Parameters

callback_func (Callable[[...], Any]) –

remove_callback(callback_func)

Removes a callback function.

Parameters

callback_func (Callable[[...], Any]) –

async stop_callbacks()

Cancels any running callback task.

Testing

class clu.testing.MockReply(command_id, user_id, flag, data={})

Bases: dict

Stores a reply written to a transport.

The data of the message is stored as part of the dictionary.

Parameters

• user_id (int) – The user ID of the client to which the reply was sent.

• command_id (int) – The command ID of the command that produced this reply.

• flag (str) – The message type flag.

• data (Dict[str, Any]) – The payload of the message.

class clu.testing.MockReplyList(actor)

Bases: list

Stores replies as MockReply objects.

clear()

Remove all items from list.

parse_reply(reply, routing_key=None)

Parses a reply and construct a MockReply, which is appended.

Parameters

• reply (bytes | str | aio_pika.Message) –

• routing_key (Optional[str]) –

async clu.testing.setup_test_actor(actor, user_id=1)

Setups an actor for testing, mocking the client transport.

Takes an actor and modifies it in two ways:

• Adds a invoke_mock_command method to it that allows to submit a command string as if it had been received from a transport.

• Mocks a client transport with user_id that is connected to the actor. Messages written to the transport are stored as MockReply in a MockReplyList that is accessible via a new actor.mock_replies attribute.

The actor is modified in place and returned.

Parameters

• actor (clu.testing.T) –

• user_id (int) –

Return type

clu.testing.T