Future Base Interface

A base interface for SKA Tango devices.

This module provides the BaseInterface and related classes that form the foundation for all future SKA Tango device interfaces, providing SKA operational state management, admin mode, health state, and long-running command support. It is meant to be used as a replacement for the existing SKABaseDevice with all deprecated parts removed.

class ska_tango_base.future._base_interface.OpStateEmitMixin[source]

A mixin class for a SKA base device or its component manager.

It provides helper methods for managing the Operational State of the device.

static _combine_op_state(**kwargs: Unpack[OpStateInput]) → DevState[source]

Combine the given operational state values into a single tango.DevState.

Parameters:: kwargs – key/values for state.
Returns:: the combined tango.DevState.

software_fault(status: str) → None

Update operational state to reflect that a software fault has occurred.

Parameters:: status – the fault message to emit as the device status.

class ska_tango_base.future._base_interface.OpStateInput[source]

Bases: TypedDict

Type for operational state input dictionary.

initialising: bool: Boolean indicating whether the device is initialising. If True, the device is in INIT state.

fault: str | None: Optional string indicating a software fault has occurred. If not None, the device is in FAULT state and the string is emitted as the device status.

class ska_tango_base.future._base_interface.OpStateSignal[source]

Bases: Signal[DevState]

Special signal for Operational State.

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.future._base_interface.BaseInterface[source]

Bases: OpStateEmitMixin, SignalBusMixin, SKADevice

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

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 set the Operational State 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. Instead, the state is driven automatically by calling init_invoked(), init_completed() and software_fault() from the appropriate places in the device implementation, which will additionally ensure that change and archive events are sent.

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

INIT: The device is initialising. Mandatory initial state for all devices.
ON: The Tango device is communicating with the system under control. Mandatory default state after initialisation has finished.
FAULT: The Tango device encountered a software fault/exception. Optional.

BaseInterface also provides the signals _admin_mode and _health_state as well as the corresponding Tango attributes adminMode and healthState, as well as healthInfo.

The adminMode Tango attribute is writable and should not be set by subclasses of this interface.

The healthState and healthInfo read-only attributes should be set by subclasses by calling report_health(). It is intended to provide information of the overall health of the device, based on the state of the component(s) it is monitoring.

_admin_mode: Signal[AdminMode]

Signal for the admin mode of the device.

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

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

_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.

_health_state

Signal for the health state of the device.

This signal cannot be written to directly. 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()

init_invoked() → None[source]: Update operational state to reflect that initialisation has started.

init_completed() → None[source]

Update operational state to reflect 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()

transform_emission_cascade(emissions: dict[str, Any], store_manager: EmissionStoreManagerProtocol) → dict[str, Any][source]

Transform a cascade of emissions.

This method is called for each emit() call, receiving all signals being emitted in that cascade. You can modify the emissions - change values, add new signals, or remove signals by not including them in the returned dict.

The transformation is applied to the whole cascade of emissions at once, so the transformation can use the relationships between signals to determine how to transform them. Signals not included in the returned dict will be suppressed and not sent to observers. New signals can be added to generate derived emissions.

Any implementation must call super().transform_emission_cascade(emissions) in order to not break the transformation chain:

Parameters:

emissions – the cascade of emissions to be transformed, as a dict mapping signal name to emitted value.
store_manager – context manager for atomically updating stored emissions

Returns:

the transformed cascade of emissions, as a dict mapping signal name to emitted 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:: mode – Admin Mode of the device.

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

ska_tango_base.future._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.

When set to ControlMode.NO_MONITOR_NO_CONTROL, the Tango device does not perform its monitor and control functions, i.e. shall not monitor and control the underlying component/device(s).

When set to ControlMode.MONITOR_NO_CONTROL, the Tango device shall perform its monitoring functions, and shall not perform its control functionality, i.e. shall not accept requests to change device configuration and execute commands on the device.

When set to ControlMode.MONITOR_AND_CONTROL, the Tango device shall perform its monitoring and controlling functions, i.e. shall monitor and control the underlying component/device(s).

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.future._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.future._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.