diff --git a/CDB/LOFAR_ConfigDb.json b/CDB/LOFAR_ConfigDb.json
index 508f4d3e3f2c96eb04cda4b2fea757a16d061124..2ba1ebe751f6397710f82e8d246c74fe9eaae8dd 100644
--- a/CDB/LOFAR_ConfigDb.json
+++ b/CDB/LOFAR_ConfigDb.json
@@ -91,6 +91,9 @@
"APSPU, APSPU_TEMP_error_R",
"UNB2, UNB2_TEMP_error_R",
"RECV, RECV_TEMP_error_R"
+ ],
+ "Shutdown_Device_List":[
+ "STAT/SDP/1", "STAT/UNB2/1", "STAT/RECV/1", "STAT/APSCT/1", "STAT/APSPU/1"
]
}
}
diff --git a/CDB/stations/simulators_ConfigDb.json b/CDB/stations/simulators_ConfigDb.json
index c2f70a5999d78628421abeea7012f5cc2079f2ac..7cd92391917029be134fdc9fd4846b5540153663 100644
--- a/CDB/stations/simulators_ConfigDb.json
+++ b/CDB/stations/simulators_ConfigDb.json
@@ -123,6 +123,9 @@
"properties": {
"Alarm_Error_List": [
"RECV, HBAT_LED_on_RW"
+ ],
+ "Shutdown_Device_List":[
+ "STAT/SDP/1", "STAT/UNB2/1", "STAT/RECV/1", "STAT/APSCT/1", "STAT/APSPU/1"
]
}
}
diff --git a/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py b/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py
index f00be30ab836b2735b59c0b9c177ee450698a102..256f78bf155b7a3a17bd70e426478dd56bfc8182 100644
--- a/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py
+++ b/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py
@@ -175,7 +175,7 @@ class CustomCollector(object):
# obtain extended info about all attributes
attr_infos = {attr_info.name: attr_info for attr_info in dev.attribute_list_query()}
- if dev.state() not in [DevState.STANDBY, DevState.ON, DevState.ALARM]:
+ if dev.state() not in [DevState.STANDBY, DevState.ON, DevState.ALARM, DevState.DISABLE]:
logger.error(f"Error processing device {device_name}: it is in state {dev.state()}")
# at least log state & status
diff --git a/tangostationcontrol/docs/source/devices/using.rst b/tangostationcontrol/docs/source/devices/using.rst
index 96881178bfa260acd182397ff4fd80a0794e3763..053d73bda33703c187f6bd9de9d7d84194647254 100644
--- a/tangostationcontrol/docs/source/devices/using.rst
+++ b/tangostationcontrol/docs/source/devices/using.rst
@@ -22,7 +22,8 @@ The state of a device is then queried with ``device.state()``. Each device can b
- ``DevState.STANDBY``: The device is initialised and ready to be configured further,
- ``DevState.ON``: The device is operational,
- ``DevState.ALARM``: The device is operational, but one or more attributes are in alarm,
-- ``DevState.FAULT``: The device is malfunctioning. Functionality cannot be counted on.
+- ``DevState.FAULT``: The device is malfunctioning. Functionality cannot be counted on,
+- ``DevState.DISABLE``: The device is not operating because its hardware has been shut down.
- The ``device.state()`` function can throw an error, if the device cannot be reached at all. For example, because it's docker container is not running. See the :ref:`docker` device on how to start it.
@@ -49,6 +50,10 @@ The state of a device is then queried with ``device.state()``. Each device can b
alarm -> fault [label = "device", color="green"];
fault -> init [label = "user", color="red"];
fault -> off [label = "user", color="red"];
+ standby -> disable [label = "user", color="green"];
+ on -> disable [label = "user", color="green"];
+ alarm -> disable [label = "user", color="green"];
+ disable -> off [label= "user", color="red"];
}
@@ -58,6 +63,8 @@ Each device provides the following commands to change the state:
:warm_boot(): Turn on the device, but do not change the hardware. Moves from ``OFF`` to ``ON``.
+:disable_hardware(): Shut down the hardware related to the device. Moves from ``STANDBY``, ``ON`` or ``ALARM`` to ``DISABLE``
+
:off(): Turn the device ``OFF`` from any state.
The following procedure is a good way to bring a device to ``ON`` from any state::
diff --git a/tangostationcontrol/setup.py b/tangostationcontrol/setup.py
new file mode 100644
index 0000000000000000000000000000000000000000..b908cbe55cb344569d32de1dfc10ca7323828dc5
--- /dev/null
+++ b/tangostationcontrol/setup.py
@@ -0,0 +1,3 @@
+import setuptools
+
+setuptools.setup()
diff --git a/tangostationcontrol/tangostationcontrol/common/states.py b/tangostationcontrol/tangostationcontrol/common/states.py
index 27bc5c481a9b7c33e347d68acf8f17b45f2f2315..cc458005621a3116b7a05839e81727d3dc796e70 100644
--- a/tangostationcontrol/tangostationcontrol/common/states.py
+++ b/tangostationcontrol/tangostationcontrol/common/states.py
@@ -6,7 +6,7 @@ OPERATIONAL_STATES = [DevState.ON, DevState.ALARM]
# States in which Initialise() has happened, and the hardware
# can thus be configured or otherwise interacted with.
-INITIALISED_STATES = OPERATIONAL_STATES + [DevState.STANDBY]
+INITIALISED_STATES = OPERATIONAL_STATES + [DevState.STANDBY, DevState.DISABLE]
# States in which most commands are allowed
DEFAULT_COMMAND_STATES = INITIALISED_STATES
diff --git a/tangostationcontrol/tangostationcontrol/devices/antennafield.py b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
index 22e5f922fbc3853bf29b865cc075b206116e89e2..4c39a6bdea1db44f52c87f2fd3a4e6add71e3d99 100644
--- a/tangostationcontrol/tangostationcontrol/devices/antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
@@ -6,11 +6,15 @@
""" AntennaField Device Server for LOFAR2.0
"""
+from enum import IntEnum
+from math import pi
+import numpy
+
+# PyTango imports
from tango import DeviceProxy, DevSource, AttrWriteType, DevVarFloatArray, DevVarLongArray
from tango.server import device_property, attribute, command
-import numpy
-from math import pi
+# Additional import
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
@@ -26,6 +30,17 @@ __all__ = ["AntennaField", "HBATToRecvMapper", "main"]
# Highest number of HBA tiles we support per AntennaField
MAX_NUMBER_OF_HBAT = 96
+class AntennaUse(IntEnum):
+ AUTO = 0 # use antenna only if its OK or SUSPICIOUS
+ ON = 1 # force antenna to be on, regardless of quality
+ OFF = 2 # force antenna to be off, regardless of quality
+
+class AntennaQuality(IntEnum):
+ OK = 0
+ SUSPICIOUS = 1
+ BROKEN = 2
+ BEYOND_REPAIR = 3
+
class mapped_attribute(attribute):
def __init__(self, mapping_attribute, dtype, max_dim_x, max_dim_y=0, access=AttrWriteType.READ, **kwargs):
@@ -70,6 +85,22 @@ class AntennaField(lofar_device):
calculated, as well as the geohash.
"""
+ # ----- Antenna states
+
+ Antenna_Quality = device_property(
+ doc="Operational quality state of each antenna",
+ dtype='DevVarUShortArray',
+ mandatory=False,
+ default_value = numpy.array([AntennaQuality.OK] * MAX_NUMBER_OF_HBAT)
+ )
+
+ Antenna_Use = device_property(
+ doc="Operational State of each antenna",
+ dtype='DevVarUShortArray',
+ mandatory=False,
+ default_value = numpy.array([AntennaUse.AUTO] * MAX_NUMBER_OF_HBAT)
+ )
+
# ----- Position information
Antenna_Field_Reference_ITRF = device_property(
@@ -164,6 +195,13 @@ class AntennaField(lofar_device):
default_value = []
)
+ Antenna_Quality_R = attribute(access=AttrWriteType.READ,
+ dtype=(numpy.uint16,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ Antenna_Use_R = attribute(access=AttrWriteType.READ,
+ dtype=(numpy.uint16,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ Antenna_Usage_Mask_R = attribute(access=AttrWriteType.READ,
+ dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT)
+
HBAT_ANT_mask_RW = mapped_attribute("ANT_mask_RW", dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT, access=AttrWriteType.READ_WRITE)
HBAT_BF_delay_steps_R = mapped_attribute("HBAT_BF_delay_steps_R", dtype=((numpy.int64,),), max_dim_x=NUMBER_OF_ELEMENTS_PER_TILE * 2, max_dim_y=MAX_NUMBER_OF_HBAT)
HBAT_BF_delay_steps_RW = mapped_attribute("HBAT_BF_delay_steps_RW", dtype=((numpy.int64,),), max_dim_x=NUMBER_OF_ELEMENTS_PER_TILE * 2, max_dim_y=MAX_NUMBER_OF_HBAT, access=AttrWriteType.READ_WRITE)
@@ -209,6 +247,19 @@ class AntennaField(lofar_device):
doc='Number of HBAT in this field',
dtype=numpy.int32)
+ def read_Antenna_Use_R(self):
+ return self.Antenna_Use
+
+ def read_Antenna_Quality_R(self):
+ return self.Antenna_Quality
+
+ def read_Antenna_Usage_Mask_R(self):
+ antenna_usage = numpy.zeros(MAX_NUMBER_OF_HBAT, dtype=bool)
+ for n in range(0, MAX_NUMBER_OF_HBAT):
+ antenna_usage[n] = (self.read_attribute('Antenna_Use_R')[n] == AntennaUse.ON
+ or (self.read_attribute('Antenna_Use_R')[n] == AntennaUse.AUTO and self.read_attribute('Antenna_Quality_R')[n] <= AntennaQuality.SUSPICIOUS))
+ return antenna_usage
+
def read_nr_tiles_R(self):
# The number of tiles should be equal to:
# * the number of elements in the HBAT_Control_to_RECV_mapping (after reshaping),
@@ -278,12 +329,6 @@ class AntennaField(lofar_device):
def read_HBAT_reference_GEOHASH_R(self):
return GEO_to_GEOHASH(self.read_HBAT_reference_GEO_R())
- @log_exceptions()
- def configure_for_initialise(self):
- super().configure_for_initialise()
- self.__setup_all_receiver_proxies()
- self.__setup_mapper()
-
def __setup_all_receiver_proxies(self):
self.recv_proxies = []
@@ -317,6 +362,43 @@ class AntennaField(lofar_device):
for idx, recv_proxy in enumerate(self.recv_proxies):
recv_proxy.write_attribute(mapped_point, mapped_value[idx])
+ # --------
+ # Overloaded functions
+ # --------
+
+ @log_exceptions()
+ def configure_for_initialise(self):
+ super().configure_for_initialise()
+ self.__setup_all_receiver_proxies()
+ self.__setup_mapper()
+
+ @log_exceptions()
+ def _prepare_hardware(self):
+ # Initialise the RCU hardware.
+ for recv_proxy in self.recv_proxies:
+ RCU_mask = recv_proxy.RCU_mask_RW
+ # Set the mask to all Trues
+ recv_proxy.RCU_mask_RW = [True] * 32
+ # Turn off the RCUs
+ recv_proxy.RCU_off()
+
+ # TODO(Stefano): restore wait attribute
+ #recv_proxy.wait_attribute("RECVTR_translator_busy_R", False, recv_proxy.RCU_On_Off_timeout)
+
+ # Restore the mask
+ recv_proxy.RCU_mask_RW = RCU_mask
+ # Turn on the RCUs
+ recv_proxy.RCU_on()
+
+ # TODO(Stefano): restore wait attribute
+ #recv_proxy.wait_attribute("RECVTR_translator_busy_R", False, recv_proxy.RCU_On_Off_timeout)
+
+ @log_exceptions()
+ def _initialise_hardware(self):
+ # Disable controlling the tiles that fall outside the mask
+ # WARN: Needed in configure_for_initialise but Tango does not allow to write attributes in INIT state
+ self.proxy.write_attribute('HBAT_ANT_mask_RW', self.read_attribute('Antenna_Usage_Mask_R'))
+
# --------
# Commands
# --------
diff --git a/tangostationcontrol/tangostationcontrol/devices/apsct.py b/tangostationcontrol/tangostationcontrol/devices/apsct.py
index 18a7f6fd7e49aeffdd1091e7e6a65766835be74a..60563f1138c125c79274b1c2f5342b7df68c02e8 100644
--- a/tangostationcontrol/tangostationcontrol/devices/apsct.py
+++ b/tangostationcontrol/tangostationcontrol/devices/apsct.py
@@ -133,6 +133,13 @@ class APSCT(opcua_device):
else:
raise Exception("200MHz signal is not locked. The subrack probably do not receive clock input or the CLK PCB is broken?")
+ def _disable_hardware(self):
+ """ Disable the APSCT hardware. """
+
+ # Turn off the APSCT
+ self.APSCT_off()
+ self.wait_attribute("APSCTTR_translator_busy_R", False, self.APSCT_On_Off_timeout)
+
# --------
# Commands
# --------
diff --git a/tangostationcontrol/tangostationcontrol/devices/apspu.py b/tangostationcontrol/tangostationcontrol/devices/apspu.py
index 7f40116222956f6066d2e00324ba18bfdaa43352..08c43f5a4362b22bd15ab0065e243f741add98ae 100644
--- a/tangostationcontrol/tangostationcontrol/devices/apspu.py
+++ b/tangostationcontrol/tangostationcontrol/devices/apspu.py
@@ -98,6 +98,9 @@ class APSPU(opcua_device):
# overloaded functions
# --------
+ def _disable_hardware(self):
+ """ Disable the APSPU hardware. """
+ super()._disable_hardware()
# --------
# Commands
diff --git a/tangostationcontrol/tangostationcontrol/devices/lofar_device.py b/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
index 3f127261973bda750d246a35bd514a868592d6ec..665caca7a4d0439e0a2195d9939210229b71cdb0 100644
--- a/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/lofar_device.py
@@ -45,18 +45,22 @@ class lofar_device(Device, metaclass=DeviceMeta):
ON = Device is fully configured, functional, controls the hardware, and is possibly actively running,
ALARM = Device is operating but one of its attributes is out of range,
FAULT = Device detected an unrecoverable error, and is thus malfunctional,
+ DISABLE = Device has shut down all its dependant hardware
OFF = Device is turned off, drops connection to the hardware,
The following state transitions are implemented:
- boot -> OFF: Triggered by tango. Device will be instantiated,
- OFF -> INIT: Triggered by device. Device will initialise (connect to hardware, other devices),
- INIT -> STANDBY: Triggered by device. Device is initialised, and is ready for additional configuration by the user,
- STANDBY -> ON: Triggered by user. Device reports to be functional,
- ON -> ALARM: Triggered by tango. Device has attribute(s) with value(s) exceeding their alarm treshold,
- * -> FAULT: Triggered by device. Device has degraded to malfunctional, for example because the connection to the hardware is lost,
- * -> FAULT: Triggered by user. Emulate a forced malfunction for integration testing purposes,
- * -> OFF: Triggered by user. Device is turned off. Triggered by the Off() command,
- FAULT -> INIT: Triggered by user. Device is reinitialised to recover from an error,
+ boot -> OFF: Triggered by tango. Device will be instantiated,
+ OFF -> INIT: Triggered by device. Device will initialise (connect to hardware, other devices),
+ INIT -> STANDBY: Triggered by device. Device is initialised, and is ready for additional configuration by the user,
+ STANDBY -> ON: Triggered by user. Device reports to be functional,
+ STANDBY -> DISABLE: Triggered by user. Device has shut down its hardware. Triggered by the disable_hardware() command,
+ ON -> DISABLE: Triggered by user. Device has shut down its hardware. Triggered by the disable_hardware() command,
+ ALARM -> DISABLE: Triggered by user. Device has shut down its hardware. Triggered by the disable_hardware() command,
+ ON -> ALARM: Triggered by tango. Device has attribute(s) with value(s) exceeding their alarm treshold,
+ * -> FAULT: Triggered by device. Device has degraded to malfunctional, for example because the connection to the hardware is lost,
+ * -> FAULT: Triggered by user. Emulate a forced malfunction for integration testing purposes,
+ * -> OFF: Triggered by user. Device is turned off. Triggered by the Off() command,
+ FAULT -> INIT: Triggered by user. Device is reinitialised to recover from an error,
The user triggers their transitions by the commands reflecting the target state (Initialise(), On(), Fault()).
"""
@@ -349,6 +353,24 @@ class lofar_device(Device, metaclass=DeviceMeta):
# This is just the command version of _initialise_hardware().
self._initialise_hardware()
+ @only_in_states(INITIALISED_STATES)
+ @fault_on_error()
+ @command()
+ @DebugIt()
+ def disable_hardware(self):
+ """ Disable the hardware related to the device. """
+
+ if self.get_state() == DevState.DISABLE:
+ # Already disabled.
+ logger.warning("Requested to go to DISABLE state, but am already in DISABLE state.")
+ return
+
+ self._disable_hardware()
+
+ # Set state to DISABLE
+ self.set_state(DevState.DISABLE)
+ self.set_status("Device is in the DISABLE state.")
+
@only_in_states(DEFAULT_COMMAND_STATES)
@command(dtype_out = DevDouble)
def max_archiving_load(self):
@@ -398,6 +420,10 @@ class lofar_device(Device, metaclass=DeviceMeta):
""" Override this method to initialise any hardware after configuring it. """
pass
+ def _disable_hardware(self):
+ """ Override this method to disable any hardware related to the device. """
+ pass
+
def read_attribute(self, attr_name):
""" Read the value of a certain attribute (directly from the hardware). """
diff --git a/tangostationcontrol/tangostationcontrol/devices/recv.py b/tangostationcontrol/tangostationcontrol/devices/recv.py
index ff2e4166fca4ba038895de25d268792274b90702..037b8adb69fa2515f5a3e2dbaa4ed00c8a2ab43b 100644
--- a/tangostationcontrol/tangostationcontrol/devices/recv.py
+++ b/tangostationcontrol/tangostationcontrol/devices/recv.py
@@ -235,6 +235,28 @@ class RECV(opcua_device):
# by a fixed amount, the average of all steps. Doing so should result
# in positive delays regardless of the pointing direction.
self.HBAT_bf_delay_offset = numpy.mean(self.HBAT_bf_delay_step_delays)
+
+ def _initialise_hardware(self):
+ """ Initialise the RCU hardware. """
+
+ # Cycle RCUs
+ self.RCU_off()
+ self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
+ self.RCU_on()
+ self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
+
+ def _disable_hardware(self):
+ """ Disable the RECV hardware. """
+
+ # Save actual mask values
+ RCU_mask = self.proxy.RCU_mask_RW
+ # Set the mask to all Trues
+ self.RCU_mask_RW = [True] * 32
+ # Turn off the RCUs
+ self.RCU_off()
+ self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
+ # Restore the mask
+ self.RCU_mask_RW = RCU_mask
# --------
# internal functions
@@ -319,15 +341,6 @@ class RECV(opcua_device):
"""
self.opcua_connection.call_method(["RCU_DTH_on"])
- def _initialise_hardware(self):
- """ Initialise the RCU hardware. """
-
- # Cycle RCUs
- self.RCU_off()
- self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
- self.RCU_on()
- self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
-
# ----------
# Run server
# ----------
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
index 309016fa7edb9256a88713eac13205cc79eae090..3a302239c317b83d1f1598a27b8fe1a5222d3858 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
@@ -37,6 +37,7 @@ class Beamlet(opcua_device):
N_BEAMLETS_CTRL = 488
N_BEAMSETS_CTRL = 2
N_POL_BF = 2
+ P_SUM = 2 # number of hops that the data of the stream has traveled to reach the BSN aligner on this node
# -----------------
# Device Properties
@@ -154,6 +155,19 @@ class Beamlet(opcua_device):
FPGA_bf_weights_yy_R = attribute_wrapper(comms_annotation=["FPGA_bf_weights_yy_R"], datatype=numpy.uint32, dims=(N_PN, A_PN, N_BEAMLETS_CTRL))
FPGA_bf_weights_yy_RW = attribute_wrapper(comms_annotation=["FPGA_bf_weights_yy_RW"], datatype=numpy.uint32, dims=(N_PN, A_PN, N_BEAMLETS_CTRL), access=AttrWriteType.READ_WRITE)
+ # boolean[N_pn][N_beamsets_ctrl][P_sum]
+ FPGA_bf_rx_align_stream_enable_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_stream_enable_R"], datatype=bool, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+ FPGA_bf_rx_align_stream_enable_RW = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_stream_enable_RW"], datatype=bool, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM), access=AttrWriteType.READ_WRITE)
+
+ # int64[N_pn][N_beamsets_ctrl][P_sum]
+ FPGA_bf_rx_align_bsn_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_bsn_R"], datatype=numpy.int64, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+
+ #int32[N_pn][N_beamsets_ctrl][P_sum]
+ FPGA_bf_rx_align_nof_packets_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_nof_packets_R"], datatype=numpy.int32, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+ FPGA_bf_rx_align_nof_valid_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_nof_valid_R"], datatype=numpy.int32, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+ FPGA_bf_rx_align_latency_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_latency_R"], datatype=numpy.int32, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+ FPGA_bf_rx_align_nof_replaced_packets_R = attribute_wrapper(comms_annotation=["FPGA_bf_rx_align_nof_replaced_packets_R"], datatype=numpy.int32, dims=(N_PN, N_BEAMSETS_CTRL, P_SUM))
+
subband_select_RW = attribute(dtype=(numpy.uint32,), max_dim_x=N_BEAMLETS_CTRL, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed")
def read_subband_select_RW(self):
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
index 0b44a14292c9e2f0e053001a6e071adcd977f978..256d156a4337bffb46e2ed085cbcc151d6d4f5b2 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
@@ -99,7 +99,7 @@ class DigitalBeam(beam_device):
def write_antenna_select_RW(self, antennas):
for input_nr, antenna_nr in enumerate(self.Input_to_Antenna_Mapping):
- if antenna_nr >= 0:
+ if self.antennafield_proxy.Antenna_Usage_Mask_R[input_nr] and antenna_nr >= 0:
self._input_select[input_nr] = antennas[antenna_nr]
# ----------
@@ -120,6 +120,7 @@ class DigitalBeam(beam_device):
self.antennafield_proxy = DeviceProxy(self.AntennaField_Device)
self.antennafield_proxy.set_source(DevSource.DEV)
+ antenna_usage_mask = self.antennafield_proxy.Antenna_Usage_Mask_R
self.beamlet_proxy = DeviceProxy(self.Beamlet_Device)
self.beamlet_proxy.set_source(DevSource.DEV)
@@ -132,7 +133,7 @@ class DigitalBeam(beam_device):
# Use reference position for any missing antennas so they always get a delay of 0
input_itrf = numpy.array([reference_itrf] * self.NUM_INPUTS)
for input_nr, antenna_nr in enumerate(self.Input_to_Antenna_Mapping):
- if antenna_nr >= 0:
+ if antenna_usage_mask[input_nr] and antenna_nr >= 0:
input_itrf[input_nr] = antenna_itrf[antenna_nr]
# a delay calculator
@@ -142,7 +143,8 @@ class DigitalBeam(beam_device):
self.relative_input_positions = input_itrf - reference_itrf
# use all antennas in the mapping for all beamlets, unless specified otherwise
- input_select = numpy.repeat((numpy.array(self.Input_to_Antenna_Mapping) >= 0), self.NUM_BEAMLETS).reshape(self.NUM_INPUTS, self.NUM_BEAMLETS)
+ input_select_mask = numpy.logical_and(numpy.array(self.Input_to_Antenna_Mapping) >= 0, antenna_usage_mask)
+ input_select = numpy.repeat(input_select_mask, self.NUM_BEAMLETS).reshape(self.NUM_INPUTS, self.NUM_BEAMLETS)
self.write_input_select_RW(input_select)
# --------
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
index b0038a0f1161c6c926f88815ebc27d0eaba53ef3..2a7ca8991f168f6bf5ba673207da6dc9dbb09817 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
@@ -283,6 +283,17 @@ class SDP(opcua_device):
# Wait for the firmware to be loaded (ignoring masked out elements)
self.wait_attribute("FPGA_boot_image_R", lambda attr: ((attr == 1) | ~wait_for).all(), 60)
+
+ def _disable_hardware(self):
+ """ Disable the SDP hardware. """
+ # Save actual mask values
+ TR_fpga_mask = self.proxy.TR_fpga_mask_RW
+ # Set the mask to all Trues
+ self.TR_fpga_mask_RW = [True] * 16
+ # Boot the boot image firmware
+ self.FPGA_boot_image_RW = [0] * self.N_pn
+ # Restore the mask
+ self.TR_fpga_mask_RW = TR_fpga_mask
# --------
# Commands
diff --git a/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py b/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
index 912753f82c655a65d785af891fbf03222db86c3b..777343c80492bd03de0f84c21d5f1a1c29e09b90 100644
--- a/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
+++ b/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
@@ -12,8 +12,8 @@ from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
-
-from tango import Util, DeviceProxy, AttributeInfoEx, AttrDataFormat, EventType
+from tango.server import command
+from tango import Util, DeviceProxy, AttributeInfoEx, AttrDataFormat, EventType, DevSource, DebugIt
from tango.server import attribute, device_property
import numpy as np
@@ -60,6 +60,12 @@ class TemperatureManager(lofar_device):
default_value=[]
)
+ Shutdown_Device_List = device_property(
+ dtype=[str],
+ mandatory=False,
+ default_value=["STAT/SDP/1", "STAT/UNB2/1", "STAT/RECV/1", "STAT/APSCT/1", "STAT/APSPU/1"]
+ )
+
# ----------
# Attributes
# ----------
@@ -87,6 +93,7 @@ class TemperatureManager(lofar_device):
# get the proxy to the device
proxy = DeviceProxy(f"{ds_inst}/{proxy_name}/{instance_number}")
+ proxy.set_source(DevSource.DEV)
# make sure the attribute is polled, otherwise we wont receive events
if not proxy.is_attribute_polled(f"{attribute_name}"):
@@ -142,19 +149,24 @@ class TemperatureManager(lofar_device):
logger.warning(f"Detected a temperature alarm for {event.device}: {event.attr_value.name} := {event.attr_value.value}")
self.auto_shutdown_hardware()
+ # --------
+ # Commands
+ # --------
+ @command()
+ @DebugIt()
def auto_shutdown_hardware(self):
"""
This function automatically shuts down all hardware devices whenever a temperature alarm is detected
- In the future there should be a strategy for turning off devices
"""
- DeviceProxy("STAT/SDP/1").off()
- DeviceProxy("STAT/UNB2/1").off()
- DeviceProxy("STAT/RECV/1").off()
- DeviceProxy("STAT/APSCT/1").off()
- DeviceProxy("STAT/APSPU/1").off()
- DeviceProxy("STAT/PSOC/1").off()
- logger.warning(f"Temperature alarm triggered auto shutdown of all hardware devices")
+ for dev_name in self.Shutdown_Device_List:
+ try:
+ proxy = DeviceProxy(dev_name)
+ proxy.disable_hardware()
+ except Exception as e:
+ logger.warning(f"Automatic hardware shutdown of device {dev_name} has failed: {e.args[0]}")
+ # TODO(Stefano): Add "STAT/PSOC/1" to the shutdown list and develop its behaviour
+ logger.warning(f"Temperature alarm triggered auto shutdown of all hardware devices")
# ----------
# Run server
diff --git a/tangostationcontrol/tangostationcontrol/devices/unb2.py b/tangostationcontrol/tangostationcontrol/devices/unb2.py
index 3168621529a30a3c9b9690f2a07883ae04a965db..8ec767588ae0e9fe333411b763d47c765da20649 100644
--- a/tangostationcontrol/tangostationcontrol/devices/unb2.py
+++ b/tangostationcontrol/tangostationcontrol/devices/unb2.py
@@ -198,6 +198,28 @@ class UNB2(opcua_device):
# overloaded functions
# --------
+ def _initialise_hardware(self):
+ """ Initialise the UNB2 hardware. """
+
+ # Cycle UNB2s
+ self.UNB2_off()
+ self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)
+ self.UNB2_on()
+ self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)
+
+ def _disable_hardware(self):
+ """ Disable the UNB2 hardware. """
+
+ # Save actual mask values
+ UNB2_mask = self.proxy.UNB2_mask_RW
+ # Set the mask to all Trues
+ self.UNB2_mask_RW = [True] * 2
+ # Turn off the uniboards
+ self.UNB2_off()
+ self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)
+ # Restore the mask
+ self.UNB2_mask_RW = UNB2_mask
+
# --------
# Commands
# --------
@@ -222,15 +244,6 @@ class UNB2(opcua_device):
"""
self.opcua_connection.call_method(["UNB2_on"])
- def _initialise_hardware(self):
- """ Initialise the UNB2 hardware. """
-
- # Cycle UNB2s
- self.UNB2_off()
- self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)
- self.UNB2_on()
- self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)
-
# ----------
# Run server
# ----------
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
index 2b34767d6c74df2199017e1a884a0e2165af7fb0..f14a2838d489b3f40a03186d565b56d75190f950 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
@@ -7,8 +7,10 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+import numpy
+from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
from .base import AbstractTestBases
class TestAntennaFieldDevice(AbstractTestBases.TestDeviceBase):
@@ -30,3 +32,41 @@ class TestAntennaFieldDevice(AbstractTestBases.TestDeviceBase):
def test_property_recv_devices_has_one_receiver(self):
result = self.proxy.get_property("RECV_devices")
self.assertSequenceEqual(result["RECV_devices"], ["STAT/RECV/1"])
+
+ def test_HBAT_ANT_mask_RW_configured_after_Antenna_Usage_Mask(self):
+ """ Verify if HBAT_ANT_mask_RW values are correctly configured from Antenna_Usage_Mask values """
+ recv_proxy = self.setup_recv_proxy()
+ antennafield_proxy = self.proxy
+ numpy.testing.assert_equal(numpy.array([True] * 96), recv_proxy.ANT_mask_RW)
+
+ antenna_qualities = numpy.array([AntennaQuality.OK] * 96)
+ antenna_use = numpy.array([AntennaUse.ON] + [AntennaUse.AUTO] * 95)
+ antenna_properties = {'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use}
+ mapping_properties = {"RECV_devices": ["STAT/RECV/1"],
+ "HBAT_Power_to_RECV_mapping": [-1, -1] * 48,
+ "HBAT_Control_to_RECV_mapping": [1, 0 , 1, 1] + [-1, -1] * 46}
+ antennafield_proxy.off()
+ antennafield_proxy.put_property(antenna_properties)
+ antennafield_proxy.put_property(mapping_properties)
+ antennafield_proxy.boot() # initialise hardware values as well
+ numpy.testing.assert_equal(numpy.array([True] * 96), antennafield_proxy.Antenna_Usage_Mask_R)
+ numpy.testing.assert_equal(numpy.array([True] * 2 + [False] * 46), antennafield_proxy.HBAT_ANT_mask_RW)
+ numpy.testing.assert_equal(numpy.array([True] * 2 + [False] * 46 + [False] * 48), recv_proxy.ANT_mask_RW)
+
+ def test_HBAT_ANT_mask_RW_configured_after_Antenna_Usage_Mask_only_one_functioning_antenna(self):
+ """ Verify if HBAT_ANT_mask_RW values are correctly configured from Antenna_Usage_Mask values (only second antenna is OK)"""
+ recv_proxy = self.setup_recv_proxy()
+ antennafield_proxy = self.proxy
+ antenna_qualities = numpy.array([AntennaQuality.BROKEN] + [AntennaQuality.OK] + [AntennaQuality.BROKEN] * 94)
+ antenna_use = numpy.array([AntennaUse.AUTO] * 96)
+ antenna_properties = {'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use}
+ mapping_properties = {"RECV_devices": ["STAT/RECV/1"],
+ "HBAT_Power_to_RECV_mapping": [-1, -1] * 48,
+ "HBAT_Control_to_RECV_mapping": [1, 0 , 1, 1] + [-1, -1] * 46}
+ antennafield_proxy.off()
+ antennafield_proxy.put_property(antenna_properties)
+ antennafield_proxy.put_property(mapping_properties)
+ antennafield_proxy.boot() # initialise hardware values as well
+ numpy.testing.assert_equal(numpy.array([False] + [True] + [False] * 94), antennafield_proxy.Antenna_Usage_Mask_R)
+ numpy.testing.assert_equal(numpy.array([False] + [True] + [False] * 46), antennafield_proxy.HBAT_ANT_mask_RW)
+ numpy.testing.assert_equal(numpy.array([False] + [True] + [False] * 46 + [False] * 48), recv_proxy.ANT_mask_RW)
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
index c62076477d0596d2fbf9993b9e81db69a15c12d7..f8857a0f52cb9c9e84394863b52c41e0cbf461b3 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_digitalbeam.py
@@ -9,6 +9,7 @@
# See LICENSE.txt for more info.
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
from .base import AbstractTestBases
@@ -16,6 +17,10 @@ import numpy
class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
+ antenna_qualities_ok = numpy.array([AntennaQuality.OK] * 96)
+ antenna_qualities_only_second = numpy.array([AntennaQuality.BROKEN] + [AntennaQuality.OK] + [AntennaQuality.BROKEN] * 94)
+ antenna_use_ok = numpy.array([AntennaUse.AUTO] * 96)
+
def setUp(self):
"""Intentionally recreate the device object in each test"""
super().setUp("STAT/DigitalBeam/1")
@@ -44,8 +49,21 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
sdp_proxy.warm_boot()
sdp_proxy.set_defaults()
return sdp_proxy
+
+ def setup_antennafield_proxy(self, antenna_qualities, antenna_use):
+ # setup AntennaField
+ NR_TILES = 48
+ antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
+ control_mapping = [[1,i] for i in range(NR_TILES)]
+ antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
+ "HBAT_Control_to_RECV_mapping": numpy.array(control_mapping).flatten(),
+ 'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use})
+ antennafield_proxy.off()
+ antennafield_proxy.boot()
+ return antennafield_proxy
def test_pointing_to_zenith(self):
+ self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
self.setup_sdp_proxy()
self.setup_recv_proxy()
# Setup beamlet configuration
@@ -61,3 +79,25 @@ class TestDeviceDigitalBeam(AbstractTestBases.TestDeviceBase):
# beam weights should now be non-zero, we don't actually check their values for correctness
self.assertNotEqual(0, sum(self.beamlet_proxy.FPGA_bf_weights_xx_yy_RW.flatten()))
+
+ def test_input_select_with_all_antennas_ok(self):
+ """ Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask """
+ antennafield_proxy = self.setup_antennafield_proxy(self.antenna_qualities_ok, self.antenna_use_ok)
+ numpy.testing.assert_equal(numpy.array([True] * 96), antennafield_proxy.Antenna_Usage_Mask_R)
+ self.setUp()
+ self.proxy.warm_boot()
+ expected_input_select = numpy.array([[True] * 488 ] * 48 + [[False] * 488] * 48) # first 48 rows are True
+ numpy.testing.assert_equal(expected_input_select, self.proxy.input_select_RW)
+ expected_antenna_select = numpy.array([[True] * 488 ] * 48)
+ numpy.testing.assert_equal(expected_antenna_select, self.proxy.antenna_select_RW)
+
+ def test_input_select_with_only_second_antenna_ok(self):
+ """ Verify if input and antenna select are correctly calculated following Antennafield.Antenna_Usage_Mask """
+ antennafield_proxy = self.setup_antennafield_proxy(self.antenna_qualities_only_second, self.antenna_use_ok)
+ numpy.testing.assert_equal(numpy.array([False] + [True] + [False] * 94), antennafield_proxy.Antenna_Usage_Mask_R)
+ self.setUp()
+ self.proxy.warm_boot()
+ expected_input_select = numpy.array([[False] * 488 ] + [[True] * 488] + [[False] * 488] * 46 + [[False] * 488] * 48) # first 48 rows are True
+ numpy.testing.assert_equal(expected_input_select, self.proxy.input_select_RW)
+ expected_antenna_select = numpy.array([[False] * 488 ] + [[True] * 488] + [[False] * 488] * 46)
+ numpy.testing.assert_equal(expected_antenna_select, self.proxy.antenna_select_RW)
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
index ce09175933edc18baf555ba091bb8b72ebdb1781..d3f70dd07cf441be3f6deb75ce8358c861987194 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
@@ -14,6 +14,7 @@ from tango import DevState, DevFailed
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
from tangostationcontrol.test.devices.test_observation_base import TestObservationBase
+from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
from .base import AbstractTestBases
class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
@@ -69,11 +70,13 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
# setup AntennaField
antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
control_mapping = [[1,i] for i in range(self.NUM_TILES)]
+ antenna_qualities = numpy.array([AntennaQuality.OK] * 96)
+ antenna_use = numpy.array([AntennaUse.AUTO] * 96)
antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
- "HBAT_Power_to_RECV_mapping": numpy.array(control_mapping).flatten()})
+ "HBAT_Control_to_RECV_mapping": numpy.array(control_mapping).flatten(),
+ 'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use})
antennafield_proxy.off()
- antennafield_proxy.warm_boot()
- antennafield_proxy.set_defaults()
+ antennafield_proxy.boot()
return antennafield_proxy
def setup_beamlet_proxy(self):
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_temperature_manager.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_temperature_manager.py
index 4ba379bb19e6e25c0406cec3301d77137719c317..3e32310e130e219b75f4c9525ae87a5143f92eee 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_temperature_manager.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_temperature_manager.py
@@ -9,6 +9,7 @@
from .base import AbstractTestBases
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
from tango._tango import DevState
+from tango import DeviceProxy
import time
@@ -20,9 +21,9 @@ class TestDeviceTemperatureManager(AbstractTestBases.TestDeviceBase):
def setUp(self):
"""Intentionally recreate the device object in each test"""
self.recv_proxy = self.setup_recv_proxy()
+ self.sdp_proxy = self.setup_sdp_proxy()
super().setUp("STAT/TemperatureManager/1")
-
def tearDown(self):
self.recv_proxy.stop_poll_attribute("HBAT_LED_on_RW")
@@ -36,11 +37,36 @@ class TestDeviceTemperatureManager(AbstractTestBases.TestDeviceBase):
self.assertTrue(recv_proxy.is_attribute_polled(f"HBAT_LED_on_RW"))
return recv_proxy
+ def setup_sdp_proxy(self):
+ # setup SDP, on which this device depends
+ sdp_proxy = TestDeviceProxy("STAT/SDP/1")
+ sdp_proxy.off()
+ sdp_proxy.warm_boot()
+ sdp_proxy.set_defaults()
+ return sdp_proxy
+
def test_alarm(self):
+ # Exclude other devices which raise a TimeoutError, since they wait for the attribute *_translator_busy_R to become False
+ # (set instead to True in this test environment)
+ self.proxy.put_property({"Shutdown_Device_List": ["STAT/SDP/1"]})
+ devices = [DeviceProxy("STAT/SDP/1")]
+
self.proxy.off()
self.proxy.initialise()
self.proxy.on()
+ self.setup_recv_proxy()
+ self.setup_sdp_proxy()
+
+ # make sure none of the devices are in the OFF or FAULT state. Any other state is fine
+ for dev in devices:
+ if dev.state() == DevState.OFF:
+ dev.warm_boot()
+ elif dev.state() == DevState.FAULT:
+ dev.off()
+ dev.warm_boot()
+ self.assertEqual(self.proxy.get_property('Shutdown_Device_List')['Shutdown_Device_List'][0], "STAT/SDP/1")
+
# Here we trigger our own change event by just using an RW attribute
self.recv_proxy.HBAT_LED_on_RW = [[False] * 32] * 96
time.sleep(2)
@@ -52,32 +78,7 @@ class TestDeviceTemperatureManager(AbstractTestBases.TestDeviceBase):
# the TEMP_MANAGER_is_alarming_R should now be True, since it should have detected the temperature alarm.
self.assertTrue(self.proxy.is_alarming_R)
-
- def test_shutdown(self):
- self.proxy.off()
- self.proxy.initialise()
- self.proxy.on()
-
- devices = [TestDeviceProxy("STAT/SDP/1"), TestDeviceProxy("STAT/UNB2/1"), self.recv_proxy,
- TestDeviceProxy("STAT/APSCT/1"), TestDeviceProxy("STAT/APSPU/1"), TestDeviceProxy("STAT/PSOC/1")]
-
- # make sure none of the devices are in the OFF state. Any other state is fine
- for dev in devices:
- if dev.state() == DevState.OFF:
- dev.initialise()
-
- # toggle the attribute to make sure we get a change event to True
- self.recv_proxy.HBAT_LED_on_RW = [[False] * 32] * 96
- self.recv_proxy.HBAT_LED_on_RW = [[True] * 32] * 96
-
- # sleeping here to make sure we've dealt with the above events
- time.sleep(2)
-
- # make sure all the devices are in the OFF state
+
+ # make sure all the hardware devices are in the DISABLE state
for dev in devices:
- self.assertEqual(DevState.OFF, dev.state())
-
-
-
-
-
+ self.assertEqual(DevState.DISABLE, dev.state())
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
index 32dfa8e322100fa16a86c1ab7d8610f617e315ed..991e03543e70c8b308319e45766f797e530a71c6 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
@@ -13,6 +13,7 @@ import datetime
import json
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
from .base import AbstractTestBases
@@ -45,11 +46,13 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
# setup AntennaField
antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
control_mapping = [[1,i] for i in range(self.NR_TILES)]
+ antenna_qualities = numpy.array([AntennaQuality.OK] * 96)
+ antenna_use = numpy.array([AntennaUse.AUTO] * 96)
antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
- "HBAT_Power_to_RECV_mapping": numpy.array(control_mapping).flatten()})
+ "HBAT_Control_to_RECV_mapping": numpy.array(control_mapping).flatten(),
+ 'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use})
antennafield_proxy.off()
- antennafield_proxy.warm_boot()
- antennafield_proxy.set_defaults()
+ antennafield_proxy.boot()
# check if AntennaField really exposes the expected number of tiles
self.assertEqual(self.NR_TILES, antennafield_proxy.nr_tiles_R)
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py
index eca4686a8c65aec51980734383eb80ee3ce1fc15..96eb83300bec3352b1c0572653f38bf96a2e77d9 100644
--- a/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py
@@ -12,7 +12,7 @@ import numpy
from tango.test_context import DeviceTestContext
from tangostationcontrol.devices import antennafield
-from tangostationcontrol.devices.antennafield import HBATToRecvMapper
+from tangostationcontrol.devices.antennafield import HBATToRecvMapper, AntennaQuality, AntennaUse
from tangostationcontrol.test import base
from tangostationcontrol.test.devices import device_base
@@ -340,6 +340,14 @@ class TestAntennafieldDevice(device_base.DeviceTestCase):
AT_PROPERTIES = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0,
'Antenna_Field_Reference_ITRF' : [3.0, 3.0, 3.0], 'Antenna_Field_Reference_ETRS' : [7.0, 7.0, 7.0]}
+ # A mapping where HBATs are all not mapped to power RCUs
+ POWER_NOT_CONNECTED = [[-1, -1]] * 48
+ # A mapping where HBATs are all not mapped to control RCUs
+ CONTROL_NOT_CONNECTED = [[-1, -1]] * 48
+ # A mapping where first two HBATs are mapped on the first Receiver.
+ # The first HBAT control line on RCU 1 and the second HBAT control line on RCU 0.
+ CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1 = [[1, 1], [1, 0]] + [[-1, -1]] * 46
+
def setUp(self):
# DeviceTestCase setUp patches lofar_device DeviceProxy
super(TestAntennafieldDevice, self).setUp()
@@ -353,4 +361,31 @@ class TestAntennafieldDevice(device_base.DeviceTestCase):
at_properties_v2 = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0, 'Antenna_Field_Reference_ETRS' : [7.0, 7.0, 7.0]}
with DeviceTestContext(antennafield.AntennaField, properties=at_properties_v2, process=True) as proxy:
self.assertNotEqual(3.0, proxy.Antenna_Field_Reference_ITRF_R[0]) # value = 6.948998835785814
-
+
+ def test_read_Antenna_Quality(self):
+ """ Verify if Antenna_Quality_R is correctly retrieved """
+ antenna_qualities = numpy.array([AntennaQuality.OK] * 96)
+ with DeviceTestContext(antennafield.AntennaField, properties=self.AT_PROPERTIES, process=True) as proxy:
+ numpy.testing.assert_equal(antenna_qualities, proxy.Antenna_Quality_R)
+
+ def test_read_Antenna_Use(self):
+ """ Verify if Antenna_Use_R is correctly retrieved """
+ antenna_use = numpy.array([AntennaUse.AUTO] * 96)
+ with DeviceTestContext(antennafield.AntennaField, properties=self.AT_PROPERTIES, process=True) as proxy:
+ numpy.testing.assert_equal(antenna_use, proxy.Antenna_Use_R)
+
+ def test_read_Antenna_Usage_Mask(self):
+ """ Verify if Antenna_Usage_Mask_R is correctly retrieved """
+ antenna_qualities = numpy.array([AntennaQuality.OK] * 96)
+ antenna_use = numpy.array([AntennaUse.ON] + [AntennaUse.AUTO] * 95)
+ antenna_properties = {'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use}
+ with DeviceTestContext(antennafield.AntennaField, properties={**self.AT_PROPERTIES, **antenna_properties}, process=True) as proxy:
+ numpy.testing.assert_equal(numpy.array([True] * 96), proxy.Antenna_Usage_Mask_R)
+
+ def test_read_Antenna_Usage_Mask_only_one_functioning_antenna(self):
+ """ Verify if Antenna_Usage_Mask_R (only first antenna is OK) is correctly retrieved """
+ antenna_qualities = numpy.array([AntennaQuality.OK] + [AntennaQuality.BROKEN] * 95)
+ antenna_use = numpy.array([AntennaUse.ON] + [AntennaUse.AUTO] * 95)
+ antenna_properties = {'Antenna_Quality': antenna_qualities, 'Antenna_Use': antenna_use}
+ with DeviceTestContext(antennafield.AntennaField, properties={**self.AT_PROPERTIES, **antenna_properties}, process=True) as proxy:
+ numpy.testing.assert_equal(numpy.array([True] + [False] * 95), proxy.Antenna_Usage_Mask_R)
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_lofar_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_lofar_device.py
index d97cf7b9ebc6625cf9568fb2db24cab4edd51bc7..38d599f9f780cccd2d506a588ae8a2596f5f5ff8 100644
--- a/tangostationcontrol/tangostationcontrol/test/devices/test_lofar_device.py
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_lofar_device.py
@@ -9,6 +9,7 @@
from tango.test_context import DeviceTestContext
from tango.server import attribute
+from tango import DevState, DevFailed
from tangostationcontrol.devices import lofar_device
@@ -45,4 +46,23 @@ class TestLofarDevice(device_base.DeviceTestCase):
proxy.initialise()
self.assertEqual(42.0, proxy.read_attribute_A)
self.assertListEqual([42.0, 43.0], proxy.read_attribute_B_array.tolist())
+
+ def test_disable_state(self):
+ with DeviceTestContext(lofar_device.lofar_device, process=True, timeout=10) as proxy:
+ proxy.initialise()
+ self.assertEqual(DevState.STANDBY, proxy.state())
+ proxy.on()
+ self.assertEqual(DevState.ON, proxy.state())
+ proxy.disable_hardware()
+ self.assertEqual(DevState.DISABLE, proxy.state())
+
+ def test_disable_state_transitions(self):
+ with DeviceTestContext(lofar_device.lofar_device, process=True, timeout=10) as proxy:
+ proxy.off()
+ with self.assertRaises(DevFailed):
+ proxy.disable_hardware()
+ proxy.warm_boot()
+ proxy.Fault()
+ with self.assertRaises(DevFailed):
+ proxy.disable_hardware()