"""Mixin to provide ControlMode for a device."""
import ska_control_model as scm
from .. import faults, type_hints
from ..software_bus import Signal, attribute_from_signal
from ._base_interface import BaseInterface
__all__ = ["ControlModeMixin"]
[docs]
class ControlModeMixin(BaseInterface):
"""
Mixin to provide a control mode for an SKA Tango device.
An SKA Tango device with a control mode can be instructed by a client to
stop controlling and/or monitoring its component. Typically, this should
only be needed for Tango devices that monitor and control a low-level
hardware, although if needed it can be implemented by higher level devices
(such as subarrays).
Subclasses must override the :meth:`start_monitoring_component` and
:meth:`stop_monitoring_component` methods to start and stop monitoring their
component when the control mode changes. These methods are respectively
called on transitions from/to ``NO_MONITOR_NO_CONTROL``.
After a call to :meth:`!init_device`, the device should *not* be
communicating with its component as if the control mode was
``NO_MONITOR_NO_CONTROL``. Any memorized value for the :attr:`controlMode`
attribute will then be written and :meth:`start_monitoring_component` will
be called if that was requested by the end user.
Subclasses should also provide is-allowed methods that check
:meth:`check_component_controlled` for every command that should be disabled when
not in the ``MONITOR_AND_CONTROL`` control mode.
"""
_control_mode = Signal[scm.future.ControlMode](
stored=True, initial_value=scm.future.ControlMode.NO_MONITOR_NO_CONTROL
)
"""
Signal for the control mode of the device.
In general, this signal should not be written to by the device and instead should
be updated by clients.
:meta public:
"""
controlMode = attribute_from_signal(
_control_mode,
name="controlMode",
dtype=scm.future.ControlMode,
doc=(
"The control mode of the device.\n\n"
"The Tango Device either does not perform its monitor and control "
"functions at all, performs only its monitoring functions, or performs "
"both its monitoring and controlling functions."
),
memorized=True,
hw_memorized=True,
)
"""
The control mode of the device.
This attribute is memorized and written by clients."""
def __read_controlMode(self) -> type_hints.ReadAttrType[scm.future.ControlMode]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_controlMode()
controlMode.read(__read_controlMode)
[docs]
def read_controlMode(self) -> type_hints.ReadAttrType[scm.future.ControlMode]:
"""
Read the control mode of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`controlMode` attribute.
"""
return self.__class__.controlMode.do_read(self)
def __write_controlMode(self, mode: scm.future.ControlMode) -> None:
"""Dispatch to write method to allow subclasses to override."""
self.write_controlMode(mode)
controlMode.write(__write_controlMode)
[docs]
def write_controlMode(self, mode: scm.future.ControlMode) -> None:
"""
Set the control mode of the device.
Subclasses can override this to change the behaviour of the
:attr:`controlMode` attribute. Most subclasses should instead override
the :meth:`start_monitoring_component`/:meth:`stop_monitoring_component`
methods.
:param mode: new control mode of the device.
"""
old_mode = self._control_mode
if mode == old_mode:
return
if old_mode == scm.future.ControlMode.NO_MONITOR_NO_CONTROL:
# The order here matters, we need to allow health reporting
# (mark_health_unknown(False)) first, so that subclasses can report
# health in start_monitoring_component()
self.mark_health_unknown(False)
try:
self.start_monitoring_component()
except NotImplementedError as ex:
self.software_fault(str(ex))
raise
if self._health_state == scm.HealthState.UNKNOWN:
self.report_health(
scm.HealthState.FAILED,
["Device implementation has not provided a health report"],
)
elif mode == scm.future.ControlMode.NO_MONITOR_NO_CONTROL:
# The order here matters, we need to disallow health reporting
# (mark_health_unknown(True)) after we have called
# stop_monitoring_component() just in case any last-minute reports
# come in while we are stopping.
try:
self.stop_monitoring_component()
except NotImplementedError as ex:
self.software_fault(str(ex))
raise
self.mark_health_unknown(True, "controlMode is NO_MONITOR_NO_CONTROL")
self._control_mode = mode
[docs]
def init_device(self) -> None:
"""Initialise the device."""
super().init_device()
self.mark_health_unknown(True, "controlMode is NO_MONITOR_NO_CONTROL")
[docs]
def start_monitoring_component(self) -> None:
"""
Start monitoring the component.
Subclasses must override this to enable component monitoring.
The health of the component should be reported with
:meth:`~._base_interface.BaseInterface.report_health`. If the
connection to the component is established asynchronously, then use
``HealthState.FAILED`` to indicate that the connect has not yet been
established.
"""
raise NotImplementedError(
"'start_monitoring_component' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'ControlModeMixin' is an abstract base class."
)
[docs]
def stop_monitoring_component(self) -> None:
"""
Stop monitoring the component.
Subclasses must override this to disable component monitoring. This function
should not return until the connection to the component has been severed.
"""
raise NotImplementedError(
"'stop_monitoring_component' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'ControlModeMixin' is an abstract base class."
)
[docs]
def check_component_controlled(self) -> None:
"""Raise an exception if the Tango device is not controlling its component."""
if self._control_mode != scm.future.ControlMode.MONITOR_AND_CONTROL:
raise faults.StateModelError(
f"Control is disabled: controlMode is {self._control_mode}"
)