Future Obs Interface

Interface for devices with an observation state.

class ska_tango_base.future._obs_interface.ObsStateEmitMixin[source]

Bases: SharingObserver

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

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

_update_obs_state(**kwargs: Unpack[ObsStateInput]) → None[source]

Emit a change in observation state.

This is a helper method for use by subclasses.

Parameters:: kwargs – key/values for state.

component_scanning() → None[source]: Update observation state to reflect that scanning is in progress.

component_not_scanning() → None[source]: Update observation state to reflect that scanning is not in progress.

component_resourced() → None[source]: Update observation state to reflect that resources are allocated.

component_unresourced() → None[source]: Update observation state to reflect that no resources are allocated.

component_configured() → None[source]: Update observation state to reflect that the subarray is configured.

component_unconfigured() → None[source]: Update observation state to reflect that the subarray is not configured.

component_obsfault() → None[source]: Update observation state to reflect that a fault has occurred.

component_no_obsfault() → None[source]: Revert the observation state fault.

class ska_tango_base.future._obs_interface.ObsInterface[source]

Bases: ObsStateEmitMixin, BaseInterface

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

_obs_state: Signal[ObsState]

Signal for the observation state of the device.

Write to this signal to report the observation state of the device to Tango clients.

obsState

Observation state attribute of the device.

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

_commanded_obs_state: CommandedObsStateSignal

Signal for the commanded observation state of the device.

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

commandedObsState

Attribute for the last commanded Observation State of the device.

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

_config_progress: Signal[int]

Signal for the configuration progress of the device.

Write to this signal to report configuration progress to Tango clients.

configurationProgress

Configuration progress attribute of the device.

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

_config_delay_expected: Signal[int]

Signal for the configuration delay expected of the device.

Write to this signal to report the expected configuration delay to Tango clients.

configurationDelayExpected

Configuration delay expected attribute of the device.

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

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_obsState() → ObsState | tuple[ObsState, float, AttrQuality][source]

Read the observation state of the device.

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

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

Check if the obsState can be read currently.

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

read_commandedObsState() → ObsState | tuple[ObsState, float, AttrQuality][source]

Read the commanded observation state of the device.

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

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

Check if the commandedObsState can be read currently.

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

read_configurationProgress() → int | tuple[int, float, AttrQuality][source]

Read the configuration progress of the device.

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

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

Check if the configurationProgress can be read currently.

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

read_configurationDelayExpected() → int | tuple[int, float, AttrQuality][source]

Read the expected configuration delay of the device.

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

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

Check if the configurationDelayExpected can be read currently.

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

ska_tango_base.future._obs_interface.standard_obs_mode(description: str | None = None, **kwargs: Any) → tuple[Signal[list[ObsMode]], attribute_from_signal][source]

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

The obsMode attribute exposes the active observation modes of the device as a list of ObsMode enumerants. The initial value is [ObsMode.IDLE]. The returned tuple should be unpacked into class-level signal and attribute declarations, for example:

_obs_mode, obsMode = standard_obs_mode()

Override description to provide a device-specific Tango attribute description.

Parameters:

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

Returns:

A (signal, attribute) tuple.