Commands

This module provides abstract base classes for device commands, and a ResultCode enum.

The following command classes are provided:

• FastCommand: implements the common pattern for fast commands; that is, commands that do not perform any blocking action. These commands call their callback to indicate that they have started, then execute their do hook, and then immediately call their callback to indicate that they have completed.

• DeviceInitCommand: Implements the common pattern for device Init

  commands. This is just a FastCommands, with a fixed signature for the __init__ method.

• SlowCommand: implements the common pattern for slow commands; that is, commands that need to perform a blocking action, such as file I/O, network I/O, waiting for a shared resource, etc. These commands call their callback to indicate that they have started, then execute their do hook. However they do not immediately call their callback to indicate completion. They assume that the do hook will launch work in an asynchronous context (such as a thread), and make it the responsibility of that asynchronous context to call the command’s completed method when it finishes.

• SubmittedSlowCommand: whereas SlowCommand makes no assumptions

  about how the command will be implemented, SubmittedSlowCommand assumes the current device structure: i.e. a command tracker, and a component manager with support for submitting tasks.

class ResultCode(value)

Python enumerated type for command result codes.

OK = 0

The command was executed successfully.

STARTED = 1

The command has been accepted and will start immediately.

QUEUED = 2

The command has been accepted and will be executed at a future time.

FAILED = 3

The command could not be executed.

UNKNOWN = 4

The status of the command is not known.

REJECTED = 5

The command execution has been rejected.

NOT_ALLOWED = 6

The command is not allowed to be executed.

ABORTED = 7

The command in progress has been aborted.

class FastCommand(logger: Optional[Logger] = None, validator: Optional[ArgumentValidator] = None)

An abstract class for Tango device server commands that execute quickly.

That is, they do not perform any blocking operation, so can be safely run synchronously.

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

class SlowCommand(callback: Optional[Callable[[bool], None]], logger: Optional[Logger] = None, validator: Optional[ArgumentValidator] = None)

An abstract class for Tango device server commands that execute slowly.

That is, they perform at least one blocking operation, such as file I/O, network I/O, waiting for a shared resources, etc. They therefore need to be run asynchronously in order to preserve throughput.

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

Exception – method does not exist

Returns:

result of command submission

class DeviceInitCommand(device: tango.server.Device, logger: Optional[Logger] = None, validator: Optional[ArgumentValidator] = None)

A SlowCommand with a fixed initialisation interface.

Although most commands have lots of flexibility in how they are initialised, device InitCommand instances are always called in the same way. This class fixes that interface. InitCommand instances should inherit from this command, rather than directly from SlowCommand, to ensure that their initialisation signature is correct.

class SubmittedSlowCommand(command_name: str, command_tracker: CommandTrackerProtocol, component_manager: BaseComponentManager, method_name: str, callback: Optional[Callable[[bool], None]] = None, logger: Optional[Logger] = None, validator: Optional[ArgumentValidator] = None)

A SlowCommand with lots of implementation-dependent boilerplate in it.

Whereas the SlowCommand is generic, and makes no assumptions about how the slow command will be executed, this SubmittedSlowCommand contains implementation-dependent information about the SKABaseDevice model, such as knowledge of the command tracker and component manager. It thus implements a lot of boilerplate code, and allows us to avoid implementing many identical commands.

Parameters:

• command_name – name of the command e.g. “Scan”. This is only used to ensure that the generated command id contains it.

• command_tracker – the device’s command tracker

• component_manager – the device’s component manager

• method_name – name of the component manager method to be invoked by the do hook

• callback – an optional callback to be called when this command starts and finishes.

• logger – a logger for this command to log with.

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.result_code.ResultCode, str]

Stateless hook for command functionality.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Returns:

A tuple containing the task status (e.g. QUEUED or REJECTED), and a string message containing a command_id (if the command has been accepted) or an informational message (if the command was rejected)

class CommandTrackerProtocol(*args, **kwargs)

All things to do with commands.

new_command(command_name: str, completed_callback: Optional[Callable[[], None]] = None) → str

Create a new command.

Parameters:

• command_name – the command name

• completed_callback – an optional callback for command completion

update_command_info(command_id: str, status: Optional[TaskStatus] = None, progress: Optional[int] = None, result: Optional[tuple[ska_control_model.result_code.ResultCode, str]] = None, exception: Optional[Exception] = None) → None

Update status information on the command.

Parameters:

• command_id – the unique command id

• status – the status of the asynchronous task

• progress – the progress of the asynchronous task

• result – the result of the completed asynchronous task

• exception – any exception caught in the running task

class ArgumentValidator

Base class for command argument validators.

validate(*args: Any, **kwargs: Any) → tuple[tuple[Any, ...], dict[str, Any]]

Parse and/or validate the call arguments.

This is a default implementation that performs no parsing or validating, and simply returns the arguments as received.

Subclasses may override this method to parse, validate and transform the arguments.

Parameters:

• args – positional args to the command

• kwargs – keyword args to the command

Returns:

an (args, kwargs) tuple with which to invoke the do() hook. This default implementation simply returns the arguments as received.

class JsonValidator(command_name: str, schema: dict[str, Any] | None, logger: Optional[Logger] = None)

An argument validator for JSON commands.

It checks that the argument is called with no more than one positional argument and no keyword arguments, unpacks the positional argument (if any) from JSON into a dictionary, and validates against JSON schema.

validate(*args: Any, **kwargs: Any) → tuple[tuple[Any, ...], dict[str, Any]]

Validate the command arguments.

Checks that there is only one positional argument and no keyword arguments; unpacks the positional argument from JSON into a dictionary; and validate against the provided JSON schema.

Parameters:

• args – positional args to the command

• kwargs – keyword args to the command

Returns:

validated args and kwargs