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 ska_tango_base.base.base_device.SKABaseDevice

A generic base device for SKA with support for long running commands.

SKABaseDevice inherits from the BaseInterface and AbstractLRCMixin, and expects a component manager to be provided by implementing the create_component_manager() method.

property component_manager: ComponentManagerT

Get the component manager.

property task_executor: TaskExecutorProtocol

Get the task executor.

Returns:

The initialised task executor.

class InitCommand

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

Warning

InitCommand is deprecated and should be set to None to acknowledge the deprecation. Override init_device() directly instead. The overridden init_device() method must call init_completed() once initialisation has finished.

Example

class MyDevice(SKABaseDevice):
    InitCommand = None

    def init_device(self) -> None:
        super().init_device()

        self.init_completed()

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": []}
   ]} ]

init_device() → None

Initialise the tango device after startup.

Subclasses overriding init_device() should set InitCommand = None and call init_completed() once initialisation has finished.

Example

class MyDevice(SKABaseDevice):
    InitCommand = None

    def init_device(self) -> None:
        super().init_device()

        self.init_completed()

init_completed() → None

Notify the state machine that initialisation has completed.

Must be called from your init_device() method when you have set InitCommand = None.

Example

class MyDevice(SKABaseDevice):
    InitCommand = None

    def init_device(self) -> None:
        super().init_device()

        self.init_completed()

on_new_shared_bus() → None

Create component manager.

delete_device() → None

Cleanup the component manager.

change_control_level(control_level: ControlLevel) → None

Stop/start communicating with the component.

_update_health_state(health_state: HealthState) → None

Update the healthState of the device and push events.

Parameters:

health_state – the new healthState value

_communication_state_changed(communication_state: CommunicationStatus) → None

Update the device about communication state to the component.

Subclasses should ensure that this is called by the communication_state_changed callback passed to their component manager.

Parameters:

communicate_state – The new communicate state

_component_state_changed(fault: bool | None = None, power: PowerState | None = None) → None

Update the device about the component’s state.

Subclasses should ensure that this is called by the component_state_callback callback passed to their component manager.

Parameters:

• fault – The new fault state of the device, or None if unchanged

• power – The new power state of the device, or None if unchanged

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.

Warning

Command objects are deprecated. Directly provide the Tango command instead, or override execute_<cmd>().

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.

Warning

Command objects are deprecated. Directly provide the Tango command instead, or override execute_<cmd>().

Parameters:

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

Returns:

the registered command object (handler) for the command

Raises:

KeyError – if the command cannot be found.

init_command_objects() → None

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

_control_mode: Signal[ska_control_model.ControlMode]

Signal for the control mode of the device.

Values are emitted for this signal whenever a client successfully changes to the controlMode attribute.

controlMode: attribute_from_signal

Control mode attribute of the device.

The control mode of the device is either REMOTE or LOCAL. Tango Device accepts only from a ‘local’ client and ignores commands and queries received from TM or any other ‘remote’ clients. The Local clients has to release LOCAL control before REMOTE clients can take control again.

read_controlMode() → ControlMode | tuple[ControlMode, float, AttrQuality]

Read the control mode of the device.

Subclasses can override this to change the behaviour of the controlMode attribute.

write_controlMode(mode: ControlMode) → None

Write the control mode of the device.

Subclasses can override this to change the behaviour of the controlMode attribute.

is_controlMode_allowed(request_type: AttReqType) → bool

Check if the controlMode can be read/written currently.

This can be overridden by subclasses to restrict when clients can access the attribute.

_simulation_mode: Signal[ska_control_model.SimulationMode]

Signal for the simulation mode of the device.

Values are emitted for this signal whenever a client successfully changes to the simulationMode attribute.

simulationMode: attribute_from_signal

Simulation mode attribute of the device.

Some devices may implement both modes, while others will have simulators that set simulationMode to True while the real devices always set simulationMode to False.

read_simulationMode() → SimulationMode | tuple[SimulationMode, float, AttrQuality]

Read the simulation mode of the device.

Subclasses can override this to change the behaviour of the simulationMode attribute.

write_simulationMode(mode: SimulationMode) → None

Write the simulation mode of the device.

Subclasses can override this to change the behaviour of the simulationMode attribute.

is_simulationMode_allowed(request_type: AttReqType) → bool

Check if the simulationMode can be read/written currently.

This can be overridden by subclasses to restrict when clients can access the attribute.

_test_mode: Signal[ska_control_model.TestMode]

Signal for the test mode of the device.

Values are emitted for this signal whenever a client successfully changes to the testMode attribute.

testMode: attribute_from_signal

Test mode attribute of the device.

Either no test mode or an indication of the test mode.

read_testMode() → TestMode | tuple[TestMode, float, AttrQuality]

Read the test mode of the device.

Subclasses can override this to change the behaviour of the testMode attribute.

write_testMode(mode: TestMode) → None

Write the test mode of the device.

Subclasses can override this to change the behaviour of the testMode attribute.

is_testMode_allowed(request_type: AttReqType) → bool

Check if the testMode can be read/written currently.

This can be overridden by subclasses to restrict when clients can access the attribute.

execute_Reset() → DevVarLongStringArrayType

Reset the device.

The default implementation of this command uses the deprecated command objects to delegate the behaviour to the “reset” method of the component manager. To avoid the deprecation warnings clients should override this method. SKA command objects deprecated provides guidelines for doing this.

See BaseInterface.execute_Reset for the expected behaviour of the :py:meth`!Reset()` command.

Returns:

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

execute_Standby() → DevVarLongStringArrayType

Put the device into standby mode.

The default implementation of this command uses the deprecated command objects to delegate the behaviour to the “standby” method of the component manager. To avoid the deprecation warnings clients should override this method. SKA command objects deprecated provides guidelines for doing this.

Returns:

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

execute_Off() → DevVarLongStringArrayType

Turn the device off.

The default implementation of this command uses the deprecated command objects to delegate the behaviour to the “off” method of the component manager. To avoid the deprecation warnings clients should override this method. SKA command objects deprecated provides guidelines for doing this.

Returns:

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

execute_On() → DevVarLongStringArrayType

Turn device on.

The default implementation of this command uses the deprecated command objects to delegate the behaviour to the “on” method of the component manager. To avoid the deprecation warnings clients should override this method. SKA command objects deprecated provides guidelines for doing this.

Returns:

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

class AbortCommand

A class for SKASubarray’s Abort() command.

__init__(command_tracker: CommandTrackerProtocol, component_manager: BaseComponentManager, callback: Callable[[bool], None] | None, logger: Logger | None = None) → None

Initialise a new AbortCommand instance.

Parameters:

• command_tracker – the device’s command tracker

• component_manager – the device’s component manager

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

• logger – a logger for this command object to use

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

Stateless hook for Abort() command functionality.

Parameters:

• args – positional arguments to the command. This command does not take any, so this should be empty.

• kwargs – keyword arguments to the command. This command does not take any, so this should be empty.

Returns:

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

execute_Abort() → DevVarLongStringArrayType

Abort any executing long running command(s) and empty the queue.

This override is provided for backwards compatible with the deprecated SKA command objects. Subclasses should override schedule_abort_task() to change the behaviour of the Abort() command. SKA command objects deprecated provides guidelines for overriding commands without using the SKA command objects.

Returns:

A tuple containing a result code and the unique ID of the command

schedule_abort_task(task_callback: TaskCallbackType) → tuple[TaskStatus, str]

Schedule an Abort task to begin executing immediately.

Subclasses should override this to change the behaviour of the Abort() command.

Parameters:

task_callback – Notified of progress of the abort command.

Returns:

A tuple containing TaskStatus.IN_PROGRESS and a message

class AbortCommandsCommand

The command class for the AbortCommand command.

__init__(component_manager: ComponentManagerT, logger: Logger | None = None) → None

Initialise a new AbortCommandsCommand instance.

Parameters:

• component_manager – contains the queue manager which manages the worker thread and the LRC attributes

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

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

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

The command class for the CheckLongRunningCommandStatus command.

__init__(command_tracker: CommandTrackerProtocol, logger: Logger | None = None) → None

Initialise a new CheckLongRunningCommandStatusCommand instance.

Parameters:

• command_tracker – command tracker

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

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

execute_CheckLongRunningCommandStatus(argin: str) → str

Check the status of a long running command by ID.

Parameters:

argin – the command id

Returns:

command status

class DebugDeviceCommand

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

__init__(device: SKABaseDevice[ComponentManagerT], logger: Logger | None = None) → None

Initialise a new instance.

Parameters:

• device – the device to which this command belongs.

• logger – a logger for this command to use.

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

class ska_tango_base.base.base_device.CommandTracker

DEPRECATED

Alias for ska_tango_base.base.command_tracker.CommandTracker. Use ska_tango_base.type_hints.CommandTrackerProtocol for type hints instead.