Deprecated Base

This base module implements functionality common to all SKA Tango devices.

Base Component Manager

This module provides an abstract component manager for SKA Tango base devices.

The basic model is:

Every Tango device has a component that it monitors and/or controls. That component could be, for example:
- Hardware such as an antenna, APIU, TPM, switch, subrack, etc.
- An external software system such as a cluster manager
- A software routine, possibly implemented within the Tango device itself
- In a hierarchical system, a pool of lower-level Tango devices.
A Tango device will usually need to establish and maintain communication with its component. This connection may be deliberately broken by the device, or it may fail.
A Tango device controls its component by issuing commands that cause the component to change behaviour and/or state; and it monitors its component by keeping track of its state.

ska_tango_base.base.base_component_manager.check_communicating(func: Wrapped) → Wrapped[source]

Return a function that checks component communication before calling a function.

Deprecated:: Implement a “is_cmd_allowed” method for the Tango command instead.

The component manager needs to have established communications with the component, in order for the function to be called.

This function is intended to be used as a decorator:

@check_communicating
def scan(self): ...

Parameters:: func – the wrapped function
Returns:: the wrapped function

ska_tango_base.base.base_component_manager.check_on(func: Wrapped) → Wrapped[source]

Return a function that checks the component state then calls another function.

Deprecated:: Implement a “is_cmd_allowed” method for the Tango command instead.

The component needs to be turned on, and not faulty, in order for the function to be called.

This function is intended to be used as a decorator:

@check_on
def scan(self): ...

Parameters:: func – the wrapped function
Returns:: the wrapped function

class ska_tango_base.base.base_component_manager.BaseComponentManager[source]

An abstract base class for a component manager for SKA Tango devices.

Deprecated:: Use the future API instead.

It supports:

Maintaining a connection to its component
Controlling its component via commands like Off(), Standby(), On(), etc.
Monitoring its component, e.g. detect that it has been turned off or on

__init__(logger: Logger, communication_state_callback: Callable[[CommunicationStatus], None] | None = None, component_state_callback: Callable[[...], None] | None = None, **state: Any) → None[source]

Initialise a new ComponentManager instance.

Parameters:

logger – the logger to be used by this manager
communication_state_callback – callback to be called when the status of communications between the component manager and its component changes.
component_state_callback – callback to be called when the monitored state of the component changes
state – key/value pairs

property max_queued_tasks: int

Get the task queue size.

This property should be overridden by subclasses if they provide a task executor.

Returns:: The task queue size

property max_executing_tasks: int

Get the max number of tasks that can be executing at once.

This property should be overridden by subclasses if they provide a task executor.

Returns:: max number of simultaneously executing tasks.

cleanup() → None[source]: Cleanup any software resources in use by the component manager.

start_communicating() → None[source]

Establish communication with the component, then start monitoring.

This is the place to do things like:

Initiate a connection to the component (if your communication is connection-oriented)
Subscribe to component events (if using “pull” model)
Start a polling loop to monitor the component (if using a “push” model)

Raises:: NotImplementedError – Not implemented it’s an abstract class

stop_communicating() → None[source]

Cease monitoring the component, and break off all communication with it.

For example,

If you are communicating over a connection, disconnect.
If you have subscribed to events, unsubscribe.
If you are running a polling loop, stop it.

Raises:: NotImplementedError – Not implemented it’s an abstract class

property communication_state: CommunicationStatus

Return the communication status of this component manager.

Returns:: status of the communication channel with the component.

_update_communication_state(communication_state: CommunicationStatus) → None[source]

Handle a change in communication status.

This is a helper method for use by subclasses.

Parameters:: communication_state – the new communication status of the component manager.

property component_state: dict[str, Any]

Return the state of this component manager’s component.

Returns:: state of the component.

_update_component_state(**kwargs: Any) → None[source]

Handle a change in component state.

This is a helper method for use by subclasses.

Parameters:: kwargs – key/values for state

off(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Turn the component off.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

standby(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Put the component into low-power standby mode.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

on(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Turn the component on.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

reset(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Reset the component (from fault state).

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

abort(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort activities on the device.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message

abort_commands(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort all tasks queued & running.

Deprecated:: Use abort_tasks() instead.
Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message

abort_tasks(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Abort all tasks queued & running.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message
Raises:: NotImplementedError – Not implemented it’s an abstract class

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[source]

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

Deprecated:: Use the future API instead.

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[source]

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][source]

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[source]

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[source]

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[source]: Create component manager.

delete_device() → None[source]: Cleanup the component manager.

change_control_level(control_level: ControlLevel) → None[source]: Stop/start communicating with the component.

_update_health_state(health_state: HealthState) → None[source]

Update the healthState of the device and push events.

Parameters:: health_state – the new healthState value

_communication_state_changed(communication_state: CommunicationStatus) → None[source]

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[source]

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[source]

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[source]

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][source]

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[source]: 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][source]

Read the control mode of the device.

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

write_controlMode(mode: ControlMode) → None[source]

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[source]

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][source]

Read the simulation mode of the device.

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

write_simulationMode(mode: SimulationMode) → None[source]

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[source]

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][source]

Read the test mode of the device.

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

write_testMode(mode: TestMode) → None[source]

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[source]

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[source]

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 :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[source]

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[source]

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[source]

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[source]

A class for SKASubarray’s Abort() command.

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

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][source]

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[source]

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][source]

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[source]

The command class for the AbortCommand command.

__init__(component_manager: ComponentManagerT, logger: Logger | None = None) → None[source]

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][source]

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[source]

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[source]

The command class for the CheckLongRunningCommandStatus command.

__init__(command_tracker: CommandTrackerProtocol, logger: Logger | None = None) → None[source]

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[source]

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[source]

Check the status of a long running command by ID.

Parameters:: argin – the command id
Returns:: command status

class DebugDeviceCommand[source]

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

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

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[source]

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[source]

Start the debugger and return the allocated port.

Parameters:: port – port to listen on
Returns:: allocated port

monkey_patch_all_methods_for_debugger() → None[source]: Monkeypatch methods that need to be patched for the debugger.

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

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

DEPRECATED

This compatibility alias is documented here. Use ska_tango_base.type_hints.CommandTrackerProtocol for type hints instead.

Base Interface

Base interfaces for SKA Tango devices.

class ska_tango_base.base.base_interface.OpStateSignal[source]

Bases: Signal[DevState]

Special signal for Operational State.

Deprecated:: Use the future API instead.

Ensures that only valid enumerants from DevState for the SKA Operational State are emitted for the signal.

__init__() → None[source]: Initialise the signal.

class ska_tango_base.base.base_interface.ControlLevel[source]

Bases: Enum

How much control the Tango device should exert on the system under control.

Deprecated:: Use the future API instead.

This enumeration should not be considered exhaustive as, in the future, we may add MONITORING_ONLY to this enumeration.

NO_CONTACT = 1: The Tango device should make no contact with the system under control.

FULL_CONTROL = 2: The Tango device should attempt to control the system under control fully.

class ska_tango_base.base.base_interface.BaseInterface[source]

Bases: SignalBusMixin, SKADevice

Provides the Tango interface for an SKA device with an Operational State.

Deprecated:: Use the future API instead.

This class only provides the Tango interface required for SKA Tango devices which support an Operational State. It is up to subclasses to override various abstract methods to provide the appropriate behaviour and to drive the Operational State Model as appropriate, except for the initial state that is set in init_device().

The Operational State of an SKA Tango device is exposed as the built-in Tango device state via set_state(). Subclasses are not expected to call set_state() themselves and should instead change the state by performing the appropriate action on the OpStateModel, which will additionally ensure that change and archive events are sent.

The Operational State only supports a subset of the DevState enumeration, with the following interpretations:

INIT: The device is initialising
DISABLE: The device is not currently monitoring or controlling the system
UNKNOWN: The device cannot determine the state of the system
OFF: The system under control is powered off
STANDBY: The system under control is in low-power standby mode
ON: The system under control is powered on
ALARM: The system under control is powered on and some attribute is raising an alarm. This is typically automatically handled by Tango.
FAULT: The system under control is in fault

The _op_state signal object will raise a ValueError if set to a DevState value other than those above.

BaseInterface similarly provides a _status signal which will update the built-in Tango status attribute with set_status() and push change and archive events whenever set.

In addition to providing the _op_state and _status signals BaseInterface also provides the signals _admin_mode, _commanded_state and _health_state as well as the corresponding Tango attributes adminMode, commandedState and healthState.

The adminMode Tango attribute is writable and should not be set by subclasses of this interface. Subclasses will be notified when a client has requested the device to stop/start controlling the system under control via the change_control_level() method, which they are expected to override.

The commandedState and healthState should be set by subclasses as appropriate, using the corresponding signals.

This interface also provides optional state transition commands Off(), Standby(), On() as well as implementations of the is_Off_allowed(), is_Standby_allowed() and is_On_allowed() methods which respect the Operational State machine.

Subclasses can provide an implementation for these commands by overriding the execute_Off(), execute_Standby() and execute_On() methods.

Additionally, the Reset() command is required and the corresponding execute_Reset() method must be overridden by all subclasses.

_admin_mode: Signal[ska_control_model.AdminMode]

Signal for the admin mode of the device.

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

Typically, subclasses will not need to listen to this signal and should instead override change_control_level().

admin_mode_model: ska_control_model.AdminModeModel: State model used to ensure that admin mode transitions are allowed.

adminMode: attribute_from_signal: Admin mode attribute of the device.

op_state_model: ska_control_model.OpStateModel: State model used to ensure that operational state transitions are allowed.

_op_state: OpStateSignal

Signal for the Operational State of the device.

Write to this signal to set the state of the Tango device.

_status: Signal[str]

Signal for the status of the device.

Write to this signal to set the status of the Tango device.

_commanded_state: CommandedStateSignal

Signal for the commanded Operational State of the device.

Write to this signal whenever a command is executed which will result in a state transition.

commandedState: attribute_from_signal

Attribute for the last commanded Operating State of the device.

This should be set by subclasses of this interface by writing to _commanded_state.

_health_state

Signal for the health state of the device.

Writing to this signal sets the reported health state of the Tango device.

Writing to this signal directly is deprecated. Use the report_health() method instead.

healthState

Health state attribute of the device.

This should be set by subclasses of this interface by calling report_health().

healthInfo

A list of reasons why the device has a particular health state.

Should be an empty list when the healthState is HealthState.OK.

This should be set by subclasses of this interface by calling report_health().

init_device() → None[source]

Initialise the tango device after startup.

Subclasses overriding init_device must call init_completed() once initialisation has finished.

Example

class MyDevice(BaseInterface):
    def init_device(self) -> None:
        super().init_device()
        ...
        self.init_completed()

on_new_shared_bus() → None[source]: Enable manual event pushing for state and status.

_init_state_model() → None[source]

Initialise the state model for the device.

init_invoked() → None[source]: Perform ‘init_invoked’ action on the OpStateModel.

init_completed() → None[source]

Notify the state machine that initialisation has completed.

Must be called from your init_device() method.

Example

class MyDevice(BaseInterface):
    def init_device(self) -> None:
        super().init_device()
        ...
        self.init_completed()

component_disconnected() → None[source]: Perform disconnected action on the OpStateModel.

component_unknown() → None[source]: Perform unknown state action on the OpStateModel.

component_off() → None[source]: Perform off action on the OpStateModel.

component_standby() → None[source]: Perform cstandby action on the OpStateModel.

component_on() → None[source]: Perform on action on the OpStateModel.

component_fault() → None[source]: Perform fault action on the OpStateModel.

component_no_fault() → None[source]: Perform fault stopped action on the OpStateModel.

_update_state(state: DevState, status: str | None = None) → None[source]

Perform Tango operations in response to a change in op state.

This helper method is passed to the op state model as a callback, so that the model can trigger actions in the Tango device.

Parameters:

state – the new state value
status – an optional new status string

_update_admin_mode(admin_mode: AdminMode) → None[source]

Update the admin mode of the device.

This helper method is passed to the admin mode model as a callback, so that the model can trigger actions in the Tango device.

Parameters:: admin_mode – the new admin mode value

read_adminMode() → AdminMode | tuple[AdminMode, float, AttrQuality][source]

Read the admin mode of the device.

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

write_adminMode(mode: AdminMode) → None[source]

Set the Admin Mode of the device.

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

Parameters:

value – Admin Mode of the device.

Raises:

ValueError – for unknown adminMode
StateModelError – for a disallowed transition

change_control_level(control_level: ControlLevel) → None[source]

Change how the device is interacting with the system under control.

Subclasses must override this method to do the following:

Stop all communication with the system under control when passed ControlLevel.NO_CONTACT. Once the Tango device has stopped communicating with the system under control, the operational state must transition to DISABLE.
Start controlling the system under control when passed ControlLevel.FULL_CONTROL. Once the Tango device has started trying to communicating with the system under control, it must transition to a state other than DISABLE. If the Tango device is trying to communicate with the system under control, but unable to, it should transition to operational state UNKNOWN.

Subclasses overriding this method should not assume that the ControlLevel is exhaustive.

is_adminMode_allowed(request_type: AttReqType) → bool[source]

Check if the adminMode can be read/written currently.

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

is_commandedState_allowed(request_type: AttReqType) → bool[source]

Check if the commandedState can be read currently.

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

read_commandedState() → str | tuple[str, float, AttrQuality][source]

Read the commanded state of the device.

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

is_healthState_allowed(request_type: AttReqType) → bool[source]

Check if the healthState can be read currently.

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

read_healthState() → HealthState | tuple[HealthState, float, AttrQuality][source]

Read the health state of the device.

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

is_healthInfo_allowed(request_type: AttReqType) → bool[source]

Check if the healthInfo can be read currently.

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

read_healthInfo() → list[str] | tuple[list[str], float, AttrQuality][source]

Read the health info of the device.

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

report_health(health_state: HealthState, health_info: list[str]) → None[source]

Report the health of the device.

The health_info should include all the reasons that the current device is reporting a DEGRADED or FAILED health_state. Each element should be a separate reason and should be as brief as possible, while still providing enough information to aid in diagnosis.

If the health_state == OK, then the health_info` must be an empty list.

health_state == UNKNOWN is not supported from this interface. Use health_state == FAILED with a descriptive health_info instead, e.g. “component <x> is unreachable”.

Parameters:

health_state – The overall health state of the device
health_info – A list of reasons for the current overall health state

Raises:

ValueError – If health_state==OK and health_info != []
ValueError – If health_state==UNKNOWN

is_Reset_allowed(request_type: LRCReqType | None = None) → bool[source]

Return whether the Reset() command may be called currently.

This method can be overridden by subclasses to change when this command is allowed.

Returns:: whether the command may be called in the current device state

execute_Reset() → DevVarLongStringArrayType[source]

Execute the standard Reset command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

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 :meth`!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(request_type: LRCReqType | None = None) → bool[source]

Return whether the Standby() command may be called currently.

This method can be overridden by subclasses to change when this command is allowed.

Returns:: whether the command may be called in the current device state

execute_Standby() → DevVarLongStringArrayType[source]

Execute the standard Standby command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should put the device in standby mode and result in the operational state transitioning to STANDBY. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

is_Off_allowed(request_type: LRCReqType | None = None) → bool[source]

Return whether the Off() command may be called currently.

This method can be overridden by subclasses to change when this command is allowed.

Returns:: whether the command may be called in the current device state

execute_Off() → DevVarLongStringArrayType[source]

Execute the standard Off command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device off and result in the operational state transitioning to OFF. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

is_On_allowed(request_type: LRCReqType | None = None) → bool[source]

Return whether the On() command may be called currently.

This method can be overridden by subclasses to change when this command is allowed.

Returns:: whether the command may be called in the current device state

execute_On() → DevVarLongStringArrayType[source]

Execute the standard On command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device on and result in the operational state transitioning to ON or ALARM. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

Reset() → DevVarLongStringArrayType[source]

Reset the device.

Subclasses should override the execute_Reset method to change the behaviour of this command.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

On() → DevVarLongStringArrayType[source]

Turn the device on.

Subclasses should override the execute_On method to change the behaviour of this command.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

Standby() → DevVarLongStringArrayType[source]

Put the device into standby mode.

Subclasses should override the execute_Standby method to change the behaviour of this command.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

Off() → DevVarLongStringArrayType[source]

Turn the device off.

Subclasses should override the execute_Off method to change the behaviour of this command.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

ska_tango_base.base.base_interface.standard_control_mode(doc: str | None = None, **kwargs: Any) → tuple[Signal[ControlMode], attribute_from_signal][source]

Return a signal and attribute pair for the optional controlMode attribute.

Deprecated:: Use the future API instead.

When set to ControlMode.LOCAL, the Tango device accepts commands only from a local client and ignores commands and queries received from TM or any other remote clients. The local client must release LOCAL control before remote clients can take control again. The returned tuple should be unpacked into class-level signal and attribute declarations, for example:

_control_mode, controlMode = standard_control_mode()

The attribute is writable and memorized. Override doc to provide a device-specific description.

Parameters:

doc – Optional override for the Tango attribute description.
kwargs – Additional keyword arguments forwarded to attribute_from_signal.

Returns:

A (signal, attribute) tuple.

ska_tango_base.base.base_interface.standard_simulation_mode(doc: str | None = None, **kwargs: Any) → tuple[Signal[SimulationMode], attribute_from_signal][source]

Return a signal and attribute pair for the optional simulationMode attribute.

When SimulationMode.TRUE, the device is using a simulator instead of real hardware. The returned tuple should be unpacked into class-level signal and attribute declarations, for example:

_simulation_mode, simulationMode = standard_simulation_mode()

The attribute is writable and memorized. Override doc to provide a device-specific description.

Parameters:

doc – Optional override for the Tango attribute description.
kwargs – Additional keyword arguments forwarded to attribute_from_signal.

Returns:

A (signal, attribute) tuple.

ska_tango_base.base.base_interface.standard_test_mode(doc: str | None = None, **kwargs: Any) → tuple[Signal[TestMode], attribute_from_signal][source]

Return a signal and attribute pair for the optional testMode attribute.

When TestMode.TEST, the device substitutes its normal operating logic with testing/stub logic, which is useful for integration testing without real hardware. The returned tuple should be unpacked into class-level signal and attribute declarations, for example:

_test_mode, testMode = standard_test_mode()

The attribute is writable and memorized. Override doc to provide a device-specific description.

Parameters:

doc – Optional override for the Tango attribute description.
kwargs – Additional keyword arguments forwarded to attribute_from_signal.

Returns:

A (signal, attribute) tuple.

Logging

Provided for backwards compatibility.

class ska_tango_base.base.logging.LoggingUtils[source]

Utility functions to aid logger configuration.

These functions are encapsulated in class to aid testing - it allows dependent functions to be mocked.

static sanitise_logging_targets(targets: list[str] | None, device_name: str) → list[str][source]

Validate and return logging targets ‘<type>::<name>’ strings.

Parameters:

targets – List of candidate logging target strings, like ‘<type>[::<name>]’ Empty and whitespace-only strings are ignored. Can also be None.
device_name – Tango device name, like ‘domain/family/member’, used for the default file name

Returns:

list of ‘<type>::<name>’ strings, with default name, if applicable

Raises:

LoggingTargetError – for invalid target string that cannot be corrected

static get_syslog_address_and_socktype(url: str) → tuple[tuple[str, int] | str, SocketKind | None][source]

Parse syslog URL and extract address and socktype parameters for SysLogHandler.

Parameters:

url –

Universal resource locator string for syslog target. Three types are supported: file path, remote UDP server, remote TCP server.

Output to a file: ‘file://<path to file>’. For example, ‘file:///dev/log’ will write to ‘/dev/log’.
Output to remote server over UDP: ‘udp://<hostname>:<port>’. For example, ‘udp://syslog.com:514’ will send to host ‘syslog.com’ on UDP port 514
Output to remote server over TCP: ‘tcp://<hostname>:<port>’. For example, ‘tcp://rsyslog.com:601’ will send to host ‘rsyslog.com’ on TCP port 601

For backwards compatibility, if the protocol prefix is missing, the type is interpreted as file. This is deprecated. For example, ‘/dev/log’ is equivalent to ‘file:///dev/log’.

Returns:

An (address, socktype) tuple.

For file types:

the address is the file path as as string
socktype is None

For UDP and TCP:

the address is tuple of (hostname, port), with hostname a string, and port an integer.
socktype is socket.SOCK_DGRAM for UDP, or socket.SOCK_STREAM for TCP.

Raises:

LoggingTargetError – for invalid url string

static create_logging_handler(target: str, tango_logger: Logger | None = None) → Any[source]

Create a Python log handler based on the target type.

Supported target types are “console”, “file”, “syslog”, “tango”.

Parameters:

target – Logging target for logger, <type>::<name>
tango_logger – Instance of tango.Logger, optional. Only required if creating a target of type “tango”.

Returns:

StreamHandler, RotatingFileHandler, SysLogHandler, or TangoLoggingServiceHandler

Raises:

LoggingTargetError – for invalid target string

static update_logging_handlers(targets: list[str], logger: Logger) → None[source]

Update a logger’s handlers.

Parameters:

targets – a list of handler names. Current handlers whose name is not included here will be removed. Names for which the logger currently does not have a handler will have handlers created and added.
logger – the logger whose handlers are to be updated.

class ska_tango_base.base.logging.TangoLoggingServiceHandler[source]

Handler that emit logs via Tango device’s logger to TLS.

__init__(tango_logger: Logger) → None[source]

Initialise a new instance.

Parameters:: tango_logger – the tango logger to use to emit logs

emit(record: LogRecord) → None[source]

Emit a log record.

Parameters:: record – the log to be emitted.