#
# 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 ska_control_model
from ska_control_model import (
ObsMode,
ObsState,
)
from ..base import BaseInterface
from ..software_bus import (
Signal,
attribute_from_signal,
)
from ..type_hints import ReadAttrType, SharingObserverProtocol
[docs]
class CommandedObsStateSignal(Signal[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.
"""
[docs]
def __init__(self) -> None:
"""Initialise the signal."""
super().__init__(stored=True, initial_value=ObsState.EMPTY)
def __set__(self, obj: SharingObserverProtocol, value: ObsState) -> None:
"""
Emit value on the bus for the signal.
:raises ValueError: if the value is not a stable observation state.
"""
if value not in [
ObsState.EMPTY,
ObsState.IDLE,
ObsState.READY,
ObsState.ABORTED,
]:
raise ValueError(f"{value} is not a stable observation state.")
super().__set__(obj, value)
[docs]
class ObsInterface(BaseInterface): # pylint: disable=abstract-method
"""Provides the Tango interface for an SKA device with an Observation State."""
_obs_state: Signal[ska_control_model.ObsState] = Signal[ObsState](
stored=True, initial_value=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=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=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`.
"""
# TODO: This ObsMode should be a list. Need to clarify exactly how this
# should behave.
_obs_mode: Signal[ska_control_model.ObsMode] = Signal[ObsMode](
stored=True, initial_value=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(
_obs_mode,
dtype=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`.
"""
def __read_obsState(self) -> ReadAttrType[ObsState]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_obsState()
obsState.read(__read_obsState)
[docs]
def read_obsState(self) -> ReadAttrType[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 __read_commandedObsState(self) -> ReadAttrType[ObsState]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_commandedObsState()
commandedObsState.read(__read_commandedObsState)
[docs]
def read_commandedObsState(self) -> ReadAttrType[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 __read_obsMode(self) -> ReadAttrType[ObsMode]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_obsMode()
obsMode.read(__read_obsMode)
[docs]
def read_obsMode(self) -> ReadAttrType[ObsMode]:
"""
Read the observation mode of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`obsMode` attribute.
"""
return self.__class__.obsMode.do_read(self)
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 __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)