"""Tango Device for Dish Structure with monitoring and control to OPCUA server."""
import os
import time
import weakref
from typing import Any, Callable, List, Optional, Tuple
from urllib.parse import urlparse
import tango
from ska_control_model import CommunicationStatus, HealthState, ResultCode, TaskStatus
from ska_tango_base import SKAController
from ska_tango_base.commands import SubmittedSlowCommand
from tango import AttrQuality, AttrWriteType, DevState
from tango.server import attribute, command, device_property, run
from ska_mid_dish_ds_manager.decorators import last_command_invoked_decorator
from ska_mid_dish_ds_manager.models.command_class import (
ReleaseAuthCommand,
StowCommand,
TrackLoadTableCommand,
)
from ska_mid_dish_ds_manager.models.constants import (
ARCHIVE_AZ_DELTA_DEG,
ARCHIVE_EL_DELTA_DEG,
BAND_POINTING_MODEL_PARAMS_LENGTH,
DEFAULT_AZIMUTH_SPEED_DPS,
DEFAULT_ELEVATION_SPEED_DPS,
DSC_MAX_POWER_LIMIT_KW,
DSC_MIN_POWER_LIMIT_KW,
ERROR_STATUS_NODES,
MAX_AZIMUTH_SPEED_DPS,
MAX_ELEVATION_SPEED_DPS,
)
from ska_mid_dish_ds_manager.models.dish_enums import (
ApplicationStateType,
BandType,
CurrentModeType,
DscCmdAuthType,
DscCtrlState,
DSCState,
DSOperatingMode,
DSPowerState,
IndexerPosition,
PointingState,
TiltOnType,
TrackInterpolationMode,
TrackProgramMode,
)
from ska_mid_dish_ds_manager.opcua_component_manager import OPCUAComponentManager
from ska_mid_dish_ds_manager.release import get_ds_manager_release_version
from ska_mid_dish_ds_manager.utils import BaseInfoIt, get_immutable_user_id
DevVarLongStringArrayType = Tuple[List[ResultCode], List[Optional[str]]]
[docs]class DSManager(SKAController): # type: ignore
"""The Dish Structure Manager."""
# Access instances for debugging and testing
instances = weakref.WeakValueDictionary() # type: ignore
# -----------------
# Device Properties
# -----------------
opcua_server_addr = "opc.tcp://ds-opcua-server-simulator-001-svc:4840/dish-structure/server/"
DSCFqdn = device_property(dtype=str, default_value=opcua_server_addr)
opcua_namespace = "http://skao.int/DS_ICD/"
DSOPCUANamespace = device_property(dtype=str, default_value=opcua_namespace)
default_node_trees_to_monitor = (
"Azimuth, CommandArbiter, Elevation, FeedIndexer, Management, Pointing, Safety, Time_cds, "
"Tracking"
)
DSNodeTreesToMonitor = device_property(dtype=str, default_value=default_node_trees_to_monitor)
[docs] class InitCommand(SKAController.InitCommand):
"""Initializes the attributes of the DSManager."""
[docs] def do(
self: SKAController.InitCommand,
*args: Any,
**kwargs: Any,
) -> tuple[ResultCode, str]:
"""Initialise DSManager.
:param args: positional arguments
:param kwargs: keyword arguments
:return: A tuple containing a return code and a string
"""
self._device._connection_state = CommunicationStatus.NOT_ESTABLISHED
self._device._version_id = get_ds_manager_release_version()
self._device.last_archived_az = 0.0
self._device.last_archived_el = 0.0
self._device._component_state_attr_map = {
"powerstate": "powerState",
"pointingstate": "pointingState",
"operatingmode": "operatingMode",
"indexerposition": "indexerPosition",
"achievedpointing": "achievedPointing",
"desiredpointingaz": "desiredPointingAz",
"desiredpointingel": "desiredPointingEl",
"currentpointing": "currentPointing",
"healthstate": "healthState",
"buildstate": "buildState",
"band0pointingmodelparams": "band0PointingModelParams",
"band1pointingmodelparams": "band1PointingModelParams",
"band2pointingmodelparams": "band2PointingModelParams",
"band3pointingmodelparams": "band3PointingModelParams",
"band4pointingmodelparams": "band4PointingModelParams",
"band5apointingmodelparams": "band5aPointingModelParams",
"band5bpointingmodelparams": "band5bPointingModelParams",
"band6pointingmodelparams": "band6PointingModelParams",
"dscstate": "dscState",
"trackinterpolationmode": "trackInterpolationMode",
"azimuthspeed": "azimuthSpeed",
"elevationspeed": "elevationSpeed",
"configuretargetlock": "configureTargetLock",
"achievedtargetlock": "achievedTargetLock",
"actstaticoffsetvaluexel": "actStaticOffsetValueXel",
"actstaticoffsetvalueel": "actStaticOffsetValueEl",
"trackprogrammode": "trackProgramMode",
"dscpowerlimitkw": "dscPowerLimitKw",
"actptactindex": "trackTableCurrentIndex",
"actptendindex": "trackTableEndIndex",
"lastcommandupdate": "lastCommandUpdate",
"lastcommandinvoked": "lastCommandInvoked",
"dsccmdauth": "dscCmdAuth",
"currentmode": "currentMode",
"applicationstate": "applicationState",
"staticoninput": "staticOnInput",
"tiltoninput": "tiltOnInput",
"ambtoninput": "ambTOnInput",
"dscctrlstate": "dscCtrlState",
"displayeddiagnosis": "displayedDiagnosis",
"hhpconnected": "hhpConnected",
}
for attr in self._device._component_state_attr_map.values():
self._device.set_change_event(attr, True, False)
self._device.set_archive_event(attr, True, False)
# Add Management.ErrorStatus attributes
for attr_name in ERROR_STATUS_NODES:
self._device._create_tango_device_attribute(
attr_name=attr_name,
comp_state_name=attr_name.lower(),
dtype=bool,
doc=ERROR_STATUS_NODES[attr_name],
# Bind attr_name as a default argument for the lambda to avoid Python's
# late-binding closure behaviour. Lambdas capture variables by reference
# (not value), without this each lambda would use the final loop value of
# attr_name.
fget=lambda *args, attr_name=attr_name, **kwargs: self._device._read_attribute(
attr_name.lower(),
False,
),
fset=None,
)
self._device.set_change_event("connectionState", True, False)
self._device.set_archive_event("connectionState", True, False)
self._device.push_change_event("healthState", HealthState.UNKNOWN)
self._device.push_archive_event("healthState", HealthState.UNKNOWN)
self._device.set_state(DevState.ON)
self._device.component_manager.start_communicating()
# Try to connect to DB and update memorized attributes with stored values
# if the TANGO_HOST is set
self._device.try_update_memorized_attributes_from_db()
# Save the device instance
# This can be used during unit testing to access the running device instance and its
# component manager to manually change values and trigger callbacks.
self._device.instances[self._device.get_name()] = self._device
(result_code, message) = super().do() # type: ignore
return (ResultCode(result_code), message)
[docs] def create_component_manager(
self: "DSManager",
) -> OPCUAComponentManager:
"""Create a component manager.
:return: the component manager
"""
parsed_fqdn = urlparse(self.DSCFqdn)
return OPCUAComponentManager(
ds_host=parsed_fqdn.hostname,
ds_port=parsed_fqdn.port,
ds_endpoint=parsed_fqdn.path,
ds_opcua_namespace=self.DSOPCUANamespace,
ds_unique_id=get_immutable_user_id(self.get_name(), self.logger),
logger=self.logger,
node_trees_to_monitor=self.DSNodeTreesToMonitor,
create_tango_device_attribute=self._create_tango_device_attribute,
communication_state_callback=self._communication_state_changed,
component_state_callback=self._component_state_changed,
update_device_state=self._update_state,
update_device_status=self._update_status,
get_attr_type_and_quality=self._get_attr_type_and_quality,
update_attr_quality=self._update_attr_quality,
)
[docs] def delete_device(self) -> None:
"""Override delete_device to stop communication with DSC before deletion."""
self.logger.info(
"DSManager device is being deleted. Stopping communication with DSC.",
)
if self.component_manager:
self.component_manager.stop_communicating()
return super().delete_device()
[docs] def init_command_objects(self) -> None:
"""Initialise the command handlers."""
super().init_command_objects()
for command_name, method_name in [
("SetStandbyMode", "set_standby_mode"),
("SetPointMode", "set_point_mode"),
("SetIndexPosition", "set_index_position"),
("Slew", "slew"),
("Track", "track"),
("TrackStop", "track_stop"),
("TrackLoadStaticOff", "track_load_static_off"),
("SetPowerMode", "set_power_mode"),
("ExecuteOPCUACommand", "execute_opcua_command"),
("Unstow", "unstow"),
("InterlockAck", "interlock_acknowledge"),
("ResetAxesCommand", "reset_axes_command"),
("TakeAuthority", "take_authority"),
("ReTakeAuthority", "retake_authority"),
]:
self.register_command_object(
command_name,
SubmittedSlowCommand(
command_name,
self._command_tracker,
self.component_manager,
method_name,
callback=None,
logger=self.logger,
),
)
self.register_command_object(
"Stow",
StowCommand(
"Stow",
self._command_tracker,
self.component_manager,
"stow",
callback=None,
logger=self.logger,
),
)
self.register_command_object(
"ReleaseAuth",
ReleaseAuthCommand(
"ReleaseAuth",
self._command_tracker,
self.component_manager,
"release_auth",
callback=None,
logger=self.logger,
),
)
self.register_command_object(
"TrackLoadTable",
TrackLoadTableCommand(self.component_manager, self.logger),
)
###########
# callbacks
###########
def _update_status(self, status: str) -> None:
"""Update the status of the device.
:param status: the new status string
"""
self.set_status(status)
self.logger.debug(status)
self.push_change_event("status")
self.push_archive_event("status")
def _get_attr_type_and_quality(
self, attribute_name: str
) -> Tuple[int, AttrQuality] | Tuple[None, None]:
"""Read the data type and quality of a Tango attribute.
:param attribute_name: the name of the Tango attribute
:return: datatype and quality for the given Tango attribute
:rtype: Tuple[int, AttrQuality] | Tuple[None, None]
"""
attr_name = self._component_state_attr_map.get(attribute_name, attribute_name)
try:
attribute_object = self.get_device_attr().get_attr_by_name(attr_name)
if attribute_object:
return attribute_object.get_data_type(), attribute_object.get_quality()
except tango.DevFailed:
pass
# No Tango attribute found
return None, None
def _update_attr_quality(
self,
attribute_name: str,
new_attribute_quality: AttrQuality,
send_event: bool = False,
) -> None:
"""Update the quality of a Tango attribute.
:param attribute_name: the name of the Tango attribute
:param new_attribute_quality: the new tango.AttrQuality
"""
self.logger.debug(
"Updating attr (%s) with quality (%s). Pushing event=%s.",
attribute_name,
new_attribute_quality,
send_event,
)
attr_name = self._component_state_attr_map.get(attribute_name, attribute_name)
attribute_object = self.get_device_attr().get_attr_by_name(attr_name)
if attribute_object:
attribute_object.set_quality(new_attribute_quality, send_event)
def _communication_state_changed(self, communication_state: CommunicationStatus) -> None:
"""Push and archive events on communication state change."""
self.push_change_event("connectionState", communication_state)
self.push_archive_event("connectionState", communication_state)
def _component_state_changed(self, *args: Any, **kwargs: Any) -> None:
if not hasattr(self, "_component_state_attr_map"):
self.logger.warning("Init not completed, but state is being updated [%s]", kwargs)
return
for comp_state_name, comp_state_value in kwargs.items():
if comp_state_value is not None:
attribute_name = self._component_state_attr_map.get(comp_state_name, None)
if attribute_name:
self.push_change_event(attribute_name, comp_state_value)
if attribute_name not in ["currentPointing", "achievedPointing"]:
self.push_archive_event(attribute_name, comp_state_value)
return
# Some special rules for when to archive currentPointing and achievedPointing
if (
abs(comp_state_value[1] - self.last_archived_az) > ARCHIVE_AZ_DELTA_DEG # type: ignore
) or (abs(comp_state_value[2] - self.last_archived_el) > ARCHIVE_EL_DELTA_DEG): # type: ignore
self.last_archived_az = comp_state_value[1]
self.last_archived_el = comp_state_value[2]
self.push_archive_event(attribute_name, comp_state_value)
def _create_tango_device_attribute(
self,
attr_name: str,
comp_state_name: str,
dtype: type,
doc: str,
fget: Callable,
fset: Optional[Callable],
abs_change: Optional[Any] = None,
) -> None:
"""Create a Tango attribute on the device and configure change and archive events.
:param attr_name: The name of the Tango attribute
:type attr_name: str
:param comp_state_name: The key for the component state
:type comp_state_name: str
:param dtype: The python data type of the attribute
:type dtype: type
:param doc: A description of the attribute
:type doc: str
:param fget: The function used to retrieve the attribute value
:type fget: Callable
:param fset: The function used to set the attribute value (if writable)
:type fset: Optional[Callable]
"""
self._component_state_attr_map[comp_state_name] = attr_name
detect = False
if abs_change:
tango_attr = attribute(
name=attr_name,
dtype=dtype,
doc=doc,
fget=fget,
fset=fset,
abs_change=abs_change,
archive_abs_change=abs_change,
)
detect = True
else:
tango_attr = attribute(
name=attr_name,
dtype=dtype,
doc=doc,
fget=fget,
fset=fset,
)
self.add_attribute(tango_attr)
# Allow filtering at tango layer only if abs_change is set
self.set_change_event(attr_name, True, detect)
# Note: This needs to be a _submit_tango_operation as tango base doesn't implement
# set_archive_event as of v1.3.1. This is not required if EnsureOmniThread is used but
# then we get errors trying to acquire the serialisation monitor.
self._submit_tango_operation("set_archive_event", attr_name, True, detect)
def _get_device_attribute_property_value(self, attribute_name: str) -> Optional[str]:
"""Read memorized attributes values from TangoDB.
:param: attribute_name: Tango attribute name
:type attribute_name: str
:return: value for the given attribute
:rtype: Optional[str]
"""
self.logger.debug("Getting attribute property value for %s.", attribute_name)
database = tango.Database()
attr_property = database.get_device_attribute_property(self.get_name(), attribute_name)
try:
attr_property_value = attr_property[attribute_name]
# Try "__value", the default and "attribute_name", like tiltOnInput which someone may
# have entered themselves
for valid_key in ["__value", attribute_name]:
if valid_key in attr_property_value:
return attr_property_value[valid_key][0]
except (KeyError, IndexError):
self.logger.exception("Error reading attribute from DB")
return None
[docs] def try_update_memorized_attributes_from_db(self) -> None:
"""Read memorized attributes values from TangoDB and update device attributes."""
if "TANGO_HOST" not in os.environ:
self.logger.debug("Not updating memorized attributes. TANGO_HOST is not set.")
return
self.logger.debug("Updating memorized attributes. Trying to read from database.")
try:
# staticOnInput
value_from_db = self._get_device_attribute_property_value("staticOnInput")
if value_from_db is not None:
self.logger.debug(
"Updating staticOnInput value with value from database %s.",
value_from_db,
)
static_on_input = value_from_db == "true"
self.component_manager._update_component_state(staticoninput=static_on_input)
# tiltOnInput
value_from_db = self._get_device_attribute_property_value("tiltOnInput")
if value_from_db is not None:
self.logger.debug(
"Updating tiltOnInput value with value from database %s.",
value_from_db,
)
tilt_on_input = int(value_from_db)
self.component_manager._update_component_state(tiltoninput=tilt_on_input)
# ambTOnInput
value_from_db = self._get_device_attribute_property_value("ambTOnInput")
if value_from_db is not None:
self.logger.debug(
"Updating ambTOnInput value with value from database %s.",
value_from_db,
)
amb_t_on_input = value_from_db == "true"
self.component_manager._update_component_state(ambtoninput=amb_t_on_input)
except tango.DevFailed:
self.logger.debug(
"Could not update memorized attributes. Failed to connect to database."
)
# -----------
# Attributes
# -----------
def _read_attribute(
self, attribute_name: str, default: Optional[Any] = None
) -> Tuple[Any, float, AttrQuality]:
"""Helper method to read an attributes value and quality.
Returning a triplet during a read allows the quality to remain as it is set. If only a
single value is returned then Tango will automatically revert the quality to ATTR_VALID.
:param attribute_name: the name of the Tango attribute
:param default: the default value if no value is found in the component state
:return: value, current time, and quality for the given Tango attribute
:rtype: Tuple[Any, float, AttrQuality]
"""
value = self.component_manager.component_state.get(attribute_name, default)
_, current_quality = self._get_attr_type_and_quality(attribute_name)
return value, time.time(), current_quality
[docs] @attribute(dtype=str)
def buildState(self) -> Tuple[str, float, AttrQuality]:
"""Get DS manager and DSC version structure."""
return self._read_attribute("buildstate")
[docs] @attribute(
dtype=CommunicationStatus,
access=AttrWriteType.READ,
doc="Displays connection status to DSC device",
)
def connectionState(self) -> CommunicationStatus:
"""Indicate connection state of device."""
return self.component_manager.communication_state
[docs] @attribute(
dtype=DSPowerState,
access=AttrWriteType.READ,
)
def powerState(self: "DSManager") -> Tuple[DSPowerState, float, AttrQuality]:
"""Indicate power state of device."""
return self._read_attribute("powerstate", DSPowerState.LOW_POWER)
[docs] @attribute(
dtype=DSOperatingMode,
access=AttrWriteType.READ,
)
def operatingMode(self: "DSManager") -> Tuple[DSOperatingMode, float, AttrQuality]:
"""Indicate operating mode of device."""
return self._read_attribute("operatingmode", DSOperatingMode.UNKNOWN)
[docs] @attribute(
dtype=IndexerPosition,
access=AttrWriteType.READ,
)
def indexerPosition(self: "DSManager") -> Tuple[IndexerPosition, float, AttrQuality]:
"""Indicate operating mode of device."""
return self._read_attribute("indexerposition", IndexerPosition.UNKNOWN)
[docs] @attribute(
dtype=PointingState,
access=AttrWriteType.READ,
)
def pointingState(self: "DSManager") -> Tuple[PointingState, float, AttrQuality]:
"""Indicate operating mode of device."""
return self._read_attribute("pointingstate", None)
[docs] @attribute(
dtype=bool,
access=AttrWriteType.READ,
)
def achievedTargetLock(self) -> Tuple[bool, float, AttrQuality]:
"""Indicate Actual “On Sky” rms tracking error of device."""
return self._read_attribute("achievedtargetlock", False)
[docs] @attribute(
dtype=DscCmdAuthType,
access=AttrWriteType.READ,
)
def dscCmdAuth(self) -> Tuple[DscCmdAuthType, float, AttrQuality]:
"""Indicate DSC command authority."""
return self._read_attribute("dsccmdauth", DscCmdAuthType.NO_AUTHORITY)
[docs] @attribute(
dtype=bool,
doc="Indicates whether the hand held pendant is connected",
access=AttrWriteType.READ,
)
def hhpConnected(self) -> Tuple[bool, float, AttrQuality]:
"""Indicate whether the hand held pendant is connected."""
return self._read_attribute("hhpconnected", False)
[docs] @attribute(
max_dim_x=3,
dtype=(float,),
doc="[0] Timestamp\n[1] Azimuth\n[2] Elevation",
access=AttrWriteType.READ,
)
def achievedPointing(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate achieved pointing of dish axis."""
return self._read_attribute("achievedpointing", [0.0, 0.0, 0.0])
[docs] @attribute(
max_dim_x=7,
dtype=(float,),
doc=(
"List consisting of pointing information:"
"\n[\n[0] TAIoffset,\n[1] Azimuth.p_Act,\n[2] Elevation.p_Act,"
"\n[3] Azimuth.p_Enc,\n[4] Elevation.p_Enc,"
"\n[5] TiltCorrVal_Az,\n[6] TiltCorrVal_El\n]"
),
access=AttrWriteType.READ,
)
def currentPointing(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Display pointing feedback."""
return self._read_attribute("currentpointing", [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0])
[docs] @attribute(
max_dim_x=2,
dtype=(float,),
doc="[0] Timestamp\n[1] Azimuth",
access=AttrWriteType.READ,
)
def desiredPointingAz(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate desired pointing of device azimuth axis."""
return self._read_attribute("desiredpointingaz", [0.0, 0.0, 0.0])
[docs] @attribute(
max_dim_x=2,
dtype=(float,),
doc="[0] Timestamp\n[1] Elevation",
access=AttrWriteType.READ,
)
def desiredPointingEl(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate desired pointing of device elevation axis."""
return self._read_attribute("desiredpointingel", [0.0, 0.0, 0.0])
[docs] @attribute(
dtype=float,
doc="Actual cross-elevation static offset (arcsec)",
access=AttrWriteType.READ,
)
def actStaticOffsetValueXel(self: "DSManager") -> Tuple[float, float, AttrQuality]:
"""Indicate actual cross-elevation static offset in arcsec."""
return self._read_attribute("actstaticoffsetvaluexel", 0.0)
[docs] @attribute(
dtype=float,
doc="Actual elevation static offset (arcsec)",
access=AttrWriteType.READ,
)
def actStaticOffsetValueEl(self: "DSManager") -> Tuple[float, float, AttrQuality]:
"""Indicate actual elevation static offset in arcsec."""
return self._read_attribute("actstaticoffsetvalueel", 0.0)
[docs] @attribute(
max_dim_x=1,
dtype=HealthState,
doc="Returns the current Health State",
access=AttrWriteType.READ,
)
def healthState(self: "DSManager") -> Tuple[HealthState, float, AttrQuality]:
"""Indicate the Health State."""
return self._read_attribute("healthstate", HealthState.UNKNOWN)
[docs] @attribute(
dtype=int,
doc="Actual used index in the track table",
access=AttrWriteType.READ,
)
def trackTableCurrentIndex(self) -> Tuple[int, float, AttrQuality]:
"""Index of current point being tracked in the track table."""
return self._read_attribute("actptactindex", 0)
[docs] @attribute(
dtype=int,
doc="End index in the track table",
access=AttrWriteType.READ,
)
def trackTableEndIndex(self) -> Tuple[int, float, AttrQuality]:
"""Index of last point in the track table."""
return self._read_attribute("actptendindex", 0)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 0 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B1.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band0PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 0 static pointing model parameters of device."""
return self._read_attribute("band0pointingmodelparams", [])
[docs] @band0PointingModelParams.write # type: ignore
def band0PointingModelParams(self: "DSManager", value):
"""Update band 0 static pointing model parameters."""
self.logger.debug("Updating band0PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.OPTICAL,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 1 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B1.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band1PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 1 static pointing model parameters of device."""
return self._read_attribute("band1pointingmodelparams", [])
[docs] @band1PointingModelParams.write # type: ignore
def band1PointingModelParams(self: "DSManager", value):
"""Update band 1 static pointing model parameters."""
self.logger.debug("Updating band1PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B1,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 2 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B2.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band2PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 2 static pointing model parameters of device."""
return self._read_attribute("band2pointingmodelparams", [])
[docs] @band2PointingModelParams.write # type: ignore
def band2PointingModelParams(self: "DSManager", value):
"""Update band 2 static pointing model parameters."""
self.logger.debug("Updating band2PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B2,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 3 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B3.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band3PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 3 static pointing model parameters of device."""
return self._read_attribute("band3pointingmodelparams", [])
[docs] @band3PointingModelParams.write # type: ignore
def band3PointingModelParams(self: "DSManager", value):
"""Update band 3 static pointing model parameters."""
self.logger.debug("Updating band3PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B3,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 4 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B4.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band4PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 4 static pointing model parameters of device."""
return self._read_attribute("band4pointingmodelparams", [])
[docs] @band4PointingModelParams.write # type: ignore
def band4PointingModelParams(self: "DSManager", value):
"""Update band 4 static pointing model parameters."""
self.logger.debug("Updating band4PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B4,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 5a pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B5a.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band5aPointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 5a static pointing model parameters of device."""
return self._read_attribute("band5apointingmodelparams", [])
[docs] @band5aPointingModelParams.write # type: ignore
def band5aPointingModelParams(self: "DSManager", value):
"""Update band 5a static pointing model parameters."""
self.logger.debug("Updating band5aPointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B5a,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 5b pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B5b.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band5bPointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 5b static pointing model parameters of device."""
return self._read_attribute("band5bpointingmodelparams", [])
[docs] @band5bPointingModelParams.write # type: ignore
def band5bPointingModelParams(self: "DSManager", value):
"""Update band 5b static pointing model parameters."""
self.logger.debug("Updating band5bPointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B5b,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
@attribute(
dtype=(float,),
max_dim_x=BAND_POINTING_MODEL_PARAMS_LENGTH,
doc="""
Parameters for (local) Band 6 pointing models used by Dish to do pointing corrections.
When writing to this attribute, the selected band for correction will be set to B6.
Band pointing model parameters are:
[0] IA, [1] CA, [2] NPAE, [3] AN, [4] AN0, [5] AW, [6] AW0, [7] ACEC, [8] ACES,
[9] ABA, [10] ABphi, [11] IE, [12] ECEC, [13] ECES, [14] HECE4,
[15] HESE4, [16] HECE8, [17] HESE8
""",
access=AttrWriteType.READ_WRITE,
)
def band6PointingModelParams(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate band 6 static pointing model parameters of device."""
return self._read_attribute("band6pointingmodelparams", [])
[docs] @band6PointingModelParams.write # type: ignore
def band6PointingModelParams(self: "DSManager", value):
"""Update band 6 static pointing model parameters."""
self.logger.debug("Updating band6PointingModelParams %s", value)
task_status, message = self.component_manager.update_pointing_model_params(
BandType.B6,
value,
)
if task_status != TaskStatus.COMPLETED:
raise ValueError(message)
[docs] @attribute(
dtype=DSCState,
doc="Dish Structure Controller State",
)
def dscState(self: "DSManager") -> Tuple[DSCState, float, AttrQuality]:
"""Indicate dish structure controller state."""
return self._read_attribute("dscstate")
@attribute(
dtype=TrackInterpolationMode,
access=AttrWriteType.READ_WRITE,
doc="Selects the type of interpolation to be used in program tracking.",
)
def trackInterpolationMode(self) -> Tuple[TrackInterpolationMode, float, AttrQuality]:
"""Indicate track interpolation mode of device."""
return self._read_attribute("trackinterpolationmode", TrackInterpolationMode.SPLINE)
[docs] @trackInterpolationMode.write # type: ignore
def trackInterpolationMode(self, value):
"""Set the trackInterpolationMode."""
self.component_manager._update_component_state(trackinterpolationmode=value)
self.push_change_event("trackInterpolationMode", value)
self.push_archive_event("trackInterpolationMode", value)
@attribute(
dtype=float,
access=AttrWriteType.READ_WRITE,
doc="Sets the Azimuth rotation speed during slew in degrees per second.",
)
def azimuthSpeed(self) -> Tuple[float, float, AttrQuality]:
"""Indicate azimuth speed for slew moves."""
return self._read_attribute("azimuthspeed", DEFAULT_AZIMUTH_SPEED_DPS)
[docs] @azimuthSpeed.write # type: ignore
def azimuthSpeed(self, value):
"""Set the azimuth speed."""
if value <= 0 or value > MAX_AZIMUTH_SPEED_DPS:
raise ValueError(
f"Invalid value for Azimuth speed, valid range is (0, {MAX_AZIMUTH_SPEED_DPS}]."
)
self.component_manager._update_component_state(azimuthspeed=value)
@attribute(
dtype=float,
access=AttrWriteType.READ_WRITE,
doc="Sets the Elevation rotation speed during slew in degrees per second.",
)
def elevationSpeed(self) -> Tuple[float, float, AttrQuality]:
"""Indicate elevation speed for slew moves."""
return self._read_attribute("elevationspeed", DEFAULT_ELEVATION_SPEED_DPS)
[docs] @elevationSpeed.write # type: ignore
def elevationSpeed(self, value):
"""Set the elevation speed."""
if value <= 0 or value > MAX_ELEVATION_SPEED_DPS:
raise ValueError(
"Invalid value for Elevation speed,"
f" valid range is (0, {MAX_ELEVATION_SPEED_DPS}]."
)
self.component_manager._update_component_state(elevationspeed=value)
@attribute(
dtype=TrackProgramMode,
access=AttrWriteType.READ_WRITE,
doc="Selects the type of tracking mode to be used.",
)
def trackProgramMode(self: "DSManager") -> Tuple[TrackProgramMode, float, AttrQuality]:
"""Indicate track mode selected."""
return self._read_attribute("trackprogrammode", TrackProgramMode.TABLE)
[docs] @trackProgramMode.write # type: ignore
def trackProgramMode(self: "DSManager", value: TrackProgramMode):
"""Set the trackProgramMode."""
self.component_manager._update_component_state(trackprogrammode=value)
@attribute(
max_dim_x=2,
dtype=(float,),
access=AttrWriteType.READ_WRITE,
doc="""
Threshold level the rms on-sky vector error between the desired
Azimuth & Elevation and actual Azimuth & Elevation.
Note:
The attributes `onSourceThreshold` (Tracking.Status.OnSourceThreshold) and
`thresholdTimePeriod` (Tracking.Status.ThresholdTimePeriod) can be accessed
via `configureTargetLock`, as elements in the
list: [onSourceThreshold, thresholdTimePeriod].
""",
)
def configureTargetLock(self: "DSManager") -> Tuple[list[float], float, AttrQuality]:
"""Indicate the configuretargetlock."""
return self._read_attribute("configuretargetlock", [0.0, 0.0])
@attribute(
dtype=float,
doc="""
DSC Power Limit (kW). Note that this attribute can also be set by calling
SetPowerMode. This value does not reflect the power limit in reality because
the current PowerLimit(kW) is not reported as it cannot be read from the DSC.
""",
access=AttrWriteType.READ_WRITE,
)
def dscPowerLimitKw(self: "DSManager") -> Tuple[float, float, AttrQuality]:
"""Represent the DSC Power Limit (kW)."""
return self._read_attribute("dscpowerlimitkw", DSC_MIN_POWER_LIMIT_KW)
[docs] @dscPowerLimitKw.write # type: ignore
def dscPowerLimitKw(self: "DSManager", value):
"""Set the DSC Power Limit (kW)."""
self.logger.debug("Updating DSC Power Limit Mode %skW", value)
if DSC_MIN_POWER_LIMIT_KW <= value <= DSC_MAX_POWER_LIMIT_KW:
self.component_manager._update_component_state(dscpowerlimitkw=value)
self.push_change_event("dscpowerlimitkw", value)
self.push_archive_event("dscpowerlimitkw", value)
else:
raise ValueError(
f"Invalid value, {value} for DSC Power Limit (kW),"
f" valid range is [{DSC_MIN_POWER_LIMIT_KW}, {DSC_MAX_POWER_LIMIT_KW}]."
)
[docs] @attribute(
dtype=(str,),
max_dim_x=25,
access=AttrWriteType.READ,
doc="This attribute tracks the last tango & opcua commands called on the Dish Structure"
" and the results of the OPCUA command",
)
def lastCommandUpdate(self: "DSManager") -> Tuple[list[str], float, AttrQuality]:
"""Reflect the last command update."""
return self._read_attribute("lastcommandupdate", [])
[docs] @attribute(
dtype=(str, str),
max_dim_x=2,
access=AttrWriteType.READ,
doc=("Stores the name and timestamp (in UNIX UTC format) of the last invoked command."),
)
def lastCommandInvoked(self: "DSManager") -> Tuple[tuple[str, str], float, AttrQuality]:
"""Return the last command invoked and its timestamp."""
return self._read_attribute("lastcommandinvoked", ("0.0", ""))
[docs] @attribute(
dtype=CurrentModeType,
access=AttrWriteType.READ,
)
def currentMode(self: "DSManager") -> Tuple[CurrentModeType, float, AttrQuality]:
"""Reflect the Current Mode node's state."""
return self._read_attribute("currentmode", None)
[docs] @attribute(
dtype=ApplicationStateType,
access=AttrWriteType.READ,
)
def applicationState(self: "DSManager") -> Tuple[ApplicationStateType, float, AttrQuality]:
"""Reflect the Application State node's state."""
return self._read_attribute("applicationstate", ApplicationStateType.Unknown)
@attribute(
dtype=bool,
access=AttrWriteType.READ_WRITE,
doc=(
"Value to be used for StaticOn when the next pointing model is applied. "
"Setting this to True will enable static pointing model corrections."
),
memorized=True,
)
def staticOnInput(self) -> Tuple[bool, float, AttrQuality]:
"""Indicate the value to be used for StaticOn when the next pointing model is applied."""
return self._read_attribute("staticoninput", False)
@attribute(
dtype=TiltOnType,
access=AttrWriteType.READ_WRITE,
doc=(
"Value to be used for TiltOn when the next pointing model is applied. "
"Options: OFF (0), TILTMETER_ONE (1), TILTMETER_TWO (2)."
),
memorized=True,
)
def tiltOnInput(self) -> Tuple[TiltOnType, float, AttrQuality]:
"""Indicate the value to be used for TiltOn when the next pointing model is applied."""
return self._read_attribute("tiltoninput", TiltOnType.OFF)
@attribute(
dtype=bool,
access=AttrWriteType.READ_WRITE,
doc=(
"Value to be used for AmbTOn when the next pointing model is applied. "
"Setting this to True will enable ambient temperature corrections."
),
memorized=True,
)
def ambTOnInput(self) -> Tuple[bool, float, AttrQuality]:
"""Indicate the value to be used for AmbTOn when the next pointing model is applied."""
return self._read_attribute("ambtoninput", False)
[docs] @attribute(
dtype=DscCtrlState,
access=AttrWriteType.READ,
)
def dscCtrlState(self: "DSManager") -> Tuple[DscCtrlState, float, AttrQuality]:
"""Indicate dsc control state."""
return self._read_attribute("dscctrlstate", DscCtrlState.NO_AUTHORITY)
[docs] @attribute(
dtype=str,
access=AttrWriteType.READ,
)
def displayedDiagnosis(self: "DSManager") -> Tuple[str, float, AttrQuality]:
"""Indicate dsc DisplayedDiagnosis."""
return self._read_attribute("displayeddiagnosis", "Not updated by Component Manager")
# -----------
# Commands
# -----------
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def SetStandbyMode(self: "DSManager") -> DevVarLongStringArrayType:
"""Triggers the Dish to transition to the STANDBY mode.
If the dish is stowed, sets it to full power mode and unstows it. Then, deactivates the
azimuth and elevation axes if they are not in standby mode before setting the dish to
low power mode.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("SetStandbyMode")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def SetPointMode(self: "DSManager") -> DevVarLongStringArrayType:
"""Triggers the Dish to transition to the POINT mode.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("SetPointMode")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def Stow(self: "DSManager") -> DevVarLongStringArrayType:
"""Transition the dish to STOW immediately bypassing the LRC queue.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("Stow")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
doc_in="Unstow the dish",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def Unstow(self: "DSManager") -> DevVarLongStringArrayType:
"""Triggers the dish to unstow.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("Unstow")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=int,
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def SetIndexPosition(self: "DSManager", band: int) -> DevVarLongStringArrayType:
"""Triggers the Dish to transition the feed indexer to the `argin` position.
If the dish is in low power mode, it is first set to full power mode.
:param band: The band which the dish is instructed to move the feed indexer to
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("SetIndexPosition")
result_code, unique_id = handler(band)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=(float,),
dtype_out="DevVarLongStringArray",
doc_in="[0]: Azimuth\n[1]: Elevation",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def Slew(self: "DSManager", values: list[float]) -> DevVarLongStringArrayType:
"""Trigger the Dish to start moving to the commanded (Az,El) position.
If the dish is in low power mode, it is first set to full power mode.
:param values: float values for the az position, el position
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("Slew")
result_code, unique_id = handler(
values, # type: ignore[has-type]
self.component_manager.component_state.get("azimuthspeed", DEFAULT_AZIMUTH_SPEED_DPS),
self.component_manager.component_state.get(
"elevationspeed", DEFAULT_ELEVATION_SPEED_DPS
),
)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def Track(self: "DSManager") -> DevVarLongStringArrayType:
"""Triggers the Dish to start tracking.
Dish will start tracking the positions present in the track table using the set
interpolation mode.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("Track")
inter_mode = self.component_manager.component_state.get(
"trackinterpolationmode", TrackInterpolationMode.SPLINE
)
result_code, unique_id = handler(
inter_mode, self.component_manager.get_current_tai_offset_with_manual_fallback
) # type: ignore[has-type]
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def TrackStop(self: "DSManager") -> DevVarLongStringArrayType:
"""Triggers the dish to stop tracking but will not apply brakes.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("TrackStop")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=(float,),
dtype_out="DevVarLongStringArray",
doc_in="""
Load (global) static tracking offsets.
The offset is loaded immediately and is not cancelled
between tracks. The static offset introduces a positional adjustment to facilitate
reference pointing and the five-point calibration. The static offsets are added the
output of the interpolator before the correction of the static pointing model.
Note: If the static pointing correction is switched off, the static offsets remain as
an offset to the Azimuth and Elevation positions and need to be set to zero manually.
Static offset parameters are:
[0] Off_Xel, [1] Off_El
""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def TrackLoadStaticOff(self: "DSManager", values: list[float]) -> DevVarLongStringArrayType:
"""Load the given static pointing model offsets.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("TrackLoadStaticOff")
result_code, unique_id = handler(values)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=(float,), # needs to be double for the timestamps
dtype_out="DevVarLongStringArray",
doc_in="""
[0]: LoadMode\n
[1]: SequenceLength,\n
[2 - end]: TrackTable, length should be a multiple of 3 values:\n
(timestamp, azimuth coordinate, elevation coordinate)
NB. Load mode enum might be different from that accepted by the dish
controller and is dynamically mapped at runtime based on OPCUA enum values.
""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def TrackLoadTable(self: "DSManager", values: list[float]) -> DevVarLongStringArrayType:
"""Load the track table with the given load mode.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("TrackLoadTable")
result_code, unique_id = handler(values)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=None,
dtype_out="DevVarLongStringArray",
doc_in="""Resets the axes errors.""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def ResetAxesCommand(self: "DSManager") -> DevVarLongStringArrayType:
"""Reset the axes errors.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("ResetAxesCommand")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=None,
dtype_out="DevVarLongStringArray",
doc_in="""Take authority.""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def TakeAuthority(self: "DSManager") -> DevVarLongStringArrayType:
"""Take authority.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("TakeAuthority")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=None,
dtype_out="DevVarLongStringArray",
doc_in="""Retake authority.""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def ReTakeAuthority(self: "DSManager") -> DevVarLongStringArrayType:
"""Retake authority.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("ReTakeAuthority")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=(float,),
dtype_out="DevVarLongStringArray",
doc_in="""
Sets mode to Low or Full Power.
The current low or full power active is reported by the powerState attribute.
The current PowerLimit (kW) is not reported as it cannot be read from the DSC.
LowPowerOn represented by a boolean where True(1.0) sets LowPowerOn to
True , and False(0.0) sets LowPowerOn to False.
[0] LowPowerOn (Boolean): True (1.0), False (0.0)\n
[1] Power limit(kW): Maximum power consumption allowed.
""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def SetPowerMode(self: "DSManager", values: list[float]) -> DevVarLongStringArrayType:
"""Trigger the Dish to transition to SetPowerMode.
:param values: float values for LowPowerOn(boolean), Power limit(kW)
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("SetPowerMode")
result_code, unique_id = handler(values)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_in=None,
dtype_out="DevDouble",
doc_in="""
Returns the current TAIoffset as read from the DSC.
If the read for the TAIoffset fails on the DSC then the TAI offset will be calculated
manually by the DSManager device.
""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def GetCurrentTAIOffset(self: "DSManager") -> float:
"""Get the current TAIoffset as read from the DSC with a manual calculation fallback.
If the read for the TAIoffset fails on the DSC then the TAI offset will be calculated
manually by the DSManager device.
:return: TAI offset
:rtype: float
"""
return self.component_manager.get_current_tai_offset_with_manual_fallback()
[docs] @command( # type: ignore[misc]
dtype_in=(str,),
dtype_out="DevVarLongStringArray",
doc_in="""
Execute an OPC-UA command on the DSC.
The argument to this command is a list of strings. With the first value being the
OPC-UA command node path (in dot notation & relative to the PLC_PRG node), followed by
the list of arguments for the given command.
""",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def ExecuteOPCUACommand(self: "DSManager", values: list[str]) -> DevVarLongStringArrayType:
"""Execute an OPC-UA command on the DSC.
:param values: string values for the command name and any required arguments
:type values: list[str]
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("ExecuteOPCUACommand")
result_code, unique_id = handler(
values, # type: ignore[has-type]
)
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
doc_in="Clear the interlock status once interlocks or E-Stops have been cleared.",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def InterlockAck(self: "DSManager") -> DevVarLongStringArrayType:
"""Clear the interlock status once interlocks or E-Stops have been cleared.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("InterlockAck")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs] @command( # type: ignore[misc]
dtype_out="DevVarLongStringArray",
doc_in="Release command authority of the dish.",
)
@BaseInfoIt(show_args=True, show_kwargs=True, show_ret=True)
@last_command_invoked_decorator
def ReleaseAuth(self: "DSManager") -> DevVarLongStringArrayType:
"""Releases command authority immediately bypassing the LRC queue.
:return: A tuple containing a return code and a string
message indicating status.
"""
handler = self.get_command_object("ReleaseAuth")
result_code, unique_id = handler()
return ([result_code], [unique_id])
[docs]def main(args: Any = None, **kwargs: Any) -> None:
"""Launch a DSManager device."""
return run((DSManager,), args=args, **kwargs)
if __name__ == "__main__":
main()