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: Logger | None = None, validator: ArgumentValidator | None = None)

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

do(*args: Any, **kwargs: Any) → tuple[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: 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) → 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(attr: tango.server.attribute) → None

Read the long running commands in the queue.

Keep track of which commands are that are currently known about. Entries are removed self._command_tracker._removal_time seconds after they have finished.

Parameters:

attr – Tango attribute being read

longRunningCommandIDsInQueue(attr: tango.server.attribute) → None

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 currently allocated. Entries are removed self._command_tracker._removal_time seconds after they have finished.

Parameters:

attr – Tango attribute being read

longRunningCommandStatus(attr: tango.server.attribute) → None

Read the status of the currently executing long running commands.

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

Parameters:

attr – Tango attribute being read

longRunningCommandInProgress(attr: tango.server.attribute) → None

Read the name(s) of the currently executing long running command(s).

Name(s) of command and possible abort in progress or empty string(s). :param attr: Tango attribute being read

longRunningCommandProgress(attr: tango.server.attribute) → None

Read the progress of the currently executing long running command(s).

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

Parameters:

attr – Tango attribute being read

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.

commandedState() → str

Read the last commanded operating state of the device.

Initial string is “None”. Only other strings it can change to is “OFF”, “STANDBY” or “ON”, following the start of the Off(), Standby(), On() or Reset() long running commands.

Returns:

commanded operating state string.

GetVersionInfo() → list[str]

Return the version information of the device.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

The result code and the command unique ID

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

Reset() → tuple[list[ResultCode], list[str]]

Reset the device.

To modify behaviour for this command, modify the do() method of the command class.

For a device that directly monitors and controls hardware, this command should put that hardware into a known state, for example by clearing buffers and loading default values into registers, or if necessary even by power-cycling and re-initialising the hardware.

Logical devices should generally implement this command to perform a sensible reset of that logical device. For example, aborting any current activities and clearing internal state.

Reset generally should not change the power state of the device or its hardware:

• If invoking Reset() from STANDBY state, the device would usually be expected to remain in STANDBY.

• If invoking Reset() from ON state, the device would usually be expected to remain in ON.

• If invoking Reset() from FAULT state, the device would usually be expected to transition to ON or remain in FAULT, depending on whether the reset was successful in clearing then fault.

Reset generally should not propagate to subservient devices. For example, a subsystem controller device should implement Reset to reset the subsystem as a whole, but that probably should not result in all of the subsystem’s hardware being power-cycled.

Returns:

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

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

Standby() → tuple[list[ResultCode], list[str]]

Put the device into standby mode.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

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

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

Off() → tuple[list[ResultCode], list[str]]

Turn the device off.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

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

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

On() → tuple[list[ResultCode], list[str]]

Turn device on.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

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

class AbortCommandsCommand(component_manager: ComponentManagerT, logger: Logger | None = None)

The command class for the AbortCommand command.

do(*args: Any, **kwargs: Any) → tuple[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.

AbortCommands() → tuple[list[ResultCode], list[str]]

Empty out long running commands in queue.

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: Logger | None = 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

CheckLongRunningCommandStatus(argin: str) → str

Check the status of a long running command by ID.

Parameters:

argin – the command id

Returns:

command status

class DebugDeviceCommand(device: tango.server.Device, logger: Logger | None = 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: MethodType) → 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

DebugDevice() → int

Enable remote debugging of this device.

To modify behaviour for this command, modify the do() method of the command class: DebugDeviceCommand.

Returns:

the port the debugger is listening on

set_state(state: 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, *args: Any) → 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

This is an “overloaded” function which can be called with multiple signatures supported. These are dispatched based on the types passed.

In the overloads below Scalar refers to any data type that can be converted to a tango scalar. Any refers to Scalar | Sequence[Scalar] | Sequence[Sequence[Scalar]].

• push_change_event(self, name: str)

  Push a device server change event for the “state” or “status”.

  Raises a tango.DevFailed if name is not “state” or “status”.

• push_change_event(self, name: str, expection: DevFailed)

  Push a device server change event for an attribute with an exception.

  exception: exception to send to client

• push_change_event(self, name: str, data: Any)

  Push a device server change event for an attribute.

  data: value to send to client

• push_change_event(self, name: str, data: Sequence[Scalar], dim_x: int)

  Push a device server change event for a spectrum attribute with truncation.

  A copy of data will be truncated before being sent to clients.

  Requires dim_x <= len(data).

  data: value to send to client dim_x: length to truncate value to

• push_change_event(self, name: str, data: Sequence[Scalar], dim_x: int, dim_y: int)

  Push a device server change event for a image attribute with reshaping.

  A copy of data will be reshaped to (dim_x, dim_y) before being sent to clients.

  Note that data must be flat.

  Requires dim_x * dim_y <= len(data).

  data: value to send to client dim_x: x dimension to reshape to dim_y: y dimension to reshape to

• push_change_event(self, name: str, str_data: str, data: bytes | str)

  Push a device server change event for an encoded attribute.

  str_data: encoding format for data data: encoded data to send

• push_change_event(self, name: str, data: Any, timestamp: float, quality: tango.AttrQuality)

  Push a device server change event for an attribute with timestamp and quality.

  data: value to send timestamp: unix timestamp quality: quality of attribute

• push_change_event(self, name: str, data: Sequence[Scalar], timestamp: float, quality: tango.AttrQuality, dim_x: int)

  Push a device server change event for a spectrum attribute with truncation, timestamp and quality.

  A copy of data will be truncated before being sent to clients.

  Requires dim_x <= len(data).

  data: value to send timestamp: unix timestamp quality: quality of attribute dim_x: length to truncate value to

• push_change_event(self, name: str, data: Scalar, timestamp: float, quality: tango.AttrQuality, dim_x: int, dim_y: int)

  Push a device server change event for a image attribute with reshaping, timestampe and quality.

  A copy of data will be reshaped to (dim_x, dim_y) before being sent to clients.

  Note that data must be flat.

  Requires dim_x * dim_y <= len(data).

  data: value to send timestamp: unix timestamp quality: quality of attribute dim_x: x dimension to reshape to dim_y: y dimension to reshape to

• push_change_event(self, name: str, str_data: str, data: bytes | str, timestamp: double, quality: tango.AttrQuality)

  Push a device server change event for a encoded attribute with timestamp and quality.

  str_data: encoding format for data data: encoded data to send timestamp: unix timestamp quality: quality of attribute

Parameters:

• name – the attribute name

• args – the arguments to dispatch on

push_archive_event(name: str, *args: Any) → 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.

This is an “overloaded” function which can be called with multiple signatures supported. These are dispatched based on the types passed.

In the overloads below Scalar refers to any data type that can be converted to a tango scalar. Any refers to Scalar | Sequence[Scalar] | Sequence[Sequence[Scalar]].

• push_archive_event(self, name: str)

  Push a device server archive event for the “state” or “status”.

  Raises a DevFailed if name is not “state” or “status”.

• push_archive_event(self, name: str, expection: DevFailed)

  Push a device server archive event for an attribute with an exception.

  exception: exception to send to client

• push_archive_event(self, name: str, data: Any)

  Push a device server archive event for an attribute.

  data: value to send to client

• push_archive_event(self, name: str, data: Sequence[Scalar], dim_x: int)

  Push a device server archive event for a spectrum attribute with truncation.

  A copy of data will be truncated before being sent to clients.

  Requires dim_x <= len(data).

  data: value to send to client dim_x: length to truncate value to

• push_archive_event(self, name: str, data: Sequence[Scalar], dim_x: int, dim_y: int)

  Push a device server archive event for a image attribute with reshaping.

  A copy of data will be reshaped to (dim_x, dim_y) before being sent to clients.

  Note that data must be flat.

  Requires dim_x * dim_y <= len(data).

  data: value to send to client dim_x: x dimension to reshape to dim_y: y dimension to reshape to

• push_archive_event(self, name: str, str_data: str, data: bytes | str)

  Push a device server archive event for an encoded attribute.

  str_data: encoding format for data data: encoded data to send

• push_archive_event(self, name: str, data: Any, timestamp: float, quality: tango.AttrQuality)

  Push a device server archive event for an attribute with timestamp and quality.

  data: value to send timestamp: unix timestamp quality: quality of attribute

• push_archive_event(self, name: str, data: Sequence[Scalar], timestamp: float, quality: tango.AttrQuality, dim_x: int)

  Push a device server archive event for a spectrum attribute with truncation, timestamp and quality.

  A copy of data will be truncated before being sent to clients.

  Requires dim_x <= len(data).

  data: value to send timestamp: unix timestamp quality: quality of attribute dim_x: length to truncate value to

• push_archive_event(self, name: str, data: Scalar, timestamp: float, quality: tango.AttrQuality, dim_x: int, dim_y: int)

  Push a device server archive event for a image attribute with reshaping, timestampe and quality.

  A copy of data will be reshaped to (dim_x, dim_y) before being sent to clients.

  Note that data must be flat.

  Requires dim_x * dim_y <= len(data).

  data: value to send timestamp: unix timestamp quality: quality of attribute dim_x: x dimension to reshape to dim_y: y dimension to reshape to

• push_archive_event(self, name: str, str_data: str, data: bytes | str, timestamp: double, quality: tango.AttrQuality)

  Push a device server archive event for a encoded attribute with timestamp and quality.

  str_data: encoding format for data data: encoded data to send timestamp: unix timestamp quality: quality of attribute

Parameters:

• name – the attribute name

• args – the arguments to dispatch on

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, TaskStatus]]], None], progress_changed_callback: Callable[[list[tuple[str, int]]], None], result_callback: Callable[[str, tuple[ResultCode, str]], None], exception_callback: Callable[[str, Exception], None] | 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: Callable[[], None] | 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: TaskStatus | None = None, progress: int | None = None, result: tuple[ResultCode, str] | None = None, exception: Exception | None = 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

has_current_thread_locked() → bool

Has CommandTracker locked the current thread for updating the LRC attributes.

Returns:

if current thread is locked by CommandTracker.

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, 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[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.

evict_command(command_id: str) → bool

Add to the list of commands not to be reported by the LRC attributes.

This is used to ensure we don’t overflow the attribute bounds when there are too many finished commands lingering for the removal_period.

Parameters:

command_id – the unique command id

Returns:

True if the command was not already evicted.