#
# 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.
"""Mixin to provide a heritable admin mode for an SKA Tango device."""
from __future__ import annotations
import concurrent.futures as _fs
import inspect
import logging
import time
import types
import typing
import ska_control_model as scm
import tango.server
from .. import type_hints, utils
from ..callback_scheduler import CallbackID, CallbackScheduler, CallbackSchedulerMixin
from ..software_bus import AttrSignal, SharingObserver, Signal, attribute_from_signal
from ._base_interface import BaseInterface
Inherit = type_hints.InheritType()
"""The inherit marker object."""
_INHERIT_VALUE = len(scm.future.AdminMode.__members__)
class _AdminModeInheritor(SharingObserver):
def __init__(
self,
parent_trl: str,
connection_timeout: float,
*,
logger: logging.Logger | None = None,
time_source: typing.Callable[[], float] = time.time,
) -> None:
self._parent_trl = parent_trl
# We store the future for _callback_id so that we only
# wait for this if we actually need to use it to unregister.
self._callback_id: _fs.Future[CallbackID] | None = None
self._disconnect_timestamp: float | None = None
self._connection_timeout = connection_timeout
self._time_source = time_source
self._last_admin_mode_from_parent: scm.future.AdminMode | None = None
if logger is None:
self._logger = logging.getLogger(__name__)
else:
self._logger = logger
def start(self, scheduler: CallbackScheduler) -> None:
if self._callback_id is None:
self._callback_id = scheduler.register_event_callback(
self._parent_trl,
"adminMode",
tango.EventType.CHANGE_EVENT,
self.on_parent_admin_mode,
)
def stop(self, scheduler: CallbackScheduler) -> None:
if self._callback_id is not None:
try:
scheduler.unregister_callback(self._callback_id.result())
except Exception:
pass
finally:
self._callback_id = None
self._last_admin_mode_from_parent = None
def on_parent_admin_mode(self, event: type_hints.EventDataType) -> None:
"""Process a adminMode event from parent."""
if event.err:
# NOTE: Here we assume any error event is a connection issue, in
# principle the parent device could send a custom error, so this
# assumption is not really correct. However, without a use case for
# pushing these custom events I struggle to know what I should do
# instead.
self._logger.warning(
"Received error event when subscribing to parent adminMode: %s",
event.errors,
)
self.notify_parent_not_connected()
else:
self.update_admin_mode(event.attr_value.value)
def update_admin_mode(
self,
admin_mode: scm.future.AdminMode,
) -> None:
self._disconnect_timestamp = None
self.shared_bus.emit("._admin_mode", admin_mode)
if self._callback_id is None:
self.shared_bus.emit(
"._parent_connection_status", scm.CommunicationStatus.DISABLED
)
else:
self._last_admin_mode_from_parent = admin_mode
self.shared_bus.emit(
"._parent_connection_status", scm.CommunicationStatus.ESTABLISHED
)
def notify_parent_not_connected(self) -> None:
if self._disconnect_timestamp is None:
self._disconnect_timestamp = self._time_source()
if (
self._last_admin_mode_from_parent is not None
and self._time_source() - self._disconnect_timestamp
< self._connection_timeout
):
# We deliberately emit another adminMode event here in case
# the write value has changed
self.shared_bus.emit("._admin_mode", self._last_admin_mode_from_parent)
self.shared_bus.emit(
"._parent_connection_status", scm.CommunicationStatus.NOT_ESTABLISHED
)
else:
self.shared_bus.emit("._admin_mode", None)
self.shared_bus.emit(
"._parent_connection_status",
(
scm.CommunicationStatus.NOT_ESTABLISHED,
None,
tango.AttrQuality.ATTR_ALARM,
),
)
[docs]
class HeritableAdminModeMixin(CallbackSchedulerMixin, BaseInterface):
ParentTRL = tango.server.device_property(
mandatory=False, default_value="", dtype="DevString"
)
"""
Device property.
Tango resource locator of parent. If not set, or an empty string, the device is
considered to not have a parent and :attr:`!adminMode` inheritance is not possible.
"""
ParentConnectionTimeout = tango.server.device_property(
mandatory=False, default_value=60, dtype="DevFloat"
)
"""
Device property.
Seconds until lack of contact from parent results in an alarm.
"""
[docs]
def init_device(self) -> None:
"""Initialise device."""
super().init_device()
# We don't want write_adminMode to wait around for event subscriptions
# to be set up. To achieve this we explicitly connect from the
# device_exported_hook thread and use this variable to keep track of
# whether that has completed yet.
#
# Lock invariant for Tango monitor:
# - If this is set and we are inheriting adminMode, then the
# _AdminModeInheritor has been started.
self._parent_admin_mode_connected = False
if self.ParentTRL:
# We will get some flickering here during a call to Init() with e.g.
# ENGINEERING memorized, but I think that is acceptable.
self.adminMode.set_write_value(_INHERIT_VALUE)
self.write_adminMode(Inherit)
else:
self.adminMode.set_write_value(scm.future.AdminMode.ENGINEERING)
self.write_adminMode(scm.future.AdminMode.ENGINEERING)
[docs]
def on_new_shared_bus(self) -> None:
"""Create adminMode inheritor."""
super().on_new_shared_bus()
self._admin_mode_inheritor = _AdminModeInheritor(
self.ParentTRL, self.ParentConnectionTimeout, logger=self.logger
)
[docs]
def device_exported_hook(self) -> None:
"""Connect to parent."""
super().device_exported_hook()
if self.ParentTRL:
self.callback_scheduler.connect_event_stream(
self.ParentTRL, "adminMode", tango.EventType.CHANGE_EVENT
)
with utils._hold_tango_monitor(self):
self._parent_admin_mode_connected = True
if self.adminMode.get_write_value() == _INHERIT_VALUE:
self._admin_mode_inheritor.start(self.callback_scheduler)
@tango.server.attribute(dtype="DevString", doc=ParentTRL.__doc__)
def parentTRL(self) -> str:
"""Return the configured TRL of the parent."""
return typing.cast(str, self.ParentTRL)
@tango.server.attribute(dtype="DevFloat", doc=ParentConnectionTimeout.__doc__)
def parentConnectionTimeout(self) -> float:
"""Return the configured parent timeout."""
return typing.cast(float, self.ParentConnectionTimeout)
_parent_connection_status = AttrSignal[scm.CommunicationStatus](
stored=True, initial_value=scm.CommunicationStatus.DISABLED
)
parentConnectionStatus = attribute_from_signal(
_parent_connection_status,
dtype=scm.CommunicationStatus,
doc="Communication status to parent device.",
)
"""Attribute reporting the connection status to the parent."""
_admin_mode = Signal[scm.future.AdminMode](stored=True)
"""Override _admin_mode signal to remove initial value.
This doc string is not marked public, so _admin_mode will be
hidden from the documentation.
The fact we are overriding this is not of interest to subclasses.
"""
adminMode = attribute_from_signal(
_admin_mode,
dtype="DevEnum",
enum_labels=list(scm.future.AdminMode.__members__.keys()) + ["INHERIT"],
description=(
"The Admin Mode of the device.\n\n"
"Set by a client to mark the device for science or only engineering."
"May be set to a special 'INHERIT' value to automatically update the "
"adminMode from the parent at parentTRL. Cannot set to 'INHERIT' if "
"parentTRL is empty or not set."
),
memorized=True,
hw_memorized=True,
)
"""Overridden admin mode attribute with additional INHERIT set point."""
def __write_adminMode(self, value: int) -> None:
converted: scm.future.AdminMode | type_hints.InheritType
if value == _INHERIT_VALUE:
converted = Inherit
else:
converted = scm.future.AdminMode(value)
self.write_adminMode(converted)
adminMode.write(__write_adminMode)
@tango.server.attribute(
dtype="DevBoolean",
description=("True if the device's adminMode is being inherited"),
)
def inheritAdminMode(self) -> bool:
"""Return True if the device is inheriting its adminMode."""
return typing.cast(bool, self.adminMode.get_write_value() == _INHERIT_VALUE)
[docs]
def write_adminMode(
self, setpoint: scm.future.AdminMode | type_hints.InheritType
) -> None:
"""
Set the Admin Mode of the device.
:param setpoint: The value to set it to
:raises tango.DevFailed: When trying to set INHERIT with no ParentTRL
"""
if setpoint is Inherit:
if not self.ParentTRL:
info = inspect.getframeinfo(
typing.cast(types.FrameType, inspect.currentframe())
)
tango.Except.throw_exception(
"STB_NotSupported",
"'INHERIT' is not supported: No ParentTRL set",
f"{info.filename}:{info.lineno}",
)
self._admin_mode_inheritor.stop(self.callback_scheduler)
if self._parent_admin_mode_connected:
# This may end up calling notify_parent_not_connected()
# if an error event is passed to on_parent_admin_mode(),
# however, if we have good data then we go straight to
# ESTABLISHED.
self._admin_mode_inheritor.start(self.callback_scheduler)
else:
self._admin_mode_inheritor.notify_parent_not_connected()
else:
self._admin_mode_inheritor.stop(self.callback_scheduler)
self._admin_mode_inheritor.update_admin_mode(
typing.cast(scm.future.AdminMode, setpoint)
)