diff --git a/tangostationcontrol/tangostationcontrol/clients/udp_receiver.py b/tangostationcontrol/tangostationcontrol/clients/udp_receiver.py
index 04e24366e193567f4b04c97716a77a5546bcadae..05252bfc6ef4504ac46ec9cf17a8434ccaaab9eb 100644
--- a/tangostationcontrol/tangostationcontrol/clients/udp_receiver.py
+++ b/tangostationcontrol/tangostationcontrol/clients/udp_receiver.py
@@ -8,7 +8,7 @@ import time
from typing import List # not needed for python3.9+, where we can use the type "list[Queue]" directly
from tangostationcontrol.clients.statistics.client_thread import StatisticsClientThread
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import MAX_ETH_FRAME_SIZE
logger = logging.getLogger()
@@ -52,7 +52,7 @@ class UDPReceiver(Thread, StatisticsClientThread):
# Number of packets we had to drop due to a full queue
"nof_packets_dropped": numpy.uint64(0),
# Packets are at most 9000 bytes, the largest payload (well, MTU) of an Ethernet Jumbo frame
- "last_packet": numpy.zeros((constants.MAX_ETH_FRAME_SIZE,), dtype=numpy.uint8),
+ "last_packet": numpy.zeros((MAX_ETH_FRAME_SIZE,), dtype=numpy.uint8),
# Timestamp of when the last packet was received
"last_packet_timestamp": numpy.uint64(0),
}
@@ -90,7 +90,7 @@ class UDPReceiver(Thread, StatisticsClientThread):
while self.stream_on:
try:
- packet, _, _, _ = self.sock.recvmsg(constants.MAX_ETH_FRAME_SIZE)
+ packet, _, _, _ = self.sock.recvmsg(MAX_ETH_FRAME_SIZE)
self.parameters["nof_packets_received"] += numpy.uint64(1)
self.parameters["nof_bytes_received"] += numpy.uint64(len(packet))
diff --git a/tangostationcontrol/tangostationcontrol/common/constants.py b/tangostationcontrol/tangostationcontrol/common/constants.py
index 2fcd3c3143cee76773e770a2ea8af6044be9632b..bbec0c55fd8777edf30cc06e946de17dfe2f7478 100644
--- a/tangostationcontrol/tangostationcontrol/common/constants.py
+++ b/tangostationcontrol/tangostationcontrol/common/constants.py
@@ -1,97 +1,96 @@
from tangostationcontrol.common.baselines import nr_baselines
-class constants:
- # number of FPGA processing nodes
- N_pn = 16
+# number of FPGA processing nodes
+N_pn = 16
- # number of polarisations per antenna (X and y polarisations)
- N_pol = 2
+# number of polarisations per antenna (X and y polarisations)
+N_pol = 2
- # antennas per FPGA
- A_pn = 6
+# antennas per FPGA
+A_pn = 6
- # signal inputs per FPGA ( A_pn * N_pol )
- S_pn = A_pn * N_pol
+# signal inputs per FPGA ( A_pn * N_pol )
+S_pn = A_pn * N_pol
- # Highest number antennas we support
- MAX_ANTENNA = 96
+# Highest number antennas we support
+MAX_ANTENNA = 96
- # Maximum number of antenna inputs we support (used to determine array sizes)
- MAX_INPUTS = 192
+# Maximum number of antenna inputs we support (used to determine array sizes)
+MAX_INPUTS = 192
- # Number of tile elements (antenna dipoles) in each HBA tile
- N_elements = 16
+# Number of tile elements (antenna dipoles) in each HBA tile
+N_elements = 16
- # number of RCU's per subrack
- N_rcu = 32
+# number of RCU's per subrack
+N_rcu = 32
- # Number of antenna inputs per RCU
- N_rcu_inp = 3
+# Number of antenna inputs per RCU
+N_rcu_inp = 3
- # number of hops that the data of the stream has traveled to reach the BSN aligner on this node
- P_sum = 2
+# number of hops that the data of the stream has traveled to reach the BSN aligner on this node
+P_sum = 2
- # the number of square correlator cells produced per FPGA for XST's (P_sq := N_pn // 2 + 1)
- P_sq = 9
+# the number of square correlator cells produced per FPGA for XST's (P_sq := N_pn // 2 + 1)
+P_sq = 9
- # total number of beamsets
- N_beamsets_sdp = 2
- # number of beamsets we control
- N_beamsets_ctrl = 1
- # The maximum amount of beamlets the SDP (and we) support
- N_beamlets_max = 976
- # number of actively controlled beamlets
- N_beamlets_ctrl = 488
+# total number of beamsets
+N_beamsets_sdp = 2
+# number of beamsets we control
+N_beamsets_ctrl = 1
+# The maximum amount of beamlets the SDP (and we) support
+N_beamlets_max = 976
+# number of actively controlled beamlets
+N_beamlets_ctrl = 488
- # Maximum number of subbands we support
- N_subbands = 512
- # Number of points per subband (the resolution)
- N_subband_res = 1024
+# Maximum number of subbands we support
+N_subbands = 512
+# Number of points per subband (the resolution)
+N_subband_res = 1024
- # main clock frequency's are 200MHz and 160MHz
- CLK_200_MHZ = 200_000_000
- CLK_160_MHZ = 160_000_000
+# main clock frequency's are 200MHz and 160MHz
+CLK_200_MHZ = 200_000_000
+CLK_160_MHZ = 160_000_000
- # Maximum number of subbands for which we collect XSTs simultaneously
- MAX_PARALLEL_SUBBANDS = 8
- # Expected block for XST's
- BLOCK_LENGTH = 12
- # Complex values are (real, imag).
- VALUES_PER_COMPLEX = 2
- # Max blocks for the BST statistics
- BST_MAX_BLOCKS = 2
- # Expected number of blocks: enough to cover all baselines without the conjugates (that is, the top-left triangle of the matrix).
- MAX_BLOCKS = nr_baselines(MAX_INPUTS // BLOCK_LENGTH)
+# Maximum number of subbands for which we collect XSTs simultaneously
+MAX_PARALLEL_SUBBANDS = 8
+# Expected block for XST's
+BLOCK_LENGTH = 12
+# Complex values are (real, imag).
+VALUES_PER_COMPLEX = 2
+# Max blocks for the BST statistics
+BST_MAX_BLOCKS = 2
+# Expected number of blocks: enough to cover all baselines without the conjugates (that is, the top-left triangle of the matrix).
+MAX_BLOCKS = nr_baselines(MAX_INPUTS // BLOCK_LENGTH)
- # UNB2 constants
- # number of uniboards in a subrack
- N_unb = 2
- # number of FPGA's in a uniboard
- N_fpga = 4
- # number of DDR modules per FPGA
- N_ddr = 2
- # number of QSFP tranceivers per uniboard
- N_qsfp = 24
+# UNB2 constants
+# number of uniboards in a subrack
+N_unb = 2
+# number of FPGA's in a uniboard
+N_fpga = 4
+# number of DDR modules per FPGA
+N_ddr = 2
+# number of QSFP tranceivers per uniboard
+N_qsfp = 24
- # the three spatial dimensions XYZ used a lot for PQR and ITRF coordinates.
- N_xyz = 3
- # amount of parameters needed for a pointing
- N_point_prop = 3
- # number of values for latitude/longitude coordinates
- N_latlong = 2
+# the three spatial dimensions XYZ used a lot for PQR and ITRF coordinates.
+N_xyz = 3
+# amount of parameters needed for a pointing
+N_point_prop = 3
+# number of values for latitude/longitude coordinates
+N_latlong = 2
- # default subband we use because of its low RFI
- DEFAULT_SUBBAND = 102
+# default subband we use because of its low RFI
+DEFAULT_SUBBAND = 102
- # Maximum array size to allocate for beam_device pointings,
- MAX_POINTINGS = N_beamlets_max
+# Maximum array size to allocate for beam_device pointings,
+MAX_POINTINGS = N_beamlets_max
- # max size for a statistic packet
- MAX_ETH_FRAME_SIZE = 9000
+# max size for a statistic packet
+MAX_ETH_FRAME_SIZE = 9000
- # The default polling period for polled attributes
- DEFAULT_POLLING_PERIOD = 1000
+# The default polling period for polled attributes
+DEFAULT_POLLING_PERIOD = 1000
diff --git a/tangostationcontrol/tangostationcontrol/devices/antennafield.py b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
index 21e5d00831bda8c81fd16e839679c7f44509b3f0..3c294d7474bb1236b4fd0d6102bb946188645de6 100644
--- a/tangostationcontrol/tangostationcontrol/devices/antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
@@ -19,7 +19,7 @@ from tango.server import device_property, attribute, command
from tangostationcontrol.common.type_checking import type_not_sequence
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_elements, MAX_INPUTS, N_pol, N_xyz, N_latlong, MAX_ANTENNA, N_rcu, N_rcu_inp
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.devices.device_decorators import fault_on_error
from tangostationcontrol.beam.geo import ETRS_to_ITRF, ITRF_to_GEO, GEO_to_GEOHASH
@@ -91,16 +91,13 @@ class AntennaField(lofar_device):
calculated, as well as the geohash.
"""
- MAX_NUMBER_OF_HBAT = constants.MAX_ANTENNA
- N_te = constants.N_elements
-
# ----- Antenna names
Antenna_Names = device_property(
doc="Name of each antenna",
dtype='DevVarStringArray',
mandatory=False,
- default_value=[f'Antenna{n + 1}' for n in range(MAX_NUMBER_OF_HBAT)]
+ default_value=[f'Antenna{n + 1}' for n in range(MAX_INPUTS)]
)
# ----- Antenna states
@@ -109,14 +106,14 @@ class AntennaField(lofar_device):
doc="Operational quality state of each antenna",
dtype='DevVarUShortArray',
mandatory=False,
- default_value=numpy.array([AntennaQuality.OK] * MAX_NUMBER_OF_HBAT)
+ default_value=numpy.array([AntennaQuality.OK] * MAX_INPUTS)
)
Antenna_Use = device_property(
doc="Operational State of each antenna",
dtype='DevVarUShortArray',
mandatory=False,
- default_value=numpy.array([AntennaUse.AUTO] * MAX_NUMBER_OF_HBAT)
+ default_value=numpy.array([AntennaUse.AUTO] * MAX_INPUTS)
)
# ----- Antenna properties
@@ -132,7 +129,7 @@ class AntennaField(lofar_device):
doc="Whether to provide power to each antenna (False for noise sources)",
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=numpy.array([False] * MAX_NUMBER_OF_HBAT)
+ default_value=numpy.array([False] * MAX_INPUTS)
)
# ----- Position information
@@ -178,7 +175,7 @@ class AntennaField(lofar_device):
doc='Rotation of each tile in the PQ plane ("horizontal") in degrees.',
dtype='DevVarFloatArray',
mandatory=False,
- default_value=[0.0] * MAX_NUMBER_OF_HBAT
+ default_value=[0.0] * MAX_INPUTS
)
PQR_to_ETRS_rotation_matrix = device_property(
@@ -203,7 +200,7 @@ class AntennaField(lofar_device):
Antenna_to_SDP_Mapping = device_property(
dtype=(numpy.int32,),
doc='The mapping of Antennas to FPGA input pairs. Each FPGA can handle 6 inputs, and SDP has 16 FPGAs. Each antenna is represented with a (fpga, input) value pair. The array is flattened, so must be reshaped upon use. An input=-1 means the antenna is unconnected.',
- default_value=numpy.array([-1] * MAX_NUMBER_OF_HBAT * 2, dtype=numpy.int32)
+ default_value=numpy.array([-1] * MAX_INPUTS * 2, dtype=numpy.int32)
)
SDP_device = device_property(
@@ -219,14 +216,14 @@ class AntennaField(lofar_device):
dtype=(numpy.int32,),
doc='The mapping of Antenna power lines to RECV mapping. Each RECV can handle 96 inputs. The Antenna number is the index and the value shows to which receiver device it is connected and on which input. The first integer is the input. The second integer is the RECV id. Example: [0, 3] = first receiver of property RECV_devices with input 3. -1 means that the Antenna is not connected. The property is stored in a one dimensional structure. It needs to be reshaped to a list of lists of two items.',
mandatory=False,
- default_value=[-1] * MAX_NUMBER_OF_HBAT * 2
+ default_value=[-1] * MAX_INPUTS * 2
)
Control_to_RECV_mapping = device_property(
dtype=(numpy.int32,),
doc='The mapping of Antenna control lines to RECV mapping. Each RECV can handle 96 inputs. The Antenna number is the index and the value shows to which receiver device it is connected and on which input. The first integer is the input. The second interger is the RECV id. Example: [1, 3] = STAT/RECV/1 with input 3. -1 means that the Antenna is not connected. The property is stored in a one dimensional structure. It needs to be reshaped to a list of lists of two items.',
mandatory=False,
- default_value=[-1] * MAX_NUMBER_OF_HBAT * 2
+ default_value=[-1] * MAX_INPUTS * 2
)
RECV_devices = device_property(
@@ -240,59 +237,59 @@ class AntennaField(lofar_device):
dtype=str)
Antenna_Names_R = attribute(access=AttrWriteType.READ,
- dtype=(str,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(str,), max_dim_x=MAX_INPUTS)
Antenna_Quality_R = attribute(doc='The quality of each antenna. 0=OK, 1=SUSPICIOUS, 2=BROKEN, 3=BEYOND_REPAIR.',
- dtype=(numpy.uint32,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(numpy.uint32,), max_dim_x=MAX_INPUTS)
Antenna_Use_R = attribute(
doc='Whether each antenna should be used. 0=AUTO, 1=ON, 2=OFF. In AUTO mode, the antenna is used if it is not BROKEN or BEYOND_REPAIR.',
- dtype=(numpy.uint32,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(numpy.uint32,), max_dim_x=MAX_INPUTS)
Antenna_Quality_str_R = attribute(doc='The quality of each antenna, as a string.',
- dtype=(str,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(str,), max_dim_x=MAX_INPUTS)
Antenna_Use_str_R = attribute(doc='Whether each antenna should be used, as a string.',
- dtype=(str,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(str,), max_dim_x=MAX_INPUTS)
Antenna_Usage_Mask_R = attribute(doc='Whether each antenna will be used.',
- dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT)
+ dtype=(bool,), max_dim_x=MAX_INPUTS)
Antenna_to_SDP_Mapping_R = attribute(doc='To which (fpga, input) pair each antenna is connected. -1=unconnected.',
- dtype=((numpy.int32,),), max_dim_x=constants.N_pol, max_dim_y=MAX_NUMBER_OF_HBAT)
+ dtype=((numpy.int32,),), max_dim_x=N_pol, max_dim_y=MAX_INPUTS)
- ANT_mask_RW = mapped_attribute("ANT_mask_RW", dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT,
+ ANT_mask_RW = mapped_attribute("ANT_mask_RW", dtype=(bool,), max_dim_x=MAX_INPUTS,
access=AttrWriteType.READ_WRITE)
- RCU_PWR_ANT_on_R = mapped_attribute("RCU_PWR_ANT_on_R", dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT)
- RCU_PWR_ANT_on_RW = mapped_attribute("RCU_PWR_ANT_on_RW", dtype=(bool,), max_dim_x=MAX_NUMBER_OF_HBAT,
+ RCU_PWR_ANT_on_R = mapped_attribute("RCU_PWR_ANT_on_R", dtype=(bool,), max_dim_x=MAX_INPUTS)
+ RCU_PWR_ANT_on_RW = mapped_attribute("RCU_PWR_ANT_on_RW", dtype=(bool,), max_dim_x=MAX_INPUTS,
access=AttrWriteType.READ_WRITE)
HBAT_BF_delay_steps_R = mapped_attribute("HBAT_BF_delay_steps_R", dtype=((numpy.int64,),),
- max_dim_x=N_te * constants.N_pol, max_dim_y=MAX_NUMBER_OF_HBAT)
+ max_dim_x=N_elements * N_pol, max_dim_y=MAX_INPUTS)
HBAT_BF_delay_steps_RW = mapped_attribute("HBAT_BF_delay_steps_RW", dtype=((numpy.int64,),),
- max_dim_x=N_te * constants.N_pol, max_dim_y=MAX_NUMBER_OF_HBAT,
+ max_dim_x=N_elements * N_pol, max_dim_y=MAX_INPUTS,
access=AttrWriteType.READ_WRITE)
- HBAT_LED_on_R = mapped_attribute("HBAT_LED_on_R", dtype=((bool,),), max_dim_x=N_te * constants.N_pol,
- max_dim_y=MAX_NUMBER_OF_HBAT)
- HBAT_LED_on_RW = mapped_attribute("HBAT_LED_on_RW", dtype=((bool,),), max_dim_x=N_te * constants.N_pol,
- max_dim_y=MAX_NUMBER_OF_HBAT, access=AttrWriteType.READ_WRITE)
+ HBAT_LED_on_R = mapped_attribute("HBAT_LED_on_R", dtype=((bool,),), max_dim_x=N_elements * N_pol,
+ max_dim_y=MAX_INPUTS)
+ HBAT_LED_on_RW = mapped_attribute("HBAT_LED_on_RW", dtype=((bool,),), max_dim_x=N_elements * N_pol,
+ max_dim_y=MAX_INPUTS, access=AttrWriteType.READ_WRITE)
HBAT_PWR_LNA_on_R = mapped_attribute("HBAT_PWR_LNA_on_R", dtype=((bool,),),
- max_dim_x=N_te * constants.N_pol, max_dim_y=MAX_NUMBER_OF_HBAT)
+ max_dim_x=N_elements * N_pol, max_dim_y=MAX_INPUTS)
HBAT_PWR_LNA_on_RW = mapped_attribute("HBAT_PWR_LNA_on_RW", dtype=((bool,),),
- max_dim_x=N_te * constants.N_pol, max_dim_y=MAX_NUMBER_OF_HBAT,
+ max_dim_x=N_elements * N_pol, max_dim_y=MAX_INPUTS,
access=AttrWriteType.READ_WRITE)
- HBAT_PWR_on_R = mapped_attribute("HBAT_PWR_on_R", dtype=((bool,),), max_dim_x=N_te * constants.N_pol,
- max_dim_y=MAX_NUMBER_OF_HBAT)
- HBAT_PWR_on_RW = mapped_attribute("HBAT_PWR_on_RW", dtype=((bool,),), max_dim_x=N_te * constants.N_pol,
- max_dim_y=MAX_NUMBER_OF_HBAT, access=AttrWriteType.READ_WRITE)
- RCU_band_select_RW = mapped_attribute("RCU_band_select_RW", dtype=(numpy.int64,), max_dim_x=MAX_NUMBER_OF_HBAT,
+ HBAT_PWR_on_R = mapped_attribute("HBAT_PWR_on_R", dtype=((bool,),), max_dim_x=N_elements * N_pol,
+ max_dim_y=MAX_INPUTS)
+ HBAT_PWR_on_RW = mapped_attribute("HBAT_PWR_on_RW", dtype=((bool,),), max_dim_x=N_elements * N_pol,
+ max_dim_y=MAX_INPUTS, access=AttrWriteType.READ_WRITE)
+ RCU_band_select_RW = mapped_attribute("RCU_band_select_RW", dtype=(numpy.int64,), max_dim_x=MAX_INPUTS,
access=AttrWriteType.READ_WRITE)
# ----- Position information
Antenna_Field_Reference_ITRF_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of antenna field, in ITRF (XYZ)',
- dtype=(numpy.float64,), max_dim_x=constants.N_xyz)
+ dtype=(numpy.float64,), max_dim_x=N_xyz)
Antenna_Field_Reference_GEO_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of antenna field, in latitude/longitude (degrees)',
- dtype=(numpy.float64,), max_dim_x=constants.N_latlong)
+ dtype=(numpy.float64,), max_dim_x=N_latlong)
Antenna_Field_Reference_GEOHASH_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of antenna field, as a geohash string',
@@ -300,20 +297,20 @@ class AntennaField(lofar_device):
HBAT_antenna_ITRF_offsets_R = attribute(access=AttrWriteType.READ,
doc='For each tile, the offsets of the antennas within that, in ITRF ("iHBADeltas"). True shape: nrtiles x 16 x 3.',
- dtype=((numpy.float64,),), max_dim_x=MAX_NUMBER_OF_HBAT * constants.N_xyz,
- max_dim_y=MAX_NUMBER_OF_HBAT)
+ dtype=((numpy.float64,),), max_dim_x=MAX_INPUTS * N_xyz,
+ max_dim_y=MAX_INPUTS)
Antenna_Reference_ITRF_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of each tile, in ITRF (XYZ)',
- dtype=((numpy.float64,),), max_dim_x=constants.N_xyz, max_dim_y=MAX_NUMBER_OF_HBAT)
+ dtype=((numpy.float64,),), max_dim_x=N_xyz, max_dim_y=MAX_INPUTS)
Antenna_Reference_GEO_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of each tile, in latitude/longitude (degrees)',
- dtype=((numpy.float64,),), max_dim_x=constants.N_latlong, max_dim_y=MAX_NUMBER_OF_HBAT)
+ dtype=((numpy.float64,),), max_dim_x=N_latlong, max_dim_y=MAX_INPUTS)
Antenna_Reference_GEOHASH_R = attribute(access=AttrWriteType.READ,
doc='Absolute reference position of each tile, as geohash strings',
- dtype=(str,), max_dim_x=MAX_NUMBER_OF_HBAT, )
+ dtype=(str,), max_dim_x=MAX_INPUTS, )
nr_antennas_R = attribute(
doc='Number of Antennas in this field',
@@ -364,10 +361,10 @@ class AntennaField(lofar_device):
def read_Antenna_Field_Reference_ITRF_R(self):
# provide ITRF field coordinates if they were configured
if self.Antenna_Field_Reference_ITRF:
- return numpy.array(self.Antenna_Field_Reference_ITRF).reshape(constants.N_xyz)
+ return numpy.array(self.Antenna_Field_Reference_ITRF).reshape(N_xyz)
# calculate them from ETRS coordinates if not, using the configured ITRF reference
- ETRS_coordinates = numpy.array(self.Antenna_Field_Reference_ETRS).reshape(constants.N_xyz)
+ ETRS_coordinates = numpy.array(self.Antenna_Field_Reference_ETRS).reshape(N_xyz)
return ETRS_to_ITRF(ETRS_coordinates, self.ITRF_Reference_Frame, self.ITRF_Reference_Epoch)
def read_Antenna_Field_Reference_GEO_R(self):
@@ -392,9 +389,9 @@ class AntennaField(lofar_device):
tiles lie on the same plane in ITRF. """
# the relative offsets between the elements is fixed in HBAT_base_antenna_offsets
- base_antenna_offsets = numpy.array(self.HBAT_base_antenna_offsets).reshape(constants.N_elements, constants.N_xyz)
+ base_antenna_offsets = numpy.array(self.HBAT_base_antenna_offsets).reshape(N_elements, N_xyz)
- PQR_to_ETRS_rotation_matrix = numpy.array(self.PQR_to_ETRS_rotation_matrix).reshape(constants.N_xyz, constants.N_xyz)
+ PQR_to_ETRS_rotation_matrix = numpy.array(self.PQR_to_ETRS_rotation_matrix).reshape(N_xyz, N_xyz)
# each tile has its own rotation angle, resulting in different offsets per tile
all_offsets = numpy.array(
@@ -404,15 +401,15 @@ class AntennaField(lofar_device):
PQR_to_ETRS_rotation_matrix)
for angle_deg in self.HBAT_PQR_rotation_angles_deg])
- return all_offsets.reshape(-1, constants.N_elements * constants.N_xyz)
+ return all_offsets.reshape(-1, N_elements * N_xyz)
def read_Antenna_Reference_ITRF_R(self):
# provide ITRF coordinates if they were configured
if self.Antenna_Reference_ITRF:
- return numpy.array(self.Antenna_Reference_ITRF).reshape(-1, constants.N_xyz)
+ return numpy.array(self.Antenna_Reference_ITRF).reshape(-1, N_xyz)
# calculate them from ETRS coordinates if not, using the configured ITRF reference
- ETRS_coordinates = numpy.array(self.Antenna_Reference_ETRS).reshape(-1, constants.N_xyz)
+ ETRS_coordinates = numpy.array(self.Antenna_Reference_ETRS).reshape(-1, N_xyz)
return ETRS_to_ITRF(ETRS_coordinates, self.ITRF_Reference_Frame, self.ITRF_Reference_Epoch)
@@ -521,10 +518,10 @@ class AntennaField(lofar_device):
def calculate_HBAT_bf_delay_steps(self, delays: numpy.ndarray):
num_tiles = self.read_nr_antennas_R()
- delays = delays.reshape(num_tiles, constants.N_elements)
+ delays = delays.reshape(num_tiles, N_elements)
- result_values = numpy.zeros((num_tiles, constants.N_elements * constants.N_pol), dtype=numpy.int64)
- control_mapping = numpy.reshape(self.Control_to_RECV_mapping, (-1, constants.N_pol))
+ result_values = numpy.zeros((num_tiles, N_elements * N_pol), dtype=numpy.int64)
+ control_mapping = numpy.reshape(self.Control_to_RECV_mapping, (-1, N_pol))
for recv_idx, recv_proxy in enumerate(self.recv_proxies):
# collect all delays for this recv_proxy
@@ -538,7 +535,7 @@ class AntennaField(lofar_device):
# convert them into delay steps
flatten_delay_steps = numpy.array(recv_proxy.calculate_HBAT_bf_delay_steps(recv_delays.flatten()),
dtype=numpy.int64)
- delay_steps = numpy.reshape(flatten_delay_steps, (-1, constants.N_elements * constants.N_pol))
+ delay_steps = numpy.reshape(flatten_delay_steps, (-1, N_elements * N_pol))
# write back into same positions we collected them from
result_values[recv_result_indices] = delay_steps
@@ -548,16 +545,16 @@ class AntennaField(lofar_device):
class AntennaToRecvMapper(object):
- _VALUE_MAP_NONE_96 = numpy.full(constants.MAX_ANTENNA, None)
- _VALUE_MAP_NONE_96_32 = numpy.full((constants.MAX_ANTENNA, constants.N_rcu), None)
+ _VALUE_MAP_NONE_96 = numpy.full(MAX_ANTENNA, None)
+ _VALUE_MAP_NONE_96_32 = numpy.full((MAX_ANTENNA, N_rcu), None)
def __init__(self, control_to_recv_mapping, power_to_recv_mapping, number_of_receivers):
number_of_antennas = len(control_to_recv_mapping)
# Reduce memory footprint of default values by creating single instance of
# common fields
- value_map_ant_32_int = numpy.zeros([number_of_antennas, constants.N_rcu], dtype=numpy.int64)
- value_map_ant_32_bool = numpy.full((number_of_antennas, constants.N_rcu), False)
+ value_map_ant_32_int = numpy.zeros([number_of_antennas, N_rcu], dtype=numpy.int64)
+ value_map_ant_32_bool = numpy.full((number_of_antennas, N_rcu), False)
value_map_ant_bool = numpy.full(number_of_antennas, False)
self._control_mapping = control_to_recv_mapping
@@ -587,16 +584,16 @@ class AntennaToRecvMapper(object):
"RCU_band_select_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96,
}
self._reshape_attributes_in = {
- "HBAT_BF_delay_steps_RW": (constants.MAX_ANTENNA, constants.N_rcu),
- "RCU_PWR_ANT_on_R": (constants.MAX_ANTENNA,),
- "RCU_PWR_ANT_on_RW": (constants.MAX_ANTENNA,),
- "RCU_band_select_RW": (constants.MAX_ANTENNA,),
+ "HBAT_BF_delay_steps_RW": (MAX_ANTENNA, N_rcu),
+ "RCU_PWR_ANT_on_R": (MAX_ANTENNA,),
+ "RCU_PWR_ANT_on_RW": (MAX_ANTENNA,),
+ "RCU_band_select_RW": (MAX_ANTENNA,),
}
self._reshape_attributes_out = {
- "HBAT_BF_delay_steps_RW": (constants.MAX_ANTENNA, constants.N_rcu),
- "RCU_PWR_ANT_on_R": (constants.N_rcu, constants.N_rcu_inp),
- "RCU_PWR_ANT_on_RW": (constants.N_rcu, constants.N_rcu_inp),
- "RCU_band_select_RW": (constants.N_rcu, constants.N_rcu_inp),
+ "HBAT_BF_delay_steps_RW": (MAX_ANTENNA, N_rcu),
+ "RCU_PWR_ANT_on_R": (N_rcu, N_rcu_inp),
+ "RCU_PWR_ANT_on_RW": (N_rcu, N_rcu_inp),
+ "RCU_band_select_RW": (N_rcu, N_rcu_inp),
}
def map_read(self, mapped_attribute: str, recv_results: List[any]) -> List[any]:
diff --git a/tangostationcontrol/tangostationcontrol/devices/apsct.py b/tangostationcontrol/tangostationcontrol/devices/apsct.py
index 72bd74350520ad022940da36d83b70dce3fd97b2..d39289c6f959cf92457d696a58fe021a5d3eda07 100644
--- a/tangostationcontrol/tangostationcontrol/devices/apsct.py
+++ b/tangostationcontrol/tangostationcontrol/devices/apsct.py
@@ -22,7 +22,7 @@ from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.common.states import DEFAULT_COMMAND_STATES
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD
from tangostationcontrol.devices.device_decorators import only_in_states
from tangostationcontrol.devices.opcua_device import opcua_device
@@ -102,7 +102,7 @@ class APSCT(opcua_device):
self.read_attribute("APSCT_PLL_160MHz_locked_R") and self.read_attribute("APSCT_PLL_160MHz_error_R")]
return any(errors)
- APSCT_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=constants.DEFAULT_POLLING_PERIOD)
+ APSCT_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=DEFAULT_POLLING_PERIOD)
APSCT_VOUT_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed")
def read_APSCT_TEMP_error_R(self):
diff --git a/tangostationcontrol/tangostationcontrol/devices/apspu.py b/tangostationcontrol/tangostationcontrol/devices/apspu.py
index 8c18ce5c42707ac9aec0e03335366fe3f097e98e..08ee51fa8d968a3fcb5aed86809294a4471ed081 100644
--- a/tangostationcontrol/tangostationcontrol/devices/apspu.py
+++ b/tangostationcontrol/tangostationcontrol/devices/apspu.py
@@ -21,7 +21,7 @@ from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
from tangostationcontrol.common.lofar_logging import device_logging_to_python
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD
__all__ = ["APSPU", "main"]
@@ -78,7 +78,7 @@ class APSPU(opcua_device):
or self.alarm_val("APSPU_FAN3_RPM_R"))
APSPU_IOUT_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed")
- APSPU_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=constants.DEFAULT_POLLING_PERIOD)
+ APSPU_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=DEFAULT_POLLING_PERIOD)
APSPU_VOUT_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed")
def read_APSPU_IOUT_error_R(self):
diff --git a/tangostationcontrol/tangostationcontrol/devices/beam_device.py b/tangostationcontrol/tangostationcontrol/devices/beam_device.py
index f0b66e2155965de8cf4f523e8bbca36ce749a56f..dc2ee888d21726ea3fdefc99b471e0d076e8ab67 100644
--- a/tangostationcontrol/tangostationcontrol/devices/beam_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/beam_device.py
@@ -21,7 +21,7 @@ from tango import AttrWriteType, DebugIt, DevVarStringArray, DevVarDoubleArray,
# Additional import
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.measures import get_measures_directory, get_available_measures_directories, download_measures, use_measures_directory, restart_python
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import MAX_POINTINGS, N_point_prop
from tangostationcontrol.common.lofar_logging import log_exceptions
from tangostationcontrol.common.states import DEFAULT_COMMAND_STATES
from tangostationcontrol.devices.device_decorators import TimeIt, only_in_states, fault_on_error
@@ -74,20 +74,20 @@ class beam_device(lofar_device):
# will be stored as self._num_pointings.
Pointing_direction_R = attribute(access=AttrWriteType.READ,
- dtype=((str,),), max_dim_x=constants.N_point_prop, max_dim_y=constants.MAX_POINTINGS,
+ dtype=((str,),), max_dim_x=N_point_prop, max_dim_y=MAX_POINTINGS,
fget=lambda self: self._pointing_direction_r)
Pointing_direction_RW = attribute(access=AttrWriteType.READ_WRITE,
- dtype=((str,),), max_dim_x=constants.N_point_prop, max_dim_y=constants.MAX_POINTINGS,
+ dtype=((str,),), max_dim_x=N_point_prop, max_dim_y=MAX_POINTINGS,
fget=lambda self: self._pointing_direction_rw)
Pointing_direction_str_R = attribute(access=AttrWriteType.READ,
doc='Pointing direction as a formatted string',
- dtype=(str,), max_dim_x=constants.MAX_POINTINGS,
+ dtype=(str,), max_dim_x=MAX_POINTINGS,
fget=lambda self: ["{0} ({1}, {2})".format(*x) for x in self._pointing_direction_r])
Pointing_timestamp_R = attribute(access=AttrWriteType.READ,
- dtype=(numpy.double,), max_dim_x=constants.MAX_POINTINGS,
+ dtype=(numpy.double,), max_dim_x=MAX_POINTINGS,
fget=lambda self: self._pointing_timestamp_r)
Tracking_enabled_R = attribute(access=AttrWriteType.READ,
@@ -224,13 +224,13 @@ class beam_device(lofar_device):
def configure_for_initialise(self, num_pointings):
super().configure_for_initialise()
- if not (0 < num_pointings <= constants.MAX_POINTINGS):
- raise ValueError(f"beam_device is configured to support 0 - {constants.MAX_POINTINGS} pointings, but {num_pointings} were requested")
+ if not (0 < num_pointings <= MAX_POINTINGS):
+ raise ValueError(f"beam_device is configured to support 0 - {MAX_POINTINGS} pointings, but {num_pointings} were requested")
# Initialise tracking control
self._num_pointings = num_pointings
self._pointing_timestamp_r = numpy.zeros(num_pointings, dtype=numpy.double)
- self._pointing_direction_r = numpy.zeros((num_pointings, constants.N_point_prop), dtype="<U32")
+ self._pointing_direction_r = numpy.zeros((num_pointings, N_point_prop), dtype="<U32")
self._pointing_direction_rw = numpy.array([["AZELGEO","0deg","90deg"]] * num_pointings, dtype="<U32")
self._tracking_enabled_rw = self.Tracking_enabled_RW_default
diff --git a/tangostationcontrol/tangostationcontrol/devices/ccd.py b/tangostationcontrol/tangostationcontrol/devices/ccd.py
index 4d0c12c1b5dd4e247a93c929bf074c7f2b14f117..ae95b6657ad96bbe8089062ba9aaf8df6c35a01d 100644
--- a/tangostationcontrol/tangostationcontrol/devices/ccd.py
+++ b/tangostationcontrol/tangostationcontrol/devices/ccd.py
@@ -22,7 +22,7 @@ from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.common.states import DEFAULT_COMMAND_STATES
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD
from tangostationcontrol.devices.device_decorators import only_in_states
from tangostationcontrol.devices.opcua_device import opcua_device
@@ -92,7 +92,7 @@ class CCD(opcua_device):
not self.read_attribute("CCD_PLL_locked_R")]
return any(errors)
- CCD_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=constants.DEFAULT_POLLING_PERIOD)
+ CCD_TEMP_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed", polling_period=DEFAULT_POLLING_PERIOD)
CCD_VOUT_error_R = attribute(dtype=bool, fisallowed="is_attribute_access_allowed")
def read_CCD_TEMP_error_R(self):
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation.py b/tangostationcontrol/tangostationcontrol/devices/observation.py
index 90b786e43f3be7b2344691d8fd9e1abb1edeecfb..cae2bd23855efab552b98521fe338986fb09e5e0 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation.py
@@ -14,7 +14,7 @@ import numpy
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.common.lofar_logging import log_exceptions
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD, MAX_ANTENNA, N_beamlets_ctrl, N_point_prop
from tangostationcontrol.devices.device_decorators import fault_on_error
from tangostationcontrol.devices.device_decorators import only_when_on
from tangostationcontrol.devices.device_decorators import only_in_states
@@ -43,15 +43,15 @@ class Observation(lofar_device):
# Attributes
- observation_running_R = attribute(dtype=numpy.float64, access=AttrWriteType.READ, polling_period=constants.DEFAULT_POLLING_PERIOD, period=constants.DEFAULT_POLLING_PERIOD,
+ observation_running_R = attribute(dtype=numpy.float64, access=AttrWriteType.READ, polling_period=DEFAULT_POLLING_PERIOD, period=DEFAULT_POLLING_PERIOD,
rel_change="1.0")
observation_id_R = attribute(dtype=numpy.int64, access=AttrWriteType.READ)
stop_time_R = attribute(dtype=numpy.float64, access=AttrWriteType.READ)
- antenna_mask_R = attribute(dtype=(numpy.int64,), max_dim_x=constants.MAX_ANTENNA, access=AttrWriteType.READ)
+ antenna_mask_R = attribute(dtype=(numpy.int64,), max_dim_x=MAX_ANTENNA, access=AttrWriteType.READ)
filter_R = attribute(dtype=numpy.str, access=AttrWriteType.READ)
- saps_subband_R = attribute(dtype=((numpy.uint32,),), max_dim_x=constants.N_beamlets_ctrl, max_dim_y=constants.N_beamlets_ctrl, access=AttrWriteType.READ)
- saps_pointing_R = attribute(dtype=((numpy.str,),), max_dim_x=constants.N_point_prop, max_dim_y=constants.N_beamlets_ctrl, access=AttrWriteType.READ)
- tile_beam_R = attribute(dtype=(numpy.str,), max_dim_x=constants.N_point_prop, access=AttrWriteType.READ)
+ saps_subband_R = attribute(dtype=((numpy.uint32,),), max_dim_x=N_beamlets_ctrl, max_dim_y=N_beamlets_ctrl, access=AttrWriteType.READ)
+ saps_pointing_R = attribute(dtype=((numpy.str,),), max_dim_x=N_point_prop, max_dim_y=N_beamlets_ctrl, access=AttrWriteType.READ)
+ tile_beam_R = attribute(dtype=(numpy.str,), max_dim_x=N_point_prop, access=AttrWriteType.READ)
first_beamlet_R = attribute(dtype=numpy.int64, access=AttrWriteType.READ)
observation_settings_RW = attribute(dtype=str, access=AttrWriteType.READ_WRITE)
@@ -243,8 +243,8 @@ class Observation(lofar_device):
retrieve the RCU band from filter name, returning the correct format for
AntennaField device
"""
- ANT_mask_RW = [False] * constants.MAX_ANTENNA
- RCU_band_select_RW = [0] * constants.MAX_ANTENNA
+ ANT_mask_RW = [False] * MAX_ANTENNA
+ RCU_band_select_RW = [0] * MAX_ANTENNA
rcu_band = self.recv_proxy.get_rcu_band_from_filter(filter_name)
for a in antenna_mask:
ANT_mask_RW[a] = True
@@ -270,10 +270,10 @@ class Observation(lofar_device):
def _apply_saps_antenna_select(self, antenna_mask:list):
""" Convert an array of antenna indexes into a boolean select array"""
- antenna_select = numpy.array([[False] * constants.N_beamlets_ctrl] * self._num_inputs)
+ antenna_select = numpy.array([[False] * N_beamlets_ctrl] * self._num_inputs)
first_beamlet = numpy.array(self.read_first_beamlet_R(), dtype=numpy.int64)
for a in antenna_mask:
- for i in range(first_beamlet, constants.N_beamlets_ctrl):
+ for i in range(first_beamlet, N_beamlets_ctrl):
antenna_select[a,i] = True
return antenna_select
diff --git a/tangostationcontrol/tangostationcontrol/devices/recv.py b/tangostationcontrol/tangostationcontrol/devices/recv.py
index 503daeb6a7d086f21cf740a7e2dbfb3d20c916cf..3c1b81ae137fdf93646b11b8b0fc061c475503b2 100644
--- a/tangostationcontrol/tangostationcontrol/devices/recv.py
+++ b/tangostationcontrol/tangostationcontrol/devices/recv.py
@@ -23,7 +23,7 @@ import numpy
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.common.states import DEFAULT_COMMAND_STATES
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_rcu, N_elements, N_pol, N_rcu_inp, DEFAULT_POLLING_PERIOD
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.device_decorators import only_in_states
from tangostationcontrol.devices.opcua_device import opcua_device
@@ -38,11 +38,6 @@ __all__ = ["RECV", "main"]
@device_logging_to_python()
class RECV(opcua_device):
- N_rcu = constants.N_rcu
- N_te = constants.N_elements
- N_pol = constants.N_pol
- N_rcu_inp = constants.N_rcu_inp
-
FILTER_RCU_DICT = {
"LBA_10_90": 1,
"LBA_10_70": 1,
@@ -151,17 +146,17 @@ class RECV(opcua_device):
# The HBAT beamformer delays represent 32 delays for each of the 96 inputs.
# The 32 delays deconstruct as delays[polarisation][dipole], and each delay is the number of 'delay steps' to apply (0.5ns for HBAT1).
HBAT_BF_delay_steps_R = attribute_wrapper(comms_annotation=["HBAT_BF_delay_steps_R"], datatype=numpy.int64,
- dims=(N_rcu * N_rcu_inp, N_te, N_pol))
+ dims=(N_rcu * N_rcu_inp, N_elements, N_pol))
HBAT_BF_delay_steps_RW = attribute_wrapper(comms_annotation=["HBAT_BF_delay_steps_RW"], datatype=numpy.int64,
- dims=(N_rcu * N_rcu_inp, N_te, N_pol), access=AttrWriteType.READ_WRITE)
- HBAT_LED_on_R = attribute_wrapper(comms_annotation=["HBAT_LED_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol))
- HBAT_LED_on_RW = attribute_wrapper(comms_annotation=["HBAT_LED_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol),
+ dims=(N_rcu * N_rcu_inp, N_elements, N_pol), access=AttrWriteType.READ_WRITE)
+ HBAT_LED_on_R = attribute_wrapper(comms_annotation=["HBAT_LED_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol))
+ HBAT_LED_on_RW = attribute_wrapper(comms_annotation=["HBAT_LED_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol),
access=AttrWriteType.READ_WRITE)
- HBAT_PWR_LNA_on_R = attribute_wrapper(comms_annotation=["HBAT_PWR_LNA_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol))
- HBAT_PWR_LNA_on_RW = attribute_wrapper(comms_annotation=["HBAT_PWR_LNA_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol),
+ HBAT_PWR_LNA_on_R = attribute_wrapper(comms_annotation=["HBAT_PWR_LNA_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol))
+ HBAT_PWR_LNA_on_RW = attribute_wrapper(comms_annotation=["HBAT_PWR_LNA_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol),
access=AttrWriteType.READ_WRITE)
- HBAT_PWR_on_R = attribute_wrapper(comms_annotation=["HBAT_PWR_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol))
- HBAT_PWR_on_RW = attribute_wrapper(comms_annotation=["HBAT_PWR_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_te, N_pol),
+ HBAT_PWR_on_R = attribute_wrapper(comms_annotation=["HBAT_PWR_on_R"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol))
+ HBAT_PWR_on_RW = attribute_wrapper(comms_annotation=["HBAT_PWR_on_RW"], datatype=bool, dims=(N_rcu * N_rcu_inp, N_elements, N_pol),
access=AttrWriteType.READ_WRITE)
RCU_ADC_locked_R = attribute_wrapper(comms_annotation=["RCU_ADC_locked_R"], datatype=bool, dims=(N_rcu, N_rcu_inp))
RCU_attenuator_dB_R = attribute_wrapper(comms_annotation=["RCU_attenuator_dB_R"], datatype=numpy.int64,
@@ -231,7 +226,7 @@ class RECV(opcua_device):
RECV_IOUT_error_R = attribute(dtype=(bool,), max_dim_x=N_rcu * N_rcu_inp, fisallowed="is_attribute_access_allowed")
RECV_TEMP_error_R = attribute(dtype=(bool,), max_dim_x=N_rcu, fisallowed="is_attribute_access_allowed",
- polling_period=constants.DEFAULT_POLLING_PERIOD)
+ polling_period=DEFAULT_POLLING_PERIOD)
RECV_VOUT_error_R = attribute(dtype=(bool,), max_dim_x=N_rcu, fisallowed="is_attribute_access_allowed")
def read_RECV_IOUT_error_R(self):
@@ -249,7 +244,7 @@ class RECV(opcua_device):
return (self.read_attribute("ANT_mask_RW") & (
self.alarm_val("RCU_PWR_ANT_VIN_R").flatten()
| self.alarm_val("RCU_PWR_ANT_VOUT_R").flatten()
- )).reshape(constants.N_rcu,constants.N_rcu_inp).any(axis=1) | (self.read_attribute("RCU_mask_RW") & (
+ )).reshape(N_rcu,N_rcu_inp).any(axis=1) | (self.read_attribute("RCU_mask_RW") & (
self.alarm_val("RCU_PWR_1V8_R")
| self.alarm_val("RCU_PWR_2V5_R")
| self.alarm_val("RCU_PWR_3V3_R")
@@ -287,7 +282,7 @@ class RECV(opcua_device):
# Save actual mask values
RCU_mask = self.proxy.RCU_mask_RW
# Set the mask to all Trues
- self.RCU_mask_RW = [True] * constants.N_rcu
+ self.RCU_mask_RW = [True] * N_rcu
# Turn off the RCUs
self.RCU_off()
self.wait_attribute("RECVTR_translator_busy_R", False, self.RCU_On_Off_timeout)
@@ -303,7 +298,7 @@ class RECV(opcua_device):
which is a value per tile per dipole per polarisation.
"""
# Duplicate delay values per polarisation
- polarised_delays = numpy.repeat(delays, constants.N_pol, axis=1) # output dims -> 96x32
+ polarised_delays = numpy.repeat(delays, N_pol, axis=1) # output dims -> 96x32
# Add signal input delay
calibrated_delays = numpy.add(polarised_delays, self.HBAT_signal_input_delays)
@@ -325,7 +320,7 @@ class RECV(opcua_device):
""" converts a signal path delay (in seconds) to an analog beam weight """
# Reshape the flattened input array, into whatever how many tiles we get
- delays = numpy.array(delays).reshape(-1, constants.N_elements)
+ delays = numpy.array(delays).reshape(-1, N_elements)
# Calculate the beam weight array
HBAT_bf_delay_steps = self._calculate_HBAT_bf_delay_steps(delays)
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
index 2db9f70dd45a9d2a11b53fe9243373851c0b117f..af151941fcbbe0104798b08dbf58341631f9c982 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
@@ -14,7 +14,7 @@ from tango import AttrWriteType, DevVarFloatArray, DevVarULongArray, DeviceProxy
# Additional import
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import log_exceptions
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_pn, A_pn, N_pol, N_beamlets_ctrl, N_beamsets_ctrl, P_sum, N_subband_res, N_subbands, DEFAULT_SUBBAND
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
@@ -29,15 +29,6 @@ logger = logging.getLogger()
class Beamlet(opcua_device):
- # redeclared these much used constants here to make the code a bit cleaner.
- N_pn = constants.N_pn
- A_pn = constants.A_pn
- N_pol = constants.N_pol
- N_beamlets_ctrl = constants.N_beamlets_ctrl
- N_beamsets_ctrl = constants.N_beamsets_ctrl
- N_pol_bf = constants.N_pol
- P_sum = constants.P_sum
-
# -----------------
# Device Properties
# -----------------
@@ -93,7 +84,7 @@ class Beamlet(opcua_device):
subband_select_RW_default = device_property(
dtype='DevVarULongArray',
mandatory=False,
- default_value =[constants.DEFAULT_SUBBAND] * N_beamlets_ctrl
+ default_value =[DEFAULT_SUBBAND] * N_beamlets_ctrl
)
FIRST_DEFAULT_SETTINGS = [
@@ -167,8 +158,8 @@ class Beamlet(opcua_device):
# cint16[N_pn][N_pol][A_pn][N_pol][N_beamlets_ctrl]
# Full Jones matrix of BF weights.
- FPGA_bf_weights_xx_xy_yx_yy_R = attribute_wrapper(comms_annotation=["FPGA_bf_weights_xx_xy_yx_yy_R"], datatype=numpy.uint32, dims=(N_pn, N_pol_bf, A_pn, N_pol, N_beamlets_ctrl))
- FPGA_bf_weights_xx_xy_yx_yy_RW = attribute_wrapper(comms_annotation=["FPGA_bf_weights_xx_xy_yx_yy_RW"], datatype=numpy.uint32, dims=(N_pn, N_pol_bf, A_pn, N_pol, N_beamlets_ctrl), access=AttrWriteType.READ_WRITE)
+ FPGA_bf_weights_xx_xy_yx_yy_R = attribute_wrapper(comms_annotation=["FPGA_bf_weights_xx_xy_yx_yy_R"], datatype=numpy.uint32, dims=(N_pn, N_pol, A_pn, N_pol, N_beamlets_ctrl))
+ FPGA_bf_weights_xx_xy_yx_yy_RW = attribute_wrapper(comms_annotation=["FPGA_bf_weights_xx_xy_yx_yy_RW"], datatype=numpy.uint32, dims=(N_pn, N_pol, A_pn, N_pol, N_beamlets_ctrl), access=AttrWriteType.READ_WRITE)
# cint16[N_pn][A_pn][N_beamlets_ctrl]
# BF weights for separate access to respectively w_xx, w_xy, w_yx, and w_yy.
@@ -208,14 +199,14 @@ class Beamlet(opcua_device):
return self.subband_select_RW_default
# We return the first setting within the mask. Convert into actual shape for a single FPGA
- mask_for_all_inputs = subbands_in_mask[0].reshape(constants.A_pn, constants.N_pol, constants.N_beamlets_ctrl)
+ mask_for_all_inputs = subbands_in_mask[0].reshape(A_pn, N_pol, N_beamlets_ctrl)
# Return the first setting (antenna, pol) within this FPGA
return mask_for_all_inputs[0,0]
def write_subband_select_RW(self, subbands):
# Use the same subband for all inputs and polarisations of a beamlet
- self.proxy.FPGA_beamlet_subband_select_RW = numpy.tile(subbands, (constants.N_pn, constants.A_pn * constants.N_pol))
+ self.proxy.FPGA_beamlet_subband_select_RW = numpy.tile(subbands, (N_pn, A_pn * N_pol))
self.cache_clear()
@@ -320,8 +311,8 @@ class Beamlet(opcua_device):
def _subband_frequencies(subbands: numpy.ndarray, clock: int, nyquist_zones: numpy.ndarray) -> numpy.ndarray:
""" Obtain the frequencies of each subband, given a clock and an antenna type. """
- subband_width = clock / constants.N_subband_res
- base_subbands = nyquist_zones * constants.N_subbands
+ subband_width = clock / N_subband_res
+ base_subbands = nyquist_zones * N_subbands
# broadcast clock across frequencies
frequencies = (subbands + base_subbands) * subband_width
@@ -340,7 +331,7 @@ class Beamlet(opcua_device):
nyquist_zones = self.sdp_proxy.nyquist_zone_R # (fpga_nr, [input_nr][pol])
# repeat nyquist zone for all beamlets, to match the shape of beamlet_subbands
- nyquist_zones = numpy.repeat(nyquist_zones, constants.N_beamlets_ctrl, axis=1)
+ nyquist_zones = numpy.repeat(nyquist_zones, N_beamlets_ctrl, axis=1)
# compute the frequency of each beamlet for each input
return self._subband_frequencies(beamlet_subbands, self.sdp_proxy.clock_RW, nyquist_zones)
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/bst.py b/tangostationcontrol/tangostationcontrol/devices/sdp/bst.py
index 3636ea25c548fdb41058abc6a48049ddd86bee4e..af885a5f4cd5f5c1c6602928385765bd2e75d55c 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/bst.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/bst.py
@@ -20,7 +20,7 @@ from lofar_station_client.statistics.collector import BSTCollector
# Own imports
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_pn, BST_MAX_BLOCKS, N_beamlets_max
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.clients.opcua_client import OPCUAConnection
from tangostationcontrol.clients.statistics.client import StatisticsClient
@@ -53,7 +53,7 @@ class BST(Statistics):
FPGA_bst_offload_enable_RW_default = device_property(
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=[True] * constants.N_pn
+ default_value=[True] * N_pn
)
FIRST_DEFAULT_SETTINGS = [
@@ -70,33 +70,33 @@ class BST(Statistics):
# ----------
# FPGA control points for BSTs
- FPGA_bst_offload_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_enable_RW"], datatype=bool, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_bst_offload_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_enable_R"], datatype=bool, dims=(constants.N_pn,))
- FPGA_bst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_eth_destination_mac_RW"], datatype=str, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_bst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_eth_destination_mac_R"], datatype=str, dims=(constants.N_pn,))
- FPGA_bst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_ip_destination_address_RW"], datatype=str, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_bst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_ip_destination_address_R"], datatype=str, dims=(constants.N_pn,))
- FPGA_bst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_bst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(constants.N_pn,))
- FPGA_bst_offload_bsn_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_bsn_R"], datatype=numpy.int64, dims=(constants.N_pn,))
-
- FPGA_bst_offload_nof_packets_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_nof_packets_R"], datatype=numpy.int32, dims=(constants.N_pn,))
- FPGA_bst_offload_nof_valid_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_nof_valid_R"], datatype=numpy.int32, dims=(constants.N_pn,))
+ FPGA_bst_offload_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_enable_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_bst_offload_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_enable_R"], datatype=bool, dims=(N_pn,))
+ FPGA_bst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_eth_destination_mac_RW"], datatype=str, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_bst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_eth_destination_mac_R"], datatype=str, dims=(N_pn,))
+ FPGA_bst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_ip_destination_address_RW"], datatype=str, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_bst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_ip_destination_address_R"], datatype=str, dims=(N_pn,))
+ FPGA_bst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_bst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(N_pn,))
+ FPGA_bst_offload_bsn_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_bst_offload_bsn_R"], datatype=numpy.int64, dims=(N_pn,))
+
+ FPGA_bst_offload_nof_packets_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_nof_packets_R"], datatype=numpy.int32, dims=(N_pn,))
+ FPGA_bst_offload_nof_valid_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_nof_valid_R"], datatype=numpy.int32, dims=(N_pn,))
# number of packets with valid payloads
- nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(N_pn,), datatype=numpy.uint64)
# number of packets with invalid payloads
- nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(N_pn,), datatype=numpy.uint64)
# latest BSTs
- bst_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "bst_values"}, dims=(constants.BST_MAX_BLOCKS, constants.N_beamlets_max), datatype=numpy.uint64)
+ bst_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "bst_values"}, dims=(BST_MAX_BLOCKS, N_beamlets_max), datatype=numpy.uint64)
# reported timestamp
# for each row in the latest BSTs
- bst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "bst_timestamps"}, dims=(constants.BST_MAX_BLOCKS,), datatype=numpy.uint64)
+ bst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "bst_timestamps"}, dims=(BST_MAX_BLOCKS,), datatype=numpy.uint64)
# ----------
# Summarising Attributes
# ----------
- FPGA_processing_error_R = attribute(dtype=(bool,), max_dim_x=constants.N_pn)
+ FPGA_processing_error_R = attribute(dtype=(bool,), max_dim_x=N_pn)
def read_FPGA_processing_error_R(self):
return self.sdp_proxy.TR_fpga_mask_RW & (
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
index b7ae61b8f7fd3d6720a4561b8c8eae1181933b98..f33b7c6df9c3c12d16a75cf36248cb50bf18e2d6 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/digitalbeam.py
@@ -13,7 +13,7 @@ from tango.server import attribute, AttrWriteType, device_property
# Additional import
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_beamlets_ctrl, MAX_ANTENNA, N_xyz, A_pn, N_pn, N_pol
from tangostationcontrol.devices.beam_device import beam_device
from tangostationcontrol.devices.device_decorators import TimeIt
from tangostationcontrol.common.lofar_logging import log_exceptions
@@ -73,10 +73,10 @@ class DigitalBeam(beam_device):
dtype=numpy.float64, fget=lambda self: self._delays.statistics["last"] or 0)
input_select_RW = attribute(doc='Selection of inputs to use for forming each beamlet. Allows selecting broken antennas.',
- dtype=((bool,),), max_dim_x=constants.N_beamlets_ctrl, max_dim_y=constants.MAX_ANTENNA, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed")
+ dtype=((bool,),), max_dim_x=N_beamlets_ctrl, max_dim_y=MAX_ANTENNA, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed")
antenna_select_RW = attribute(doc='Selection of antennas desired to use for forming each beamlet (= a subset of input_select of the configured antennas). Unselects broken antennas.',
- dtype=((bool,),), max_dim_x=constants.N_beamlets_ctrl, max_dim_y=constants.MAX_ANTENNA, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed")
+ dtype=((bool,),), max_dim_x=N_beamlets_ctrl, max_dim_y=MAX_ANTENNA, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed")
nr_inputs_R = attribute(doc='Number of configured inputs from the associated antenna field.',
dtype=numpy.uint32, fget="nr_inputs")
@@ -94,7 +94,7 @@ class DigitalBeam(beam_device):
def read_antenna_select_RW(self):
# select only the rows from self.__input_select for which a mapping onto antennas is defined.
- antenna_select = [[False] * constants.N_beamlets_ctrl] * self.nr_inputs()
+ antenna_select = [[False] * N_beamlets_ctrl] * self.nr_inputs()
for antenna_nr, (fpga_nr, input_nr) in enumerate(self.antennafield_proxy.Antenna_to_SDP_Mapping_R):
if input_nr >= 0:
@@ -126,7 +126,7 @@ class DigitalBeam(beam_device):
@log_exceptions()
def configure_for_initialise(self):
- super().configure_for_initialise(constants.N_beamlets_ctrl)
+ super().configure_for_initialise(N_beamlets_ctrl)
# Set a reference of RECV device that is correlated to this BEAM device
util = Util.instance()
@@ -140,14 +140,14 @@ class DigitalBeam(beam_device):
# Retrieve positions from RECV device
reference_itrf = self.antennafield_proxy.Antenna_Field_Reference_ITRF_R
- antenna_itrf = self.antennafield_proxy.Antenna_Reference_ITRF_R.reshape(-1, constants.N_xyz)
+ antenna_itrf = self.antennafield_proxy.Antenna_Reference_ITRF_R.reshape(-1, N_xyz)
# Generate positions for all FPGA inputs.
# Use reference position for any missing antennas so they always get a delay of 0
- input_itrf = numpy.array([reference_itrf] * constants.MAX_ANTENNA)
+ input_itrf = numpy.array([reference_itrf] * MAX_ANTENNA)
for antenna_nr, (fpga_nr, input_nr) in enumerate(self.antennafield_proxy.Antenna_to_SDP_Mapping_R):
if input_nr >= 0:
- input_itrf[fpga_nr * constants.A_pn + input_nr] = antenna_itrf[antenna_nr]
+ input_itrf[fpga_nr * A_pn + input_nr] = antenna_itrf[antenna_nr]
# a delay calculator
@@ -157,8 +157,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
- self.write_input_select_RW(numpy.zeros((constants.MAX_ANTENNA, constants.N_beamlets_ctrl), dtype=bool))
- self.write_antenna_select_RW(numpy.ones((self.nr_inputs(), constants.N_beamlets_ctrl), dtype=bool))
+ self.write_input_select_RW(numpy.zeros((MAX_ANTENNA, N_beamlets_ctrl), dtype=bool))
+ self.write_antenna_select_RW(numpy.ones((self.nr_inputs(), N_beamlets_ctrl), dtype=bool))
# --------
# internal functions
@@ -172,7 +172,7 @@ class DigitalBeam(beam_device):
Returns delays[antenna][beamlet]
"""
- delays = numpy.zeros((constants.MAX_ANTENNA, constants.N_beamlets_ctrl), dtype=numpy.float64)
+ delays = numpy.zeros((MAX_ANTENNA, N_beamlets_ctrl), dtype=numpy.float64)
d = self.delay_calculator
d.set_measure_time(timestamp)
@@ -186,7 +186,7 @@ class DigitalBeam(beam_device):
""" Converts an array with dimensions [antenna][beamlet] -> [fpga_nr][input_nr][pol_nr][beamlet]
by repeating the values for both polarisations. """
- assert arr.shape == (constants.MAX_ANTENNA, constants.N_beamlets_ctrl)
+ assert arr.shape == (MAX_ANTENNA, N_beamlets_ctrl)
# Each antenna maps on [fpga_nr][input_nr][0] and [fpga_nr][input_nr][1], and we work
# with [antenna][beamlet], so we have to interleave copies of the input array per beamlet
@@ -200,7 +200,7 @@ class DigitalBeam(beam_device):
result = result.reshape(result.shape[0] * 2, result.shape[1] // 2)
# cast into (fpga_nr, [input_nr][pol_nr][beamlet])
- result = result.reshape((constants.N_pn, constants.A_pn * constants.N_pol * constants.N_beamlets_ctrl))
+ result = result.reshape((N_pn, A_pn * N_pol * N_beamlets_ctrl))
return result
@@ -217,7 +217,7 @@ class DigitalBeam(beam_device):
fpga_delays = self._map_inputs_on_polarised_inputs(antenna_delays)
beam_weights = self.beamlet_proxy.calculate_bf_weights(fpga_delays.flatten())
- beam_weights = beam_weights.reshape((constants.N_pn, constants.A_pn * constants.N_pol * constants.N_beamlets_ctrl))
+ beam_weights = beam_weights.reshape((N_pn, A_pn * N_pol * N_beamlets_ctrl))
return beam_weights
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
index 1e1b24f233444b1a6214344078d55a1f8ec4d682..ef02614641aa156513129b088db1f00abe534a16 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
@@ -20,7 +20,7 @@ from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import S_pn, N_pn, CLK_200_MHZ, CLK_160_MHZ, N_subband_res, N_subbands, DEFAULT_SUBBAND, N_beamsets_ctrl, DEFAULT_POLLING_PERIOD
import numpy
@@ -30,11 +30,6 @@ __all__ = ["SDP", "main"]
@device_logging_to_python()
class SDP(opcua_device):
- # redeclared frequently used constants
- S_pn = constants.S_pn
- N_pn = constants.N_pn
-
-
# -----------------
# Device Properties
# -----------------
@@ -69,7 +64,7 @@ class SDP(opcua_device):
dtype='DevVarDoubleArray',
mandatory=False,
# Emit a signal on subband 102
- default_value=[[constants.DEFAULT_SUBBAND * float(constants.CLK_200_MHZ) / constants.N_subband_res] * S_pn] * N_pn
+ default_value=[[DEFAULT_SUBBAND * float(CLK_200_MHZ) / N_subband_res] * S_pn] * N_pn
)
FPGA_wg_phase_RW_default = device_property(
@@ -93,13 +88,13 @@ class SDP(opcua_device):
FPGA_subband_weights_RW_default = device_property(
dtype='DevVarULongArray',
mandatory=False,
- default_value=[[8192] * S_pn * constants.N_subbands] * N_pn
+ default_value=[[8192] * S_pn * N_subbands] * N_pn
)
clock_RW_default = device_property(
dtype='DevULong',
mandatory=False,
- default_value = constants.CLK_200_MHZ
+ default_value = CLK_200_MHZ
)
TRANSLATOR_DEFAULT_SETTINGS = [
@@ -129,8 +124,8 @@ class SDP(opcua_device):
FPGA_ring_use_cable_to_next_rn_RW = attribute_wrapper(comms_annotation=["FPGA_ring_use_cable_to_next_rn_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
FPGA_ring_use_cable_to_previous_rn_R = attribute_wrapper(comms_annotation=["FPGA_ring_use_cable_to_previous_rn_R"], datatype=bool, dims=(N_pn,))
FPGA_ring_use_cable_to_previous_rn_RW = attribute_wrapper(comms_annotation=["FPGA_ring_use_cable_to_previous_rn_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_scrap_R = attribute_wrapper(comms_annotation=["FPGA_scrap_R"], datatype=numpy.int32, dims=(N_pn, constants.N_subbands))
- FPGA_scrap_RW = attribute_wrapper(comms_annotation=["FPGA_scrap_RW"], datatype=numpy.int32, dims=(N_pn, constants.N_subbands), access=AttrWriteType.READ_WRITE)
+ FPGA_scrap_R = attribute_wrapper(comms_annotation=["FPGA_scrap_R"], datatype=numpy.int32, dims=(N_pn, N_subbands))
+ FPGA_scrap_RW = attribute_wrapper(comms_annotation=["FPGA_scrap_RW"], datatype=numpy.int32, dims=(N_pn, N_subbands), access=AttrWriteType.READ_WRITE)
FPGA_sdp_info_antenna_band_index_R = attribute_wrapper(comms_annotation=["FPGA_sdp_info_antenna_band_index_R"], datatype=numpy.uint32, dims=(N_pn,))
FPGA_sdp_info_block_period_R = attribute_wrapper(comms_annotation=["FPGA_sdp_info_block_period_R"], datatype=numpy.uint32, dims=(N_pn,))
FPGA_sdp_info_f_adc_R = attribute_wrapper(comms_annotation=["FPGA_sdp_info_f_adc_R"], datatype=numpy.uint32, dims=(N_pn,))
@@ -141,8 +136,8 @@ class SDP(opcua_device):
FPGA_sdp_info_observation_id_RW = attribute_wrapper(comms_annotation=["FPGA_sdp_info_observation_id_RW"], datatype=numpy.uint32, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
FPGA_sdp_info_station_id_R = attribute_wrapper(comms_annotation=["FPGA_sdp_info_station_id_R"], datatype=numpy.uint32, dims=(N_pn,))
FPGA_sdp_info_station_id_RW = attribute_wrapper(comms_annotation=["FPGA_sdp_info_station_id_RW"], datatype=numpy.uint32, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_subband_weights_R = attribute_wrapper(comms_annotation=["FPGA_subband_weights_R"], datatype=numpy.uint32, dims=(N_pn, S_pn, constants.N_subbands))
- FPGA_subband_weights_RW = attribute_wrapper(comms_annotation=["FPGA_subband_weights_RW"], datatype=numpy.uint32, dims=(N_pn, S_pn, constants.N_subbands), access=AttrWriteType.READ_WRITE)
+ FPGA_subband_weights_R = attribute_wrapper(comms_annotation=["FPGA_subband_weights_R"], datatype=numpy.uint32, dims=(N_pn, S_pn, N_subbands))
+ FPGA_subband_weights_RW = attribute_wrapper(comms_annotation=["FPGA_subband_weights_RW"], datatype=numpy.uint32, dims=(N_pn, S_pn, N_subbands), access=AttrWriteType.READ_WRITE)
FPGA_time_since_last_pps_R = attribute_wrapper(comms_annotation=["FPGA_time_since_last_pps_R"], datatype=numpy.float_, dims=(N_pn,))
FPGA_temp_R = attribute_wrapper(comms_annotation=["FPGA_temp_R"], datatype=numpy.float_, dims=(N_pn,))
FPGA_wg_amplitude_R = attribute_wrapper(comms_annotation=["FPGA_wg_amplitude_R"], datatype=numpy.float_, dims=(N_pn, S_pn))
@@ -179,7 +174,7 @@ class SDP(opcua_device):
FPGA_signal_input_samples_delay_R = attribute_wrapper(comms_annotation=["FPGA_signal_input_samples_delay_R"], datatype=numpy.uint32, dims=(N_pn, S_pn))
FPGA_signal_input_samples_delay_RW = attribute_wrapper(comms_annotation=["FPGA_signal_input_samples_delay_RW"], datatype=numpy.uint32, dims=(N_pn, S_pn), access=AttrWriteType.READ_WRITE)
- FPGA_bst_offload_bsn_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_bsn_R"], datatype=numpy.int64, dims=(N_pn, constants.N_beamsets_ctrl))
+ FPGA_bst_offload_bsn_R = attribute_wrapper(comms_annotation=["FPGA_bst_offload_bsn_R"], datatype=numpy.int64, dims=(N_pn, N_beamsets_ctrl))
antenna_type_RW = attribute(doc='Type of antenna (LBA or HBA) attached to each input of the FPGAs',
dtype=((str,),), max_dim_y=N_pn, max_dim_x=S_pn,
@@ -187,10 +182,10 @@ class SDP(opcua_device):
nyquist_zone_R = attribute(doc='Nyquist zone of each input.',
dtype=((numpy.uint32,),), max_dim_y=N_pn, max_dim_x=S_pn,
fisallowed="is_attribute_access_allowed",
- polling_period=constants.DEFAULT_POLLING_PERIOD, abs_change=1)
+ polling_period=DEFAULT_POLLING_PERIOD, abs_change=1)
clock_RW = attribute(doc='Configured sampling clock (Hz)',
dtype=numpy.uint32, access=AttrWriteType.READ_WRITE, fisallowed="is_attribute_access_allowed",
- polling_period=constants.DEFAULT_POLLING_PERIOD, abs_change=1)
+ polling_period=DEFAULT_POLLING_PERIOD, abs_change=1)
def read_antenna_type_RW(self):
return self._antenna_type
@@ -250,7 +245,7 @@ class SDP(opcua_device):
return numpy.uint32(clocks_in_mask[0]) if clocks_in_mask else self.clock_RW_default
def write_clock_RW(self, clock):
- if clock not in (constants.CLK_160_MHZ, constants.CLK_200_MHZ):
+ if clock not in (CLK_160_MHZ, CLK_200_MHZ):
raise ValueError(f"Unsupported clock frequency: {clock}")
# Tell all FPGAs to use this clock
@@ -303,7 +298,7 @@ class SDP(opcua_device):
wait_for = ~(self.read_attribute("TR_fpga_communication_error_R")) & self.read_attribute("TR_fpga_mask_R")
# Order the correct firmare to be loaded
- self.proxy.FPGA_boot_image_RW = [1] * constants.N_pn
+ self.proxy.FPGA_boot_image_RW = [1] * N_pn
# 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)
@@ -313,9 +308,9 @@ class SDP(opcua_device):
# 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] * constants.N_pn
+ self.TR_fpga_mask_RW = [True] * N_pn
# Boot the boot image firmware
- self.FPGA_boot_image_RW = [0] * constants.N_pn
+ self.FPGA_boot_image_RW = [0] * N_pn
# Restore the mask
self.TR_fpga_mask_RW = TR_fpga_mask
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/sst.py b/tangostationcontrol/tangostationcontrol/devices/sdp/sst.py
index 6afaa1b6df9e50fa327b8ed5791f7bab86003ab6..52faf1a74030a5a443baa25456969cf2d1ce2d42 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/sst.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/sst.py
@@ -17,7 +17,7 @@ from tango import AttrWriteType
# Additional import
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_pn, MAX_INPUTS, N_subbands
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.clients.opcua_client import OPCUAConnection
from tangostationcontrol.clients.statistics.client import StatisticsClient
@@ -55,13 +55,13 @@ class SST(Statistics):
FPGA_sst_offload_enable_RW_default = device_property(
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=[True] * constants.N_pn
+ default_value=[True] * N_pn
)
FPGA_sst_offload_weighted_subbands_RW_default = device_property(
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=[True] * constants.N_pn
+ default_value=[True] * N_pn
)
FIRST_DEFAULT_SETTINGS = [
@@ -80,39 +80,39 @@ class SST(Statistics):
# ----------
# FPGA control points for SSTs
- FPGA_sst_offload_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_enable_RW"], datatype=bool, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_sst_offload_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_enable_R"], datatype=bool, dims=(constants.N_pn,))
- FPGA_sst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_eth_destination_mac_RW"], datatype=str, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_sst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_eth_destination_mac_R"], datatype=str, dims=(constants.N_pn,))
- FPGA_sst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_ip_destination_address_RW"], datatype=str, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_sst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_ip_destination_address_R"], datatype=str, dims=(constants.N_pn,))
- FPGA_sst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_sst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(constants.N_pn,))
- FPGA_sst_offload_bsn_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_bsn_R"], datatype=numpy.int64, dims=(constants.N_pn,))
- FPGA_sst_offload_weighted_subbands_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_weighted_subbands_RW"], datatype=bool, dims=(constants.N_pn,), access=AttrWriteType.READ_WRITE)
- FPGA_sst_offload_weighted_subbands_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_weighted_subbands_R"], datatype=bool, dims=(constants.N_pn,))
-
- FPGA_sst_offload_nof_packets_R = attribute_wrapper(comms_annotation=["FPGA_sst_offload_nof_packets_R"], datatype=numpy.int32, dims=(constants.N_pn,))
- FPGA_sst_offload_nof_valid_R = attribute_wrapper(comms_annotation=["FPGA_sst_offload_nof_valid_R"], datatype=numpy.int32, dims=(constants.N_pn,))
+ FPGA_sst_offload_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_enable_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_sst_offload_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_enable_R"], datatype=bool, dims=(N_pn,))
+ FPGA_sst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_eth_destination_mac_RW"], datatype=str, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_sst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_eth_destination_mac_R"], datatype=str, dims=(N_pn,))
+ FPGA_sst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_ip_destination_address_RW"], datatype=str, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_sst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_ip_destination_address_R"], datatype=str, dims=(N_pn,))
+ FPGA_sst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_sst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(N_pn,))
+ FPGA_sst_offload_bsn_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_bsn_R"], datatype=numpy.int64, dims=(N_pn,))
+ FPGA_sst_offload_weighted_subbands_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_weighted_subbands_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
+ FPGA_sst_offload_weighted_subbands_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_sst_offload_weighted_subbands_R"], datatype=bool, dims=(N_pn,))
+
+ FPGA_sst_offload_nof_packets_R = attribute_wrapper(comms_annotation=["FPGA_sst_offload_nof_packets_R"], datatype=numpy.int32, dims=(N_pn,))
+ FPGA_sst_offload_nof_valid_R = attribute_wrapper(comms_annotation=["FPGA_sst_offload_nof_valid_R"], datatype=numpy.int32, dims=(N_pn,))
# number of packets with valid payloads
- nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(N_pn,), datatype=numpy.uint64)
# number of packets with invalid payloads
- nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(N_pn,), datatype=numpy.uint64)
# latest SSTs
- sst_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "sst_values"}, dims=(constants.MAX_INPUTS, constants.N_subbands), datatype=numpy.uint64)
+ sst_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "sst_values"}, dims=(MAX_INPUTS, N_subbands), datatype=numpy.uint64)
# reported timestamp
# for each row in the latest SSTs
- sst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "sst_timestamps"}, dims=(constants.MAX_INPUTS,), datatype=numpy.uint64)
+ sst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "sst_timestamps"}, dims=(MAX_INPUTS,), datatype=numpy.uint64)
# integration interval for each row in the latest SSTs
- integration_interval_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "integration_intervals"}, dims=(constants.MAX_INPUTS,), datatype=numpy.float32)
+ integration_interval_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "integration_intervals"}, dims=(MAX_INPUTS,), datatype=numpy.float32)
# whether the subband data was calibrated by the SDP (that is, were subband weights applied)
- subbands_calibrated_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "subbands_calibrated"}, dims=(constants.MAX_INPUTS,), datatype=bool)
+ subbands_calibrated_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "subbands_calibrated"}, dims=(MAX_INPUTS,), datatype=bool)
# ----------
# Summarising Attributes
# ----------
- FPGA_processing_error_R = attribute(dtype=(bool,), max_dim_x=constants.N_pn)
+ FPGA_processing_error_R = attribute(dtype=(bool,), max_dim_x=N_pn)
def read_FPGA_processing_error_R(self):
return self.sdp_proxy.TR_fpga_mask_RW & (
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/statistics.py b/tangostationcontrol/tangostationcontrol/devices/sdp/statistics.py
index 4f1d62947356e90173fcdd4095a8088dcb29576d..8e2bc497a752176a96bbd7d996448a15fac36c78 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/statistics.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/statistics.py
@@ -22,7 +22,7 @@ from tangostationcontrol.clients.statistics.client import StatisticsClient
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
from tangostationcontrol.common.lofar_logging import log_exceptions
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import MAX_ETH_FRAME_SIZE
import logging
@@ -67,7 +67,7 @@ class Statistics(opcua_device):
# number of UDP packets that were dropped because we couldn't keep up with processing
nof_packets_dropped_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "udp", "parameter": "nof_packets_dropped"}, datatype=numpy.uint64)
# last packet we processed
- last_packet_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "udp", "parameter": "last_packet"}, dims=(constants.MAX_ETH_FRAME_SIZE,), datatype=numpy.uint8)
+ last_packet_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "udp", "parameter": "last_packet"}, dims=(MAX_ETH_FRAME_SIZE,), datatype=numpy.uint8)
# when last packet was received
last_packet_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "udp", "parameter": "last_packet_timestamp"}, datatype=numpy.uint64)
@@ -87,7 +87,7 @@ class Statistics(opcua_device):
# number of invalid (non-SST) packets received
nof_invalid_packets_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_invalid_packets"}, datatype=numpy.uint64)
# last packet that could not be parsed
- last_invalid_packet_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "last_invalid_packet"}, dims=(constants.MAX_ETH_FRAME_SIZE,), datatype=numpy.uint8)
+ last_invalid_packet_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "last_invalid_packet"}, dims=(MAX_ETH_FRAME_SIZE,), datatype=numpy.uint8)
# what the last exception was
last_invalid_packet_exception_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "last_invalid_packet_exception"}, datatype=str)
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/xst.py b/tangostationcontrol/tangostationcontrol/devices/sdp/xst.py
index 89d1722281111e6a83386e8f2e37a1d9ab539b79..f39bbb3843377f77229e4a91f3fcde23443f3ad0 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/xst.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/xst.py
@@ -19,7 +19,7 @@ from lofar_station_client.statistics.collector import XSTCollector
# Additional import
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_pn, P_sq, DEFAULT_SUBBAND, MAX_PARALLEL_SUBBANDS, MAX_BLOCKS, BLOCK_LENGTH, VALUES_PER_COMPLEX, MAX_INPUTS
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.clients.opcua_client import OPCUAConnection
from tangostationcontrol.clients.statistics.client import StatisticsClient
@@ -33,9 +33,6 @@ __all__ = ["XST", "main"]
class XST(Statistics):
- N_pn = constants.N_pn
- P_sq = constants.P_sq
-
STATISTICS_COLLECTOR_CLASS = XSTCollector
# -----------------
@@ -60,25 +57,25 @@ class XST(Statistics):
FPGA_xst_processing_enable_RW_default = device_property(
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=[True] * constants.N_pn
+ default_value=[True] * N_pn
)
FPGA_xst_subband_select_RW_default = device_property(
dtype='DevVarULongArray',
mandatory=False,
- default_value=[[0,constants.DEFAULT_SUBBAND,0,0,0,0,0,0]] * constants.N_pn
+ default_value=[[0,DEFAULT_SUBBAND,0,0,0,0,0,0]] * N_pn
)
FPGA_xst_integration_interval_RW_default = device_property(
dtype='DevVarDoubleArray',
mandatory=False,
- default_value=[1.0] * constants.N_pn
+ default_value=[1.0] * N_pn
)
FPGA_xst_offload_enable_RW_default = device_property(
dtype='DevVarBooleanArray',
mandatory=False,
- default_value=[True] * constants.N_pn
+ default_value=[True] * N_pn
)
FIRST_DEFAULT_SETTINGS = [
@@ -111,8 +108,8 @@ class XST(Statistics):
FPGA_xst_offload_bsn_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_offload_bsn_R"], datatype=numpy.int64, dims=(N_pn,))
FPGA_xst_processing_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_processing_enable_RW"], datatype=bool, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
FPGA_xst_processing_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_processing_enable_R"], datatype=bool, dims=(N_pn,))
- FPGA_xst_subband_select_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_subband_select_RW"], datatype=numpy.uint32, dims=(constants.MAX_PARALLEL_SUBBANDS,N_pn), access=AttrWriteType.READ_WRITE)
- FPGA_xst_subband_select_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_subband_select_R"], datatype=numpy.uint32, dims=(constants.MAX_PARALLEL_SUBBANDS,N_pn))
+ FPGA_xst_subband_select_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_subband_select_RW"], datatype=numpy.uint32, dims=(MAX_PARALLEL_SUBBANDS,N_pn), access=AttrWriteType.READ_WRITE)
+ FPGA_xst_subband_select_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_subband_select_R"], datatype=numpy.uint32, dims=(MAX_PARALLEL_SUBBANDS,N_pn))
FPGA_xst_offload_nof_crosslets_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_offload_nof_crosslets_RW"], datatype=numpy.uint32, dims=(N_pn,), access=AttrWriteType.READ_WRITE)
FPGA_xst_offload_nof_crosslets_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_offload_nof_crosslets_R"], datatype=numpy.uint32, dims=(N_pn,))
@@ -149,78 +146,78 @@ class XST(Statistics):
FPGA_xst_ring_tx_latency_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["FPGA_xst_ring_tx_latency_R"], datatype=numpy.int32, dims=(N_pn,N_pn))
# number of packets with valid payloads
- nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_valid_payloads_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_valid_payloads"}, dims=(N_pn,), datatype=numpy.uint64)
# number of packets with invalid payloads
- nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(constants.N_pn,), datatype=numpy.uint64)
+ nof_payload_errors_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "nof_payload_errors"}, dims=(N_pn,), datatype=numpy.uint64)
# latest XSTs
- xst_blocks_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_blocks", "reshape": True}, dims=(constants.MAX_PARALLEL_SUBBANDS, constants.MAX_BLOCKS, constants.BLOCK_LENGTH, constants.BLOCK_LENGTH, constants.VALUES_PER_COMPLEX), datatype=numpy.int64)
+ xst_blocks_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_blocks", "reshape": True}, dims=(MAX_PARALLEL_SUBBANDS, MAX_BLOCKS, BLOCK_LENGTH, BLOCK_LENGTH, VALUES_PER_COMPLEX), datatype=numpy.int64)
# whether the values in the block are conjugated and transposed
- xst_conjugated_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_conjugated", "reshape": True}, dims=(constants.MAX_PARALLEL_SUBBANDS, XSTCollector.MAX_BLOCKS), datatype=bool)
+ xst_conjugated_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_conjugated", "reshape": True}, dims=(MAX_PARALLEL_SUBBANDS, MAX_BLOCKS), datatype=bool)
# reported timestamp for each subband in the latest XSTs
- xst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_timestamps"}, dims=(constants.MAX_PARALLEL_SUBBANDS,), datatype=numpy.uint64)
+ xst_timestamp_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_timestamps"}, dims=(MAX_PARALLEL_SUBBANDS,), datatype=numpy.uint64)
# which subband the XSTs describe
- xst_subbands_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_subbands"}, dims=(constants.MAX_PARALLEL_SUBBANDS,), datatype=numpy.uint16)
+ xst_subbands_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_subbands"}, dims=(MAX_PARALLEL_SUBBANDS,), datatype=numpy.uint16)
# integration interval for each subband in the latest XSTs
- xst_integration_interval_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_integration_intervals"}, dims=(constants.MAX_PARALLEL_SUBBANDS,), datatype=numpy.float32)
+ xst_integration_interval_R = attribute_wrapper(comms_id=StatisticsClient, comms_annotation={"type": "statistics", "parameter": "xst_integration_intervals"}, dims=(MAX_PARALLEL_SUBBANDS,), datatype=numpy.float32)
# xst_R, but as a matrix of subband x (input x input)
- xst_real_R = attribute(max_dim_x=constants.MAX_INPUTS * constants.MAX_INPUTS, max_dim_y=constants.MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
- xst_imag_R = attribute(max_dim_x=constants.MAX_INPUTS * constants.MAX_INPUTS, max_dim_y=constants.MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
- xst_power_R = attribute(max_dim_x=constants.MAX_INPUTS * constants.MAX_INPUTS, max_dim_y=constants.MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
- xst_phase_R = attribute(max_dim_x=constants.MAX_INPUTS * constants.MAX_INPUTS, max_dim_y=constants.MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
+ xst_real_R = attribute(max_dim_x=MAX_INPUTS * MAX_INPUTS, max_dim_y=MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
+ xst_imag_R = attribute(max_dim_x=MAX_INPUTS * MAX_INPUTS, max_dim_y=MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
+ xst_power_R = attribute(max_dim_x=MAX_INPUTS * MAX_INPUTS, max_dim_y=MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
+ xst_phase_R = attribute(max_dim_x=MAX_INPUTS * MAX_INPUTS, max_dim_y=MAX_PARALLEL_SUBBANDS, dtype=((numpy.float32,),))
def read_xst_real_R(self):
- return numpy.real(self.statistics_client.collector.xst_values()).reshape(constants.MAX_PARALLEL_SUBBANDS, constants.MAX_INPUTS * constants.MAX_INPUTS)
+ return numpy.real(self.statistics_client.collector.xst_values()).reshape(MAX_PARALLEL_SUBBANDS, MAX_INPUTS * MAX_INPUTS)
def read_xst_imag_R(self):
- return numpy.imag(self.statistics_client.collector.xst_values()).reshape(constants.MAX_PARALLEL_SUBBANDS, constants.MAX_INPUTS * constants.MAX_INPUTS)
+ return numpy.imag(self.statistics_client.collector.xst_values()).reshape(MAX_PARALLEL_SUBBANDS, MAX_INPUTS * MAX_INPUTS)
def read_xst_power_R(self):
- return numpy.abs(self.statistics_client.collector.xst_values()).reshape(constants.MAX_PARALLEL_SUBBANDS, constants.MAX_INPUTS * constants.MAX_INPUTS)
+ return numpy.abs(self.statistics_client.collector.xst_values()).reshape(MAX_PARALLEL_SUBBANDS, MAX_INPUTS * MAX_INPUTS)
def read_xst_phase_R(self):
- return numpy.angle(self.statistics_client.collector.xst_values()).reshape(constants.MAX_PARALLEL_SUBBANDS, constants.MAX_INPUTS * constants.MAX_INPUTS)
+ return numpy.angle(self.statistics_client.collector.xst_values()).reshape(MAX_PARALLEL_SUBBANDS, MAX_INPUTS * MAX_INPUTS)
# xst_R, but as a matrix of input x input, for each specific subband index
- xst_0_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(0))
- xst_0_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(0))
- xst_0_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(0))
- xst_0_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(0))
-
- xst_1_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(1))
- xst_1_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(1))
- xst_1_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(1))
- xst_1_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(1))
-
- xst_2_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(2))
- xst_2_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(2))
- xst_2_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(2))
- xst_2_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(2))
-
- xst_3_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(3))
- xst_3_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(3))
- xst_3_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(3))
- xst_3_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(3))
-
- xst_4_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(4))
- xst_4_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(4))
- xst_4_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(4))
- xst_4_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(4))
-
- xst_5_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(5))
- xst_5_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(5))
- xst_5_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(5))
- xst_5_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(5))
-
- xst_6_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(6))
- xst_6_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(6))
- xst_6_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(6))
- xst_6_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(6))
-
- xst_7_real_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(7))
- xst_7_imag_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(7))
- xst_7_power_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(7))
- xst_7_phase_R = attribute(max_dim_x=constants.MAX_INPUTS, max_dim_y=constants.MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(7))
+ xst_0_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(0))
+ xst_0_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(0))
+ xst_0_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(0))
+ xst_0_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(0))
+
+ xst_1_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(1))
+ xst_1_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(1))
+ xst_1_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(1))
+ xst_1_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(1))
+
+ xst_2_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(2))
+ xst_2_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(2))
+ xst_2_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(2))
+ xst_2_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(2))
+
+ xst_3_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(3))
+ xst_3_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(3))
+ xst_3_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(3))
+ xst_3_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(3))
+
+ xst_4_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(4))
+ xst_4_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(4))
+ xst_4_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(4))
+ xst_4_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(4))
+
+ xst_5_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(5))
+ xst_5_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(5))
+ xst_5_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(5))
+ xst_5_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(5))
+
+ xst_6_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(6))
+ xst_6_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(6))
+ xst_6_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(6))
+ xst_6_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(6))
+
+ xst_7_real_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_real_R(7))
+ xst_7_imag_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_imag_R(7))
+ xst_7_power_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_power_R(7))
+ xst_7_phase_R = attribute(max_dim_x=MAX_INPUTS, max_dim_y=MAX_INPUTS, dtype=((numpy.float32,),), fget = lambda self: self.read_xst_N_phase_R(7))
def read_xst_N_real_R(self, subband_idx):
return numpy.real(self.statistics_client.collector.xst_values([subband_idx])[0])
diff --git a/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py b/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
index 0a8d067d233cbb55d20f33a4e19da461a9a79b16..8947e440364ac38d6da1fdd7ef5182018c91de9b 100644
--- a/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
+++ b/tangostationcontrol/tangostationcontrol/devices/temperature_manager.py
@@ -9,7 +9,7 @@
# Additional import
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
@@ -122,7 +122,7 @@ class TemperatureManager(lofar_device):
dev_attr.proxy.unsubscribe_event(dev_attr.subscription_id)
del dev_attr
- is_alarming_R = attribute(dtype=bool, polling_period=constants.DEFAULT_POLLING_PERIOD, fisallowed="is_attribute_access_allowed")
+ is_alarming_R = attribute(dtype=bool, polling_period=DEFAULT_POLLING_PERIOD, fisallowed="is_attribute_access_allowed")
def read_is_alarming_R(self):
# return whether any of the devices are alarming
diff --git a/tangostationcontrol/tangostationcontrol/devices/tilebeam.py b/tangostationcontrol/tangostationcontrol/devices/tilebeam.py
index 1207eaab54560684d89dbe9e8c2726fd960b8c85..1d6a59e352f2ce0995b34dae5de74179847a27f1 100644
--- a/tangostationcontrol/tangostationcontrol/devices/tilebeam.py
+++ b/tangostationcontrol/tangostationcontrol/devices/tilebeam.py
@@ -16,7 +16,7 @@ from tango import Util
# Additional import
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_elements, N_xyz, N_elements, N_pol
from tangostationcontrol.beam.delays import Delays
from tangostationcontrol.devices.beam_device import beam_device
from tangostationcontrol.devices.device_decorators import TimeIt
@@ -58,7 +58,7 @@ class TileBeam(beam_device):
# Retrieve positions from AntennaField device
Antenna_Reference_itrf = self.antennafield_proxy.Antenna_Reference_itrf_R
- HBAT_antenna_itrf_offsets = self.antennafield_proxy.HBAT_antenna_itrf_offsets_R.reshape(self._nr_tiles, constants.N_elements, constants.N_xyz)
+ HBAT_antenna_itrf_offsets = self.antennafield_proxy.HBAT_antenna_itrf_offsets_R.reshape(self._nr_tiles, N_elements, N_xyz)
# a delay calculator for each tile
self.HBAT_delay_calculators = [Delays(reference_itrf) for reference_itrf in Antenna_Reference_itrf]
@@ -77,7 +77,7 @@ class TileBeam(beam_device):
Returns delays[tile][element]
"""
- delays = numpy.zeros((self._nr_tiles, constants.N_elements), dtype=numpy.float64)
+ delays = numpy.zeros((self._nr_tiles, N_elements), dtype=numpy.float64)
for tile in range(self._nr_tiles):
# initialise delay calculator
@@ -107,7 +107,7 @@ class TileBeam(beam_device):
@TimeIt()
def _apply_weights(self, pointing_direction: numpy.array, timestamp: datetime.datetime, bf_delay_steps: numpy.array):
# Write weights to RECV through the AntennaToRecvMapper
- self.antennafield_proxy.HBAT_bf_delay_steps_RW = bf_delay_steps.reshape(self._nr_tiles, constants.N_elements * constants.N_pol)
+ self.antennafield_proxy.HBAT_bf_delay_steps_RW = bf_delay_steps.reshape(self._nr_tiles, N_elements * N_pol)
# Record where we now point to, now that we've updated the weights.
# Only the entries within the mask have been updated
diff --git a/tangostationcontrol/tangostationcontrol/devices/unb2.py b/tangostationcontrol/tangostationcontrol/devices/unb2.py
index 66ae6cc3775a70d16a0df3242a75d01f0a1a3f48..28716a92e666d48bb0780340e1f243f21e5b765e 100644
--- a/tangostationcontrol/tangostationcontrol/devices/unb2.py
+++ b/tangostationcontrol/tangostationcontrol/devices/unb2.py
@@ -17,7 +17,7 @@ from tango import AttrWriteType, DebugIt
# Additional import
from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.common.constants import constants
+from tangostationcontrol.common.constants import N_unb, N_fpga, N_ddr, N_qsfp, DEFAULT_POLLING_PERIOD
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
from tangostationcontrol.common.lofar_logging import device_logging_to_python
@@ -31,12 +31,6 @@ __all__ = ["UNB2", "main"]
@device_logging_to_python()
class UNB2(opcua_device):
- # redeclared commonly used constants for convenience
- N_unb = constants.N_unb
- N_fpga = constants.N_fpga
- N_ddr = constants.N_ddr
- N_qsfp = constants.N_qsfp
-
# -----------------
# Device Properties
# -----------------
@@ -141,7 +135,7 @@ class UNB2(opcua_device):
)
UNB2_IOUT_error_R = attribute(dtype=(bool,), max_dim_x=N_unb, fisallowed="is_attribute_access_allowed")
- UNB2_TEMP_error_R = attribute(dtype=(bool,), max_dim_x=N_unb, fisallowed="is_attribute_access_allowed", polling_period=constants.DEFAULT_POLLING_PERIOD)
+ UNB2_TEMP_error_R = attribute(dtype=(bool,), max_dim_x=N_unb, fisallowed="is_attribute_access_allowed", polling_period=DEFAULT_POLLING_PERIOD)
UNB2_VOUT_error_R = attribute(dtype=(bool,), max_dim_x=N_unb, fisallowed="is_attribute_access_allowed")
def read_UNB2_IOUT_error_R(self):
@@ -211,7 +205,7 @@ class UNB2(opcua_device):
# Save actual mask values
UNB2_mask = self.proxy.UNB2_mask_RW
# Set the mask to all Trues
- self.UNB2_mask_RW = [True] * constants.N_unb
+ self.UNB2_mask_RW = [True] * N_unb
# Turn off the uniboards
self.UNB2_off()
self.wait_attribute("UNB2TR_translator_busy_R", False, self.UNB2_On_Off_timeout)