# mypy: disable-error-code="union-attr"
# mypy: disable-error-code="no-redef"
# mypy: disable-error-code="index"
# flake8: noqa:C901
"""Module that contains CommandManager where commands to DSC are managed."""
import enum
import logging
import sys
import threading
from time import perf_counter
from typing import Any, Callable, Optional, Tuple, Union
from asyncua import ua
from asyncua.ua.uaerrors import UaError
from ska_control_model import ResultCode, TaskStatus
from ska_mid_dish_steering_control import CmdReturn, SteeringControlUnit
from ska_mid_dish_steering_control import ResultCode as SCUResultCode
from ska_tango_base.executor import TaskExecutorComponentManager
from ska_mid_dish_ds_manager.models.constants import (
AZ_AXIS_STATE_OPCUA_PATH,
BAND_POINTING_MODEL_PARAMS_LENGTH,
DSC_CMD_AUTH_PATH,
DSC_MAX_POWER_LIMIT_KW,
DSC_MIN_POWER_LIMIT_KW,
DSC_STATE_OPCUA_PATH,
EL_AXIS_STATE_OPCUA_PATH,
HORN_ACTIVE_PATH,
HORN_CHECK_MAX_TRIES,
HORN_CHECK_PERIOD_SEC,
IMMUTABLE_ID_PATH,
LOW_POWER_ACTIVE_OPCUA_PATH,
OBFUSCATED_ID_HASH_LEN_TRUNCATED,
TRACK_START_LEAD_TIME,
)
from ska_mid_dish_ds_manager.models.dish_enums import (
AxisStateType,
BandType,
DscCmdAuthType,
DSCState,
DSOperatingMode,
DSPowerState,
IndexerPosition,
TiltOnType,
TrackInterpolationMode,
TrackTableLoadMode,
)
from ska_mid_dish_ds_manager.utils import (
MAX_AZIMUTH,
MAX_ELEVATION,
MIN_AZIMUTH,
MIN_ELEVATION,
are_points_within_dish_limits,
is_point_within_dish_limits,
)
[docs]def update_task_status(task_callback: Optional[Callable], **task_statuses: Any) -> None:
"""Wraps the task callback."""
if task_callback:
task_callback(**task_statuses)
[docs]class CommandNotAuthorized(UaError):
"""Exception for incorrect command authorization."""
[docs]def track_updates_decorator(func: Callable) -> Any:
"""Decorator to track updates with the DSC."""
def track_updates_wrapper(self: "CommandManager", *args: Any, **kwargs: Any) -> Any:
"""Wrapper function to track updates."""
self.updates_list = []
self.updates_list.append(f"command: {func.__name__}")
result = func(self, *args, **kwargs)
if self.update_component_state:
self.update_component_state(lastcommandupdate=self.updates_list)
return result
return track_updates_wrapper
OK_CODES = {
SCUResultCode.COMMAND_DONE,
SCUResultCode.COMMAND_ACTIVATED,
}
[docs]class CommandManager:
"""Wrap all the commands to DSC."""
[docs] def __init__(
self,
logger: logging.Logger,
immutable_user_id: str,
scu: Union[SteeringControlUnit, None],
submit_task: Callable,
update_component_state: Callable,
command_progress_callback: Optional[Callable] = None,
) -> None:
"""Wrap all the commands to DSC.
:param logger: logger
:type logger: logging.Logger
:param immutable_user_id: The immutable user ID for the DSC
:type immutable_user_id: str
:param scu: SteeringControlUnit instance, None if it is not established yet.
:type scu: SteeringControlUnit
:param submit_task: Submit a task to the task executor
:type submit_task: Callable
"""
self.logger = logger
self._immutable_user_id = immutable_user_id
self.scu = scu
self.submit_task = submit_task
self.update_component_state: Optional[Callable[..., None]] = update_component_state
self.updates_list: list[str,] = []
self._command_progress_callback = command_progress_callback
# --------------
# Helper methods
# --------------
def _report_task_progress(self, progress_msg: str) -> None:
"""Wraps the progress callback to update device status."""
if self._command_progress_callback:
self._command_progress_callback(progress_msg)
def _reformat_track_load_table_log_msg_args(
self,
load_mode: int,
sequence_length: int,
timestamp_data: list[float],
az_data: list[float],
el_data: list[float],
) -> str:
"""Truncate TrackLoadTable command args sequences.
This is done by removing the padded zeroes in the supplied tracktable
to be shown in the log message.
:param load_mode: The tracktable load mode value.
:type load_mode: int
:param sequence_length: The number of valid data entries in the seqeunces.
:type sequence_length: int
:param timestamp_data: List of tracktable timestamp values.
:type timestamp_data: list[float]
:param az_data: List of tracktable azimuth values.
:type az_data: list[float]
:param el_data: List of tracktable elevation values.
:type el_data: list[float]
:return: Formatted string with truncated arguments and zero padding count.
:rtype: str
"""
zero_padding_count = len(timestamp_data) - sequence_length
truncated_ts_data = timestamp_data[:sequence_length]
truncated_az_data = az_data[:sequence_length]
truncated_elev_data = el_data[:sequence_length]
truncated_args = (
f"[{load_mode}, {sequence_length}, {truncated_ts_data},"
f"{truncated_az_data}, {truncated_elev_data}]. Track table data "
f"padded with [{zero_padding_count}] zeroes"
)
return truncated_args
def _call_opc_ua_method(
self,
node_path: str,
command_name: str,
*args: Any,
task_abort_event: Optional[threading.Event] = None,
wait_for_horn: bool = True,
) -> CmdReturn: # type: ignore
"""Execute a OPCUA method.
:param node_path: The OPCUA node path
:type node_path: str
:param command_name: The originating command name
:type command_name: str
:param args: The arguments to pass to the command
:type args: Any
:param task_abort_event: Optional event to wait for the horn to go off
:type task_abort_event: threading.Event, optional
:raises RuntimeError: Something went wrong
:raises CommandNotAuthorized: Could not get auth
:return: The result
:rtype: CmdReturn
"""
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
try:
if command_name != "TrackLoadTable":
task_progress_log = (
f"_call_opc_ua_method for [{node_path}] called with "
f"args [{args}] from [{command_name}] command"
)
else:
task_progress_log = (
f"_call_opc_ua_method for [{node_path}] called with args "
f"[{self._reformat_track_load_table_log_msg_args(*args)}] from "
f"[{command_name}] command"
)
self._report_task_progress(task_progress_log)
self.updates_list.append(task_progress_log)
dsccmdauth_value = self.scu.attributes.get(DSC_CMD_AUTH_PATH)
if dsccmdauth_value is None:
self.updates_list.append(f"attribute read: {DSC_CMD_AUTH_PATH}, result: (None)")
raise RuntimeError(f"Attribute {DSC_CMD_AUTH_PATH} not found in SCU attributes")
# Sometimes dsccmdauth_value.value is None, defaulting to NO_AUTHORITY in that case
if dsccmdauth_value.value is None:
self.logger.warning("dsccmdauth_value.value is None, when it should be an int")
dsccmdauth = DscCmdAuthType.NO_AUTHORITY
else:
dsccmdauth = DscCmdAuthType(dsccmdauth_value.value)
self.updates_list.append(
f"attribute read: {DSC_CMD_AUTH_PATH}, result: {dsccmdauth.name}:({dsccmdauth})"
)
# Handle case where the command authority is held by a higher priority client
if dsccmdauth in [DscCmdAuthType.HHP, DscCmdAuthType.EGUI]:
status_msg = (
"Command not executed because DscCmdAuthority is held by a "
f"higher priority client, {dsccmdauth.name}."
)
raise CommandNotAuthorized(status_msg)
# Grab authority if LMC does not have it
if dsccmdauth == DscCmdAuthType.NO_AUTHORITY:
dsccmdauth = self._take_authority(task_abort_event, wait_for_horn=wait_for_horn)
if dsccmdauth == DscCmdAuthType.LMC:
command_result = self._execute_scu_command(node_path, *args)
if command_result[0] == SCUResultCode.NO_CMD_AUTH:
self.logger.warning(
"Received NO_CMD_AUTH from SCU when trying to execute command. "
"Attempting to re-take authority."
)
dsccmdauth = self._retake_authority(
task_abort_event, wait_for_horn=wait_for_horn
)
command_result = self._execute_scu_command(node_path, *args)
if command_result[0] not in OK_CODES:
status_msg = (
f"Command [{node_path}] failed with result: {command_result[0].name}."
)
raise RuntimeError(status_msg)
else:
return command_result
# Future proofing for other authority types
status_msg = f"Unknown DscCmdAuthority type [{dsccmdauth.name}]."
raise RuntimeError(status_msg)
# Catch any errors and raise RuntimeError since only RuntimeError is handled higher up
except RuntimeError:
raise
except Exception as exc:
self.logger.exception(exc)
raise RuntimeError(f"Caught unexpected error calling OPC UA method, {exc}") from exc
def _execute_scu_command(self, node_path: str, *args: Any) -> CmdReturn:
command_start_time = perf_counter()
command_result = self.scu.commands[node_path](*args)
command_end_time = perf_counter()
self.updates_list.append(
f"opcua command timing: [{node_path}] executed in"
f" [{(command_end_time - command_start_time):.2f}] seconds"
)
self.updates_list.append(f"opcua result: {command_result[0].name}:({command_result[0]})")
return command_result
# ----------------
# command handlers
# ----------------
[docs] def set_standby_mode(
self,
component_state: dict,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to STANDBY mode.
:param component_state: the current component state
:type component_state: dict
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._set_standby_mode,
args=[
component_state,
],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _set_standby_mode(
self,
component_state: dict,
task_abort_event: Any,
task_callback: Optional[Callable] = None,
) -> None:
"""Transition the dish to STANDBY mode.
:param component_state: the current component state
:type component_state: dict
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called SetStandbyMode")
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
current_power_limit = component_state.get("dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW)
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
self._report_task_progress(
f"Low Power Mode called using DSC Power Limit: {current_power_limit}kW"
)
try:
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
self.updates_list.append(
f"attribute read: {DSC_STATE_OPCUA_PATH}, "
f"result: {DSCState(dsc_state).name}:({dsc_state})"
)
# Use the scu attribute so that when we call '.value' it reads directly from the DSC to
# get the latest value
low_power_active = self.scu.attributes[LOW_POWER_ACTIVE_OPCUA_PATH]
self.updates_list.append(
f"attribute read: {LOW_POWER_ACTIVE_OPCUA_PATH}, "
f"result: ({bool(low_power_active.value)})"
)
az_axis_state = self.scu.attributes[AZ_AXIS_STATE_OPCUA_PATH]
self.updates_list.append(
f"attribute read: {AZ_AXIS_STATE_OPCUA_PATH}, result: "
f"{AxisStateType(az_axis_state.value).name}:({az_axis_state.value})"
)
el_axis_state = self.scu.attributes[EL_AXIS_STATE_OPCUA_PATH]
self.updates_list.append(
f"attribute read: {EL_AXIS_STATE_OPCUA_PATH}, result: "
f"{AxisStateType(el_axis_state.value).name}:({el_axis_state.value})"
)
# Handle transition from DSC STOW state
if dsc_state == ua.DscStateType.Stowed.value:
# The dish needs to be in full power mode in order to call Stow(False)
if low_power_active.value:
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"SetStandbyMode",
False,
current_power_limit,
task_abort_event=task_abort_event,
)
self._call_opc_ua_method(
"Management.Commands.Stow",
"SetStandbyMode",
False,
task_abort_event=task_abort_event,
)
az_axis_state = self.scu.attributes[AZ_AXIS_STATE_OPCUA_PATH].value
self.updates_list.append(
f"attribute read: {AZ_AXIS_STATE_OPCUA_PATH}, result: "
f"{AxisStateType(az_axis_state).name}:({az_axis_state})"
)
el_axis_state = self.scu.attributes[EL_AXIS_STATE_OPCUA_PATH].value
self.updates_list.append(
f"attribute read: {EL_AXIS_STATE_OPCUA_PATH}, result: "
f"{AxisStateType(el_axis_state).name}:({el_axis_state})"
)
# Set the DSC power state to LOW_POWER
if not low_power_active.value:
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"SetStandbyMode",
True,
current_power_limit,
task_abort_event=task_abort_event,
)
# Finally, deactivate all axis, if not already deactivated
if (az_axis_state != ua.AxisStateType.Standby) or (
el_axis_state != ua.AxisStateType.Standby
):
self._call_opc_ua_method(
"Management.Commands.DeActivate",
"SetStandbyMode",
ua.UInt16(ua.AxisSelectType.AzEl),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="SetStandbyMode Called"
)
[docs] def reset_axes_command(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Reset the errors on dish structure axes.
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._reset_axes_command,
args=[],
task_callback=task_callback,
)
return status, response
[docs] def take_authority(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Take authority on dish structure.
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._take_authority_task,
args=[],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _take_authority_task(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Take authority on dish structure.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
dsccmdauth = self._take_authority(task_abort_event)
update_task_status(
task_callback,
status=TaskStatus.COMPLETED,
result=f"TakeAuthority succeeded. New authority: {dsccmdauth.name}",
)
return
except (CommandNotAuthorized, RuntimeError) as err:
self.logger.error(err)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(err))
return
[docs] def retake_authority(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Retake authority on dish structure.
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._retake_authority_task,
args=[],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _retake_authority_task(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Retake authority on dish structure.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
# Check if dscCmdAuth is LMC first before trying to retake authority.
dsccmdauth_value = self.scu.attributes.get(DSC_CMD_AUTH_PATH)
dsccmdauth = DscCmdAuthType(dsccmdauth_value.value)
if dsccmdauth != DscCmdAuthType.LMC:
status_msg = "Can only retake authority when LMC has authority. "
status_msg += f"Current authority: [{dsccmdauth.name}]."
update_task_status(task_callback, status=TaskStatus.FAILED, result=status_msg)
return
dsccmdauth = self._retake_authority(task_abort_event)
update_task_status(
task_callback,
status=TaskStatus.COMPLETED,
result=f"RetakeAuthority succeeded. {dsccmdauth.name} has authority.",
)
return
except (CommandNotAuthorized, RuntimeError) as err:
self.logger.error(err)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(err))
return
@track_updates_decorator
def _reset_axes_command(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Reset the errors on dish structure axes.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called ResetAxesCommand")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
"Management.Commands.Reset",
"ResetAxes",
ua.UInt16(1),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(task_callback, status=TaskStatus.COMPLETED, result="ResetAxes Called")
[docs] def set_power_mode(
self,
component_state_callback: Callable,
values: list[float],
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to SetPowerMode.
:param values: (0.0 or 1.0) and limit(kW)] with 1.0 representing True
:type values: list(float)
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
if len(values) != 2:
return (
TaskStatus.REJECTED,
f"Expected 2 arguments [(0.0 or 1.0) and limit(kW)] with 1.0 representing True "
f"and 0.0 representing False. Got {len(values)} arguments.",
)
# Check if the boolean number is 0.0 or 1.0
boolean_value = values[0]
if boolean_value not in [0.0, 1.0]:
return (
TaskStatus.REJECTED,
f"Invalid boolean number: {boolean_value}. Expected 0.0 or 1.0.",
)
boolean_arg = bool(values[0])
limit = values[1]
if limit > DSC_MAX_POWER_LIMIT_KW or limit < DSC_MIN_POWER_LIMIT_KW:
return (
TaskStatus.REJECTED,
f"Invalid dsc power limit: {limit}. Expected a value in this range"
f" [{DSC_MIN_POWER_LIMIT_KW}, {DSC_MAX_POWER_LIMIT_KW}].",
)
status, response = self.submit_task(
self._set_power_mode,
args=[component_state_callback, boolean_arg, limit],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _set_power_mode(
self,
component_state_callback: Callable,
boolean_arg: bool,
limit: float,
task_abort_event: Any,
task_callback: Optional[Callable] = None,
) -> None:
"""Transition the dish to SetPowerMode.
:param boolean_arg: Low power active or not
:type boolean_arg: bool
:param limit: The power limit in Kw
:type limit: float
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called SetPowerMode with args [%s, %s]", boolean_arg, limit)
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
component_state_callback(dscpowerlimitkw=limit)
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"SetPowerMode",
boolean_arg,
limit,
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="SetPowerMode Called"
)
[docs] def set_point_mode(
self,
component_state: dict,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to POINT mode.
:param component_state: the current component state
:type component_state: dict
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
allowed_init_states_to_point_mode = [
ua.DscStateType.Standstill.value,
ua.DscStateType.Standby.value,
ua.DscStateType.Stowed.value,
ua.DscStateType.Locked.value,
ua.DscStateType.Locked_Stowed.value,
]
if dsc_state not in allowed_init_states_to_point_mode:
current_op_mode = component_state.get("operatingmode", DSOperatingMode.UNKNOWN)
return (
TaskStatus.REJECTED,
"Request to call SetPointMode rejected for current"
f" DS operating mode: {DSOperatingMode(current_op_mode).name},"
f" DSCState: {DSCState(dsc_state).name}."
" SetPointMode call allowed in DS operating modes STOW, STANDBY,"
" or LOCKED",
)
status, response = self.submit_task(
self._set_point_mode,
args=[
component_state,
],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _set_point_mode(
self,
component_state: dict,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Transition the dish to POINT mode.
:param component_state: the current component state
:type component_state: dict
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called SetPointMode")
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
current_power_limit = component_state.get("dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW)
update_task_status(
task_callback,
status=TaskStatus.IN_PROGRESS,
)
try:
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
# Complete immediately if the Dish is already in Standstill state
if dsc_state == ua.DscStateType.Standstill.value:
current_op_mode = component_state.get("operatingmode", DSOperatingMode.UNKNOWN)
update_task_status(
task_callback,
status=TaskStatus.COMPLETED,
result="Dish is already in desired state."
f" DS operating mode: {DSOperatingMode(current_op_mode).name},"
f" DSCState: {DSCState(dsc_state).name}.",
)
return
# 1: Set the Dish power state to Full power
low_power_active = self.scu.attributes[ # type: ignore[union-attr]
LOW_POWER_ACTIVE_OPCUA_PATH
].value
self.updates_list.append(
f"attribute read: {LOW_POWER_ACTIVE_OPCUA_PATH}, "
f"result: <{bool(low_power_active)}>"
)
if low_power_active:
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"SetPointMode",
False,
current_power_limit,
task_abort_event=task_abort_event,
)
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
self.updates_list.append(
f"attribute read: {DSC_STATE_OPCUA_PATH}, "
f"result: <{DSCState(dsc_state).name}:{dsc_state}>"
)
# 2: Unstow the dish if Stowed
if dsc_state == ua.DscStateType.Stowed.value:
self._call_opc_ua_method(
"Management.Commands.Stow",
"SetPointMode",
False,
task_abort_event=task_abort_event,
)
# Finally, activate both axis
self._call_opc_ua_method(
"Management.Commands.Activate",
"SetPointMode",
ua.UInt16(ua.AxisSelectType.AzEl),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="SetPointMode Called"
)
[docs] def release_auth(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Release command authority.
:return: The result of the command execution.
"""
not_connected_resp = "Not connected to the Dish Structure Controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self._release_auth(task_abort_event=None, task_callback=task_callback)
return status, response
@track_updates_decorator
def _release_auth(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Release command authority by immediately bypassing the LRC queue.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
self.logger.info("Called ReleaseAuth")
if self.scu is None:
msg = "SCU is not initialized"
update_task_status(task_callback, status=TaskStatus.FAILED, result=msg)
return TaskStatus.FAILED, msg
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
# Check if we have authority first
dsccmdauth_path = DSC_CMD_AUTH_PATH
dsccmdauth_value = self.scu.attributes.get(dsccmdauth_path)
if dsccmdauth_value is None:
self.updates_list.append(f"attribute read: {dsccmdauth_path}, result: (None)")
raise RuntimeError(f"Attribute {dsccmdauth_path} not found in SCU attributes")
dsccmdauth = DscCmdAuthType(dsccmdauth_value.value)
self.updates_list.append(
f"attribute read: {dsccmdauth_path}, result: {dsccmdauth.name}:({dsccmdauth})"
)
if dsccmdauth == DscCmdAuthType.LMC:
code, msg = self.scu.release_authority()
self.logger.debug("Releasing authority result as user [%s] [%s]", code, msg)
if code == SCUResultCode.COMMAND_DONE:
self.updates_list.append(f"Successfully released authority: {msg}")
else:
self.updates_list.append(f"Failed to release authority: {msg}")
raise RuntimeError(msg)
else:
msg = "LMC does not have authority."
self.updates_list.append(msg)
raise RuntimeError(msg)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return (
TaskStatus.FAILED,
"Release authority command not successful on the Dish Structure Controller",
)
update_task_status(task_callback, status=TaskStatus.COMPLETED, result="ReleaseAuth Called")
return (
TaskStatus.COMPLETED,
"The Release authority command has been called on the Dish Structure Controller",
)
[docs] def stow(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to STOW immediately bypassing the LRC queue.
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish Structure Controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
# As DS LOCKED is implemented as a safety state all movement is
# to be prohibited while the dish is locked
locked_dsc_states = [ua.DscStateType.Locked.value, ua.DscStateType.Locked_Stowed.value]
if dsc_state in locked_dsc_states:
return (
TaskStatus.REJECTED,
"Stow request failure due to current"
" Dish Structure State LOCKED or LOCKED & STOWED",
)
status, response = self._stow(task_abort_event=None, task_callback=task_callback)
return status, response
@track_updates_decorator
def _stow(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to STOW immediately bypassing the LRC queue.
Stow is only rejected if the current DSC state is LOCKED or LOCKED & STOWED
as a safety measure.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
self.logger.info("Called Stow")
def _run_stow() -> None:
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
"Management.Commands.Stow",
"Stow",
True,
task_abort_event=task_abort_event,
)
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="Stow Called"
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
threading.Thread(target=_run_stow).start()
return (
TaskStatus.IN_PROGRESS,
"The Stow command is being called on the Dish Structure Controller",
)
[docs] def unstow(
self,
component_state: dict,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Request the dish to unstow.
:param component_state: the current component state
:type component_state: dict
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish Structure Controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._unstow, args=[component_state], task_callback=task_callback
)
return status, response
@track_updates_decorator
def _unstow(
self,
component_state: dict,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Request the dish to unstow.
:param component_state: the current component state
:type component_state: dict
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called Unstow")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
result = "The Unstow command has been called on the Dish Structure Controller"
try:
dsc_state = self.scu.attributes[DSC_STATE_OPCUA_PATH].value # type: ignore[union-attr]
self.updates_list.append(
f"attribute read: {DSC_STATE_OPCUA_PATH}, "
f"result: {DSCState(dsc_state).name}:({dsc_state})"
)
if dsc_state == ua.DscStateType.Stowed.value:
# The dish needs to be in full power mode in order to call Stow(False)
low_power_active = self.scu.attributes[ # type: ignore[union-attr]
LOW_POWER_ACTIVE_OPCUA_PATH
].value
self.updates_list.append(
f"attribute read: {LOW_POWER_ACTIVE_OPCUA_PATH}, "
f"result: ({bool(low_power_active)})"
)
if low_power_active:
current_power_limit = component_state.get(
"dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW
)
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"Unstow",
False,
current_power_limit,
task_abort_event=task_abort_event,
)
self._call_opc_ua_method(
"Management.Commands.Stow",
"Unstow",
False,
task_abort_event=task_abort_event,
)
else:
result = "Dish is not stowed"
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(task_callback, status=TaskStatus.COMPLETED, result=result)
[docs] def set_index_position(
self,
component_state: dict,
band: IndexerPosition,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Set the indexer position on the dish.
:param band: The indexer position to move to.
:type band: IndexerPosition
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._set_index_position, args=[component_state, band], task_callback=task_callback
)
return status, response
@track_updates_decorator
def _set_index_position(
self,
component_state: dict,
band: IndexerPosition,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Set the indexer position on the dish.
:param band: The band to select.
:type band: IndexerPosition
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called SetIndexPosition")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
ua_band = ua.UInt16(self.indexer_position_to_band_type(band))
if ua_band is None:
update_task_status(
task_callback,
status=TaskStatus.REJECTED,
result=f"SetIndexPosition Rejected. Invalid band given ({str(band)})",
)
return
try:
# The dish needs to be in full power mode in order to call Move2Band
if component_state.get("powerstate", DSPowerState.LOW_POWER) == DSPowerState.LOW_POWER:
current_power_limit = component_state.get(
"dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW
)
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"SetIndexPosition",
False,
current_power_limit,
task_abort_event=task_abort_event,
)
self._call_opc_ua_method(
"Management.Commands.Move2Band",
"SetIndexPosition",
ua_band,
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="SetIndexPosition Called"
)
[docs] def slew(
self,
component_state: dict,
values: list[float],
speed_az: float,
speed_el: float,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to SLEW.
:param values: The slew coordinates
:type values: list[float]
:param speed_az: Az axis speed
:type speed_az: float
:param speed_el: El Axis speed
:type speed_el: float
:param component_state: the current component state
:type component_state: dict
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
if len(values) != 2:
update_task_status(task_callback, status=TaskStatus.REJECTED)
return (
TaskStatus.REJECTED,
f"Expected array of length 2 ([pos_az, pos_el])"
f" but got array of length {len(values)}.",
)
pos_az = values[0]
pos_el = values[1]
if not is_point_within_dish_limits(pos_az, pos_el):
update_task_status(task_callback, status=TaskStatus.REJECTED)
return (
TaskStatus.REJECTED,
f"Target point (Az: {pos_az}, El: {pos_el}) is not within the dishes limits"
f" (Az: [{MIN_AZIMUTH}, {MAX_AZIMUTH}], El: [{MIN_ELEVATION}, {MAX_ELEVATION}]))",
)
status, response = self.submit_task(
self._slew,
args=[pos_az, pos_el, speed_az, speed_el, component_state],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _slew(
self,
pos_az: float,
pos_el: float,
speed_az: float,
speed_el: float,
component_state: dict,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Transition the dish to SLEW.
:param pos_az: Az coordinate
:type pos_az: float
:param pos_el: El coordinate
:type pos_el: float
:param speed_az: Az speed
:type speed_az: float
:param speed_el: El speed
:type speed_el: float
:param component_state: the current component state
:type component_state: dict
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info(
"Called Slew with args [%s, %s, %s, %s]", pos_az, pos_el, speed_az, speed_el
)
current_power_limit = component_state.get("dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW)
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
self._report_task_progress(
f"Slew called with Azimuth speed: {speed_az} deg/s, "
f"Elevation speed: {speed_el} deg/s and DSC Power Limit: {current_power_limit}kW"
)
try:
if component_state.get("powerstate", DSPowerState.LOW_POWER) == DSPowerState.LOW_POWER:
self._call_opc_ua_method(
"Management.Commands.SetPowerMode",
"Slew",
False,
current_power_limit,
task_abort_event=task_abort_event,
)
self._call_opc_ua_method(
"Management.Commands.Slew2AbsAzEl",
"Slew",
float(pos_az),
float(pos_el),
float(speed_az),
float(speed_el),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(task_callback, status=TaskStatus.COMPLETED, result="Slew Called")
[docs] def track(
self,
interpolation_mode: TrackInterpolationMode,
get_current_tai_offset: Callable,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Transition the dish to TRACK.
:param interpolation_mode: The interpolation mode
:type interpolation_mode: TrackInterpolationMode
:param get_current_tai_offset: Callback to get the current tai offset
:type get_current_tai_offset: Callable
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._track,
args=[interpolation_mode, get_current_tai_offset],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _track(
self,
interpolation_mode: TrackInterpolationMode,
get_current_tai_offset: Callable,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Transition the dish to TRACK.
:param interpolation_mode: The DSC interpolation mode
:type interpolation_mode: TrackInterpolationMode
:param get_current_tai_offset: Callback to get the current tai offset
:type get_current_tai_offset: Callable
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called track")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
if interpolation_mode == TrackInterpolationMode.NEWTON:
interpol = ua.InterpolType.Newton
else:
interpol = ua.InterpolType.Spline
start_time_tai = get_current_tai_offset() + TRACK_START_LEAD_TIME
try:
self._call_opc_ua_method(
"Tracking.Commands.TrackStart",
"Track",
ua.UInt16(interpol),
start_time_tai,
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(task_callback, status=TaskStatus.COMPLETED, result="Track Called")
[docs] def track_stop(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Stop tracking."""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(self._track_stop, args=[], task_callback=task_callback)
return status, response
@track_updates_decorator
def _track_stop(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Stop tracking.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called TrackStop")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
"Management.Commands.Stop",
"TrackStop",
ua.UInt16(ua.AxisSelectType.AzEl),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(task_callback, status=TaskStatus.COMPLETED, result="TrackStop Called")
[docs] def track_load_static_off(
self,
values: list[float],
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Load the static pointing model offsets.
:param values: The static offset values
:type values: List
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
if len(values) != 2:
return (
TaskStatus.REJECTED,
f"Expected 2 arguments (off_xel, off_el) but got {len(values)} arg(s).",
)
off_xel = values[0]
off_el = values[1]
status, response = self.submit_task(
self._track_load_static_off, args=[off_xel, off_el], task_callback=task_callback
)
return status, response
@track_updates_decorator
def _track_load_static_off(
self,
off_xel: float,
off_el: float,
task_abort_event: Any,
task_callback: Optional[Callable] = None,
) -> None:
"""Load the static pointing model offsets.
:param off_xel: The XEl offset
:type off_xel: float
:param off_el: The El offset
:type off_el: float
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called TrackLoadStaticOff")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
"Tracking.Commands.TrackLoadStaticOff",
"TrackLoadStaticOff",
ua.Double(off_xel),
ua.Double(off_el),
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result="TrackLoadStaticOff Called"
)
# https://confluence.skatelescope.org/display/SWSI/MID+Dish+Structure+Control
[docs] def indexer_position_to_band_type(self, band_type: IndexerPosition) -> Union[None, enum.Enum]:
"""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"/>
mapping = {
IndexerPosition.OPTICAL: ua.BandType.Optical,
IndexerPosition.B1: ua.BandType.Band_1,
IndexerPosition.B2: ua.BandType.Band_2,
IndexerPosition.B3: ua.BandType.Band_3,
IndexerPosition.B4: ua.BandType.Band_4,
IndexerPosition.B5a: ua.BandType.Band_5a,
IndexerPosition.B5b: ua.BandType.Band_5b,
IndexerPosition.B6: ua.BandType.Band_6,
}
return mapping.get(band_type, None)
[docs] @track_updates_decorator
def track_load_table(
self,
values: list[float],
) -> Tuple[ResultCode, str]:
"""Load the track table."""
# perform validation checks
if len(values) < 5:
return (
ResultCode.REJECTED,
f"Length of argument ({len(values)}) is not as expected (>4).",
)
# Try to get load mode from ua, if it fails fall back to local mapping
try:
# This dynamic mapping is used to accommodate any changes in the controller
# implementation
track_load_mode_map = {
TrackTableLoadMode.NEW: ua.LoadModeType.New,
TrackTableLoadMode.APPEND: ua.LoadModeType.Append,
TrackTableLoadMode.RESET: ua.LoadModeType.Reset,
}
load_mode = track_load_mode_map.get(TrackTableLoadMode(int(values[0])))
except Exception:
self.logger.exception(
"Failed to load ua type for load mode mapping. "
"Falling back to locally defined mapping."
)
load_mode = TrackTableLoadMode(int(values[0]))
sequence_length = int(values[1])
track_table = values[2:]
# Check that the track_table length is a multiple of 3
if len(track_table) % 3 != 0:
return (
ResultCode.REJECTED,
f"Length of table ({len(track_table)}) is not a multiple of 3 "
"(timestamp, azimuth coordinate, elevation coordinate) as expected.",
)
# Check the given sequence length matches the given track table
sequences_in_table = len(track_table) // 3
if sequences_in_table != sequence_length:
return (
ResultCode.REJECTED,
f"Sequence length ({sequence_length}) does not match "
f"number of sequences in table ({sequences_in_table})",
)
tai_offsets = []
azimuths = []
elevations = []
for i in range(sequence_length):
timestamp_index = i * 3
az_index = i * 3 + 1
el_index = i * 3 + 2
tai_offsets.append(track_table[timestamp_index])
azimuths.append(track_table[az_index])
elevations.append(track_table[el_index])
if not are_points_within_dish_limits(list(zip(azimuths, elevations))):
return (
ResultCode.REJECTED,
"Track Table point(s) are not within the dishes limits"
f" (Az: [{MIN_AZIMUTH}, {MAX_AZIMUTH}],"
f" El: [{MIN_ELEVATION}, {MAX_ELEVATION}])",
)
# send the request to the opcua server
self.logger.info("Called TrackLoadTable")
try:
if sequence_length < 1000:
padding_required = 1000 - sequence_length
tai_offsets.extend([0] * padding_required)
azimuths.extend([0] * padding_required)
elevations.extend([0] * padding_required)
cast_tai = [ua.Double(val) for val in tai_offsets]
cast_az = [ua.Double(val) for val in azimuths]
cast_el = [ua.Double(val) for val in elevations]
self._call_opc_ua_method(
"Tracking.Commands.TrackLoadTable",
"TrackLoadTable",
ua.UInt16(load_mode),
ua.UInt16(sequence_length),
cast_tai,
cast_az,
cast_el,
wait_for_horn=False,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
return (
ResultCode.FAILED,
f"TrackLoadTable call on the DS Controller failed with message {ua_error}",
)
return ResultCode.OK, "TrackLoadTable successfully triggered the DS Controller"
[docs] def update_pointing_model_params(
self,
static_on: bool,
tilt_on: TiltOnType,
amb_temp_on: bool,
band: BandType,
values: list[float],
) -> tuple:
"""Update the threshold levels for AZ/EL pointing error between actual and desired values.
:param static_on: Flag indicating if static pointing model correction is active.
:type static_on: bool
:param tilt_on: Flag indicating if tilt correction is active.
:type tilt_on: bool
:param amb_temp_on: Flag indicating if ambient temperature correction is active.
:type amb_temp_on: bool
:param band: The Band type for which the pointing model parameters will be applied.
:type band: BandType
:param values:
A list containing exactly two elements:
- `on_source_threshold` (float): The threshold value for the pointing error.
- `threshold_time_period` (float): The time period over which the threshold is applied.
:type values: list[float]
:return: Task status
:rtype: Tuple[TaskStatus, str]
:raises ValueError: If the input list does not contain exactly two elements.
"""
# The argument value is a list of 18 floats
if len(values) != BAND_POINTING_MODEL_PARAMS_LENGTH:
return (
TaskStatus.FAILED,
f"Expected {BAND_POINTING_MODEL_PARAMS_LENGTH} arguments but got"
f" {len(values)} arg(s).",
)
try:
band = ua.UInt16(band)
cast_values = []
for value in values:
cast_values.append(ua.Double(value))
args = [
static_on,
ua.UInt16(tilt_on),
amb_temp_on,
band,
]
self.logger.info("Calling PmCorrOnOff with args %s", args)
self._call_opc_ua_method(
"Pointing.Commands.PmCorrOnOff", "PmCorrOnOff", *args, wait_for_horn=False
)
args = [band, *cast_values]
self.logger.info("Calling StaticPmSetup with args %s", args)
self._call_opc_ua_method(
"Pointing.Commands.StaticPmSetup", "StaticPmSetup", *args, wait_for_horn=False
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
return TaskStatus.FAILED, "Failed to update static pointing model parameters."
return TaskStatus.COMPLETED, "Successfully updated static pointing model parameters."
[docs] def update_threshold_levels(self, values: list[float]) -> tuple:
"""Update the threshold levels for AZ/EL pointing error between actual and desired values.
:param values:
A list containing exactly two elements:
- `on_source_threshold` (float): The threshold value for the pointing error.
- `threshold_time_period` (float): The time period over which the threshold is applied.
:type values: list[float]
:return: Task status
:rtype: Tuple[TaskStatus, str]
:return:TaskStatus
:rtype: tuple[bool, str]
:raises ValueError: If the input list does not contain exactly two elements.
"""
if len(values) != 2:
return (
TaskStatus.FAILED,
"Expected 2 arguments (on_source_threshold, threshold_time_period)"
f" but got {len(values)} arg(s).",
)
on_source_threshold = values[0]
threshold_time_period = values[1]
try:
self._call_opc_ua_method(
"Management.Commands.SetOnSourceThreshold",
"SetOnSourceThreshold",
on_source_threshold,
threshold_time_period,
wait_for_horn=False,
)
return TaskStatus.COMPLETED, "Successfully updated the threshold level."
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
return TaskStatus.FAILED, "Updating the threshold level was unsuccessful."
[docs] def interlock_acknowledge(
self,
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Request the dish to clear the interlock status.
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._interlock_acknowledge, args=[], task_callback=task_callback
)
return status, response
@track_updates_decorator
def _interlock_acknowledge(
self,
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Request the dish to clear the interlock status.
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called InterlockAck")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
"Safety.Commands.InterlockAck",
"InterlockAck",
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback,
status=TaskStatus.COMPLETED,
result="The InterlockAck command has been called on the Dish Structure Controller",
)
[docs] def execute_opcua_command(
self,
command_path: str,
command_args: list[Any],
task_callback: Optional[Callable] = None, # type: ignore
) -> Tuple[TaskStatus, str]:
"""Execute an OPC-UA command on the DSC.
:param command_path: The opcua path to the command node.
:type command_path: str
:param command_args: The list of args for the requested command.
:type command_args: list[Any]
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
:return: Task status
:rtype: Tuple[TaskStatus, str]
"""
not_connected_resp = "Not connected to the Dish structure controller."
if not self.scu:
update_task_status(task_callback, status=TaskStatus.FAILED, result=not_connected_resp)
return TaskStatus.FAILED, not_connected_resp
status, response = self.submit_task(
self._execute_opcua_command,
args=[command_path, command_args],
task_callback=task_callback,
)
return status, response
@track_updates_decorator
def _execute_opcua_command(
self,
command_path: str,
command_args: list[Any],
task_abort_event: Any,
task_callback: Optional[Callable] = None, # type: ignore
) -> None:
"""Execute an OPC-UA command on the DSC.
:param command_path: The opcua path to the command node.
:type command_path: str
:param command_args: The list of args for the requested command.
:type command_args: list[Any]
:param task_abort_event: Event to indicate that the method in progress should exit.
:type task_abort_event: threading.Event
:param task_callback: Update LRC status, defaults to None
:type task_callback: Optional[Callable], optional
"""
self.logger.info("Called ExecuteOPCUACommand")
update_task_status(task_callback, status=TaskStatus.IN_PROGRESS)
try:
self._call_opc_ua_method(
command_path,
"ExecuteOPCUACommand",
*command_args,
task_abort_event=task_abort_event,
)
except (CommandNotAuthorized, RuntimeError) as ua_error:
self.logger.exception(ua_error)
update_task_status(task_callback, status=TaskStatus.FAILED, result=str(ua_error))
return
update_task_status(
task_callback, status=TaskStatus.COMPLETED, result=f"{command_path} Called"
)
def _wait_for_horn(self, event: Optional[threading.Event] = None) -> None:
"""Wait for the warning alarm to turn off.
:param event: Optional event to wait for the horn to go off
:type event: threading.Event, optional
:raises RuntimeError: If the SCU is not connected or if the horn does not turn off within
HORN_CHECK_MAX_TRIES.
"""
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
# If no event is provided, create a new local event.
# Note: If event is None, the new event will never be set from outside.
# This is intentional for internal waiting logic.
evt = event or threading.Event()
try_counter = 0
while not evt.wait(HORN_CHECK_PERIOD_SEC):
wa_horn_on_val = self.scu.attributes.get(HORN_ACTIVE_PATH).value
self.updates_list.append(
f"attribute read: {HORN_ACTIVE_PATH}, result: {wa_horn_on_val}"
)
if not wa_horn_on_val:
self.logger.debug("Horn is off.")
break
try_counter += 1
failure_msg = "Could not take authority successfully, horn did not go off."
if try_counter >= HORN_CHECK_MAX_TRIES:
raise RuntimeError(failure_msg)
def _take_authority(
self, event: Optional[threading.Event] = None, wait_for_horn: bool = True
) -> DscCmdAuthType:
"""Take authority from the SCU.
:param event: Optional event to wait for the horn to go off
:type event: threading.Event, optional
:raises CommandNotAuthorized: If the authority cannot be taken
:raises RuntimeError: If the SCU is not connected
:return: The authority type taken
"""
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
code, msg = self.scu.take_authority(DscCmdAuthType.LMC, self._immutable_user_id)
truncated_id = self._immutable_user_id[:OBFUSCATED_ID_HASH_LEN_TRUNCATED]
self.updates_list.append(
"opcua command: CommandArbiter.Commands.TakeAuth , "
f"args ({DscCmdAuthType.LMC}, {truncated_id}...,), "
f"opcua result: {msg}:({code})"
)
self.logger.debug("Taking authority result is [%s] [%s]", code, msg)
# Cannot take auth
if code != SCUResultCode.COMMAND_DONE:
status_msg = f"Could not get device DSC authorization [{msg}]."
raise RuntimeError(status_msg)
elif wait_for_horn:
self._wait_for_horn(event)
# Double check we have authority
dsccmdauth_value = self.scu.attributes.get(DSC_CMD_AUTH_PATH)
if dsccmdauth_value is None:
self.updates_list.append(f"attribute read: {DSC_CMD_AUTH_PATH}, result: (None)")
raise RuntimeError(f"Attribute {DSC_CMD_AUTH_PATH} not found in SCU attributes")
dsccmdauth = DscCmdAuthType(dsccmdauth_value.value)
if dsccmdauth != DscCmdAuthType.LMC:
status_msg = f"Unexpected DscCmdAuthority type [{dsccmdauth.name}]."
raise RuntimeError(status_msg)
return dsccmdauth
def _retake_authority(
self, event: Optional[threading.Event] = None, wait_for_horn: bool = True
) -> DscCmdAuthType:
"""Re-take authority from the DSC.
:param event: Optional event to wait for the horn to go off
:type event: threading.Event, optional
:raises CommandNotAuthorized: If the authority cannot be taken
:raises RuntimeError: If the SCU is not connected
:return: The authority type taken
"""
if not self.scu:
raise RuntimeError("Communication with the DSC not established")
code, msg = self.scu.retake_authority(self._immutable_user_id)
truncated_id = self._immutable_user_id[:OBFUSCATED_ID_HASH_LEN_TRUNCATED]
self.updates_list.append(
"opcua command: CommandArbiter.Commands.RetakeAuth, "
f"args ({truncated_id}...,), "
f"opcua result: {msg}:({code})"
)
self.logger.debug("Retaking authority result is [%s] [%s]", code, msg)
# Cannot retake auth
if code != SCUResultCode.COMMAND_DONE:
# log id on dsc for debugging purposes
stored_immutable_id = self.scu.attributes.get(IMMUTABLE_ID_PATH)
if stored_immutable_id is not None:
self.logger.debug(
"Stored immutable id on the DSC is "
f"[{stored_immutable_id.value}] not [{self._immutable_user_id}] "
"as expected."
)
status_msg = f"Could not get device DSC authorization [{msg}]."
raise RuntimeError(status_msg)
elif wait_for_horn:
self._wait_for_horn(event)
# Double check we have authority
dsccmdauth_value = self.scu.attributes.get(DSC_CMD_AUTH_PATH)
if dsccmdauth_value is None:
self.updates_list.append(f"attribute read: {DSC_CMD_AUTH_PATH}, result: (None)")
raise RuntimeError(f"Attribute {DSC_CMD_AUTH_PATH} not found in SCU attributes")
dsccmdauth = DscCmdAuthType(dsccmdauth_value.value)
if dsccmdauth != DscCmdAuthType.LMC:
status_msg = f"Unexpected DscCmdAuthority type [{dsccmdauth.name}]."
raise RuntimeError(status_msg)
return dsccmdauth
[docs]def main() -> None:
"""For local testing."""
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler(sys.stdout))
scu = SteeringControlUnit(
"127.0.0.1", 4840, endpoint="/dish-structure/server/", namespace="http://skao.int/DS_ICD/"
)
IMMUTABLE_ID = "ds-standalone-id"
scu.connect_and_setup()
scu.take_authority("LMC", IMMUTABLE_ID)
task_executor_cm = TaskExecutorComponentManager(logger=logger)
cm_update_component_state = (
task_executor_cm._update_component_state # pylint: disable=protected-access, useless-suppression
)
com_manager = CommandManager(
logger, IMMUTABLE_ID, scu, task_executor_cm.submit_task, cm_update_component_state
)
com_manager.set_standby_mode({"dscpowerlimitkw": 10.0})
if __name__ == "__main__":
main()