#
# 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.
"""This module implements testMode and test mode overrides as a mix-in class."""
from __future__ import annotations
import json
from collections.abc import Callable, Iterable
from functools import partial, wraps
from typing import (
Any,
Concatenate,
ParamSpec,
TypeVar,
)
import ska_control_model
from ska_control_model import HealthState, TestMode
from tango import AttReqType, AttributeProxy
from .base.base_interface import standard_test_mode
from .software_bus import (
Signal,
SignalBusMixin,
attribute_from_signal,
listen_to_signal,
)
from .type_hints import ReadAttrType
[docs]
class TestModeOverrideMixin(SignalBusMixin):
"""
Add Test Mode Attribute Overrides to a Tango device.
When testMode is TEST, clients can override the value
of attributes using the test_mode_overrides attribute.
"""
[docs]
def init_device(self: TestModeOverrideMixin) -> None:
"""Add our variables to the class we are extending."""
super().init_device()
self._test_mode_overrides = {}
self._test_mode_current_overrides: set[str] = set()
self._test_mode_overrides_changed: Callable[[], None] | None = None
self._test_mode_overrides = {}
self._test_mode_enum_attrs = {
"healthState": HealthState,
}
def _get_override_value(
self: TestModeOverrideMixin, attr_name: str, default: Any = None
) -> Any:
"""
Read a value from our overrides, use a default value when not overridden.
Used where we use possibly-overridden internal values within the device server
(i.e. reading member variables, not via the Tango attribute read mechanism).
e.g.
``my_thing = self._get_override_value("thing", self._my_thing_true_value)``
:param attr_name: Tango Attribute name.
:param default: Default value to return if no override in effect.
:returns: Active override value or ``default``.
"""
if (
self._test_mode != TestMode.TEST
or attr_name not in self._test_mode_overrides
):
return default
return self._override_value_convert(
attr_name, self._test_mode_overrides[attr_name]
)
_test_mode: Signal[ska_control_model.TestMode]
"""
Signal for the test mode of the device.
Values are emitted for this signal whenever a client successfully changes to
the testMode attribute.
:meta public:
"""
testMode: attribute_from_signal
"""
Test mode attribute of the device.
Either no test mode or an indication of the test mode.
"""
_test_mode, testMode = standard_test_mode()
def __read_testMode(self) -> ReadAttrType[TestMode]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_testMode()
testMode.read(__read_testMode)
[docs]
def read_testMode(self) -> ReadAttrType[TestMode]:
"""
Read the test mode of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`testMode` attribute.
"""
return self.__class__.testMode.do_read(self)
def __write_testMode(self, mode: TestMode) -> None:
"""Dispatch to write method to allow subclasses to override."""
return self.write_testMode(mode)
testMode.write(__write_testMode)
[docs]
def write_testMode(self, mode: TestMode) -> None:
"""
Write the test mode of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`testMode` attribute.
"""
self.__class__.testMode.do_write(self, mode)
_test_mode_overrides = Signal[dict[str, Any]](stored=True)
test_mode_overrides = attribute_from_signal(
_test_mode_overrides,
dtype=str,
to_tango=json.dumps,
write_to_signal=True,
doc="Attribute value overrides (JSON dict)",
)
@test_mode_overrides.from_tango
def __test_mode_overrides(self, value: str) -> dict[str, Any]:
value_dict = json.loads(value)
assert isinstance(value_dict, dict), "expected JSON-encoded dict"
return value_dict
def __is_test_mode_overrides_allowed(self, request_type: AttReqType) -> bool:
"""Dispatch to write method to allow subclasses to override."""
return self.is_test_mode_overrides_allowed(request_type)
test_mode_overrides.is_allowed(__is_test_mode_overrides_allowed)
[docs]
def is_test_mode_overrides_allowed(
self: TestModeOverrideMixin, request_type: AttReqType
) -> bool:
"""
Control access to test_mode_overrides attribute.
Writes to the attribute are allowed only if test mode is active.
:param request_type: Attribute request type
:returns: If in test mode
"""
if request_type == AttReqType.READ_REQ:
return True
return self._test_mode == TestMode.TEST
def __read_test_mode_overrides(self: TestModeOverrideMixin) -> ReadAttrType[str]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_test_mode_overrides()
test_mode_overrides.read(__read_test_mode_overrides)
[docs]
def read_test_mode_overrides(self: TestModeOverrideMixin) -> ReadAttrType[str]:
"""
Read the current override configuration.
:return: JSON-encoded dictionary (attribute name: value)
"""
return self.__class__.test_mode_overrides.do_read(self)
def __write_test_mode_overrides(
self: TestModeOverrideMixin, overrides: str
) -> None:
"""Dispatch to read method to allow subclasses to override."""
return self.write_test_mode_overrides(overrides)
test_mode_overrides.write(__write_test_mode_overrides)
[docs]
def write_test_mode_overrides(self: TestModeOverrideMixin, overrides: str) -> None:
"""
Read the override configuration.
:param value_str: JSON-encoded dict of overrides (attribute name: value)
"""
self.__class__.test_mode_overrides.do_write(self, overrides)
# call downstream callback function to deal with override changes
if self._test_mode_overrides_changed is not None:
self._test_mode_overrides_changed()
@listen_to_signal("_test_mode_overrides")
def __update_overrides(self, value_dict: dict[str, Any]) -> None:
overrides_being_removed = self._test_mode_current_overrides - value_dict.keys()
self._test_mode_current_overrides = set(value_dict.keys())
# we could call _override_value_convert on incoming values here, but I prefer to
# leave as-is, so the user can read back the same thing they wrote in
self._push_events_overrides_removed(overrides_being_removed)
# send events for all overrides
# only *need* to send new or changed overrides but that's annoying to determine
# i.e. premature optimisation
for attr_name, value in value_dict.items():
converted = self._override_value_convert(attr_name, value)
attr_cfg = self.get_device_attr().get_attr_by_name(attr_name)
if attr_cfg.is_change_event():
self.push_change_event(attr_name, converted)
if attr_cfg.is_archive_event():
self.push_archive_event(attr_name, converted)
def _push_events_overrides_removed(
self: TestModeOverrideMixin, attrs_to_refresh: Iterable[str]
) -> None:
"""
Push true value events for attributes that were previously overridden.
:param attrs_to_refresh: Names of our attributes that are no longer overridden
"""
for attr_name in attrs_to_refresh:
# Read configuration of attribute
attr_cfg = self.get_device_attr().get_attr_by_name(attr_name)
manual_event = attr_cfg.is_change_event() or attr_cfg.is_archive_event()
if not manual_event:
continue
# Read current state of attribute
attr = AttributeProxy(f"{self.get_name()}/{attr_name}").read()
if attr_cfg.is_change_event():
self.push_change_event(attr_name, attr.value, attr.time, attr.quality)
if attr_cfg.is_archive_event():
self.push_archive_event(attr_name, attr.value, attr.time, attr.quality)
def _override_value_convert(self, attr_name: str, value: Any) -> Any:
"""
Automatically convert types for attr overrides (e.g. enum label -> int).
:param attr_name: Attribute name
:param value: Value to convert
:return: Converted value
"""
if attr_name in self._test_mode_enum_attrs and isinstance(value, str):
return self._test_mode_enum_attrs[attr_name][value]
# default to no conversion
return value
C = TypeVar("C", bound=TestModeOverrideMixin)
P = ParamSpec("P")
[docs]
def overridable(
func: Callable[Concatenate[C, P], Any] | None = None, *, name: str | None = None
) -> Callable[Concatenate[C, P], Any]:
"""
Decorate attribute with test mode overrides.
:param func: Tango attribute
:return: Overridden value or original function
"""
if func is None:
return partial(overridable, name=name)
attr_name = name if name else func.__name__.removeprefix("read_")
@wraps(func)
def override_attr_in_test_mode(
self: C, /, *args: P.args, **kwargs: P.kwargs
) -> Any:
"""
Override attribute when test mode is active and value specified.
:param self: Tango device with TestModeMixin
:param args: Any positional arguments
:param kwargs: Any keyword arguments
:return: Tango attribute
"""
if self._test_mode == TestMode.TEST and attr_name in self._test_mode_overrides:
return self._override_value_convert(
attr_name, self._test_mode_overrides[attr_name]
)
# Test Mode not active, normal attribute behaviour
return func(self, *args, **kwargs)
return override_attr_in_test_mode