#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""Interface for devices with an observation state."""
import typing
import warnings
import ska_control_model as scm
import tango
from ..base import BaseInterface
from ..software_bus import NoValue, NoValueType, Signal, attribute_from_signal
from ..type_hints import ReadAttrType, SharingObserverProtocol
__all__ = ["ObsInterface", "standard_obs_mode"]
class CommandedObsStateSignal(Signal[scm.ObsState]):
"""
Special signal for the commanded observation state.
Ensures that only stable enumerants from :py:class:`~ska_control_model.ObsState` for
a commanded observation state are emitted for the signal.
"""
def __init__(self) -> None:
"""Initialise the signal."""
super().__init__(stored=True, initial_value=scm.ObsState.EMPTY)
def __set__(
self, obj: SharingObserverProtocol, value: scm.ObsState | None | NoValueType
) -> None:
"""
Emit value on the bus for the signal.
:raises ValueError: if the value is not a stable observation state.
"""
if value not in [
None,
NoValue,
scm.ObsState.EMPTY,
scm.ObsState.IDLE,
scm.ObsState.READY,
scm.ObsState.ABORTED,
]:
raise ValueError(f"{value} is not a stable observation state.")
super().__set__(obj, value)
[docs]
class ObsInterface(BaseInterface):
"""
Provides the Tango interface for an SKA device with an Observation State.
.. warning ::
The default scalar :py:attr:`obsMode` attribute provided by this class has been
deprecated since ska-tango-base 1.5.0. Use the :py:meth:`standard_obs_mode`
method to override the scalar attribute and its signal with a list version
in a subclass.
**Example** ::
class MyDevice(ObsInterface):
_obs_mode, obsMode = standard_obs_mode()
"""
_obs_state: Signal[scm.ObsState] = Signal[scm.ObsState](
stored=True, initial_value=scm.ObsState.EMPTY
)
"""Signal for the observation state of the device.
Write to this signal to report the observation state of the device to Tango clients.
:meta public:
"""
obsState = attribute_from_signal(
_obs_state,
dtype=scm.ObsState,
description="The Observation State of the device.",
)
"""Observation state attribute of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_obs_state`.
"""
_commanded_obs_state: CommandedObsStateSignal = CommandedObsStateSignal()
"""Signal for the commanded observation state of the device.
Write to this signal whenever a command is executed which will result in
a obs state transition.
:meta public:
"""
commandedObsState = attribute_from_signal(
_commanded_obs_state,
dtype=scm.ObsState,
description="""
The last commanded stable Observation State of the device.
Initial value is EMPTY. The only stable states it can
change to are EMPTY, IDLE, READY or ABORTED, following the start
of any state transition command.
""",
)
"""
Attribute for the last commanded Observation State of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_commanded_obs_state`.
"""
_obs_mode: Signal[scm.ObsMode] | Signal[list[scm.ObsMode]] = Signal[scm.ObsMode](
stored=True, initial_value=scm.ObsMode.IDLE
)
"""Signal for the observation mode of the device.
Write to this signal to report observation mode of the device to Tango clients.
:meta public:
"""
obsMode: attribute_from_signal | None = attribute_from_signal(
_obs_mode,
dtype=scm.ObsMode,
description="The Observation Mode of the device.",
)
"""
Observation mode attribute of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_obs_mode`.
"""
_config_progress: Signal[int] = Signal[int](stored=True, initial_value=0)
"""Signal for the configuration progress of the device.
Write to this signal to report configuration progress to Tango clients.
:meta public:
"""
configurationProgress = attribute_from_signal(
_config_progress,
dtype="uint16",
unit="%",
max_value=100,
min_value=0,
description="The percentage configuration progress of the device.",
)
"""
Configuration progress attribute of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_config_progress`.
"""
_config_delay_expected: Signal[int] = Signal[int](stored=True, initial_value=0)
"""Signal for the configuration delay expected of the device.
Write to this signal to report the expected configuration delay to Tango clients.
:meta public:
"""
configurationDelayExpected = attribute_from_signal(
_config_delay_expected,
dtype="uint16",
unit="seconds",
description="The expected configuration delay of the device in seconds.",
)
"""
Configuration delay expected attribute of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_config_delay_expected`.
"""
[docs]
def init_device(self) -> None:
"""Initialise the tango device after startup."""
super().init_device()
if self.__class__.obsMode is not None:
obs_mode_config = self.get_attribute_config("obsMode")[0]
if obs_mode_config.data_format == tango.AttrDataFormat.SCALAR:
warnings.warn(
"The scalar 'obsMode' attribute provided by 'ObsInterface' has "
"been deprecated since ska-tango-base 1.5.0. "
"Use the 'ska_tango_base.obs.standard_obs_mode' method to override "
"the scalar attribute and its signal with a list version in the "
f"'{self.__class__.__name__}' subclass: "
"'_obs_mode, obsMode = standard_obs_mode()'",
DeprecationWarning,
)
def __read_obsState(self) -> ReadAttrType[scm.ObsState]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_obsState()
obsState.read(__read_obsState)
[docs]
def read_obsState(self) -> ReadAttrType[scm.ObsState]:
"""
Read the observation state of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`obsState` attribute.
"""
return self.__class__.obsState.do_read(self)
def __is_obsState_allowed(self, request_type: tango.AttReqType) -> bool:
return self.is_obsState_allowed(request_type)
[docs]
def is_obsState_allowed(self, request_type: tango.AttReqType) -> bool:
"""
Check if the obsState can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
obsState.fisallowed = __is_obsState_allowed
def __read_commandedObsState(self) -> ReadAttrType[scm.ObsState]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_commandedObsState()
commandedObsState.read(__read_commandedObsState)
[docs]
def read_commandedObsState(self) -> ReadAttrType[scm.ObsState]:
"""
Read the commanded observation state of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`commandedObsState` attribute.
"""
return self.__class__.commandedObsState.do_read(self)
def __is_commandedObsState_allowed(self, request_type: tango.AttReqType) -> bool:
return self.is_commandedObsState_allowed(request_type)
[docs]
def is_commandedObsState_allowed(self, request_type: tango.AttReqType) -> bool:
"""
Check if the commandedObsState can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
commandedObsState.fisallowed = __is_commandedObsState_allowed
def __read_obsMode(self) -> ReadAttrType[scm.ObsMode]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_obsMode()
typing.cast(attribute_from_signal, obsMode).read(__read_obsMode)
[docs]
def read_obsMode(self) -> ReadAttrType[scm.ObsMode]:
"""
Read the observation mode of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`obsMode` attribute.
"""
assert self.__class__.obsMode is not None
return self.__class__.obsMode.do_read(self)
def __is_obsMode_allowed(self, request_type: tango.AttReqType) -> bool:
return self.is_obsMode_allowed(request_type)
[docs]
def is_obsMode_allowed(self, request_type: tango.AttReqType) -> bool:
"""
Check if the obsMode can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
typing.cast(attribute_from_signal, obsMode).fisallowed = __is_obsMode_allowed
def __read_configurationProgress(self) -> ReadAttrType[int]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_configurationProgress()
configurationProgress.read(__read_configurationProgress)
[docs]
def read_configurationProgress(self) -> ReadAttrType[int]:
"""
Read the configuration progress of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`configurationProgress` attribute.
"""
return self.__class__.configurationProgress.do_read(self)
def __is_configurationProgress_allowed(
self, request_type: tango.AttReqType
) -> bool:
return self.is_configurationProgress_allowed(request_type)
[docs]
def is_configurationProgress_allowed(self, request_type: tango.AttReqType) -> bool:
"""
Check if the configurationProgress can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
configurationProgress.fisallowed = __is_configurationProgress_allowed
def __read_configurationDelayExpected(self) -> ReadAttrType[int]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_configurationDelayExpected()
configurationDelayExpected.read(__read_configurationDelayExpected)
[docs]
def read_configurationDelayExpected(self) -> ReadAttrType[int]:
"""
Read the expected configuration delay of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`configurationDelayExpected` attribute.
"""
return self.__class__.configurationDelayExpected.do_read(self)
def __is_configurationDelayExpected_allowed(
self, request_type: tango.AttReqType
) -> bool:
return self.is_configurationDelayExpected_allowed(request_type)
[docs]
def is_configurationDelayExpected_allowed(
self, request_type: tango.AttReqType
) -> bool:
"""
Check if the configurationDelayExpected can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
configurationDelayExpected.fisallowed = __is_configurationDelayExpected_allowed
[docs]
def _update_obs_state(self, obs_state: scm.ObsState) -> None:
"""
Perform Tango operations in response to a change in obsState.
This helper method is passed to the observation state model as a
callback, so that the model can trigger actions in the Tango
device.
:param obs_state: the new obs_state value
:meta public:
"""
self.logger.debug(f"ObsState={repr(scm.ObsState(obs_state))}")
self._obs_state = obs_state
[docs]
def standard_obs_mode(
description: str | None = None,
**kwargs: typing.Any,
) -> tuple[Signal[list[scm.ObsMode]], attribute_from_signal]:
"""
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 :py:class:`~ska_control_model.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.
:param description: Optional override for the Tango attribute description.
:param kwargs: Additional keyword arguments forwarded to
:py:func:`~ska_tango_base.software_bus.attribute_from_signal`.
:return: A ``(signal, attribute)`` tuple.
"""
signal = Signal[list[scm.ObsMode]](
name="_obs_mode", stored=True, initial_value=[scm.ObsMode.IDLE]
)
if description is None:
description = "The Observation Modes of the device."
attr = attribute_from_signal(
signal,
description=description,
name="obsMode",
dtype=(scm.ObsMode,),
max_dim_x=len(scm.ObsMode),
**kwargs,
)
return signal, attr