OPCUA Component Manager

A Component Manager for an OPCUA device.

class OPCUAComponentManager(ds_host, ds_port, ds_endpoint, ds_opcua_namespace, ds_unique_id, logger, node_trees_to_monitor, *args, create_tango_device_attribute=None, update_device_state=None, update_device_status=None, get_attr_type_and_quality=None, update_attr_quality=None, **kwargs)

Component manager for an OPCUA device.

__init__(ds_host, ds_port, ds_endpoint, ds_opcua_namespace, ds_unique_id, logger, node_trees_to_monitor, *args, create_tango_device_attribute=None, update_device_state=None, update_device_status=None, get_attr_type_and_quality=None, update_attr_quality=None, **kwargs)

Init the component manager.

You need to call start_communication to connect to the OPCUA server. This wll: - Set up the subscriptions - Create the attributes and commands that map to the OPCUA device - Start a thread that will consume the subscription events and update the component state. We only store what’s needed for LMC in the component state.

stop_communication will: - Disconnect from the OPCUA server - Stop the subscription thread and delete the Thread instance - Delete the SCU instance

Parameters:

• ds_host (int) – The host for the OPCUA server

• ds_host – The port for the OPCUA server

• ds_endpoint (str) – The OPCUA server connection endpoint

• ds_opcua_namespace (str) – The connection namespace

• ds_unique_id (str) – The unique ID for the DSC used for taking authority

• logger (logging.Logger) – logger

• node_trees_to_monitor (str) – String of comma separated node trees to monitor

• create_tango_device_attribute (Callable) – Callback to create an attribute on the Tango device

• update_device_state (Optional[Callable]) – Callback to update the Tango device DevState

add_attribute_for_opcua_node(opcua_node_path, dtype)

Set up a tango attribute required for monitoring an opcua node.

Parameters:

• opcua_node_path (str) – The path to the OPC UA node to be monitored

• dtype (type) – The python data type of the node to be monitored

Return type:

None

event_handler_cb(scu=None, sub_event=None)

React to the events from the subscriptions _or_ the connection attempt.

Sub event example: {‘name’: ‘Management.Status.DscState’, ‘node’: Node(NodeId(Identifier=6188, NamespaceIndex=2, NodeIdType=<NodeIdType.Numeric: 2>)), ‘value’: 1, ‘source_timestamp’: datetime.datetime(2024, 11, 1, 11, 14, 47, 148242, tzinfo=datetime.timezone.utc), ‘server_timestamp’: 1730459687.148249, ‘data’: DataChangeNotification <asyncua.common.subscription.SubscriptionItemData object at 0x11544a2c0>, MonitoredItemNotification(ClientHandle=201, Value=DataValue(Value=Variant(Value=1, VariantType=<VariantType.Int32: 6>, Dimensions=None, is_array=False),StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2024, 11, 1, 11, 14, 47, 148242, tzinfo=datetime.timezone.utc), ServerTimestamp=datetime.datetime(2024, 11, 1, 11, 14, 47, 148249, tzinfo=datetime.timezone.utc), SourcePicoseconds=None, ServerPicoseconds=None)))}}

Parameters:

• scu (SteeringControlUnit | None, optional) – SteeringControlUnit instance, defaults to None

• sub_event (OPCUA event, optional) – OPCUA event, defaults to None | dict

Return type:

None

execute_opcua_command(command_arguments, *args, **kwargs)

Convert the given list of string arguments and call the OPC-UA command.

Parameters:

command_path (str) – Name of the command as stored in the sculib ‘commands’ list.

Return type:

Tuple[TaskStatus, str]

get_current_tai_offset_with_manual_fallback()

Get the current TAIoffset as read from the DSC with a manual calculation fallback.

First try to read the offset from the DSC, and fallback to calculating the offset manually.

Returns:

TAI offset

Return type:

float

interlock_acknowledge(*args, **kwargs)

Call interlock_acknowledge on the DSC.

Return type:

Tuple[TaskStatus, str]

release_auth(*args, **kwargs)

Call the release authority on the DSC.

Return type:

Tuple[TaskStatus, str]

reset_axes_command(*args, **kwargs)

Call the reset axes on the DSC.

Return type:

Tuple[TaskStatus, str]

retake_authority(*args, **kwargs)

Call the retake authority on the DSC.

Return type:

Tuple[TaskStatus, str]

set_index_position(*args, **kwargs)

Call set_index_position on the DSC.

Return type:

Tuple[TaskStatus, str]

set_point_mode(*args, **kwargs)

Call set_point_mode on the DSC.

Return type:

Tuple[TaskStatus, str]

set_power_mode(*args, **kwargs)

Call set_power_mode on the DSC.

Return type:

Tuple[TaskStatus, str]

set_standby_mode(*args, **kwargs)

Call set_standby_mode on the DSC.

Return type:

Tuple[TaskStatus, str]

setup_subscriptions()

Set up the subscriptions.

Should be called after connection has been established.

DscState sub is used to react to loss of connectivity to the device.

Return type:

None

slew(*args, **kwargs)

Call slew on the DSC.

Return type:

Tuple[TaskStatus, str]

start_communicating()

Start communication with the OPCUA device.

Retry until successful in a thread so as not to block the Tango device

Return type:

None

stop_communicating()

Gracefully stop all communication, clean up resources, and reset the conn state.

Return type:

None

stow(*args, **kwargs)

Call stow on the DSC.

Return type:

Tuple[TaskStatus, str]

take_authority(*args, **kwargs)

Call the take authority on the DSC.

Return type:

Tuple[TaskStatus, str]

track(*args, **kwargs)

Call track on the DSC.

Return type:

Tuple[TaskStatus, str]

track_load_static_off(*args, **kwargs)

Call track_load_static_off on the DSC.

Return type:

Tuple[TaskStatus, str]

track_stop(*args, **kwargs)

Call track_stop on the DSC.

Return type:

Tuple[TaskStatus, str]

unstow(*args, **kwargs)

Call unstow on the DSC.

Return type:

Tuple[TaskStatus, str]

update_pointing_model_params(band, values)

Update the static pointing model parameters on the Dish Structure.

Return type:

tuple

update_threshold_levels(values)

Update the threshold level between actual and desire AZ/EL pointing error.

Return type:

tuple

main()

For local testing.

Return type:

None