diff --git a/tangostationcontrol/test/common/test_calibration.py b/tangostationcontrol/test/common/test_calibration.py
index db7bb9ee10412a8ee35f7541fe7b58d6e92dfc1f..74ec919c3af773efb86b3c7c38d8722324866321 100644
--- a/tangostationcontrol/test/common/test_calibration.py
+++ b/tangostationcontrol/test/common/test_calibration.py
@@ -6,7 +6,7 @@ from unittest.mock import patch, Mock, call, PropertyMock
from test import base
import numpy
-from numpy.testing import assert_array_equal, assert_equal
+from numpy.testing import assert_array_equal
from tangostationcontrol.common import consul
from tangostationcontrol.common.calibration import (
@@ -17,7 +17,7 @@ from tangostationcontrol.common.calibration import (
CalibrationTable,
)
from tangostationcontrol.common.constants import S_pn, N_subbands, N_pn, SDP_UNIT_WEIGHT
-from tangostationcontrol.common.sdp import complex_to_weights, are_subbands_decreasing
+from tangostationcontrol.common.sdp import complex_to_weights
class MockMinio:
@@ -75,11 +75,8 @@ class TestCalibrationManager(base.TestCase):
]
)
- @patch("tangostationcontrol.common.calibration.read_hdf5")
- def test_calibrate_subband_weights(self, hdf_reader, _):
- """Test whether calibration values are correctly applied to weights"""
-
- antenna_field_mock = Mock(
+ def _setup_mock_antennafield(self):
+ return Mock(
Antenna_to_SDP_Mapping_R=numpy.array(
[[1, 1], [1, 2]], dtype=numpy.int32
).reshape(-1, 2),
@@ -88,6 +85,19 @@ class TestCalibrationManager(base.TestCase):
antenna_type_R="HBA",
**{"name.return_value": "Stat/AFH/HBA0"},
)
+
+ def _setup_mock_caltable(self):
+ return Mock(
+ observation_station="unittest-station",
+ antennas={
+ "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
+ "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
+ },
+ )
+
+ def _test_calibration(self, nyquist_zone, hdf_reader):
+ """Common logic of calibration tests"""
+ antenna_field_mock = self._setup_mock_antennafield()
subband_weights = numpy.array([[SDP_UNIT_WEIGHT] * S_pn * N_subbands] * N_pn)
def subband_weights_side_effect(new_value=None):
@@ -97,23 +107,16 @@ class TestCalibrationManager(base.TestCase):
return subband_weights
sdp_mock = Mock(
- nyquist_zone_RW=numpy.array([[0] * N_pn] * S_pn),
+ nyquist_zone_RW=nyquist_zone,
FPGA_spectral_inversion_R=numpy.array([[0] * N_pn] * S_pn),
)
subband_property_mock = PropertyMock(side_effect=subband_weights_side_effect)
type(sdp_mock).FPGA_subband_weights_RW = subband_property_mock
- caltable_mock = Mock(
- observation_station="unittest-station",
- antennas={
- "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
- "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
- },
- )
+ caltable_mock = self._setup_mock_caltable()
hdf_reader.return_value.__enter__.return_value = caltable_mock
sut = CalibrationManager("http://server:1234", "unittest-station")
sut.calibrate_subband_weights(antenna_field_mock, sdp_mock)
-
hdf_reader.assert_has_calls(
[
call(
@@ -130,6 +133,14 @@ class TestCalibrationManager(base.TestCase):
call().__exit__(None, None, None),
]
)
+ return subband_weights
+
+ @patch("tangostationcontrol.common.calibration.read_hdf5")
+ def test_calibrate_subband_weights(self, hdf_reader, _):
+ """Test whether calibration values are correctly applied to weights"""
+ nyquist_zone = numpy.array([[0] * N_pn] * S_pn)
+ subband_weights = self._test_calibration(nyquist_zone, hdf_reader)
+
assert_array_equal(
subband_weights[1, 1024:1536],
complex_to_weights(numpy.arange(0, 512)),
@@ -151,73 +162,10 @@ class TestCalibrationManager(base.TestCase):
def test_calibrate_reverse_order(self, hdf_reader, _):
"""Test whether calibration values are applied in decreasing order
when subband frequencies are decreasing"""
- antenna_field_mock = Mock(
- Antenna_to_SDP_Mapping_R=numpy.array(
- [[1, 1], [1, 2]], dtype=numpy.int32
- ).reshape(-1, 2),
- Antenna_Names_R=[f"T{n + 1}" for n in range(2)],
- RCU_band_select_RW=numpy.array([[1, 1], [2, 2]]),
- **{"name.return_value": "Stat/AFH/HBA0"},
- )
- subband_weights = numpy.array([[SDP_UNIT_WEIGHT] * S_pn * N_subbands] * N_pn)
-
- def subband_weights_side_effect(new_value=None):
- nonlocal subband_weights
- if new_value is not None:
- subband_weights = new_value
- return subband_weights
-
- sdp_mock = Mock(
- nyquist_zone_RW=numpy.array(
- [[0] * N_pn] + [[1] * N_pn] + [[0] * N_pn] * (S_pn - 2)
- ),
- FPGA_spectral_inversion_R=numpy.array([[0] * N_pn] * S_pn),
- )
- subband_property_mock = PropertyMock(side_effect=subband_weights_side_effect)
- type(sdp_mock).FPGA_subband_weights_RW = subband_property_mock
- caltable_mock = Mock(
- observation_station="unittest-station",
- antennas={
- "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
- "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
- },
- )
- hdf_reader.return_value.__enter__.return_value = caltable_mock
-
- sut = CalibrationManager("http://server:1234", "unittest-station")
- sut.calibrate_subband_weights(antenna_field_mock, sdp_mock)
-
- hdf_reader.assert_has_calls(
- [
- call(
- f"{sut._tmp_dir.name}/CalTable-unittest-station-HBA-200MHz.h5",
- CalibrationTable,
- ),
- call().__enter__(),
- call().__exit__(None, None, None),
- call(
- f"{sut._tmp_dir.name}/CalTable-unittest-station-HBA-150MHz.h5",
- CalibrationTable,
- ),
- call().__enter__(),
- call().__exit__(None, None, None),
- ]
- )
-
- assert_equal(
- are_subbands_decreasing(
- sdp_mock.nyquist_zone_RW[0, 0],
- sdp_mock.FPGA_spectral_inversion_R[0, 0],
- ),
- False,
- )
- assert_equal(
- are_subbands_decreasing(
- sdp_mock.nyquist_zone_RW[1, 0],
- sdp_mock.FPGA_spectral_inversion_R[0, 0],
- ),
- True,
+ nyquist_zone = numpy.array(
+ [[0] * N_pn] + [[1] * N_pn] + [[0] * N_pn] * (S_pn - 2)
)
+ subband_weights = self._test_calibration(nyquist_zone, hdf_reader)
assert_array_equal(
subband_weights[1, 1024:1536],