Base Device

This module implements a generic base model and device for SKA.

It exposes the generic attributes, properties and commands of an SKA device.

class SKABaseDevice(*args: Any, **kwargs: Any)

A generic base device for SKA.

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

A class for the SKABaseDevice’s init_device() “command”.

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

Stateless hook for device initialisation.

Parameters:

• args – positional arguments to this do method

• kwargs – keyword arguments to this do method

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

SkaLevel

Device property.

Indication of importance of the device in the SKA hierarchy to support drill-down navigation: 1..6, with 1 highest.

GroupDefinitions

Device property.

Each string in the list is a JSON serialised dict defining the group_name, devices and subgroups in the group. A Tango Group object is created for each item in the list, according to the hierarchy defined. This provides easy access to the managed devices in bulk, or individually.

The general format of the list is as follows, with optional devices and subgroups keys:

[ {"group_name": "<name>",
   "devices": ["<dev name>", ...]},
  {"group_name": "<name>",
   "devices": ["<dev name>", "<dev name>", ...],
   "subgroups" : [{<nested group>},
                    {<nested group>}, ...]},
  ...
  ]

For example, a hierarchy of racks, servers and switches:

[ {"group_name": "servers",
   "devices": ["elt/server/1", "elt/server/2",
                 "elt/server/3", "elt/server/4"]},
  {"group_name": "switches",
   "devices": ["elt/switch/A", "elt/switch/B"]},
  {"group_name": "pdus",
   "devices": ["elt/pdu/rackA", "elt/pdu/rackB"]},
  {"group_name": "racks",
   "subgroups": [
        {"group_name": "rackA",
         "devices": ["elt/server/1", "elt/server/2",
                       "elt/switch/A", "elt/pdu/rackA"]},
        {"group_name": "rackB",
         "devices": ["elt/server/3", "elt/server/4",
                       "elt/switch/B", "elt/pdu/rackB"],
         "subgroups": []}
   ]} ]

LoggingLevelDefault

Device property.

Default logging level at device startup. See LoggingLevel

LoggingTargetsDefault

Device property.

Default logging targets at device startup. See the project readme for details.

init_device() → None

Initialise the tango device after startup.

Subclasses that have no need to override the default implementation of state management may leave init_device() alone. Override the do() method on the nested class InitCommand instead.

set_logging_level(value: LoggingLevel) → None

Set the logging level for the device.

Both the Python logger and the Tango logger are updated.

Parameters:

value – Logging level for logger

Raises:

LoggingLevelError – for invalid value

set_logging_targets(targets: list[str]) → None

Set the additional logging targets for the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults.

Parameters:

targets – Logging targets for logger

create_component_manager() → ComponentManagerT

Create and return a component manager for this device.

Raises:

NotImplementedError – for no implementation

register_command_object(command_name: str, command_object: Union[FastCommand[Any], SlowCommand[Any]]) → None

Register an object as a handler for a command.

Parameters:

• command_name – name of the command for which the object is being registered

• command_object – the object that will handle invocations of the given command

get_command_object(command_name: str) → Union[FastCommand[Any], SlowCommand[Any]]

Return the command object (handler) for a given command.

Parameters:

command_name – name of the command for which a command object (handler) is sought

Returns:

the registered command object (handler) for the command

init_command_objects() → None

Register command objects (handlers) for this device’s commands.

buildState() → str

Read the Build State of the device.

Returns:

the build state of the device

versionId() → str

Read the Version Id of the device.

Returns:

the version id of the device

loggingLevel(value: LoggingLevel) → None

Set the logging level for the device.

Both the Python logger and the Tango logger are updated.

Parameters:

value – Logging level for logger

loggingTargets(value: list[str]) → None

Set the additional logging targets for the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults.

Parameters:

value – Logging targets for logger

healthState() → HealthState

Read the Health State of the device.

It interprets the current device condition and condition of all managed devices to set this. Most possibly an aggregate attribute.

Returns:

Health State of the device

adminMode(value: AdminMode) → None

Set the Admin Mode of the device.

Parameters:

value – Admin Mode of the device.

Raises:

ValueError – for unknown adminMode

controlMode(value: ControlMode) → None

Set the Control Mode of the device.

Parameters:

value – Control mode value

simulationMode(value: SimulationMode) → None

Set the Simulation Mode of the device.

Parameters:

value – SimulationMode

testMode(value: TestMode) → None

Set the Test Mode of the device.

Parameters:

value – Test Mode

longRunningCommandsInQueue() → list[str]

Read the long running commands in the queue.

Keep track of which commands are in the queue. Pop off from front as they complete.

Returns:

tasks in the queue

longRunningCommandIDsInQueue() → list[str]

Read the IDs of the long running commands in the queue.

Every client that executes a command will receive a command ID as response. Keep track of IDs in the queue. Pop off from front as they complete.

Returns:

unique ids for the enqueued commands

longRunningCommandStatus() → list[str]

Read the status of the currently executing long running commands.

ID, status pair of the currently executing command. Clients can subscribe to on_change event and wait for the ID they are interested in.

Returns:

ID, status pairs of the currently executing commands

longRunningCommandProgress() → list[str]

Read the progress of the currently executing long running command.

ID, progress of the currently executing command. Clients can subscribe to on_change event and wait for the ID they are interested in.

Returns:

ID, progress of the currently executing command.

longRunningCommandResult() → tuple[str, str]

Read the result of the completed long running command.

Reports unique_id, json-encoded result. Clients can subscribe to on_change event and wait for the ID they are interested in.

Returns:

ID, result.

is_Reset_allowed() → bool

Return whether the Reset command may be called in the current device state.

Returns:

whether the command may be called in the current device state

is_Standby_allowed() → bool

Return whether the Standby command may be called in the current device state.

Returns:

whether the command may be called in the current device state

is_Off_allowed() → bool

Return whether the Off command may be called in the current device state.

Returns:

whether the command may be called in the current device state

is_On_allowed() → bool

Return whether the On command may be called in the current device state.

Returns:

whether the command may be called in the current device state

class AbortCommandsCommand(component_manager: ComponentManagerT, logger: Optional[Logger] = None)

The command class for the AbortCommand command.

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

Abort long running commands.

Abort the currently executing LRC and remove all enqueued LRCs.

Parameters:

• args – positional arguments to this do method

• kwargs – keyword arguments to this do method

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

class CheckLongRunningCommandStatusCommand(command_tracker: CommandTracker, logger: Optional[Logger] = None)

The command class for the CheckLongRunningCommandStatus command.

do(*args: Any, **kwargs: Any) → str

Determine the status of the command ID passed in, if any.

• Check command_result to see if it’s finished.

• Check command_status to see if it’s in progress

• Check command_ids_in_queue to see if it’s queued

Parameters:

• args – positional arguments to this do method. There should be only one: the command_id.

• kwargs – keyword arguments to this do method

Returns:

The string of the TaskStatus

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

A class for the SKABaseDevice’s DebugDevice() command.

do(*args: Any, **kwargs: Any) → int

Stateless hook for device DebugDevice() command.

Starts the debugpy debugger listening for remote connections (via Debugger Adaptor Protocol), and patches all methods so that they can be debugged.

If the debugger is already listening, additional execution of this command will trigger a breakpoint.

Parameters:

• args – positional arguments to this do method

• kwargs – keyword arguments to this do method

Returns:

The TCP port the debugger is listening on.

start_debugger_and_get_port(port: int) → int

Start the debugger and return the allocated port.

Parameters:

port – port to listen on

Returns:

allocated port

monkey_patch_all_methods_for_debugger() → None

Monkeypatch methods that need to be patched for the debugger.

get_all_methods() → list[tuple[object, str, Any]]

Return a list of the device’s methods.

Returns:

list of device methods

static method_must_be_patched_for_debugger(owner: object, method: method) → bool

Determine if methods are worth debugging.

The goal is to find all the user’s Python methods, but not the lower level PyTango device and Boost extension methods. The types.FunctionType check excludes the Boost methods.

Parameters:

• owner – owner

• method – the name

Returns:

True if the method contains more than the skipped modules.

patch_method_for_debugger(owner: object, name: str, method: Callable[[...], Any]) → None

Ensure method calls trigger the debugger.

Most methods in a device are executed by calls from threads spawned by the cppTango layer. These threads are not known to Python, so we have to explicitly inform the debugger about them.

Parameters:

• owner – owner

• name – the name

• method – method

set_state(state: tango.DevState) → None

Set the device server state.

This is dependent on whether the set state call has been actioned from a native python thread or a tango omni thread

Parameters:

state – the new device state

set_status(status: str) → None

Set the device server status string.

This is dependent on whether the set status call has been actioned from a native python thread or a tango omni thread

Parameters:

status – the new device status

push_change_event(name: str, value: Optional[Any] = None) → None

Push a device server change event.

This is dependent on whether the push_change_event call has been actioned from a native python thread or a tango omni thread

Parameters:

• name – the event name

• value – the event value

push_archive_event(name: str, value: Optional[Any] = None) → None

Push a device server archive event.

This is dependent on whether the push_archive_event call has been actioned from a native python thread or a tango omnithread.

Parameters:

• name – the event name

• value – the event value

add_attribute(*args: Any, **kwargs: Any) → None

Add a device attribute.

This is dependent on whether the push_archive_event call has been actioned from a native python thread or a tango omni thread

Parameters:

• args – positional args

• kwargs – keyword args

set_change_event(name: str, implemented: bool, detect: bool = True) → None

Set an attribute’s change event.

This is dependent on whether the push_archive_event call has been actioned from a native python thread or a tango omni thread

Parameters:

• name – name of the attribute

• implemented – whether the device pushes change events

• detect – whether the Tango layer should verify the change event property

ExecutePendingOperations() → None

Execute any Tango operations that have been pushed on the queue.

The poll time is initially 5ms, to circumvent the problem of device initialisation, but is reset to 100ms after the first pass.

main(*args: str, **kwargs: str) → int

Entry point for module.

Parameters:

• args – positional arguments

• kwargs – named arguments

Returns:

exit code

class CommandTracker(queue_changed_callback: Callable[[list[tuple[str, str]]], None], status_changed_callback: Callable[[list[tuple[str, ska_control_model.task_status.TaskStatus]]], None], progress_changed_callback: Callable[[list[tuple[str, int]]], None], result_callback: Callable[[str, tuple[ska_control_model.result_code.ResultCode, str]], None], exception_callback: Optional[Callable[[str, Exception], None]] = None, removal_time: float = 10.0)

A class for keeping track of the state and progress of 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

Returns:

a unique command id

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

property commands_in_queue: list[tuple[str, str]]

Return a list of commands in the queue.

Returns:

a list of (command_id, command_name) tuples, ordered by when invoked.

property command_statuses: list[tuple[str, ska_control_model.task_status.TaskStatus]]

Return a list of command statuses for commands in the queue.

Returns:

a list of (command_id, status) tuples, ordered by when invoked.

property command_progresses: list[tuple[str, int]]

Return a list of command progresses for commands in the queue.

Returns:

a list of (command_id, progress) tuples, ordered by when invoked.

property command_result: tuple[str, tuple[ska_control_model.result_code.ResultCode, str] | None] | None

Return the result of the most recently completed command.

Returns:

a (command_id, result) tuple. If no command has completed yet, then None.

property command_exception: tuple[str, Exception] | None

Return the most recent exception, if any.

Returns:

a (command_id, exception) tuple. If no command has raised an uncaught exception, then None.

get_command_status(command_id: str) → TaskStatus

Return the current status of a running command.

Parameters:

command_id – the unique command id

Returns:

a status of the asynchronous task.