# mypy: disable-error-code="attr-defined"
# mypy: disable-error-code="no-redef"
"""A Component Manager for an OPCUA device."""
import dataclasses
import json
import logging
import sys
import time
from importlib.resources import as_file
from queue import Empty, Queue
from shutil import copyfile
from threading import Event, Thread
from typing import Any, Callable, Optional, Tuple, Union
from ska_control_model import CommunicationStatus, HealthState, TaskStatus
from ska_mid_dish_steering_control import SteeringControlUnit
from ska_mid_dish_steering_control.sculib import DEFAULT_NODES_CACHE_PATH
from ska_tango_base.executor import TaskExecutorComponentManager
from tango import AttrQuality
from ska_mid_dish_ds_manager import release
from ska_mid_dish_ds_manager.command_manager import CommandManager
from ska_mid_dish_ds_manager.models.constants import (
DEFAULT_AZIMUTH_SPEED_DPS,
DEFAULT_ELEVATION_SPEED_DPS,
DEFAULT_PYTHON_TYPE_VALUES,
DEFAULT_TANGO_TYPE_VALUES,
DSC_IMMUTABLE_USER_ID_MAX_LEN,
DSC_MIN_POWER_LIMIT_KW,
DYNAMIC_NODES_ABS_CHANGE,
ERROR_STATUS_NODES,
LOGGED_COMPONENT_STATES,
OPCUA_NODE_CACHE,
TAI_OFFSET_OPCUA_PATH,
TANGO_VALUE_QUALITY_SKIP,
)
from ska_mid_dish_ds_manager.models.data_classes import (
DscBuildStateDataclass,
DsmBuildStateDataclass,
)
from ska_mid_dish_ds_manager.models.dish_enums import (
ApplicationStateType,
BandType,
DscCmdAuthType,
DscCtrlState,
TiltOnType,
TrackInterpolationMode,
TrackProgramMode,
)
from ska_mid_dish_ds_manager.subscription_manager import SubscriptionManager
from ska_mid_dish_ds_manager.utils import (
SubscriptionEvent,
convert_opcua_type_to_python_type,
generate_component_state_key_from_opcua_path,
generate_tango_name_from_opcua_path,
get_current_tai_timestamp,
safe_string_casting,
)
logging.getLogger("asyncua").setLevel(logging.WARN)
logging.getLogger("ska_mid_dish_steering_control").setLevel(logging.WARN)
logging.getLogger("ska-mid-ds-scu").setLevel(logging.WARN)
[docs]class OPCUAComponentManager(TaskExecutorComponentManager):
"""Component manager for an OPCUA device."""
[docs] def __init__(
self: "OPCUAComponentManager",
ds_host: str,
ds_port: int,
ds_endpoint: str,
ds_opcua_namespace: str,
ds_unique_id: str,
logger: logging.Logger,
node_trees_to_monitor: str,
*args: Any,
create_tango_device_attribute: Optional[Callable] = None,
update_device_state: Optional[Callable] = None,
update_device_status: Optional[Callable] = None,
get_attr_type_and_quality: Optional[Callable] = None,
update_attr_quality: Optional[Callable] = None,
**kwargs: Any,
) -> None:
"""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
:param ds_host: The host for the OPCUA server
:type ds_host: str
:param ds_host: The port for the OPCUA server
:type ds_host: int
:param ds_endpoint: The OPCUA server connection endpoint
:type ds_endpoint: str
:param ds_opcua_namespace: The connection namespace
:type ds_opcua_namespace: str
:param ds_unique_id: The unique ID for the DSC used for taking authority
:type ds_unique_id: str
:param logger: logger
:type logger: logging.Logger
:param node_trees_to_monitor: String of comma separated node trees to monitor
:type node_trees_to_monitor: str
:param create_tango_device_attribute: Callback to create an attribute on the Tango device
:type create_tango_device_attribute: Callable
:param update_device_state: Callback to update the Tango device DevState
:type create_tango_device_attribute: Callable
"""
self.ds_host = ds_host
self.ds_port = ds_port
self.ds_endpoint = ds_endpoint
self.ds_opcua_namespace = ds_opcua_namespace
self.logger = logger
if len(ds_unique_id) > DSC_IMMUTABLE_USER_ID_MAX_LEN:
ds_unique_id = ds_unique_id[:DSC_IMMUTABLE_USER_ID_MAX_LEN]
self.logger.warning(
"DSC immutable user ID is too long. Truncating to %d characters.",
DSC_IMMUTABLE_USER_ID_MAX_LEN,
)
self.ds_unique_id = ds_unique_id
self._create_tango_device_attribute = create_tango_device_attribute
self._node_trees_to_monitor = [
node_name.strip() for node_name in node_trees_to_monitor.split(",")
]
self.tango_opcua_attr_mapper: dict = {}
# Event handling
self._consuming_thread: Thread | None = None
self._stop_consumer_event: Event = Event()
self._events_queue: Queue = Queue()
# Connection handling
self._connect_thread: Thread | None = None
self._stop_connect_event = Event()
self._connect_recovery_intiated = False
self._copy_node_cache_to_tmp_rw_dir()
self.scu: Union[SteeringControlUnit, None] = None
self.subscription_manager = SubscriptionManager(self.logger, update_device_state)
self.command_manager: CommandManager = CommandManager(
logger=self.logger,
immutable_user_id=self.ds_unique_id,
scu=self.scu,
submit_task=self.submit_task,
update_component_state=self._update_component_state,
command_progress_callback=update_device_status,
)
self._get_attr_type_and_quality = get_attr_type_and_quality
self._update_attr_quality = update_attr_quality
init_component_state = dict(
# Subscribed on OPCUA
powerstate=None,
dscstate=None,
dsccmdauth=DscCmdAuthType.NO_AUTHORITY,
pointingstate=None,
operatingmode=None,
indexerposition=None,
desiredpointingaz=None,
desiredpointingel=None,
achievedpointing=None,
currentpointing=[0, 0, 0, 0, 0, 0, 0],
currentmode=None,
applicationstate=ApplicationStateType.Unknown,
healthstate=HealthState.OK,
actptactindex=0,
actptendindex=0,
achievedtargetlock=None,
actstaticoffsetvaluexel=None,
actstaticoffsetvalueel=None,
hhpconnected=None,
band0pointingmodelparams=[],
band1pointingmodelparams=[],
band2pointingmodelparams=[],
band3pointingmodelparams=[],
band4pointingmodelparams=[],
band5apointingmodelparams=[],
band5bpointingmodelparams=[],
band6pointingmodelparams=[],
configuretargetlock=[0.0, 0.0],
onsourcethreshold=0.0,
thresholdtimeperiod=0.0,
staticoninput=False,
tiltoninput=TiltOnType.OFF,
ambtoninput=False,
# Read off the OPCUA as needed
buildstate="",
# State managed separately from OPCUA
trackinterpolationmode=TrackInterpolationMode.SPLINE,
azimuthspeed=DEFAULT_AZIMUTH_SPEED_DPS,
elevationspeed=DEFAULT_ELEVATION_SPEED_DPS,
dscpowerlimitkw=DSC_MIN_POWER_LIMIT_KW,
trackprogrammode=TrackProgramMode.TABLE,
lastcommandupdate=[],
lastcommandinvoked=("0.0", ""),
dscctrlstate=DscCtrlState.NO_AUTHORITY,
displayeddiagnosis="Not updated from OPCUA device",
)
error_status_defaults = {name.lower(): False for name in ERROR_STATUS_NODES}
init_component_state.update(error_status_defaults)
super().__init__(
logger,
*args,
**init_component_state,
**kwargs,
)
# Trigger initial Astropy import to avoid first-call latency later
get_current_tai_timestamp()
def _update_component_state(self, **kwargs: Any) -> None:
"""Log the new component state and handle quality of values."""
for key, new_value in kwargs.items():
component_state_name = key.lower()
# Log a subset of the attribute changes
if component_state_name in LOGGED_COMPONENT_STATES:
self.logger.debug("Updating ds manager component state with [%s]", kwargs)
# Handle value updates and quality for a subset of Tango attributes
if component_state_name not in TANGO_VALUE_QUALITY_SKIP:
self._handle_tango_value_and_quality(component_state_name, new_value, kwargs)
super()._update_component_state(**kwargs)
def _handle_tango_value_and_quality(
self, component_state_name: str, new_value: Any, kwargs: dict
) -> None:
"""Apply Tango-specific rules to component state updates.
This method adapts incoming component state values for Tango attributes by:
- Converting None values to Tango-safe default values defined in DEFAULT_TANGO_TYPE_VALUES.
- Managing attribute quality transitions between ATTR_VALID and ATTR_INVALID based on the
presence or absence of a valid value.
:param component_state_name: The name of the Tango attribute
:type component_state_name: str
:param new_value: The new value for the attribute
:type new_value: Any
:param kwargs: The dictionary of component state updates passed to
``_update_component_state``. This method may modify its contents
in-place to ensure Tango-compatible values are propagated.
:type kwargs: dict
"""
if self._get_attr_type_and_quality is None:
return
tango_dtype, current_quality = self._get_attr_type_and_quality(component_state_name)
if tango_dtype is None:
return
if new_value is None:
if current_quality != AttrQuality.ATTR_INVALID and self._update_attr_quality:
self._update_attr_quality(
component_state_name, AttrQuality.ATTR_INVALID, send_event=True
)
kwargs[component_state_name] = DEFAULT_TANGO_TYPE_VALUES.get(tango_dtype, "")
elif current_quality != AttrQuality.ATTR_VALID and self._update_attr_quality:
# Pushing an event with ATTR_VALID causes a segmentation fault
# TODO: (12/02/2026 Devon) Investigate what is causing this behavior, a base Tango
# Device doesn't cause this.
self._update_attr_quality(
component_state_name, AttrQuality.ATTR_VALID, send_event=False
)
# Push an event manually so that there is an event if we go from a None value
# (ATTR_INVALID) (which we then set a default value in the component state) to
# ATTR_VALID with the same value as the default. This is needed since tango base wont
# push it because of this check "if self._component_state[key] != value:"
self._component_state_callback(**{component_state_name: new_value})
[docs] def add_attribute_for_opcua_node(self, opcua_node_path: str, dtype: type) -> None:
"""Set up a tango attribute required for monitoring an opcua node.
:param opcua_node_path: The path to the OPC UA node to be monitored
:type opcua_node_path: str
:param dtype: The python data type of the node to be monitored
:type dtype: type
"""
if not self.scu:
self.logger.warning("Could not add attribute. SCU instance is None.")
return
if self._create_tango_device_attribute:
tango_attr_name = generate_tango_name_from_opcua_path(opcua_node_path)
component_state_name = generate_component_state_key_from_opcua_path(opcua_node_path)
with self._component_state_lock:
self._component_state[component_state_name] = DEFAULT_PYTHON_TYPE_VALUES.get(
dtype, ""
)
# pylint: disable=unused-argument
def get_component_state_value(*args: tuple, **kwargs: dict) -> Any:
"""Get the component state value associated with the opcua node."""
attr_quality = AttrQuality.ATTR_VALID
if self._get_attr_type_and_quality:
_, attr_quality = self._get_attr_type_and_quality(component_state_name)
return self._component_state[component_state_name], time.time(), attr_quality
self._create_tango_device_attribute(
attr_name=tango_attr_name,
comp_state_name=component_state_name,
dtype=dtype,
doc=f"Read the node {opcua_node_path} on the dish structure controller",
fget=get_component_state_value,
fset=None,
abs_change=DYNAMIC_NODES_ABS_CHANGE.get(opcua_node_path, None),
)
[docs] def setup_subscriptions(self) -> None:
"""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.
"""
if not self.scu:
raise RuntimeError("SCU has not been initialised")
self.subscription_manager.setup_subscriptions(
self.scu,
self._events_queue,
self.add_attribute_for_opcua_node,
self._node_trees_to_monitor,
)
# Update the communication state to established
self._update_communication_state(CommunicationStatus.ESTABLISHED)
self._connect_recovery_intiated = False
def _lost_connection(self, *args: Any, **kwargs: Any) -> None:
"""Handle a detected connection loss.
Stop communication and attempt to reconnect.
"""
self.logger.info("Lost connection. args: %s, kwargs: %s", args, kwargs)
if not self._connect_recovery_intiated:
self._connect_recovery_intiated = True
self.stop_communicating()
self.start_communicating()
[docs] def event_handler_cb(
self, scu: Union[SteeringControlUnit, None] = None, sub_event: dict | None = None
) -> 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)))}}
:param scu: SteeringControlUnit instance, defaults to None
:type scu: SteeringControlUnit | None, optional
:param sub_event: OPCUA event, defaults to None | dict
:type sub_event: OPCUA event, optional
"""
if sub_event:
# Convert the event to `SubscriptionEvent` for better ergonomics
# May remove in future to do less processing per event
subs_event = SubscriptionEvent(**sub_event)
mon_node = self.subscription_manager.get_monitored_node(subs_event.name)
if mon_node is None:
self.logger.warning(
"Got event callback for node not in monitored nodes. %s.", subs_event.name
)
return
# Convert the val if needed. E.g to an Enum.
if mon_node.convert_to_component_state_val:
converted_value = mon_node.convert_to_component_state_val(subs_event)
component_state_name = mon_node.component_state_name
self._update_component_state(**{component_state_name: converted_value})
# Do any additional calculations and update
for state_calc in mon_node.calculate_updates:
update_vals = state_calc(subs_event, self.component_state)
if update_vals:
self._update_component_state(**update_vals)
self.subscription_manager.mark_attribute_as_received(mon_node.node_path)
if scu:
self.scu = scu
self.command_manager.scu = self.scu
self.setup_subscriptions()
self._update_build_state_info()
def _attempt_connection(self) -> None:
"""Attempt connection to the OPCUA until successful."""
scu = SteeringControlUnit(
host=self.ds_host,
port=self.ds_port,
endpoint=self.ds_endpoint,
namespace=self.ds_opcua_namespace,
use_nodes_cache=True,
connection_lost_callback=self._lost_connection,
)
while (not self._stop_connect_event.is_set()) and (not scu.is_connected()):
try:
scu.connect_and_setup()
self._events_queue.put(scu)
break
except Exception:
# If anything goes wrong we need to continue
scu.disconnect()
self.logger.error(
(
"Could not connect to OPCUA on host [%s] port [%s]"
"endpoint [%s] namespace [%s], retrying in 1 sec"
),
self.ds_host,
self.ds_port,
self.ds_endpoint,
self.ds_opcua_namespace,
)
self._stop_connect_event.wait(timeout=1)
def _stop_connection_thread(self) -> None:
"""Stop the connection attempt thread if in progress."""
if self._connect_thread and self._connect_thread.is_alive():
self._stop_connect_event.set()
self._connect_thread.join()
self._stop_connect_event.clear()
self._connect_thread = None
def _stop_consumer_thread(self) -> None:
"""Stop the consuming thread if in progress."""
if self._consuming_thread and self._consuming_thread.is_alive():
self._stop_consumer_event.set()
self._consuming_thread.join()
self._stop_consumer_event.clear()
self._consuming_thread = None
[docs] def start_communicating(self: "OPCUAComponentManager") -> None:
"""Start communication with the OPCUA device.
Retry until successful in a thread so as not to block the Tango device
"""
self._update_communication_state(CommunicationStatus.NOT_ESTABLISHED)
self._stop_connection_thread()
# Start the connection attempt thread
self._connect_thread = Thread(target=self._attempt_connection)
self._connect_thread.start()
self._stop_consumer_thread()
# Start the consuming thread if in progress
self._consuming_thread = Thread(target=self._consume_events)
self._consuming_thread.start()
[docs] def stop_communicating(self: "OPCUAComponentManager") -> None:
"""Gracefully stop all communication, clean up resources, and reset the conn state."""
self._update_communication_state(CommunicationStatus.DISABLED)
self.abort_tasks()
self._stop_connection_thread()
self._stop_consumer_thread()
# Clean up SCU objects if left in the queue
while not self._events_queue.empty():
try:
item = self._events_queue.get_nowait()
if isinstance(item, SteeringControlUnit):
item.disconnect_and_cleanup()
except Empty:
break
if self.scu:
self.scu.disconnect_and_cleanup()
self.scu = None
def _consume_events(self) -> None:
"""Consume thread events.
These events are generated by subscriptions as well the connection operation.
The callback is called with the appropriate method set.
"""
# Note: This is NOT run with EnsureOmniThread() as it causes errors trying to acquire
# the serialisation monitor lock when trying to add tango attributes dynamically.
while not self._stop_consumer_event.is_set():
try:
item = self._events_queue.get(timeout=0.5)
if isinstance(item, SteeringControlUnit):
self.event_handler_cb(scu=item)
else:
self.event_handler_cb(sub_event=item)
except Empty:
self._stop_consumer_event.wait(0.5)
except Exception:
# If anything goes wrong we need to continue
self.logger.exception("Handling the event failed")
self._stop_consumer_event.wait(0.5)
def _update_build_state_info(self) -> None:
"""Update the combined Dish manager and DSC buildState structure."""
if not self.scu:
self.logger.warning("Could not update build state info. SCU instance is None.")
return
try:
# Get buildState data nodes
dish_id = self.scu.attributes["Management.NamePlate.DishId"].value
dsc_ser_num = self.scu.attributes["Management.NamePlate.DishStructureSerialNo"].value
dsc_software_ver = self.scu.attributes["Management.NamePlate.DscSoftwareVersion"].value
dsc_icd_ver = self.scu.attributes["Management.NamePlate.IcdVersion"].value
# Construct DSC build state info struct
dsc_build_state = DscBuildStateDataclass(
"Dish Structure Controller",
dish_id,
dsc_ser_num,
dsc_software_ver,
dsc_icd_ver,
)
except KeyError:
self.logger.exception("No build state info")
dsc_build_state = DscBuildStateDataclass(
"Failed to retrieve buildState data for Dish structure controller",
"",
"",
"",
"",
)
# Construct dish manager version info
dm_build_state = DsmBuildStateDataclass(
release.name,
release.get_ds_manager_release_version(),
f"opc.tcp://{self.ds_host}:{self.ds_port}{self.ds_endpoint}",
dsc_build_state,
)
dsc_build_state_json = json.dumps(dataclasses.asdict(dm_build_state), indent=4)
self._update_component_state(buildstate=dsc_build_state_json)
[docs] def set_standby_mode(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call set_standby_mode on the DSC."""
return self.command_manager.set_standby_mode(self._component_state, *args, **kwargs)
[docs] def set_point_mode(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call set_point_mode on the DSC."""
return self.command_manager.set_point_mode(self._component_state, *args, **kwargs)
[docs] def set_index_position(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call set_index_position on the DSC."""
return self.command_manager.set_index_position(self._component_state, *args, **kwargs)
[docs] def slew(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call slew on the DSC."""
return self.command_manager.slew(self._component_state, *args, **kwargs)
[docs] def take_authority(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call the take authority on the DSC."""
return self.command_manager.take_authority(*args, **kwargs)
[docs] def retake_authority(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call the retake authority on the DSC."""
return self.command_manager.retake_authority(*args, **kwargs)
[docs] def reset_axes_command(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call the reset axes on the DSC."""
return self.command_manager.reset_axes_command(*args, **kwargs)
[docs] def track(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call track on the DSC."""
return self.command_manager.track(*args, **kwargs)
[docs] def track_stop(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call track_stop on the DSC."""
return self.command_manager.track_stop(*args, **kwargs)
[docs] def track_load_static_off(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call track_load_static_off on the DSC."""
return self.command_manager.track_load_static_off(*args, **kwargs)
[docs] def set_power_mode(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call set_power_mode on the DSC."""
return self.command_manager.set_power_mode(self._update_component_state, *args, **kwargs)
[docs] def stow(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call stow on the DSC."""
return self.command_manager.stow(*args, **kwargs)
[docs] def execute_opcua_command(
self, command_arguments: list[str], *args: Any, **kwargs: Any
) -> Tuple[TaskStatus, str]:
"""Convert the given list of string arguments and call the OPC-UA command.
:param command_path: Name of the command as stored in the sculib 'commands' list.
:type command_path: str
"""
if not self.scu:
return TaskStatus.REJECTED, "Device is not connected to DSC."
if len(command_arguments) < 1:
return (
TaskStatus.REJECTED,
"Tango command argument requires at least 1 value. See command docstring for "
"more details.",
)
command_path = command_arguments[0]
command_args = command_arguments[1:]
if command_path not in self.scu.commands:
return TaskStatus.REJECTED, "Requested command is not available."
# scu.commands contains callables, scu.nodes contains the opcua nodes
command_node, _ = self.scu.nodes[command_path]
required_command_arglist = self.scu.get_command_arguments(command_node)
# Don't consider sessionID argument for commands.
# Note: TakeAuth doesn't require a sessionID.
if "takeauth" not in command_path.lower():
required_command_arglist = required_command_arglist[1:] # remove sessionID
num_opcua_args = len(required_command_arglist)
if len(command_args) != num_opcua_args:
return (
TaskStatus.REJECTED,
f"Requested command requires {num_opcua_args} argument(s), but {len(command_args)}"
" given.",
)
# Convert the given args which are strings into the python dtype required for the command
converted_args = []
for (ua_arg_name, ua_arg_dtype), given_arg in zip(required_command_arglist, command_args):
python_dtype = convert_opcua_type_to_python_type(ua_arg_dtype)
try:
converted_arg = safe_string_casting(given_arg, python_dtype)
converted_args.append(converted_arg)
except ValueError:
return (
TaskStatus.REJECTED,
f"Given argument ('{given_arg}') could not be cast to required type"
f" ({ua_arg_dtype}) for parameter ({ua_arg_name})",
)
return self.command_manager.execute_opcua_command(
command_path, converted_args, *args, **kwargs
)
[docs] def unstow(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call unstow on the DSC."""
return self.command_manager.unstow(self._component_state, *args, **kwargs)
[docs] def interlock_acknowledge(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call interlock_acknowledge on the DSC."""
return self.command_manager.interlock_acknowledge(*args, **kwargs)
[docs] def update_pointing_model_params(
self,
band: BandType,
values: list[float],
) -> tuple:
"""Update the static pointing model parameters on the Dish Structure."""
return self.command_manager.update_pointing_model_params(
static_on=self._component_state.get("staticoninput", False),
tilt_on=self._component_state.get("tiltoninput", TiltOnType.OFF),
amb_temp_on=self._component_state.get("ambtoninput", False),
band=band,
values=values,
)
[docs] def update_threshold_levels(self, values: list[float]) -> tuple:
"""Update the threshold level between actual and desire AZ/EL pointing error."""
return self.command_manager.update_threshold_levels(values)
[docs] def get_current_tai_offset_with_manual_fallback(self) -> float:
"""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.
:return: TAI offset
:rtype: float
"""
# Try read value from DSC
if self.scu:
if TAI_OFFSET_OPCUA_PATH in self.scu.attributes:
dsc_value = self.scu.attributes[TAI_OFFSET_OPCUA_PATH].value
if dsc_value:
return dsc_value
# Calculate manually
return get_current_tai_timestamp()
def _copy_node_cache_to_tmp_rw_dir(self) -> None:
"""Copy node cache file to writable location."""
try:
with as_file(OPCUA_NODE_CACHE) as src_path:
copyfile(src_path, DEFAULT_NODES_CACHE_PATH)
except Exception as exc:
self.logger.error(f"Failed to copy node cache file: {exc}")
[docs] def release_auth(self, *args: Any, **kwargs: Any) -> Tuple[TaskStatus, str]:
"""Call the release authority on the DSC."""
return self.command_manager.release_auth(*args, **kwargs)
[docs]def main() -> None:
"""For local testing."""
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler(sys.stdout))
IMMUTABLE_USER_ID = "test_user_id"
cm = OPCUAComponentManager(
"127.0.0.1",
4840,
"/dish-structure/server/",
"http://skao.int/DS_ICD/",
IMMUTABLE_USER_ID,
logger,
"",
)
cm.start_communicating()
time.sleep(2)
cm.stop_communicating()
if __name__ == "__main__":
main()