diff --git a/tangostationcontrol/docs/source/calibration.rst b/tangostationcontrol/docs/source/calibration.rst
new file mode 100644
index 0000000000000000000000000000000000000000..b38b38d5c6fb5ccb3341d7ab60bfbc40ca9a7e42
--- /dev/null
+++ b/tangostationcontrol/docs/source/calibration.rst
@@ -0,0 +1,155 @@
+Calibration
+=============================
+
+The signal chain has to be tuned in order to precisely combine the sky signals received by the antennas.
+
+.. _casacore-measures:
+
+Emperedes & Geodetic calibration
+----------------------------------
+
+The ``TileBeam`` and ``DigitalBeam`` devices use `python-casacore <https://casacore.github.io/python-casacore/index.html>`_ to compute the direction of a given pointing with respect to our antennas and reference positions. Casacore in turn uses *measures* tables for the precise measurements of celestial positions, geodetical information, and time calibrations (f.e. leap seconds). These tables need to be installed and periodically updated to maintain the pointing accuracy:
+
+:measures_directory_R: Directory of the active set of measures tables. The directory name includes the timestamp denoting their age.
+
+ :type: ``str``
+
+:measures_directories_available_R: List of installed sets of measures tables.
+
+ :type: ``str[64]``
+
+:download_measures(): Download (but do not activate) the latest measures tables from ftp://ftp.astron.nl. Returns the directory name in which the measures were installed.
+
+ :returns: ``str``
+
+:use_measures(dir): Activate the measures tables in the provided directory. This necessitates turning off and restarting the TileBeam device, so the command will always appear to fail. Turn the device back and the selected measures tables will be active.
+
+ :returns: ``(does not return)``
+
+Instrumental calibration
+-----------------------------
+
+The antennas receive analog signals, which are delayed and attenuated (dampened) differently per antenna by the processing chain. These effects are caused most notably by:
+
+* The cables from antennas to RCUs,
+* The the RCU's processing chain,
+* The cables from RCU to RECV.
+
+We need to calibrate for the relative differences between the antennas, dominated by the cables from antennas to RCUs. To do so, we allow the station cabling to be configured, which provides us with a first-order model. From the configuration, the software computes default calibration values. The fine tuning of this calibration is to be overwritten by explicitly provided *calibration tables*, which are obtained through measurements. The calibration tables will be able to correct for any remaining differences in delay, phase, and power of the antennas.
+
+The instrument calibration depends on the frequency of the observed signal. We recognise several reference frequencies for which correction values are measured:
+
++--------------+----------------+------------------+-----------------------------+---------------------+
+| Antenna type | Frequency band | ``sdp.clock_RW`` | ``recv.RCU_band_select_RW`` | Reference frequency |
++==============+================+==================+=============================+=====================+
+| LBA | 10 - 90 MHz | (any) | 1 | 50 MHz |
++--------------+----------------+------------------+-----------------------------+---------------------+
+| LBA | 30 - 90 MHz | (any) | 2 | 50 MHz |
++--------------+----------------+------------------+-----------------------------+---------------------+
+| HBA | 110 - 190 MHz | 200_000_000 | 2 | 150 MHz |
++--------------+----------------+------------------+-----------------------------+---------------------+
+| HBA | 170 - 230 MHz | 160_000_000 | 1 | 200 MHz |
++--------------+----------------+------------------+-----------------------------+---------------------+
+| HBA | 210 - 240 MHz | 200_000_000 | 4 | 250 MHz |
++--------------+----------------+------------------+-----------------------------+---------------------+
+
+Instrument configuration
+`````````````````````````````
+
+The properties of the cables are modelled per cable type in the dictionary ``common.cables.cable_types``, and are named according to their length (f.e. "50m", "80m", etc). Which cables are used is configured in the properties of the ``AntennaField`` device:
+
+:Antenna_Cables: Names of the cable types used to connect each antenna (f.e. "50m", "80m", "120m"). Both polarisations of an antenna are assumed to be connected with the same type of cable.
+
+ :type: ``str[N_ant]``
+
+Information about the cabling is subsequently exposed through the following attributes:
+
+:Antenna_Cables_R: Which cable types are configured for each antenna (see the ``Antenna_Cables`` property).
+
+ :type: ``str[N_ant]``
+
+:Antenna_Cables_Delay_R: The delay introduced by the cabling for each antenna, in seconds.
+
+ :type: ``float[N_ant]``
+
+:Antenna_Cables_Loss_R: The attenuation introduced by the cabling for each antenna, in dB. The value returned depends on the current frequency that we observe from the antenna.
+
+ :type: ``float[N_ant]``
+
+Recalibrating the station
+`````````````````````````````
+
+The calibration values must be explicitly applied as part of initialising the RECV and SDP to process the antennas of an antenna field. These values are frequency dependent, so must be reapplied whenever the SDP clock changes, or whenever the RCU band selection does. The ``AntennaField`` device provides the following commands to apply the calibration:
+
+:calibrate_recv(): Compute and upload calibration settings to the RECV device(s) that process antennas of the antenna field.
+
+:calibrate_sdp(): Compute and upload calibration settings to the SDP device(s) that process antennas of the antenna field.
+
+Calibration explained
+`````````````````````````````
+
+We equalise the signals of the different antennas to compensate for the delay and attenuation effects, in two steps: coarse and fine. The following table describes what is corrected for where:
+
++------------+------------+---------------------------------------------+------------------------------+
+| Effect | Granuality | Compensation | How |
++============+============+=============================================+==============================+
+| Delay | Coarse | ``sdp.FPGA_signal_input_samples_delay_RW`` | Delaying using a ring buffer |
++------------+------------+---------------------------------------------+------------------------------+
+| Delay | Fine | ``sdp.FPGA_subband_weights_RW`` | Phase shifts |
++------------+------------+---------------------------------------------+------------------------------+
+| Attenuation| Coarse | ``recv.RCU_attenuator_dB_RW`` | Dampening whole dBs |
++------------+------------+---------------------------------------------+------------------------------+
+| Attenuation| Fine | ``sdp.FPGA_subband_weights_RW`` | Amplitude scaling |
++------------+------------+---------------------------------------------+------------------------------+
+
+The *coarse delay compensation* is done in SDP, by delaying all inputs to line up with the latest arriving one. The FPGAs do this through a *sample shift*, in which the samples from each input is delayed a fixed number of samples. At the 200 MHz clock, samples are 5 ns. The sample shift aligns the inputs with a remaining difference of +/- 2.5 ns.
+
+This remainder is corrected for in the *fine delay compensation*, by shifting the phases of each input backwards. A phase shift is frequency dependent (``-2pi * frequency * delay``), and is thus applied at the higher frequency resolution after creating subbands. The ``FPGA_subband_weights_RW`` in SDP allows us to configure a complex correction factor for each subband from each input. A phase shift ``phi`` is converted into a complex factor through ``cos(phi) + i * sin(phi)``.
+
+.. note:: The delay compensation shifts all antenna signals by a fixed amount: the number of samples to delay to line up with the longest cable. Yet we mark those signals as "now" in SDP. This introduces a temporal shift of the order of 200ns. This is deemed acceptable, as after the station FFT (that creates the subbands), we have 5.12ms samples, which is an order of magnitude higher time scale.
+
+The *coarse loss compensation* is done in RECV on the RCU, which can attenuate each input an integer number of decibels. We attenuate each signal to line up with the weakest. The remaining attenuation is +/- 0.5 dB.
+
+The remainder is corrected for in the *fine loss compensation*, by applying an amplitude scaling factor (``10^(-dB/10)``) as part of the complex ``FPGA_subband_weights_RW`` (see above). This scaling factor is the same for all subbands.
+
+The ``AntennaField`` device provides the following attributes that provide the results of the above calculations per antenna:
+
+:Calibration_SDP_Signal_Input_Samples_Delay_R: The number of samples that each antenna must be delayed in SDP to line up.
+
+ :type: ``uint32[N_ant]``
+
+:Calibration_RCU_Attenuation_dB_R: The amount of dB that each antenna must be dampened on the RCU to line up.
+
+ :type: ``uint32[N_ant]``
+
+:Calibration_SDP_Subband_Weights_Default_R: The computed remaining delay & loss compensation, to be applied as subband weights in SDP. Each antenna is represented by a triplet (delay_seconds, phase_offset, amplitude_scaling) that can be used to compute the complex correction factor for each subband.
+
+ :type: ``float[N_ant][3]``
+
+:Calibration_SDP_Subband_Weights_R: The remaining delay & loss compensation to be applied. Returns rows from the preconfigured calibration table if available, falling back to rows from the computed ``Calibration_SDP_Subband_Weights_Default_R`` table otherwise.
+
+ :type: ``float[N_ant][3]``
+
+The preconfigured calibration tables are described below.
+
+Calibration configuration
+`````````````````````````````
+
+The software models cannot account for all required corrections. We assume that the software correctly computes the coarse compensation, but that additional tuning of the fine compensations is needed. Actual measurements must be used to derive the precise correction factors for each antenna. The ``AntennaField`` device allows the following properties to be used to replace the computed ``Calibration_SDP_Subband_Weights_Default_R`` table for the fine correction factors:
+
+:Calibration_SDP_Subband_Weights_50MHz: The measured correction factors for each input, at the 50 MHz reference frequency. Each antenna is represented by a triplet (delay_seconds, phase_offset, amplitude_scaling).
+
+ :type: ``float[N_ant][3]``
+
+:Calibration_SDP_Subband_Weights_150MHz: The measured correction factors for each input, at the 150 MHz reference frequency. Each antenna is represented by a triplet (delay_seconds, phase_offset, amplitude_scaling).
+
+ :type: ``float[N_ant][3]``
+
+:Calibration_SDP_Subband_Weights_200MHz: The measured correction factors for each input, at the 200 MHz reference frequency. Each antenna is represented by a triplet (delay_seconds, phase_offset, amplitude_scaling).
+
+ :type: ``float[N_ant][3]``
+
+:Calibration_SDP_Subband_Weights_250MHz: The measured correction factors for each input, at the 250 MHz reference frequency. Each antenna is represented by a triplet (delay_seconds, phase_offset, amplitude_scaling).
+
+ :type: ``float[N_ant][3]``
+
diff --git a/tangostationcontrol/docs/source/devices/tilebeam-digitalbeam.rst b/tangostationcontrol/docs/source/devices/tilebeam-digitalbeam.rst
index c6701097b7976b55ce2d8fae9813bb842199b134..b1ed94a34593669fa352371dc586eb51529112be 100644
--- a/tangostationcontrol/docs/source/devices/tilebeam-digitalbeam.rst
+++ b/tangostationcontrol/docs/source/devices/tilebeam-digitalbeam.rst
@@ -84,26 +84,7 @@ The beam steering is responsible for pointing the beams at a target, by converti
:returns: ``None``
-Celestial and geographical models
-"""""""""""""""""""""""""""""""""
-
-We use `python-casacore <https://casacore.github.io/python-casacore/index.html>`_ to compute the direction of a given pointing with respect to our antennas and reference positions. Casacore in turn uses *measures* tables for the precise measurements of celestial positions, geodetical information, and time calibrations (f.e. leap seconds). These tables need to be installed and periodically updated to maintain the pointing accuracy:
-
-:measures_directory_R: Directory of the active set of measures tables. The directory name includes the timestamp denoting their age.
-
- :type: ``str``
-
-:measures_directories_available_R: List of installed sets of measures tables.
-
- :type: ``str[64]``
-
-:download_measures(): Download (but do not activate) the latest measures tables from ftp://ftp.astron.nl. Returns the directory name in which the measures were installed.
-
- :returns: ``str``
-
-:use_measures(dir): Activate the measures tables in the provided directory. This necessitates turning off and restarting the TileBeam device, so the command will always appear to fail. Turn the device back and the selected measures tables will be active.
-
- :returns: ``(does not return)``
+The direction of each pointing is derived using *casacore*, which must be periodically calibrated, see also :ref:`casacore-measures`.
Timing
"""""""""""""""""""""""""""""""""
diff --git a/tangostationcontrol/docs/source/index.rst b/tangostationcontrol/docs/source/index.rst
index 4c56206dfce3ab37b07f103a963d06846ccd8f12..55da77372087956bb057b6e340f9c40f72be5612 100644
--- a/tangostationcontrol/docs/source/index.rst
+++ b/tangostationcontrol/docs/source/index.rst
@@ -34,6 +34,7 @@ Even without having access to any LOFAR2.0 hardware, you can install the full st
devices/configure
configure_station
signal_chain
+ calibration
broken_hardware
developer
faq
diff --git a/tangostationcontrol/tangostationcontrol/common/cables.py b/tangostationcontrol/tangostationcontrol/common/cables.py
new file mode 100644
index 0000000000000000000000000000000000000000..b48123a26615a1c7d2ca2d97ffdea686968609f2
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/common/cables.py
@@ -0,0 +1,52 @@
+from dataclasses import dataclass
+
+@dataclass(frozen=True)
+class CableType:
+ """ A cable used in LOFAR, with its properties. """
+ name: str
+ length: int
+ delay: float
+ loss: dict
+
+ def speed(self):
+ """ Return the speed of the signal in this cable, in m/s. """
+
+ return self.length / self.delay
+
+ def get_loss(self, antenna_type: str, rcu_band_select: int) -> float:
+ """ Get the appropiate loss value (in dB), for the given
+ antenna type and RCU band selection. """
+
+ if antenna_type == "LBA":
+ if rcu_band_select == 1:
+ return self.loss[50]
+ elif rcu_band_select == 2:
+ return self.loss[50]
+ else:
+ raise ValueError(f"Unsupported RCU band selection for LBA: {rcu_band_select}")
+ elif antenna_type == "HBA":
+ if rcu_band_select == 1:
+ return self.loss[200]
+ elif rcu_band_select == 2:
+ return self.loss[150]
+ elif rcu_band_select == 4:
+ return self.loss[250]
+ else:
+ raise ValueError(f"Unsupported RCU band selection for HBA: {rcu_band_select}")
+
+ raise ValueError(f"Unsupported antenna type: {antenna_type}")
+
+# Global list of all known cable types.
+#
+# NB: The LOFAR1 equivalents of these tables are:
+# - MAC/Deployment/data/StaticMetaData/CableDelays/
+# - MAC/Deployment/data/StaticMetaData/CableAttenuation.conf
+cable_types = {}
+cable_types[ "0m"] = CableType(name= "0m", length= 0, delay=000.0000e-9, loss={50: 0.00, 150: 0.00, 200: 0.00, 250: 0.00})
+cable_types[ "50m"] = CableType(name= "50m", length= 50, delay=199.2573e-9, loss={50: 2.05, 150: 3.64, 200: 4.24, 250: 4.46})
+cable_types[ "80m"] = CableType(name= "80m", length= 80, delay=326.9640e-9, loss={50: 3.32, 150: 5.87, 200: 6.82, 250: 7.19})
+cable_types[ "85m"] = CableType(name= "85m", length= 85, delay=342.5133e-9, loss={50: 3.53, 150: 6.22, 200: 7.21, 250: 7.58})
+cable_types["115m"] = CableType(name="115m", length=115, delay=465.5254e-9, loss={50: 4.74, 150: 8.35, 200: 9.70, 250: 10.18})
+cable_types["120m"] = CableType(name="120m", length=120, delay=493.8617e-9, loss={50: 4.85, 150: 8.55, 200: 9.92, 250: 10.42}) # used on CS030
+cable_types["130m"] = CableType(name="130m", length=130, delay=530.6981e-9, loss={50: 5.40, 150: 9.52, 200: 11.06, 250: 11.61})
+
diff --git a/tangostationcontrol/tangostationcontrol/common/calibration.py b/tangostationcontrol/tangostationcontrol/common/calibration.py
new file mode 100644
index 0000000000000000000000000000000000000000..e08810bab69072db16f62c3cf9cf6a64b2862f98
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/common/calibration.py
@@ -0,0 +1,74 @@
+import numpy
+
+def delay_compensation(delays_seconds: numpy.ndarray, clock: int):
+ """ Return the delay compensation required to line up
+ signals that are delayed by "delays" seconds. The returned values
+ are the delay to apply, in samples (coarse) and remaining seconds
+ (fine), as a tuple (samples, remainder).
+
+ The coarse delay is to be applied in sdp.FPGA_signal_input_delays_RW,
+ the fine delay is to be incorporated into sdp.FPGA_subband_weights_RW.
+
+ Note that the remainder is -1/2 * sample <= remainder <= 1/2 * sample.
+
+ Applying this correction equalises the signal across all inputs to be delayed
+ max(round(delay_samples)) samples, instead of their value in delay_seconds.
+ So we do _not exactly_ delay all signals to match the longest.
+ """
+
+ # NB: signal_* are the amount of delay the signal obtained in our processing
+ # chain, while input_* are the amount of (negative) delay to apply
+ # to compensate.
+
+ # compute the coarse correction, in samples
+ signal_delays_samples = numpy.round(delays_seconds * clock).astype(numpy.uint32)
+
+ # correct for the coarse delay by delaying the other signals to line up
+ # we cannot configure a negative number of samples, so we must delay
+ # all of them sufficiently as well.
+ #
+ # This introduces a constant shift in timing for all samples,
+ # as we shift all of them to obtain a non-negative delay.
+ input_delays_samples = max(signal_delays_samples) - signal_delays_samples
+
+ # compute the remainder, in seconds
+ signal_delays_subsample_seconds = delays_seconds - signal_delays_samples / clock
+ input_delays_subsample_seconds = -signal_delays_subsample_seconds
+
+ return (input_delays_samples, input_delays_subsample_seconds)
+
+def dB_to_factor(dB: numpy.ndarray) -> numpy.ndarray:
+ """ Convert values in decibel (dB) into their equivalent scaling factors. """
+ return 10 ** (dB / 10)
+
+def loss_compensation(losses_dB: numpy.ndarray):
+ """ Return the attenuation required to line up
+ signals that are dampened by "lossed_dB" decibel.
+
+ Returned are the signal attenuations in whole dBs (coarse), and
+ the remaining scaling (as a factor), as a tuple (whole dBs, remainder).
+
+ The coarse attenuation is to be applied in recv.RCU_attenuation_dB_RW,
+ the fine scaling is to be incorporated into sdp.FPGA_subband_weights_RW.
+
+ Applying this correction equalises the signal across the inputs
+ to be dampened max(round(losses_dB)) instead of their value
+ in losses_dB. So we do _not_ fully dampen towards the weakest signal.
+ """
+
+ # NB: signal_* are the amount of loss the signal obtained in our processing
+ # chain, while input_* are the amount of (dampening) attenuation to apply
+ # to compensate.
+
+ # compute the coarse correction, in samples
+ signal_attenuation_integer_dB = numpy.round(losses_dB).astype(numpy.uint32)
+
+ # correct for the coarse loss by dampening the signals to line up.
+ input_attenuation_integer_dB = max(signal_attenuation_integer_dB) - signal_attenuation_integer_dB
+
+ # compute the remainder, as a scaling factor
+ signal_loss_remainder_dB = losses_dB - signal_attenuation_integer_dB
+ input_attenuation_remainder_dB = -signal_loss_remainder_dB
+ input_attenuation_remainder_factor = dB_to_factor(input_attenuation_remainder_dB)
+
+ return (input_attenuation_integer_dB, input_attenuation_remainder_factor)
diff --git a/tangostationcontrol/tangostationcontrol/devices/antennafield.py b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
index b57d3b9622e74967a7b1adf3f3f3bf2ddfabd12a..a805f0dffd94a45cf8ec8f49f2ca60bb8727f077 100644
--- a/tangostationcontrol/tangostationcontrol/devices/antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
@@ -19,9 +19,12 @@ 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 N_elements, MAX_ANTENNA, N_pol, N_xyz, N_latlong, N_rcu, N_rcu_inp
+from tangostationcontrol.common.constants import N_elements, MAX_ANTENNA, N_pol, N_xyz, N_latlong, N_rcu, N_rcu_inp, N_pn, S_pn, N_subbands
+from tangostationcontrol.common.cables import cable_types
+from tangostationcontrol.common.calibration import delay_compensation, loss_compensation
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.devices.device_decorators import fault_on_error
+from tangostationcontrol.devices.sdp.sdp import SDP
from tangostationcontrol.beam.geo import ETRS_to_ITRF, ITRF_to_GEO, GEO_to_GEOHASH
from tangostationcontrol.beam.hba_tile import HBATAntennaOffsets
@@ -132,6 +135,37 @@ class AntennaField(lofar_device):
default_value=numpy.array([False] * MAX_ANTENNA)
)
+ Antenna_Cables = device_property(
+ doc=f"Which cables connect each antenna to the RCU. Both polarisations are assumed to be connected using the same type of cable. Needs to be any of ({', '.join(cable_types.keys())}).",
+ dtype='DevVarStringArray',
+ mandatory=False,
+ default_value = numpy.array(["0m"] * MAX_ANTENNA)
+ )
+
+ Calibration_SDP_Subband_Weights_50MHz = device_property(
+ doc=f"Measured calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna, at 50 MHz. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet.",
+ dtype='DevVarFloatArray',
+ mandatory=False
+ )
+
+ Calibration_SDP_Subband_Weights_150MHz = device_property(
+ doc=f"Measured calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna, at 150 MHz. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet.",
+ dtype='DevVarFloatArray',
+ mandatory=False
+ )
+
+ Calibration_SDP_Subband_Weights_200MHz = device_property(
+ doc=f"Measured calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna, at 200 MHz. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet.",
+ dtype='DevVarFloatArray',
+ mandatory=False
+ )
+
+ Calibration_SDP_Subband_Weights_250MHz = device_property(
+ doc=f"Measured calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna, at 250 MHz. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet.",
+ dtype='DevVarFloatArray',
+ mandatory=False
+ )
+
# ----- Position information
Antenna_Field_Reference_ITRF = device_property(
@@ -233,11 +267,42 @@ class AntennaField(lofar_device):
default_value=[]
)
+ # ----- Generic information
+
Antenna_Type_R = attribute(doc='The type of antenna in this field (LBA or HBA).',
dtype=str)
-
Antenna_Names_R = attribute(access=AttrWriteType.READ,
dtype=(str,), max_dim_x=MAX_ANTENNA)
+ Antenna_to_SDP_Mapping_R = attribute(doc='To which (fpga, input) pair each antenna is connected. -1=unconnected.',
+ dtype=((numpy.int32,),), max_dim_x=N_pol, max_dim_y=MAX_ANTENNA)
+
+ # ----- Cable information (between antenna and RCU)
+
+ Antenna_Cables_R = attribute(doc=f"Which cables connect each antenna to the RCU. Both polarisations are assumed to be connected using the same type of cable. Needs to be any of ({', '.join(cable_types.keys())}).",
+ dtype=(str,), max_dim_x=MAX_ANTENNA)
+ Antenna_Cables_Delay_R = attribute(doc=f"Delay caused by the cable between antenna and RCU, in seconds.",
+ dtype=(numpy.float64,), max_dim_x=MAX_ANTENNA, unit="s")
+ Antenna_Cables_Loss_R = attribute(doc=f"Loss caused by the cable between antenna and RCU, in dB.",
+ dtype=(numpy.float64,), max_dim_x=MAX_ANTENNA, unit="dB")
+
+ # ----- Calibration information
+
+ Calibration_SDP_Signal_Input_Samples_Delay_R = attribute(doc=f"Number of samples that each antenna signal should be delayed to line up. To be applied on sdp.FPGA_signal_input_samples_delay_RW.",
+ dtype=(numpy.uint32,), max_dim_x=MAX_ANTENNA, unit="samples")
+
+ Calibration_RCU_Attenuation_dB_R = attribute(doc=f"Amount of dB with which each antenna signal must be adjusted to line up. To be applied on recv.RCU_attenuator_dB_RW.",
+ dtype=(numpy.uint32,), max_dim_x=MAX_ANTENNA, unit="dB")
+
+ Calibration_SDP_Subband_Weights_Default_R = attribute(
+ doc=f"Computed calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet.",
+ dtype=((numpy.float64,),), max_dim_y=MAX_ANTENNA, max_dim_x=3)
+
+ Calibration_SDP_Subband_Weights_R = attribute(
+ doc=f"Calibration values for the sdp.FPGA_subband_weights_RW columns of each antenna. Each antenna is represented by a (delay, phase_offset, amplitude_scaling) triplet. Returns the measured values from Calibration_SDP_Subband_Weights_XXXMHz if available, and values from Calibration_SDP_Subband_Weights_Default_R otherwise.",
+ dtype=((numpy.float64,),), max_dim_y=MAX_ANTENNA, max_dim_x=3)
+
+ # ----- Quality and usage information
+
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_ANTENNA)
Antenna_Use_R = attribute(
@@ -252,8 +317,7 @@ class AntennaField(lofar_device):
dtype=(bool,), max_dim_x=MAX_ANTENNA)
- Antenna_to_SDP_Mapping_R = attribute(doc='To which (fpga, input) pair each antenna is connected. -1=unconnected.',
- dtype=((numpy.int32,),), max_dim_x=N_pol, max_dim_y=MAX_ANTENNA)
+ # ----- Attributes mapped on RECV
ANT_mask_RW = mapped_attribute("ANT_mask_RW", dtype=(bool,), max_dim_x=MAX_ANTENNA,
access=AttrWriteType.READ_WRITE)
@@ -280,6 +344,8 @@ class AntennaField(lofar_device):
max_dim_y=MAX_ANTENNA, access=AttrWriteType.READ_WRITE)
RCU_band_select_RW = mapped_attribute("RCU_band_select_RW", dtype=(numpy.int64,), max_dim_x=MAX_ANTENNA,
access=AttrWriteType.READ_WRITE)
+ RCU_attenuator_dB_RW = mapped_attribute("RCU_attenuator_dB_RW", dtype=(numpy.int64,), max_dim_x=MAX_ANTENNA,
+ access=AttrWriteType.READ_WRITE)
# ----- Position information
@@ -322,6 +388,88 @@ class AntennaField(lofar_device):
def read_Antenna_Names_R(self):
return self.Antenna_Names
+ def read_Antenna_Cables_R(self):
+ return self.Antenna_Cables
+
+ def read_Antenna_Cables_Delay_R(self):
+ return numpy.array([cable_types[antenna].delay for antenna in self.Antenna_Cables])
+
+ def read_Antenna_Cables_Loss_R(self):
+ rcu_bands = self.read_attribute("RCU_band_select_RW")
+ return numpy.array([cable_types[antenna].get_loss(self.Antenna_Type, rcu_band) for antenna, rcu_band in zip(self.Antenna_Cables, rcu_bands)])
+
+ def read_Calibration_SDP_Signal_Input_Samples_Delay_R(self):
+ # Correct for signal delays in the cables
+ signal_delay_seconds = self.read_attribute("Antenna_Cables_Delay_R")
+
+ # compute the required compensation
+ clock = self.sdp_proxy.clock_RW
+ input_delay_samples, _ = delay_compensation(signal_delay_seconds, clock)
+
+ # return the delay to apply (in samples)
+ return input_delay_samples
+
+ def read_Calibration_SDP_Subband_Weights_Default_R(self):
+ # ----- Delay
+
+ # correct for signal delays in the cables
+ signal_delay_seconds = self.read_attribute("Antenna_Cables_Delay_R")
+
+ # compute the required compensation
+ clock = self.sdp_proxy.clock_RW
+ _, input_delay_subsample_seconds = delay_compensation(signal_delay_seconds, clock)
+
+ # ----- Phase offsets
+
+ # we don't have any
+ phase_offsets = numpy.zeros((self.read_attribute("nr_antennas_R"),),dtype=numpy.float64)
+
+ # ----- Amplitude
+
+ # correct for signal loss in the cables
+ signal_delay_loss = self.read_attribute("Antenna_Cables_Loss_R")
+
+ # return fine scaling to apply
+ _, input_attenuation_remaining_factor = loss_compensation(signal_delay_loss)
+
+ # Return as (delay, phase_offset, amplitude) triplet per antenna
+ return numpy.stack((input_delay_subsample_seconds, phase_offsets, input_attenuation_remaining_factor), axis=1)
+
+ def read_Calibration_SDP_Subband_Weights_R(self):
+ # default table to use if no calibration table is available
+ default_table = self.read_attribute("Calibration_SDP_Subband_Weights_Default_R")
+
+ # construct selector for the right calibration table
+ if self.Antenna_Type == "LBA":
+ rcu_band_to_caltable = {
+ 1: self.Calibration_SDP_Subband_Weights_50MHz or default_table,
+ 2: self.Calibration_SDP_Subband_Weights_50MHz or default_table,
+ }
+ else: # HBA
+ rcu_band_to_caltable = {
+ 2: self.Calibration_SDP_Subband_Weights_150MHz or default_table,
+ 1: self.Calibration_SDP_Subband_Weights_200MHz or default_table,
+ 4: self.Calibration_SDP_Subband_Weights_250MHz or default_table,
+ }
+
+ rcu_bands = self.read_attribute("RCU_band_select_RW")
+
+ # construct the subband weights based on the rcu_band of each antenna,
+ # combining the relevant tables.
+ subband_weights = numpy.zeros((self.read_attribute("nr_antennas_R"), 3), dtype=numpy.float64)
+ for antenna_nr, rcu_band in enumerate(rcu_bands):
+ subband_weights[antenna_nr, :] = rcu_band_to_caltable[rcu_band][antenna_nr, :]
+
+ return subband_weights
+
+ def read_Calibration_RCU_Attenuation_dB_R(self):
+ # Correct for signal loss in the cables
+ signal_delay_loss = self.read_attribute("Antenna_Cables_Loss_R")
+
+ # return coarse attenuation to apply
+ input_attenuation_integer_dB, _ = loss_compensation(signal_delay_loss)
+ return input_attenuation_integer_dB
+
def read_Antenna_Use_R(self):
return self.Antenna_Use
@@ -514,6 +662,82 @@ class AntennaField(lofar_device):
self.sdp_proxy.antenna_type_RW = tuple(sdp_antenna_type)
+ @command()
+ def calibrate_recv(self):
+ """ Calibrate RECV for our antennas.
+
+ Run whenever the following changes:
+ sdp.clock_RW
+ antennafield.RCU_band_select_RW
+ """
+
+ # -----------------------------------------------------------
+ # Set signal-input attenuations to compensate for
+ # differences in cable length.
+ # -----------------------------------------------------------
+
+ rcu_attenuator_db = self.read_attribute("Calibration_RCU_Attenuation_dB_R")
+ self.proxy.write_attribute("RCU_attenuator_dB_RW", rcu_attenuator_db)
+
+ @command()
+ def calibrate_sdp(self):
+ """ Calibrate SDP for our antennas.
+
+ Run whenever the following changes:
+ sdp.clock_RW
+ antennafield.RCU_band_select_RW
+ """
+
+ # Mapping [antenna] -> [fpga][input]
+ antenna_to_sdp_mapping = self.read_attribute("Antenna_to_SDP_Mapping_R")
+
+ # -----------------------------------------------------------
+ # Set coarse delay compensation by delaying the samples.
+ # -----------------------------------------------------------
+
+ # The delay to apply, in samples [antenna]
+ input_samples_delay = self.read_attribute("Calibration_SDP_Signal_Input_Samples_Delay_R")
+
+ # read-modify-write on [fpga][(input, polarisation)]
+ fpga_signal_input_samples_delay = self.sdp_proxy.FPGA_signal_input_samples_delay_RW
+ for antenna_nr, (fpga_nr, input_nr) in enumerate(antenna_to_sdp_mapping):
+ if input_nr == -1:
+ # skip unconnected antennas
+ continue
+
+ # set for X polarisation
+ fpga_signal_input_samples_delay[fpga_nr, input_nr * N_pol + 0] = input_samples_delay[antenna_nr]
+ # set for Y polarisation
+ fpga_signal_input_samples_delay[fpga_nr, input_nr * N_pol + 1] = input_samples_delay[antenna_nr]
+ self.sdp_proxy.FPGA_signal_input_samples_delay_RW = fpga_signal_input_samples_delay
+
+ # -----------------------------------------------------------
+ # Compute calibration of subband weights for the remaining
+ # delay and loss corrections.
+ # -----------------------------------------------------------
+
+ # obtain caltable
+ caltable = self.read_attribute("Calibration_SDP_Subband_Weights_R")
+
+ # obtain frequency information
+ clock = self.sdp_proxy.clock_RW
+ nyquist_zone = self.sdp_proxy.nyquist_zone_R # NB: for all inputs, unmapped, not just ours
+
+ # read-modify-write on [fpga][(input, polarisation)]
+ fpga_subband_weights = self.sdp_proxy.FPGA_subband_weights_RW.reshape(N_pn, S_pn, N_subbands)
+ for antenna_nr, (fpga_nr, input_nr) in enumerate(antenna_to_sdp_mapping):
+ if input_nr == -1:
+ # skip unconnected antennas
+ continue
+
+ # compute weights
+ weights = SDP.subband_weights(caltable[antenna_nr][0], caltable[antenna_nr][1], caltable[antenna_nr][2], clock, nyquist_zone[fpga_nr, input_nr])
+
+ # set weights
+ fpga_subband_weights[fpga_nr, input_nr, :] = weights
+ self.sdp_proxy.FPGA_subband_weights_RW = fpga_subband_weights.reshape(N_pn, S_pn * N_subbands)
+
+
@command(dtype_in=DevVarFloatArray, dtype_out=DevVarLongArray)
def calculate_HBAT_bf_delay_steps(self, delays: numpy.ndarray):
num_tiles = self.read_nr_antennas_R()
@@ -572,7 +796,8 @@ class AntennaToRecvMapper(object):
"HBAT_PWR_on_RW": value_map_ant_32_bool,
"RCU_PWR_ANT_on_R": value_map_ant_bool,
"RCU_PWR_ANT_on_RW": value_map_ant_bool,
- "RCU_band_select_RW": numpy.zeros(number_of_antennas, dtype=numpy.int64)
+ "RCU_band_select_RW": numpy.zeros(number_of_antennas, dtype=numpy.int64),
+ "RCU_attenuator_dB_RW": numpy.zeros(number_of_antennas, dtype=numpy.int64),
}
self._masked_value_mapping_write = {
"ANT_mask_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96,
@@ -582,18 +807,21 @@ class AntennaToRecvMapper(object):
"HBAT_PWR_on_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96_32,
"RCU_PWR_ANT_on_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96,
"RCU_band_select_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96,
+ "RCU_attenuator_dB_RW": AntennaToRecvMapper._VALUE_MAP_NONE_96,
}
self._reshape_attributes_in = {
"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,),
+ "RCU_attenuator_dB_RW": (MAX_ANTENNA,),
}
self._reshape_attributes_out = {
"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),
+ "RCU_attenuator_dB_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/beam_device.py b/tangostationcontrol/tangostationcontrol/devices/beam_device.py
index dc2ee888d21726ea3fdefc99b471e0d076e8ab67..41e0b7f70cc1432e345f9420a058f5bafe0ab275 100644
--- a/tangostationcontrol/tangostationcontrol/devices/beam_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/beam_device.py
@@ -468,10 +468,10 @@ class BeamTracker():
else:
next_update_in = now + datetime.timedelta(seconds=now.timestamp() % self.interval)
+ # sleep until the next update, or when interrupted (this releases the lock, allowing for notification)
+ # note that we need wait_for as conditions can be triggered multiple times in succession
+ self.update_condition.wait_for(lambda: self.done or self.stale_pointing, self._get_sleep_time())
+
# update pointing at requested time
self.stale_pointing = False
self.update_pointing_callback(next_update_in)
-
- # sleep until the next update, or when interrupted (this releases the lock, allowing for notification)
- # note that we need wait_for as conditions can be triggered multiple times in succession
- self.update_condition.wait_for(lambda: self.done or self.stale_pointing, self._get_sleep_time())
diff --git a/tangostationcontrol/tangostationcontrol/devices/recv.py b/tangostationcontrol/tangostationcontrol/devices/recv.py
index 3c1b81ae137fdf93646b11b8b0fc061c475503b2..e53fcbd4035b433019aacc45197f94bb1ef1f293 100644
--- a/tangostationcontrol/tangostationcontrol/devices/recv.py
+++ b/tangostationcontrol/tangostationcontrol/devices/recv.py
@@ -75,7 +75,7 @@ class RECV(opcua_device):
RCU_band_select_RW_default = device_property(
dtype='DevVarLong64Array',
mandatory=False,
- default_value=[0] * N_rcu * N_rcu_inp
+ default_value=[1] * N_rcu * N_rcu_inp
)
RCU_PWR_ANT_on_RW_default = device_property(
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
index 5fd05975cc88c0828e3166ef65fd589174a15896..68cb16d1aea06aa45e427acedbdb1d57694cf798 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/beamlet.py
@@ -14,9 +14,10 @@ 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 N_pn, A_pn, N_pol, N_beamlets_ctrl, N_beamsets_ctrl, P_sum, N_subband_res, N_subbands, DEFAULT_SUBBAND
+from tangostationcontrol.common.constants import N_pn, A_pn, N_pol, N_beamlets_ctrl, N_beamsets_ctrl, P_sum, DEFAULT_SUBBAND
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
+from tangostationcontrol.devices.sdp.common import phases_to_weights, subband_frequencies
import numpy
from functools import lru_cache
@@ -282,43 +283,6 @@ class Beamlet(opcua_device):
BF_UNIT_WEIGHT = 2**14
- @staticmethod
- def _phases_to_bf_weights(phases: numpy.ndarray):
- """ Convert differences in phase (in radians) into FPGA weights (complex numbers packed into uint32). """
-
- # flatten array and restore its shape later, which makes running over all elements a lot easier
- orig_shape = phases.shape
- phases = phases.flatten()
-
- # Convert array values in complex numbers
- real = Beamlet.BF_UNIT_WEIGHT * numpy.cos(phases)
- imag = Beamlet.BF_UNIT_WEIGHT * numpy.sin(phases)
-
- # Interleave into (real, imag) pairs, and store as int16
- # see also https://stackoverflow.com/questions/5347065/interweaving-two-numpy-arrays/5347492
- # Round to nearest integer instead of rounding down
- real_imag = numpy.empty(phases.size * 2, dtype=numpy.int16)
- real_imag[0::2] = numpy.round(real)
- real_imag[1::2] = numpy.round(imag)
-
- # Cast each (real, imag) pair into an uint32, which brings the array size
- # back to the original.
- bf_weights = real_imag.view(numpy.uint32)
-
- return bf_weights.reshape(orig_shape)
-
- @staticmethod
- 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 / N_subband_res
- base_subbands = nyquist_zones * N_subbands
-
- # broadcast clock across frequencies
- frequencies = (subbands + base_subbands) * subband_width
-
- return frequencies
-
@lru_cache() # this function requires large hardware reads for values that don't change often
def _beamlet_frequencies(self):
""" Obtain the frequencies (in Hz) of each subband that is selected for each input and beamlet.
@@ -334,7 +298,7 @@ class Beamlet(opcua_device):
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)
+ return subband_frequencies(beamlet_subbands, self.sdp_proxy.clock_RW, nyquist_zones)
@staticmethod
def _calculate_bf_weights(delays: numpy.ndarray, beamlet_frequencies: numpy.ndarray):
@@ -345,11 +309,11 @@ class Beamlet(opcua_device):
# compute the phases
- # applying a delay means rotating *backwards*
+ # correcting for a delay means rotating *backwards*
beamlet_phases = (-2.0 * numpy.pi) * beamlet_frequencies * delays
# convert to weights
- bf_weights = Beamlet._phases_to_bf_weights(beamlet_phases)
+ bf_weights = phases_to_weights(beamlet_phases, Beamlet.BF_UNIT_WEIGHT)
return bf_weights
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/common.py b/tangostationcontrol/tangostationcontrol/devices/sdp/common.py
new file mode 100644
index 0000000000000000000000000000000000000000..891dbb648b1417f5f8504e878658fbe6773bf66b
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/common.py
@@ -0,0 +1,65 @@
+from tangostationcontrol.common.constants import N_subbands, N_subband_res, VALUES_PER_COMPLEX
+
+import numpy
+from ctypes import c_short
+
+def subband_frequencies(subbands: numpy.ndarray, clock: int, nyquist_zones: numpy.ndarray) -> numpy.ndarray:
+ """ Obtain the frequencies of multiple subbands, given a clock and a nyquist zone for each subband. """
+
+ subband_width = clock / N_subband_res
+ base_subbands = nyquist_zones * N_subbands
+
+ # broadcast clock across frequencies
+ frequencies = (subbands + base_subbands) * subband_width
+
+ return frequencies
+
+def subband_frequency(subband: int, clock: float, nyquist_zone: int) -> int:
+ """ Obtain the frequencies of a subband, given a clock and a nyquist zone. """
+
+ # just use the interface for multiple subbands to avoid code duplication
+ return subband_frequencies(numpy.array(subband), clock, numpy.array(nyquist_zone)).item()
+
+def phases_to_weights(phases: numpy.ndarray, unit_weight: float, amplitudes: numpy.ndarray = None) -> numpy.ndarray:
+ """ Convert phases (in radians) into FPGA weights (complex numbers packed into uint32). """
+
+ # The FPGA accepts weights as a 16-bit (imag,real) complex pair packed into an uint32.
+
+ # flatten array and restore its shape later, which makes running over all elements a lot easier
+ orig_shape = phases.shape
+ phases = phases.flatten()
+
+ # Convert array values in complex numbers
+ real = unit_weight * numpy.cos(phases)
+ imag = unit_weight * numpy.sin(phases)
+
+ if amplitudes is not None:
+ real *= amplitudes
+ imag *= amplitudes
+
+ # Interleave into (real, imag) pairs, and store as int16
+ # see also https://stackoverflow.com/questions/5347065/interweaving-two-numpy-arrays/5347492
+ # Round to nearest integer instead of rounding down
+ real_imag = numpy.empty(phases.size * VALUES_PER_COMPLEX, dtype=numpy.int16)
+ real_imag[0::2] = numpy.round(real)
+ real_imag[1::2] = numpy.round(imag)
+
+ # Cast each (real, imag) pair into an uint32, which brings the array size
+ # back to the original.
+ weights = real_imag.view(numpy.uint32)
+
+ return weights.reshape(orig_shape)
+
+def weight_to_complex(weight: numpy.uint32, unit: int) -> complex:
+ """Unpack an FPGA weight (uint32) into a complex number.
+
+ unit: the weight value representing a weight of 1.0."""
+
+ # A weight is a (real, imag) pair, stored as int16 packed into an uint32 value
+
+ # obtain the upper 16 bits, but make sure we interpret it as c_short to get the correct sign
+ imag = c_short(weight >> 16).value
+ # isolate the lower 16 bits, and interpret it as c_short to get the correct sign
+ real = c_short(weight & 0xFFFF).value
+
+ return (real + 1j * imag) / unit
diff --git a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
index ded37fe8b296445aa3a7a0a82a44b1ce41a5742a..677b82e0f3fd9e241ab12c32433ad7d45a44dcde 100644
--- a/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
+++ b/tangostationcontrol/tangostationcontrol/devices/sdp/sdp.py
@@ -18,6 +18,7 @@ from tango import AttrWriteType
# Additional import
from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
from tangostationcontrol.devices.opcua_device import opcua_device
+from tangostationcontrol.devices.sdp.common import phases_to_weights, subband_frequency
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
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
@@ -318,6 +319,28 @@ class SDP(opcua_device):
# Commands
# --------
+ # --------
+ # Support functions
+ # --------
+
+ # A weight representing no scaling in FPGA_subband_weights_R(W)
+ SUBBAND_UNIT_WEIGHT = 2**13
+
+ @staticmethod
+ def subband_weights(delay_seconds: float, phase_offset: float, amplitude_scaling: float, clock: int, nyquist_zone: int) -> numpy.ndarray:
+ """ Compute a FPGA_subband_weights_RW row for all subbands for a single input. """
+
+ subband_phases = numpy.zeros((N_subbands,), dtype=numpy.float64)
+
+ for subband_nr in range(N_subbands):
+ frequency = subband_frequency(subband_nr, clock, nyquist_zone)
+
+ # correct for the delay by rotating the signal *back*
+ subband_phases[subband_nr] = (-2.0 * numpy.pi) * frequency * delay_seconds + phase_offset
+
+ # convert phases to their complex equivalent, as FPGA weights (cint16 packed into uint32)
+ return phases_to_weights(subband_phases, SDP.SUBBAND_UNIT_WEIGHT, numpy.array([amplitude_scaling] * N_subbands))
+
# ----------
# Run server
# ----------
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/base.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/base.py
index a38fa52aa1814501d2eaa429e258cf6d960c05d8..7a61a457e941da98c3695e91573a5a4424102db3 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/base.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/base.py
@@ -33,10 +33,20 @@ class AbstractTestBases:
# make sure the device starts in Off
self.proxy.Off()
+ # make a backup of the properties, in case they're changed
+ # NB: "or {}" is needed to deal with devices that have no properties.
+ self.original_properties = self.proxy.get_property(self.proxy.get_property_list("*") or {}) or {}
+
self.addCleanup(TestDeviceProxy.test_device_turn_off, self.name)
+ self.addCleanup(self.restore_properties)
super().setUp()
+ def restore_properties(self):
+ """Restore the properties as they were before the test."""
+
+ self.proxy.put_property(self.original_properties)
+
def test_device_fetch_state(self):
"""Test if we can successfully fetch state"""
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
index 3835ab5c16d99acd90e4128597f4bd63ac4d6c9d..14002bca5b81ccfec845bf3291b986c6c8fd34da 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_antennafield.py
@@ -12,8 +12,10 @@ import numpy
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
from tangostationcontrol.devices.antennafield import AntennaQuality, AntennaUse
+from tangostationcontrol.devices.sdp.common import weight_to_complex
+from tangostationcontrol.devices.sdp.sdp import SDP
from .base import AbstractTestBases
-from tangostationcontrol.common.constants import N_elements, MAX_ANTENNA, N_pol, N_rcu, N_rcu_inp, DEFAULT_N_HBA_TILES
+from tangostationcontrol.common.constants import N_elements, MAX_ANTENNA, N_pol, N_rcu, N_rcu_inp, DEFAULT_N_HBA_TILES, CLK_200_MHZ, N_pn, S_pn, N_subbands
class TestAntennaFieldDevice(AbstractTestBases.TestDeviceBase):
@@ -26,10 +28,14 @@ class TestAntennaFieldDevice(AbstractTestBases.TestDeviceBase):
self.recv_proxy = self.setup_recv_proxy()
self.sdp_proxy = self.setup_sdp_proxy()
- self.addCleanup(self.restore_antennafield)
self.addCleanup(self.shutdown_recv)
self.addCleanup(self.shutdown_sdp)
+ # configure the frequencies, which allows access
+ # to the calibration attributes and commands
+ self.sdp_proxy.clock_RW = CLK_200_MHZ
+ self.recv_proxy.RCU_band_select_RW = [[1] * N_rcu_inp] * N_rcu
+
def restore_antennafield(self):
self.proxy.put_property({
"RECV_devices": ["STAT/RECV/1"],
@@ -287,3 +293,71 @@ class TestAntennaFieldDevice(AbstractTestBases.TestDeviceBase):
# Verify device did not enter FAULT state
self.assertEqual(DevState.ON, antennafield_proxy.state())
+
+ def test_calibrate_recv(self):
+ calibration_properties = {
+ "Antenna_Type": ["LBA"],
+ "Antenna_Cables": ["50m","80m"] * (DEFAULT_N_HBA_TILES // 2),
+ }
+
+ antennafield_proxy = self.proxy
+ antennafield_proxy.off()
+ antennafield_proxy.put_property(calibration_properties)
+ antennafield_proxy.boot()
+
+ # calibrate
+ antennafield_proxy.calibrate_recv()
+
+ # check the results
+ rcu_attenuator_db = antennafield_proxy.RCU_attenuator_dB_RW
+
+ # values should be the same for the same cable length
+ self.assertEqual(1, len(set(rcu_attenuator_db[0::2])), msg=f"rcu_attenuator_db={rcu_attenuator_db}")
+ self.assertEqual(1, len(set(rcu_attenuator_db[1::2])), msg=f"rcu_attenuator_db={rcu_attenuator_db}")
+ # value should be larger for the shorter cable, as those signals need damping
+ self.assertGreater(rcu_attenuator_db[0], rcu_attenuator_db[1])
+ # longest cable should require no damping
+ self.assertEqual(0, rcu_attenuator_db[1])
+
+ def test_calibrate_sdp(self):
+ calibration_properties = {
+ "Antenna_Type": ["LBA"],
+ "Antenna_Cables": ["50m","80m"] * (DEFAULT_N_HBA_TILES // 2),
+ "Antenna_to_SDP_Mapping": [0, 1, 0, 0] + [-1, -1] * (DEFAULT_N_HBA_TILES - 2),
+ }
+
+ antennafield_proxy = self.proxy
+ antennafield_proxy.off()
+ antennafield_proxy.put_property(calibration_properties)
+ antennafield_proxy.boot()
+
+ # calibrate
+ antennafield_proxy.calibrate_sdp()
+
+ # check the results
+ # antenna #0 is on FPGA 0, input 2 and 3,
+ # antenna #1 is on FPGA 0, input 0 and 1
+ signal_input_samples_delay = self.sdp_proxy.FPGA_signal_input_samples_delay_RW
+
+ # delays should be equal for both polarisations
+ self.assertEqual(signal_input_samples_delay[0,0], signal_input_samples_delay[0,1])
+ self.assertEqual(signal_input_samples_delay[0,2], signal_input_samples_delay[0,3])
+
+ # antenna #0 is shorter, so should have a greater delay
+ self.assertGreater(signal_input_samples_delay[0,2], signal_input_samples_delay[0,0], msg=f"{signal_input_samples_delay}")
+ # antenna #1 is longest, so should have delay 0
+ self.assertEqual(0, signal_input_samples_delay[0,0])
+
+ # the subband weights depend on the frequency of the subband,
+ # and on the exact delay and loss differences between the cables.
+ # rather than repeating the computations from the code,
+ # we implement this as a regression test.
+ subband_weights = self.sdp_proxy.FPGA_subband_weights_RW.reshape(N_pn, S_pn, N_subbands)
+
+ def to_complex(weight):
+ return weight_to_complex(weight, SDP.SUBBAND_UNIT_WEIGHT)
+
+ self.assertAlmostEqual(0.929 + 0j, to_complex(subband_weights[0, 0, 0]), places=3)
+ self.assertAlmostEqual(0.309 + 0.876j, to_complex(subband_weights[0, 0, 511]), places=3)
+ self.assertAlmostEqual(0.989 + 0j, to_complex(subband_weights[0, 1, 0]), places=3)
+ self.assertAlmostEqual(0.883 - 0.444j, to_complex(subband_weights[0, 1, 511]), places=3)
diff --git a/tangostationcontrol/tangostationcontrol/test/common/test_cables.py b/tangostationcontrol/tangostationcontrol/test/common/test_cables.py
new file mode 100644
index 0000000000000000000000000000000000000000..04fd3653a10229e8ab2cf89810516a535a23dab8
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/common/test_cables.py
@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tangostationcontrol.common import cables
+
+from tangostationcontrol.test import base
+
+class TestCables(base.TestCase):
+ def test_cable_names(self):
+ """ Test whether cable names match their key in the cable_types dictionary. """
+
+ for name, cable in cables.cable_types.items():
+ self.assertEqual(name, cable.name)
+
+ def test_cable_speeds(self):
+ """ Test whether cables are transporting signals at 80% - 100% the speed of light,
+ which is a property of all our cables. """
+
+ speed_of_light = 299_792_458
+
+ for cable in cables.cable_types.values():
+ if cable.length > 0:
+ self.assertLess(80, cable.speed() / speed_of_light * 100, msg=f"Cable {cable.name}")
+ self.assertGreater(100, cable.speed() / speed_of_light * 100, msg=f"Cable {cable.name}")
+
+ def test_cable_loss_increases_with_frequency(self):
+ """ Test whether cable losses increase with frequency for each cable. """
+
+ for cable in cables.cable_types.values():
+ if cable.length == 0:
+ self.assertEqual(0.0, cable.loss[50])
+ self.assertEqual(0.0, cable.loss[150])
+ self.assertEqual(0.0, cable.loss[200])
+ self.assertEqual(0.0, cable.loss[250])
+ else:
+ self.assertLess(cable.loss[50], cable.loss[150], msg=f"Cable {cable.name}")
+ self.assertLess(cable.loss[150], cable.loss[200], msg=f"Cable {cable.name}")
+ self.assertLess(cable.loss[200], cable.loss[250], msg=f"Cable {cable.name}")
+
diff --git a/tangostationcontrol/tangostationcontrol/test/common/test_calibration.py b/tangostationcontrol/tangostationcontrol/test/common/test_calibration.py
new file mode 100644
index 0000000000000000000000000000000000000000..632a388f3c1daf1d6820e941e64f6d1788eb4c5a
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/common/test_calibration.py
@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tangostationcontrol.common.calibration import delay_compensation, loss_compensation, dB_to_factor
+
+from tangostationcontrol.test import base
+
+import numpy
+
+class TestCalibration(base.TestCase):
+ def test_dB_to_factor(self):
+ # Throw some known values at it
+ self.assertAlmostEqual(1.0, dB_to_factor(0.0), places=7)
+ self.assertAlmostEqual(2.0, dB_to_factor(3.0), places=2)
+ self.assertAlmostEqual(10.0, dB_to_factor(10.0), places=7)
+
+class TestLossCompensation(base.TestCase):
+ def test_integer_losses_no_remainder(self):
+ losses = [1.0, 2.0, 3.0, 4.0]
+
+ attenuation_integer_dB, remainder_factor = loss_compensation(numpy.array(losses))
+
+ # verify that there is no remainder
+ self.assertTrue(numpy.all(remainder_factor == 1.0), msg=f"attenuation_integer_dB = {attenuation_integer_dB}, remainder_factor = {remainder_factor}")
+
+ def test_loss_compensation_lines_up(self):
+ """ Test whether signals line up after the computed delay compensation. """
+
+ losses = [1.0, 2.0, 3.0, 4.0]
+
+ attenuation_integer_dB, _ = loss_compensation(numpy.array(losses))
+
+ # sample_shift and delay_samples together should line everything up
+ effective_attenuation = losses + attenuation_integer_dB
+
+ # all values must be lined up equally
+ self.assertEqual(1, len(set(effective_attenuation)), msg=f"effective_attenuation = {effective_attenuation}, attenuation_integer_dB = {attenuation_integer_dB}, losses = {losses}")
+
+ def test_loss_compensation_remainder(self):
+ """ Test correctness of the loss compensation remainders. """
+
+ # losses in dB we want to compensate for. they all round to the same integer value
+ losses = [0.75, 1.0, 1.25]
+
+ attenuation_integer_dB, remainder_factor = loss_compensation(numpy.array(losses))
+
+ # should not result in any sample shifts
+ self.assertEqual(0, attenuation_integer_dB[0])
+ self.assertEqual(0, attenuation_integer_dB[1])
+ self.assertEqual(0, attenuation_integer_dB[2])
+
+ # remainder should correspond with differences.
+ # NB: these are the factors to apply to line up the signals.
+ self.assertAlmostEqual(dB_to_factor(+0.25), remainder_factor[0])
+ self.assertAlmostEqual(dB_to_factor( 0.0), remainder_factor[1])
+ self.assertAlmostEqual(dB_to_factor(-0.25), remainder_factor[2])
+
+
+class TestDelayCompensation(base.TestCase):
+ def _compute_delay_compensation(self, delays_samples: list):
+ # convert to seconds (200 MHz clock => 5 ns samples)
+ clock = 200_000_000
+ delays_seconds = numpy.array(delays_samples) / clock
+
+ # compute delay compensation
+ return delay_compensation(delays_seconds, clock)
+
+ def test_whole_sample_shifts_no_remainder(self):
+ """ Test whether delay compensation indeed has no remainder if we shift whole samples. """
+
+ # delay to compensate for, in samples
+ delay_samples = [1, 2, 3, 4]
+
+ _, remainder_seconds = self._compute_delay_compensation(delay_samples)
+
+ # verify that there is no remainder
+ self.assertTrue(numpy.all(remainder_seconds == 0.0), msg=f"{remainder_seconds}")
+
+ def test_sample_shifts_line_up(self):
+ """ Test whether signals line up after the computed delay compensation. """
+
+ # delay to compensate for, in samples
+ delay_samples = [1, 2, 3, 4]
+
+ sample_shift, _ = self._compute_delay_compensation(delay_samples)
+
+ # sample_shift and delay_samples together should line everything up
+ effective_signal_delay = delay_samples + sample_shift
+
+ # all values must be lined up equally
+ self.assertEqual(1, len(set(effective_signal_delay)), msg=f"effective_signal_delay = {effective_signal_delay}, sample_shift = {sample_shift}, delay_samples = {delay_samples}")
+
+ def test_delay_compensation_remainder(self):
+ """ Test correctness of the delay compensation remainders. """
+
+ # delays in samples we want to compensate for. they all round to the same sample
+ delay_samples = [0.75, 1.0, 1.25]
+
+ sample_shift, remainder_seconds = self._compute_delay_compensation(delay_samples)
+
+ # should not result in any sample shifts
+ self.assertEqual(0, sample_shift[0])
+ self.assertEqual(0, sample_shift[1])
+ self.assertEqual(0, sample_shift[2])
+
+ # remainder should correspond with differences.
+ # NB: these are the remainders to apply to line up the signals.
+ self.assertAlmostEqual(+0.25, remainder_seconds[0] / 5e-9)
+ self.assertAlmostEqual( 0.00, remainder_seconds[1] / 5e-9)
+ self.assertAlmostEqual(-0.25, remainder_seconds[2] / 5e-9)
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_beamlet_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_beamlet_device.py
index 1432b2f715b15ff640d80ba07fda3cd0fc4f8a83..1fa9c8b096a409d642e303e4cff0dca6ca6a54dd 100644
--- a/tangostationcontrol/tangostationcontrol/test/devices/test_beamlet_device.py
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_beamlet_device.py
@@ -7,39 +7,20 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
+from tangostationcontrol.common.constants import CLK_200_MHZ
from tangostationcontrol.devices.sdp.beamlet import Beamlet
-from tangostationcontrol.common.constants import CLK_200_MHZ, CLK_160_MHZ, DEFAULT_SUBBAND
+from tangostationcontrol.devices.sdp.common import weight_to_complex
import numpy
import numpy.testing
-from ctypes import c_short
from tangostationcontrol.test import base
# unpack into 16-bit complex
def to_complex(uint32):
- imag = c_short(uint32 >> 16).value
- real = c_short(uint32 & 0xFFFF).value
- unit = Beamlet.BF_UNIT_WEIGHT
-
- return (real + 1j * imag) / unit
+ return weight_to_complex(uint32, Beamlet.BF_UNIT_WEIGHT)
class TestBeamletDevice(base.TestCase):
- def test_phases_to_bf_weights(self):
- # offer nice 0, 90, 180, 270, 360 degrees
- phases = numpy.array([0.0, numpy.pi / 2, numpy.pi, numpy.pi * 1.5, numpy.pi * 2])
- unit = Beamlet.BF_UNIT_WEIGHT
-
- bf_weights = Beamlet._phases_to_bf_weights(phases)
-
- # check whether the complex representation is also along the right axes and
- # has the right amplitude
- self.assertEqual(to_complex(bf_weights[0]), 1 + 0j, msg=f"bf_weights = {bf_weights}")
- self.assertEqual(to_complex(bf_weights[1]), 0 + 1j, msg=f"bf_weights = {bf_weights}")
- self.assertEqual(to_complex(bf_weights[2]), -1 + 0j, msg=f"bf_weights = {bf_weights}")
- self.assertEqual(to_complex(bf_weights[3]), 0 - 1j, msg=f"bf_weights = {bf_weights}")
- self.assertEqual(to_complex(bf_weights[4]), 1 + 0j, msg=f"bf_weights = {bf_weights}")
-
def test_calculate_bf_weights_small_numbers(self):
# 2 beamlets, 3 antennas. The antennas are 1 second apart.
delays = numpy.array([
@@ -92,42 +73,3 @@ class TestBeamletDevice(base.TestCase):
self.assertEqual(to_complex(bf_weights[0][2]), -1 + 0j, msg=f"bf_weights = {bf_weights}")
self.assertEqual(to_complex(bf_weights[0][3]), 0 + 1j, msg=f"bf_weights = {bf_weights}")
self.assertEqual(to_complex(bf_weights[0][4]), 1 + 0j, msg=f"bf_weights = {bf_weights}")
-
- def test_subband_frequencies(self):
- subbands = numpy.array([
- [0, 1, DEFAULT_SUBBAND],
- ])
-
- nyquist_zones_0 = numpy.zeros(subbands.shape)
- nyquist_zones_1 = numpy.ones(subbands.shape)
- nyquist_zones_2 = numpy.ones(subbands.shape) * 2
-
- # for reference values, see https://proxy.lofar.eu/rtsm/tests/
-
- lba_frequencies = Beamlet._subband_frequencies(subbands, CLK_160_MHZ, nyquist_zones_0)
- self.assertAlmostEqual(lba_frequencies[0][0], 0.0000000e6)
- self.assertAlmostEqual(lba_frequencies[0][1], 0.1562500e6)
- self.assertAlmostEqual(lba_frequencies[0][2], 15.9375000e6)
-
- lba_frequencies = Beamlet._subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_0)
- self.assertAlmostEqual(lba_frequencies[0][0], 0.0000000e6)
- self.assertAlmostEqual(lba_frequencies[0][1], 0.1953125e6)
- self.assertAlmostEqual(lba_frequencies[0][2], 19.9218750e6)
-
- # Nyquist zone 1 is not used in 160 MHz
-
- hba_low_frequencies = Beamlet._subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_1)
- self.assertAlmostEqual(hba_low_frequencies[0][0], 100.0000000e6)
- self.assertAlmostEqual(hba_low_frequencies[0][1], 100.1953125e6)
- self.assertAlmostEqual(hba_low_frequencies[0][2], 119.9218750e6)
-
- hba_high_frequencies = Beamlet._subband_frequencies(subbands, CLK_160_MHZ, nyquist_zones_2)
- self.assertAlmostEqual(hba_high_frequencies[0][0], 160.0000000e6)
- self.assertAlmostEqual(hba_high_frequencies[0][1], 160.1562500e6)
- self.assertAlmostEqual(hba_high_frequencies[0][2], 175.9375000e6)
-
- hba_high_frequencies = Beamlet._subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_2)
- self.assertAlmostEqual(hba_high_frequencies[0][0], 200.0000000e6)
- self.assertAlmostEqual(hba_high_frequencies[0][1], 200.1953125e6)
- self.assertAlmostEqual(hba_high_frequencies[0][2], 219.9218750e6)
-
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_sdp_common.py b/tangostationcontrol/tangostationcontrol/test/devices/test_sdp_common.py
new file mode 100644
index 0000000000000000000000000000000000000000..84655351499ef7b117fedf28e5f119b4c0905523
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_sdp_common.py
@@ -0,0 +1,107 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tangostationcontrol.common.constants import CLK_200_MHZ, CLK_160_MHZ
+from tangostationcontrol.devices.sdp.common import phases_to_weights, subband_frequencies, subband_frequency, weight_to_complex
+
+import numpy
+
+from tangostationcontrol.test import base
+
+class TestSDPCommon(base.TestCase):
+ def test_subband_frequencies(self):
+ subbands = numpy.array([
+ [0, 1, 102],
+ ])
+
+ nyquist_zones_0 = numpy.zeros(subbands.shape)
+ nyquist_zones_1 = numpy.ones(subbands.shape)
+ nyquist_zones_2 = numpy.ones(subbands.shape) * 2
+
+ # for reference values, see https://proxy.lofar.eu/rtsm/tests/
+
+ lba_frequencies = subband_frequencies(subbands, CLK_160_MHZ, nyquist_zones_0)
+ self.assertAlmostEqual(lba_frequencies[0][0], 0.0000000e6)
+ self.assertAlmostEqual(lba_frequencies[0][1], 0.1562500e6)
+ self.assertAlmostEqual(lba_frequencies[0][2], 15.9375000e6)
+
+ lba_frequencies = subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_0)
+ self.assertAlmostEqual(lba_frequencies[0][0], 0.0000000e6)
+ self.assertAlmostEqual(lba_frequencies[0][1], 0.1953125e6)
+ self.assertAlmostEqual(lba_frequencies[0][2], 19.9218750e6)
+
+ # Nyquist zone 1 is not used in 160 MHz
+
+ hba_low_frequencies = subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_1)
+ self.assertAlmostEqual(hba_low_frequencies[0][0], 100.0000000e6)
+ self.assertAlmostEqual(hba_low_frequencies[0][1], 100.1953125e6)
+ self.assertAlmostEqual(hba_low_frequencies[0][2], 119.9218750e6)
+
+ hba_high_frequencies = subband_frequencies(subbands, CLK_160_MHZ, nyquist_zones_2)
+ self.assertAlmostEqual(hba_high_frequencies[0][0], 160.0000000e6)
+ self.assertAlmostEqual(hba_high_frequencies[0][1], 160.1562500e6)
+ self.assertAlmostEqual(hba_high_frequencies[0][2], 175.9375000e6)
+
+ hba_high_frequencies = subband_frequencies(subbands, CLK_200_MHZ, nyquist_zones_2)
+ self.assertAlmostEqual(hba_high_frequencies[0][0], 200.0000000e6)
+ self.assertAlmostEqual(hba_high_frequencies[0][1], 200.1953125e6)
+ self.assertAlmostEqual(hba_high_frequencies[0][2], 219.9218750e6)
+
+ def test_subband_frequency(self):
+ # most coverage is in test_subband_frequencies, we only need to test
+ # whether the right parameter types are accepted.
+ hba_low_frequency = subband_frequency(102, CLK_200_MHZ, 1)
+ self.assertAlmostEqual(hba_low_frequency, 119.9218750e6)
+
+ def test_phases_to_weights(self):
+ # offer nice 0, 90, 180, 270, 360 degrees
+ phases = numpy.array([0.0, numpy.pi / 2, numpy.pi, numpy.pi * 1.5, numpy.pi * 2])
+ unit = 2**14
+
+ sdp_weights = phases_to_weights(phases, unit)
+
+ # check whether the complex representation is also along the right axes and
+ # has the right amplitude
+ self.assertEqual(weight_to_complex(sdp_weights[0], unit), 1 + 0j, msg=f"sdp_weights = {sdp_weights}")
+ self.assertEqual(weight_to_complex(sdp_weights[1], unit), 0 + 1j, msg=f"sdp_weights = {sdp_weights}")
+ self.assertEqual(weight_to_complex(sdp_weights[2], unit), -1 + 0j, msg=f"sdp_weights = {sdp_weights}")
+ self.assertEqual(weight_to_complex(sdp_weights[3], unit), 0 - 1j, msg=f"sdp_weights = {sdp_weights}")
+ self.assertEqual(weight_to_complex(sdp_weights[4], unit), 1 + 0j, msg=f"sdp_weights = {sdp_weights}")
+
+ def test_phases_to_weights_with_amplitude(self):
+ # offer nice 0, 90, 180, 270, 360 degrees
+ phases = numpy.array([0.0, numpy.pi / 2, numpy.pi, numpy.pi * 1.5, numpy.pi * 2])
+ amplitudes = numpy.array([0.1, 0.2, 0.3, 0.4, 0.5])
+ unit = 2**14
+
+ sdp_weights = phases_to_weights(phases, unit, amplitudes)
+
+ # check whether the complex representation is also along the right axes and
+ # has the right amplitude
+ self.assertAlmostEqual(weight_to_complex(sdp_weights[0], unit), 0.1 + 0j, places=3, msg=f"sdp_weights = {sdp_weights}")
+ self.assertAlmostEqual(weight_to_complex(sdp_weights[1], unit), 0 + 0.2j, places=3, msg=f"sdp_weights = {sdp_weights}")
+ self.assertAlmostEqual(weight_to_complex(sdp_weights[2], unit), -0.3 + 0j, places=3, msg=f"sdp_weights = {sdp_weights}")
+ self.assertAlmostEqual(weight_to_complex(sdp_weights[3], unit), 0 - 0.4j, places=3, msg=f"sdp_weights = {sdp_weights}")
+ self.assertAlmostEqual(weight_to_complex(sdp_weights[4], unit), 0.5 + 0j, places=3, msg=f"sdp_weights = {sdp_weights}")
+
+ def test_weight_to_complex(self):
+ unit = 8192
+
+ # some trivial values
+ self.assertEqual(1, weight_to_complex(unit, unit))
+ self.assertEqual(1j, weight_to_complex(unit << 16, unit))
+
+ # some constructed values
+ def complex_to_weight(c: complex) -> numpy.uint32:
+ return numpy.array([c.real * unit, c.imag * unit],dtype=numpy.int16).view(numpy.uint32).item()
+
+ self.assertEqual(-1, weight_to_complex(complex_to_weight(-1), unit))
+ self.assertEqual(-1j, weight_to_complex(complex_to_weight(-1j), unit))
+ self.assertEqual(2 - 3j, weight_to_complex(complex_to_weight(2 - 3j), unit))
+ self.assertEqual(0.5 - 0.25j, weight_to_complex(complex_to_weight(0.5 - 0.25j), unit))