"""This module has some helper methods to manage subscriptions."""
import logging
from dataclasses import dataclass, field
from functools import partial
from itertools import product
from queue import Queue
from typing import Any, Callable, Optional, Tuple
from asyncua import ua
from ska_control_model import HealthState
from ska_mid_dish_steering_control import SteeringControlUnit
from ska_mid_dish_steering_control.sculib import AttrDict
from tango import DevState
from ska_mid_dish_ds_manager.models.constants import (
DYNAMIC_PUBLISH_PERIOD_MS,
ERROR_STATUS_NODES,
IGNORED_NODES,
POINTING_EVENT_PUBLISH_PERIOD_S,
POINTING_PUBLISH_PERIOD_MS,
STATIC_PUBLISH_PERIOD_MS,
)
from ska_mid_dish_ds_manager.models.dish_enums import (
ApplicationStateType,
BandType,
CurrentModeType,
DscCmdAuthType,
DscCtrlState,
DSCState,
DSOperatingMode,
DSPowerState,
IndexerPosition,
PointingState,
)
from ska_mid_dish_ds_manager.models.node_mappings import (
dsc_state_to_ds_mode_map,
)
from ska_mid_dish_ds_manager.utils import (
SubscriptionEvent,
generate_component_state_key_from_opcua_path,
get_attribute_data_type_as_python_type,
get_tai_timestamp_from_datetime,
)
[docs]@dataclass
class MonitoredNode:
"""Dataclass to store node subscription information."""
node_path: str # Dot seperated name, E.g Management.Commands.SetPowerMode
component_state_name: str # Key for the component state
# Method to do a conversion E.g 1 -> DSCState.STANDBY
convert_to_component_state_val: Optional[Callable] = None
# Methods for side effects
calculate_updates: Tuple = field(default_factory=tuple)
[docs]@dataclass
class MonitoredNodeGroup:
"""Dataclass to store groups of nodes and their subscription rates."""
subscription_rate: int = STATIC_PUBLISH_PERIOD_MS
nodes: dict[str, MonitoredNode] = field(default_factory=dict)
buffer_samples: bool = True
[docs]class SubscriptionManager:
"""Utility class for subscriptions."""
[docs] def __init__(
self, logger: logging.Logger, update_device_state: Optional[Callable] = None
) -> None:
"""Init SubscriptionManager.
:param logger: logger
:type logger: logging.Logger
:param update_device_state: Callback to update the tango device state
:type update_device_state: Callable
"""
self.logger = logger
self._update_device_state = update_device_state
self._awaited_node_updates: set = set()
self.monitored_node_groups: list[MonitoredNodeGroup] = []
self._dsc_static_pointing_model_params_order = (
"IA",
"CA",
"NPAE",
"AN",
"AN0",
"AW",
"AW0",
"ACEC",
"ACES",
"ABA",
"ABphi",
"IE",
"ECEC",
"ECES",
"HECE4",
"HESE4",
"HECE8",
"HESE8",
)
self._dsc_static_pointing_model_bands = (0, 1, 2, 3, 4, 5, 6, 7)
[docs] def mark_attribute_as_received(self, node_path: str) -> None:
"""Mark an attribute as received and update dev state to RUNNING if all are received.
:param opcua_node_path: The path to the OPC UA node which has been received
:type opcua_node_path: str
"""
if node_path in self._awaited_node_updates:
self._awaited_node_updates.remove(node_path)
if len(self._awaited_node_updates) == 0 and self._update_device_state is not None:
self.logger.info("Setting device DevState to RUNNING")
self._update_device_state(DevState.RUNNING)
@staticmethod
def _compute_band_type_to_indexer_position(
subs_event: SubscriptionEvent, component_state: dict
) -> dict:
"""Calculate indexerposition.
:param subs_event: The event
:type subs_event: SubscriptionEvent
:return: The component state values to be updated.
:rtype: dict
"""
# Map the OPC-UA BandType enum to the IndexerPosition enum
# <Field Name="Optical" Value="0"/>
# <Field Name="Band_1" Value="1"/>
# <Field Name="Band_2" Value="2"/>
# <Field Name="Band_3" Value="3"/>
# <Field Name="Band_4" Value="4"/>
# <Field Name="Band_5a" Value="5"/>
# <Field Name="Band_5b" Value="6"/>
# <Field Name="Band_6" Value="7"/>
# <Field Name="Unknown" Value="8"/>
mapping = {
ua.BandType.Optical: IndexerPosition.OPTICAL,
ua.BandType.Band_1: IndexerPosition.B1,
ua.BandType.Band_2: IndexerPosition.B2,
ua.BandType.Band_3: IndexerPosition.B3,
ua.BandType.Band_4: IndexerPosition.B4,
ua.BandType.Band_5a: IndexerPosition.B5a,
ua.BandType.Band_5b: IndexerPosition.B5b,
ua.BandType.Band_6: IndexerPosition.B6,
ua.BandType.Unknown: IndexerPosition.MOVING,
}
updates = {"indexerposition": mapping.get(subs_event.value, IndexerPosition.UNKNOWN)}
return updates
@staticmethod
def _compute_health_state(_subs_event: SubscriptionEvent, component_state: dict) -> dict:
"""Calculate health_state.
:param _subs_event: The event, not used here.
:type _subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: The component state values to be updated.
:rtype: dict
"""
new_health_state = HealthState.FAILED
if (
component_state["currentmode"] == CurrentModeType.SYSTEMMODE_BB
and component_state["applicationstate"] == ApplicationStateType.Run
):
new_health_state = HealthState.OK
return {"healthstate": new_health_state}
@staticmethod
def _compute_pointing_updates(subs_event: SubscriptionEvent, component_state: dict) -> dict:
"""Calculate `currentpointing` and `achievedpointing` updates.
:param subs_event: The event
:type subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: The component state values to be updated.
:rtype: dict
"""
updates = {}
stored_current_pointing = component_state["currentpointing"]
if subs_event.value[0] - stored_current_pointing[0] >= POINTING_EVENT_PUBLISH_PERIOD_S:
updates["currentpointing"] = subs_event.value
new_achieved_pointing = [subs_event.value[0], subs_event.value[1], subs_event.value[2]]
updates["achievedpointing"] = new_achieved_pointing
return updates
@staticmethod
def _compute_pointing_state(subs_event: SubscriptionEvent, component_state: dict) -> dict:
"""Calculate pointingstate.
:param subs_event: The event, not used
:type subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: The component state values to be updated.
:rtype: dict
"""
updates = {}
pointing_states = {
DSCState.STANDSTILL: PointingState.READY,
DSCState.SLEW: PointingState.SLEW,
DSCState.TRACK: PointingState.TRACK,
}
if component_state["dscstate"] in pointing_states:
updates["pointingstate"] = pointing_states[component_state["dscstate"]]
else:
updates["pointingstate"] = PointingState.UNKNOWN
return updates
def _compute_operating_mode(
self, subs_event: SubscriptionEvent, component_state: dict
) -> dict:
"""Compute the operating mode.
:param subs_event: The event, not used
:type subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: Values to update the component state with.
:rtype: dict
"""
dsc_state: DSCState = component_state.get("dscstate", DSCState.UNKNOWN)
return {"operatingmode": dsc_state_to_ds_mode_map.get(dsc_state, DSOperatingMode.UNKNOWN)}
def _compute_configure_target_lock(
self, subs_event: SubscriptionEvent, component_state: dict
) -> dict:
"""Compute the configureTargetLock attribute.
:param subs_event: The event, not used
:type subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: Values to update the component state with.
:rtype: dict
"""
updated_configure_target_lock = [
component_state.get("onsourcethreshold", 0.0),
component_state.get("thresholdtimeperiod", 0.0),
]
return {"configuretargetlock": updated_configure_target_lock}
def _compute_dsc_ctrl_state(
self, subs_event: SubscriptionEvent, component_state: dict
) -> dict:
"""Compute the DSC Control State.
:param subs_event: The event, not used
:type subs_event: SubscriptionEvent
:param component_state: The current component state.
:type component_state: dict
:return: Values to update the component state with.
:rtype: dict
"""
dsc_state: DSCState = component_state.get("dscstate", DSCState.UNKNOWN)
dsc_auth: DscCmdAuthType = component_state.get("dsccmdauth", DscCtrlState.NO_AUTHORITY)
# DSC Control State follows the following heirarchy (highest precedence):
# 1. LOCKED/ LOCKED_STOWED
# 2. HHP
# 3. EGUI
# 4. LMC
# 5. NO_AUTHORITY (Extra case added to account if the dscState is not in a "locked"
# state and no authority has been taken)
# Option 1: Check if the DSC reports the DSCState as being LOCKED/ LOCKED_STOWED
# Note: These are the only 2 dscStates that influence the DSC Control State.
if dsc_state in (DSCState.LOCKED, DSCState.LOCKED_STOWED):
return {"dscctrlstate": DscCtrlState.LOCKED} # LOCKED
# Determine Control State using dscCmdAuth (options 2 - 5)
match dsc_auth:
case DscCmdAuthType.HHP:
return {"dscctrlstate": DscCtrlState.MANUAL_CONTROL}
case DscCmdAuthType.EGUI:
return {
"dscctrlstate": DscCtrlState.ENGINEERING_CONTROL
} # Note: we only have 1 mapping for EGUI (ICD has 2)
case DscCmdAuthType.LMC:
return {"dscctrlstate": DscCtrlState.REMOTE_CONTROL}
case _:
return {"dscctrlstate": DscCtrlState.NO_AUTHORITY} # No Auth (Extra case)
def _compute_band_update(
self,
component_state_name: str,
sub_coeff: str,
subs_event: SubscriptionEvent,
component_state: dict,
) -> dict:
"""Update the specified component state's coefficient value based on a subscription event.
:param component_state_name: The key in `component_state` whose values need to be updated.
:param sub_coeff: The coefficient to be updated, identified by its name.
:param subs_event (SubscriptionEvent): An event containing the new # noqa: RST206
value to update the coefficient with.
:param component_state: A dictionary containing the current state of the components, where
the key is the component name and the value is a list of coefficient values.
:return: Updated component state
"""
updated_values = list(component_state[component_state_name])
if len(updated_values) < len(self._dsc_static_pointing_model_params_order):
updated_values = [0.0] * len(self._dsc_static_pointing_model_params_order)
for i, coeff in enumerate(self._dsc_static_pointing_model_params_order):
if sub_coeff == coeff:
updated_values[i] = subs_event.value
break
return {component_state_name: updated_values}
# pylint: disable=too-many-locals
[docs] def build_static_monitored_nodes(
self, available_attributes: AttrDict
) -> list[MonitoredNodeGroup]:
"""Build up the information used for subscriptions.
:param available_attributes: The list of available attributes from the sculib
:type available_attributes: AttrDict
Returns:
{'Management.Status.DscState': MonitoredNode, ....}
E.g
node_names["Management.Status.DscState"] = MonitoredNode(
The dictionary key.
"Management.Status.DscState",
The path of the node.
"dscstate",
The component state name.
lambda sub_event: DSCState(sub_event.value),
A method that gets an event value. Used to do simple conversions like int to Enum
(self._compute_pointing_state, self._compute_operating_mode),
Handle any side effects. E.g if the DSCState changes then the pointing state and
operatingmode needs to change.)
:return: Return the information needed for subscriptions.
:rtype: list[dict[str, MonitoredNode], int]
"""
expected_static_nodes = {
"Management.Status.DscState": MonitoredNode(
"Management.Status.DscState",
"dscstate",
lambda sub_event: DSCState(sub_event.value),
(
self._compute_pointing_state,
self._compute_operating_mode,
self._compute_dsc_ctrl_state,
),
),
"Management.PowerStatus.LowPowerActive": MonitoredNode(
"Management.PowerStatus.LowPowerActive",
"powerstate",
lambda sub_event: (
DSPowerState.LOW_POWER if sub_event.value else DSPowerState.FULL_POWER
),
(self._compute_operating_mode,),
),
"System.CurrentMode": MonitoredNode(
"System.CurrentMode",
"currentmode",
lambda sub_event: CurrentModeType(sub_event.value),
(self._compute_health_state,),
),
"Logic.Application.ApplicationState": MonitoredNode(
"Logic.Application.ApplicationState",
"applicationstate",
lambda sub_event: ApplicationStateType(sub_event.value),
(self._compute_health_state,),
),
"CommandArbiter.Status.DscCmdAuthority": MonitoredNode(
"CommandArbiter.Status.DscCmdAuthority",
"dsccmdauth",
lambda sub_event: DscCmdAuthType(sub_event.value),
(self._compute_dsc_ctrl_state,),
),
"Management.Status.OnSource": MonitoredNode(
"Management.Status.OnSource",
"achievedtargetlock",
lambda sub_event: sub_event.value,
(),
),
"Management.Status.FiPos": MonitoredNode(
"Management.Status.FiPos",
"indexerposition",
None,
(self._compute_band_type_to_indexer_position,),
),
"Management.Status.HHP_Cncted": MonitoredNode(
"Management.Status.HHP_Cncted",
"hhpconnected",
lambda sub_event: sub_event.value,
(),
),
"Tracking.Status.act_statOffset_value_El": MonitoredNode(
"Tracking.Status.act_statOffset_value_El",
"actstaticoffsetvalueel",
lambda sub_event: sub_event.value,
(self._compute_operating_mode,),
),
"Tracking.Status.act_statOffset_value_Xel": MonitoredNode(
"Tracking.Status.act_statOffset_value_Xel",
"actstaticoffsetvaluexel",
lambda sub_event: sub_event.value,
(self._compute_operating_mode,),
),
"Tracking.Status.OnSourceThreshold": MonitoredNode(
"Tracking.Status.OnSourceThreshold",
"onsourcethreshold",
lambda sub_event: sub_event.value,
(self._compute_configure_target_lock,),
),
"Tracking.Status.ThresholdTimePeriod": MonitoredNode(
"Tracking.Status.ThresholdTimePeriod",
"thresholdtimeperiod",
lambda sub_event: sub_event.value,
(self._compute_configure_target_lock,),
),
"System.DisplayedDiagnosis": MonitoredNode(
"System.DisplayedDiagnosis",
"displayeddiagnosis",
lambda sub_event: str(sub_event.value),
(),
),
}
for error_node in ERROR_STATUS_NODES:
node_path = f"Management.ErrorStatus.{error_node}"
expected_static_nodes[node_path] = MonitoredNode(
node_path,
error_node.lower(),
lambda sub_event: sub_event.value,
(),
)
expected_pointing_nodes = {
"Pointing.Status.CurrentPointing": MonitoredNode(
"Pointing.Status.CurrentPointing",
"currentpointing",
None,
(self._compute_pointing_updates,),
),
"Tracking.Status.p_desired_Az": MonitoredNode(
"Tracking.Status.p_desired_Az",
"desiredpointingaz",
lambda sub_event: (
get_tai_timestamp_from_datetime(sub_event.source_timestamp),
sub_event.value,
),
(),
),
"Tracking.Status.p_desired_El": MonitoredNode(
"Tracking.Status.p_desired_El",
"desiredpointingel",
lambda sub_event: (
get_tai_timestamp_from_datetime(sub_event.source_timestamp),
sub_event.value,
),
(),
),
"Tracking.TableStatus.act_pt_act_index": MonitoredNode(
"Tracking.TableStatus.act_pt_act_index",
"actptactindex",
lambda sub_event: sub_event.value,
(),
),
"Tracking.TableStatus.act_pt_end_index": MonitoredNode(
"Tracking.TableStatus.act_pt_end_index",
"actptendindex",
lambda sub_event: sub_event.value,
(),
),
}
# Add expected static pointing model parameters to expected_static_nodes
spm_param_index_style = None
if "Pointing.StaticPmParameters.StaticPmParameters[0].IA" in available_attributes:
spm_param_index_style = "[%s]"
elif "Pointing.StaticPmParameters.StaticPmParameters_0.IA" in available_attributes:
spm_param_index_style = "_%s"
else:
self.logger.warning("Could not find StaticPmParameters in available nodes.")
if spm_param_index_style:
# Add all the band pointing model parameter nodes
for bandtype_index, coeff in product(
self._dsc_static_pointing_model_bands, self._dsc_static_pointing_model_params_order
):
band_param = BandType(bandtype_index)
if band_param == BandType.OPTICAL:
band_param_name = "0"
else:
band_param_name = band_param.name.lower()[1:]
band_comp_state_name = f"band{band_param_name}pointingmodelparams"
spm_param_index = spm_param_index_style % bandtype_index
path = f"Pointing.StaticPmParameters.StaticPmParameters{spm_param_index}.{coeff}"
expected_static_nodes[path] = MonitoredNode(
path,
band_comp_state_name,
None,
(
partial(
self._compute_band_update,
band_comp_state_name,
coeff,
),
),
)
monitored_node_groups: list[MonitoredNodeGroup] = [
MonitoredNodeGroup(STATIC_PUBLISH_PERIOD_MS, expected_static_nodes),
MonitoredNodeGroup(
POINTING_PUBLISH_PERIOD_MS, expected_pointing_nodes, buffer_samples=False
),
]
# Ensure that the expected nodes are in the list of available nodes
for group_index, mon_node_group in enumerate(monitored_node_groups):
found_nodes_in_group = {}
for mon_node_key, mon_node in mon_node_group.nodes.items():
if mon_node.node_path in available_attributes:
found_nodes_in_group[mon_node_key] = mon_node
else:
self.logger.warning(
"Expected node %s not found in list of available nodes.",
mon_node.node_path,
)
monitored_node_groups[group_index].nodes = found_nodes_in_group
return monitored_node_groups
[docs] def get_monitored_node(self, node_path: str) -> MonitoredNode | None:
"""Get a monitored node based on its node path.
:param node_path: The node path of the desired node
:type node_path: str
:return: The found MonitoredNode | None.
:rtype: MonitoredNode | None
"""
for mon_node_group in self.monitored_node_groups:
if node_path in mon_node_group.nodes:
return mon_node_group.nodes[node_path]
return None
[docs] def get_all_attributes_not_monitored(
self, node_trees_to_monitor: list[str], scu: SteeringControlUnit
) -> tuple[list[MonitoredNode], list[type | None]]:
"""Get a list of all available attributes under the given list nodes.
:param node_trees_to_monitor: The list of root nodes to monitor all children attributes.
:type node_trees_to_monitor: list[str]
:param scu: SteeringControlUnit instance
:type scu: SteeringControlUnit
"""
unmonitored_attributes_paths: list[str] = []
for available_attr_node_path in scu.attributes:
# Ensure the node starts from one of the desired root paths
for tree_to_monitor in node_trees_to_monitor:
if available_attr_node_path.lower().startswith(tree_to_monitor.lower()):
if self.get_monitored_node(available_attr_node_path) is None:
unmonitored_attributes_paths.append(available_attr_node_path)
break
# Generate list of MonitoredNodes from unmonitored_attributes_paths
unmonitored_nodes: list[MonitoredNode] = []
node_datatypes: list[type | None] = []
def handle_dynamic_sub_event(dtype: type | None, sub_event: Any) -> Any | None:
"""Convert event to expected data type.
:param dtype: python data type
:param sub_event: received tango event
:return: value in appropriate python type
"""
try:
# Pointing.StaticPmParameters.IndexMin has a list of 1 value
if (
dtype
and dtype is int
and isinstance(sub_event.value, list)
and len(sub_event.value) == 1
):
return dtype(sub_event.value[0])
return dtype(sub_event.value) if sub_event.value is not None and dtype else None
except TypeError:
self.logger.error(
"Failed to convert sub value (%s) to excepted datatype (%s) from (%s)",
sub_event.value,
dtype,
sub_event.name,
)
return None
for unmonitored_attr_path in unmonitored_attributes_paths:
if unmonitored_attr_path in IGNORED_NODES:
continue
component_state_name = generate_component_state_key_from_opcua_path(
unmonitored_attr_path
)
python_datatype = get_attribute_data_type_as_python_type(
scu,
unmonitored_attr_path,
)
sub_event_cb = partial(handle_dynamic_sub_event, python_datatype)
unmonitored_nodes.append(
MonitoredNode(
unmonitored_attr_path,
component_state_name,
sub_event_cb,
(),
)
)
node_datatypes.append(python_datatype)
return unmonitored_nodes, node_datatypes
[docs] def setup_subscriptions(
self,
scu: SteeringControlUnit,
events_queue: Queue,
add_attribute_for_opcua_node: Callable,
node_trees_to_monitor: list[str],
) -> None:
"""Get a monitored node based on its node path.
:param scu: The SCULib
:type scu: SteeringControlUnit
:param events_queue: The queue for datachange events
:type events_queue: Queue
:param add_attribute_for_opcua_node: Callback to create a tango attribute required for
monitoring an opcua node.
:type add_attribute_for_opcua_node: Callable
:param node_trees_to_monitor: The list of root nodes to monitor all children attributes.
:type node_trees_to_monitor: list[str]
"""
monitored_node_groups = self.build_static_monitored_nodes(scu.attributes)
self.monitored_node_groups = monitored_node_groups
if len(node_trees_to_monitor) > 0:
(
dynamically_exposed_attributes,
attribute_datatypes,
) = self.get_all_attributes_not_monitored(
node_trees_to_monitor=node_trees_to_monitor, scu=scu
)
# Note: Groups can have a max of 10000 nodes and there is a max of 20 groups per
# session.
dynamic_nodes_group = MonitoredNodeGroup(DYNAMIC_PUBLISH_PERIOD_MS, {})
# Create tango device attributes from this list (need a callback)
for node, dtype in zip(dynamically_exposed_attributes, attribute_datatypes):
try:
add_attribute_for_opcua_node(node.node_path, dtype)
dynamic_nodes_group.nodes[node.node_path] = node
except Exception:
self.logger.exception(
"Failed to subscribe to node at path: %s.", node.node_path
)
# Add to monitored_node_groups
self.monitored_node_groups.append(dynamic_nodes_group)
self.logger.debug("Added %s nodes dynamically", len(dynamic_nodes_group.nodes))
# Clear the set of awaited updates and add each node_path in the monitored groups to it
# so that when all have been marked as received, using the 'mark_attribute_as_received'
# command, then we can transition the device to RUNNING
self._awaited_node_updates.clear()
for mon_node_group in self.monitored_node_groups:
node_paths = []
for mon_node in mon_node_group.nodes.values():
node_paths.append(mon_node.node_path)
self._awaited_node_updates.add(mon_node.node_path)
node_paths = [mon_node.node_path for mon_node in mon_node_group.nodes.values()]
if len(node_paths) > 0:
scu.subscribe(
node_paths,
mon_node_group.subscription_rate,
events_queue,
buffer_samples=mon_node_group.buffer_samples,
)