Subscription Manager

This module has some helper methods to manage subscriptions.

class MonitoredNode(node_path, component_state_name, convert_to_component_state_val=None, calculate_updates=<factory>)[source]

Dataclass to store node subscription information.

__init__(node_path, component_state_name, convert_to_component_state_val=None, calculate_updates=<factory>)
class MonitoredNodeGroup(subscription_rate=100, nodes=<factory>, buffer_samples=True)[source]

Dataclass to store groups of nodes and their subscription rates.

__init__(subscription_rate=100, nodes=<factory>, buffer_samples=True)
class SubscriptionManager(logger, update_device_state=None)[source]

Utility class for subscriptions.

__init__(logger, update_device_state=None)[source]

Init SubscriptionManager.

Parameters:

  • logger (logging.Logger) – logger

  • update_device_state (Callable) – Callback to update the tango device state

build_static_monitored_nodes(available_attributes)[source]

Build up the information used for subscriptions.

Parameters:

available_attributes (AttrDict) – The list of available attributes from the sculib

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.)

Returns:

Return the information needed for subscriptions.

Return type:

list[dict[str, MonitoredNode], int]

get_all_attributes_not_monitored(node_trees_to_monitor, scu)[source]

Get a list of all available attributes under the given list nodes.

Parameters:

  • node_trees_to_monitor (list[str]) – The list of root nodes to monitor all children attributes.

  • scu (SteeringControlUnit) – SteeringControlUnit instance

Return type:

tuple[list[MonitoredNode], list[Optional[type]]]

get_monitored_node(node_path)[source]

Get a monitored node based on its node path.

Parameters:

node_path (str) – The node path of the desired node

Returns:

The found MonitoredNode | None.

Return type:

MonitoredNode | None

mark_attribute_as_received(node_path)[source]

Mark an attribute as received and update dev state to RUNNING if all are received.

Parameters:

opcua_node_path (str) – The path to the OPC UA node which has been received

Return type:

None

setup_subscriptions(scu, events_queue, add_attribute_for_opcua_node, node_trees_to_monitor)[source]

Get a monitored node based on its node path.

Parameters:

  • scu (SteeringControlUnit) – The SCULib

  • events_queue (Queue) – The queue for datachange events

  • add_attribute_for_opcua_node (Callable) – Callback to create a tango attribute required for monitoring an opcua node.

  • node_trees_to_monitor (list[str]) – The list of root nodes to monitor all children attributes.

Return type:

None