diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index d9ac77aa5ec97f22b96c152ed3a2a7d37d99c75b..89caf1660c68b93b00f763479d85fc0b51dce948 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -115,7 +115,7 @@ docker_build_image:
- if: $CI_PIPELINE_SOURCE == "merge_request_event"
changes:
- docker-compose/$IMAGE.yml
- - docker-compose/$IMAGE/*
+ - docker-compose/$IMAGE/**/*
- docker-compose/.env
- if: ($CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH) || $CI_COMMIT_TAG
script:
diff --git a/CDB/stations/hba_core.json b/CDB/stations/hba_core.json
index 480d5ac4f28eae7380a7b321a29d407ab2bb826c..051bf2968c5836827c4611d4258879f37052bb7e 100644
--- a/CDB/stations/hba_core.json
+++ b/CDB/stations/hba_core.json
@@ -163,6 +163,12 @@
],
"Antenna_Type": [
"HBA"
+ ],
+ "HBAT_single_element_selection_GENERIC_201512": [
+ "0", "10", "4", "3", "14", "0",
+ "5", "5", "3", "13", "10", "3",
+ "12", "2", "7", "15", "6", "14",
+ "7", "5", "7", "9", "0", "15"
]
}
},
@@ -182,6 +188,12 @@
],
"Antenna_Type": [
"HBA"
+ ],
+ "HBAT_single_element_selection_GENERIC_201512": [
+ "0", "10", "4", "3", "14", "0",
+ "5", "5", "3", "13", "10", "3",
+ "12", "2", "7", "15", "6", "14",
+ "7", "5", "7", "9", "0", "15"
]
}
}
diff --git a/CDB/stations/hba_remote.json b/CDB/stations/hba_remote.json
index 8f47468f858d741c5c638c1bfa76af13ecf2936d..1c31f41f71a0863157e5a7327a8c69253f683060 100644
--- a/CDB/stations/hba_remote.json
+++ b/CDB/stations/hba_remote.json
@@ -100,6 +100,16 @@
],
"Antenna_Type": [
"HBA"
+ ],
+ "HBAT_single_element_selection_GENERIC_201512": [
+ "0", "13", "12", "4", "11", "11",
+ "7", "8", "2", "7", "11", "2",
+ "10", "2", "6", "3", "8", "3",
+ "1", "7", "1", "15", "13", "1",
+ "11", "1", "12", "7", "10", "15",
+ "8", "2", "12", "13", "9", "13",
+ "4", "5", "5", "12", "5", "5",
+ "9", "11", "15", "12", "2", "15"
]
}
}
diff --git a/docker-compose/http-json-schemas/definitions/dithering.json b/docker-compose/http-json-schemas/definitions/dithering.json
new file mode 100644
index 0000000000000000000000000000000000000000..87a6077d24f4bd8878ba0a66ba1572417ce44adb
--- /dev/null
+++ b/docker-compose/http-json-schemas/definitions/dithering.json
@@ -0,0 +1,26 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "description": "Settings for adding dithering to the signal to increase its linearity",
+ "required": [
+ "enabled"
+ ],
+ "properties": {
+ "enabled": {
+ "type": "boolean",
+ "default": false
+ },
+ "power": {
+ "type": "number",
+ "description": "Power of dithering signal, in dBm",
+ "minimum": -25.0,
+ "maximum": -4.0,
+ "default": -4.0
+ },
+ "frequency": {
+ "type": "number",
+ "description": "Frequency of dithering signal, in Hz",
+ "default": 102000000
+ }
+ }
+}
diff --git a/docker-compose/http-json-schemas/definitions/hba.json b/docker-compose/http-json-schemas/definitions/hba.json
new file mode 100644
index 0000000000000000000000000000000000000000..5d1a319fc18c58dc45e720cfbe1d616d7fcb3d88
--- /dev/null
+++ b/docker-compose/http-json-schemas/definitions/hba.json
@@ -0,0 +1,26 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "tile_beam"
+ ],
+ "properties": {
+ "tile_beam": {
+ "$ref": "pointing.json"
+ },
+ "DAB_filter": {
+ "type": "boolean",
+ "default": false,
+ "description": "Enable hardware filter on DAB frequencies"
+ },
+ "element_selection": {
+ "type": "string",
+ "default": "ALL",
+ "description": "Which element(s) to enable in each tile",
+ "enum": [
+ "ALL",
+ "GENERIC_201512"
+ ]
+ }
+ }
+}
diff --git a/docker-compose/http-json-schemas/definitions/observation-settings.json b/docker-compose/http-json-schemas/definitions/observation-settings.json
index 50a6590f3d0fc671121ffa867349585b56d4d4ec..e6bdf42a3787cd05d819df380871b480c96bd3fc 100644
--- a/docker-compose/http-json-schemas/definitions/observation-settings.json
+++ b/docker-compose/http-json-schemas/definitions/observation-settings.json
@@ -52,6 +52,9 @@
"SPARSE_ODD"
]
},
+ "dithering": {
+ "$ref": "dithering.json"
+ },
"filter": {
"type": "string",
"enum": [
@@ -71,13 +74,13 @@
"$ref": "sap.json"
}
},
- "tile_beam": {
- "$ref": "pointing.json"
- },
"first_beamlet": {
"type": "number",
"default": 0,
"minimum": 0
+ },
+ "HBA": {
+ "$ref": "hba.json"
}
}
}
diff --git a/tangostationcontrol/docs/source/observing.rst b/tangostationcontrol/docs/source/observing.rst
index a207accf044f52ff808feac09e2912611d1535b1..6586a25efbaa2b52fa5bd4bf2f35070a09fb4e19 100644
--- a/tangostationcontrol/docs/source/observing.rst
+++ b/tangostationcontrol/docs/source/observing.rst
@@ -15,11 +15,22 @@ To observe with a station, you must construct the observation's specifications,
"antenna_field": "HBA",
"antenna_set": "ALL",
"filter": "HBA_210_250",
+ "dithering": {
+ "enabled": true,
+ "power": -4.0,
+ "frequency": 102000000
+ },
"SAPs": [{
"subbands": [10, 20, 30],
- "pointing": { "angle1": 1.5, "angle2": 0, "direction_type": "J2000" }
+ "pointing": { "angle1": 1.0, "angle2": 0, "direction_type": "J2000" }
+ }, {
+ "subbands": [40, 50, 60],
+ "pointing": { "angle1": 2.0, "angle2": 0, "direction_type": "J2000" }
}],
- "tile_beam": { "angle1": 1.5, "angle2": 0, "direction_type": "J2000" }
+ "HBA": {
+ "DAB_filter": true,
+ "tile_beam": { "angle1": 1.5, "angle2": 0, "direction_type": "J2000" }
+ }
}
import json
@@ -28,32 +39,40 @@ To observe with a station, you must construct the observation's specifications,
The above specification contains the following parameters:
-+--------------------+-----------------------------------------------------------------------------------------+
-| Parameter | Description |
-+====================+=========================================================================================+
-| ``observation_id`` | User-specified unique reference to this observation. |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``start_time`` | automatically start observing when this timestamp is reached. |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``stop_time`` | automatically stop observing when this timestamp is reached. |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``antenna_field`` | Which antenna field to use (LBA, HBA, HBA0, HBA1). |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``antenna_set`` | Which subset of antennas to use (ALL, INNER, OUTER, EVEN, ODD). |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``filter`` | Which band filter to use (LBA_10_90, LBA_30_70, HBA_110_190, HBA_170_230, HBA_210_250). |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``SAPs`` | List of pointings and frequencies (subbands) to track and beam form. |
-+--------------------+-----------------------------------------------------------------------------------------+
-| ``tile_beam`` | Pointing to track with the HBA tiles (optional). |
-+--------------------+-----------------------------------------------------------------------------------------+
++-------------------------+-----------------------------------------------------------------------------------------+
+| Parameter | Description |
++=========================+=========================================================================================+
+| ``observation_id`` | User-specified unique reference to this observation. |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``start_time`` | automatically start observing when this timestamp is reached. (optional) |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``stop_time`` | automatically stop observing when this timestamp is reached. |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``antenna_field`` | Which antenna field to use (LBA, HBA, HBA0, HBA1). |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``antenna_set`` | Which subset of antennas to use (ALL, INNER, OUTER, EVEN, ODD). |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``filter`` | Which band filter to use (LBA_10_90, LBA_30_70, HBA_110_190, HBA_170_230, HBA_210_250). |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``dithering.enabled`` | Whether to add analog dithering noise to increase linearity. (optional) |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``dithering.power`` | Power (in dB) to apply for dithering (-4.0 to -25.0). (optional) |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``dithering.frequency`` | Dithering frequency (in Hz). (optional) |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``SAPs`` | List of pointings and frequencies (subbands) to track and beam form. |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``HBA.DAB_filter`` | Enable the analog filter on the RCUs for DAB radio frequencies. (optional) |
++-------------------------+-----------------------------------------------------------------------------------------+
+| ``HBA.tile_beam`` | Pointing to track with the HBA tiles (optional). (specify for HBA) |
++-------------------------+-----------------------------------------------------------------------------------------+
This will configure the specified antenna field (f.e. ``HBA``) as follows:
* ``STAT/DigitalBeam/HBA`` is configured to beam form the antennas in the specified ``antenna_set``, track all pointings given in ``SAPs[x].pointing``, and produce beamlets for all subbands in ``SAPs[x].subbands``. The beamlets mirror the subbands in the order in which they are specified,
* The ``observation_id`` is used to annotate the beamlet data produced by this observation,
* ``STAT/AntennaField/HBA`` is configured to use the specified ``filter`` for the RCUs,
-* ``STAT/TileBeam/HBA`` is configured to beam form all tiles, tracking the given ``tile_beam`` pointing.
+* ``STAT/TileBeam/HBA`` is configured to beam form all HBA tiles, tracking the given ``tile_beam`` pointing.
Observation Output
````````````````````````
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_observation.py b/tangostationcontrol/integration_test/default/devices/test_device_observation.py
index a21df5d396b4f976b66c3bece1796ee3f9ce5009..6aa3527b937681a7d0dc6718f84ef6bc7b61e195 100644
--- a/tangostationcontrol/integration_test/default/devices/test_device_observation.py
+++ b/tangostationcontrol/integration_test/default/devices/test_device_observation.py
@@ -9,6 +9,8 @@ from tango import DevState, DevFailed
from tangostationcontrol.common.constants import (
N_beamlets_ctrl,
+ N_elements,
+ N_pol,
MAX_ANTENNA,
CS001_TILES,
)
@@ -132,6 +134,9 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
"010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101",
"111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111",
],
+ "HBAT_single_element_selection_GENERIC_201512": [
+ str(n % 16) for n in range(CS001_TILES)
+ ],
}
)
antennafield_proxy.off()
@@ -230,12 +235,18 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
)
] * len(data["SAPs"][0]["subbands"])
tile_beam = [
- str(data["tile_beam"]["direction_type"]),
- f"{data['tile_beam']['angle1']}rad",
- f"{data['tile_beam']['angle2']}rad",
+ str(data["HBA"]["tile_beam"]["direction_type"]),
+ f"{data['HBA']['tile_beam']['angle1']}rad",
+ f"{data['HBA']['tile_beam']['angle2']}rad",
]
first_beamlet = data["first_beamlet"]
+ dithering = data["dithering"]["enabled"]
+ dithering_power = data["dithering"]["power"]
+ dithering_frequency = data["dithering"]["frequency"]
+
+ dab_filter = data["HBA"]["DAB_filter"]
+
self.proxy.off()
self.proxy.observation_settings_RW = self.VALID_JSON
self.proxy.Initialise()
@@ -248,9 +259,14 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
self.assertEqual(filter_, self.proxy.filter_R)
self.assertListEqual(saps_subband, self.proxy.saps_subband_R.tolist())
self.assertListEqual(saps_pointing, list(self.proxy.saps_pointing_R))
- self.assertListEqual(tile_beam, list(self.proxy.tile_beam_R))
+ self.assertEqual(dab_filter, self.proxy.HBA_DAB_filter_R)
+ self.assertListEqual(tile_beam, list(self.proxy.HBA_tile_beam_R))
self.assertEqual(first_beamlet, self.proxy.first_beamlet_R)
+ self.assertEqual(dithering, self.proxy.dithering_enabled_R)
+ self.assertEqual(dithering_power, self.proxy.dithering_power_R)
+ self.assertEqual(dithering_frequency, self.proxy.dithering_frequency_R)
+
def test_apply_antennafield_settings(self):
"""Test that attribute filter is correctly applied"""
self.setup_stationmanager_proxy()
@@ -338,3 +354,35 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
tilebeam_directions,
["J2000", "0.0261799rad", "0rad"],
)
+
+ def test_apply_element_selection(self):
+ # failing
+ """Test that attribute element_selection is correctly applied"""
+ antennafield_proxy = self.setup_antennafield_proxy()
+ antennafield_proxy.HBAT_PWR_on_RW = numpy.ones(
+ (CS001_TILES, N_elements * N_pol), dtype=bool
+ )
+ antennafield_proxy.HBAT_PWR_LNA_on_RW = numpy.zeros(
+ (CS001_TILES, N_elements * N_pol), dtype=bool
+ )
+
+ self.proxy.off()
+ self.proxy.observation_settings_RW = self.VALID_JSON
+ self.proxy.Initialise()
+ self.proxy.On()
+
+ # construct an array with "True" at the expected spots
+ expected = numpy.zeros((CS001_TILES, N_elements, N_pol), dtype=bool)
+ for n in range(CS001_TILES):
+ expected[n, n % 16, :] = True
+
+ # collapse all values for a tile into a single dimension,
+ # like HBAT_PWR_(LNA_)on_RW will return
+ expected = expected.reshape(CS001_TILES, -1)
+
+ self.assertListEqual(
+ antennafield_proxy.HBAT_PWR_on_RW.tolist(), expected.tolist()
+ )
+ self.assertListEqual(
+ antennafield_proxy.HBAT_PWR_LNA_on_RW.tolist(), expected.tolist()
+ )
diff --git a/tangostationcontrol/tangostationcontrol/configuration/__init__.py b/tangostationcontrol/tangostationcontrol/configuration/__init__.py
index de4c16c013f4e80a96ea95465e47a454c7902bec..735a5ee255aa6d660458612587af18bb5f01110a 100644
--- a/tangostationcontrol/tangostationcontrol/configuration/__init__.py
+++ b/tangostationcontrol/tangostationcontrol/configuration/__init__.py
@@ -4,9 +4,13 @@
from .observation_settings import ObservationSettings
from .pointing import Pointing
from .sap import Sap
+from .hba import HBA
+from .dithering import Dithering
__all__ = [
"ObservationSettings",
"Pointing",
"Sap",
+ "HBA",
+ "Dithering",
]
diff --git a/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py b/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
index a922f9e82a409e50df8f9826d455e8de72ab26e7..026be5be9c924a59591fe72717d8e0ec7e37e917 100644
--- a/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
+++ b/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
@@ -7,11 +7,17 @@ from jsonschema.exceptions import ValidationError
def _from_json_hook_t(primary: Type):
- from tangostationcontrol.configuration import Pointing, Sap, ObservationSettings
+ from tangostationcontrol.configuration import (
+ Pointing,
+ Sap,
+ ObservationSettings,
+ HBA,
+ Dithering,
+ )
def actual_hook(json_dct):
primary_ex = None
- for t in [Pointing, Sap, ObservationSettings]:
+ for t in [Pointing, Sap, HBA, Dithering, ObservationSettings]:
try:
t.get_validator().validate(json_dct)
except ValidationError as ex:
diff --git a/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py b/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
index eb5fcc3928e3d7b8ea36f4aee1ce4f4e57b2a23f..3b879ff2af4145204b9796788bdb7fa817c70278 100644
--- a/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
+++ b/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
@@ -57,6 +57,8 @@ class _ConfigurationBase(ABC):
@staticmethod
def _class_to_url(cls_name):
+ cls_name = cls_name.replace("HBA", "hba")
+ cls_name = cls_name.replace("LBA", "lba")
return re.sub(r"(?<!^)(?=[A-Z])", "-", cls_name).lower()
@classmethod
diff --git a/tangostationcontrol/tangostationcontrol/configuration/dithering.py b/tangostationcontrol/tangostationcontrol/configuration/dithering.py
new file mode 100644
index 0000000000000000000000000000000000000000..750a6e463358e34b56151baf001e46ce24638530
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/dithering.py
@@ -0,0 +1,35 @@
+# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
+# SPDX-License-Identifier: Apache-2.0
+
+from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
+
+
+class Dithering(_ConfigurationBase):
+ VALIDATOR = None
+
+ def __init__(
+ self,
+ enabled: bool,
+ power: float | None,
+ frequency: float | None,
+ ):
+ self.enabled = enabled
+ self.power = power
+ self.frequency = frequency
+
+ def __iter__(self):
+ yield from {
+ "enabled": self.enabled,
+ }.items()
+ if self.power is not None:
+ yield "power", self.power
+ if self.frequency is not None:
+ yield "frequency", self.frequency
+
+ @staticmethod
+ def to_object(json_dct) -> "Dithering":
+ return Dithering(
+ json_dct["enabled"],
+ json_dct.get("power"),
+ json_dct.get("frequency"),
+ )
diff --git a/tangostationcontrol/tangostationcontrol/configuration/hba.py b/tangostationcontrol/tangostationcontrol/configuration/hba.py
new file mode 100644
index 0000000000000000000000000000000000000000..ecd65aecf06c074edf124a194fc0ba7d2f5c91eb
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/hba.py
@@ -0,0 +1,35 @@
+# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
+# SPDX-License-Identifier: Apache-2.0
+
+from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
+from tangostationcontrol.configuration.pointing import Pointing
+
+
+class HBA(_ConfigurationBase):
+ def __init__(
+ self,
+ tile_beam: Pointing,
+ DAB_filter: bool | None,
+ element_selection: str | None = "ALL",
+ ):
+ self.tile_beam = tile_beam
+ self.DAB_filter = DAB_filter
+ self.element_selection = element_selection
+
+ def __iter__(self):
+ yield from {
+ "tile_beam": dict(self.tile_beam),
+ }.items()
+
+ if self.DAB_filter is not None:
+ yield "DAB_filter", self.DAB_filter
+ if self.element_selection is not None:
+ yield "element_selection", self.element_selection
+
+ @staticmethod
+ def to_object(json_dct) -> "HBA":
+ return HBA(
+ json_dct["tile_beam"],
+ json_dct.get("DAB_filter"),
+ json_dct.get("element_selection"),
+ )
diff --git a/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
index b8f49369f062899c95c2b95369ab9a506acaf4dd..40d64922187fb1bcec7cb7c30578e0922d1ce1ae 100644
--- a/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
+++ b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
@@ -5,7 +5,8 @@ from datetime import datetime
from typing import Sequence
from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
-from tangostationcontrol.configuration.pointing import Pointing
+from tangostationcontrol.configuration.dithering import Dithering
+from tangostationcontrol.configuration.hba import HBA
from tangostationcontrol.configuration.sap import Sap
@@ -19,9 +20,10 @@ class ObservationSettings(_ConfigurationBase):
antenna_set: str,
filter: str,
SAPs: Sequence[Sap],
- tile_beam: Pointing | None = None,
+ HBA: HBA | None = None,
first_beamlet: int = 0,
lead_time: float | None = None,
+ dithering: Dithering | None = None,
):
self.observation_id = observation_id
self.start_time = start_time
@@ -30,9 +32,10 @@ class ObservationSettings(_ConfigurationBase):
self.antenna_set = antenna_set
self.filter = filter
self.SAPs = SAPs
- self.tile_beam = tile_beam
+ self.HBA = HBA
self.first_beamlet = first_beamlet
self.lead_time = lead_time
+ self.dithering = dithering
def __iter__(self):
yield "observation_id", self.observation_id
@@ -45,11 +48,13 @@ class ObservationSettings(_ConfigurationBase):
"filter": self.filter,
"SAPs": [dict(s) for s in self.SAPs],
}.items()
- if self.tile_beam:
- yield "tile_beam", dict(self.tile_beam)
+ if self.HBA:
+ yield "HBA", dict(self.HBA)
yield "first_beamlet", self.first_beamlet
if self.lead_time is not None:
yield "lead_time", self.lead_time
+ if self.dithering is not None:
+ yield "dithering", dict(self.dithering)
@staticmethod
def to_object(json_dct) -> "ObservationSettings":
@@ -63,7 +68,8 @@ class ObservationSettings(_ConfigurationBase):
json_dct["antenna_set"],
json_dct["filter"],
json_dct["SAPs"],
- json_dct["tile_beam"] if "tile_beam" in json_dct else None,
- json_dct["first_beamlet"] if "first_beamlet" in json_dct else 0,
- json_dct["lead_time"] if "lead_time" in json_dct else None,
+ json_dct.get("HBA"),
+ json_dct.get("first_beamlet", 0),
+ json_dct.get("lead_time"),
+ json_dct.get("dithering"),
)
diff --git a/tangostationcontrol/tangostationcontrol/devices/antennafield.py b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
index d4b8dc33666f22b94b4f02e7f8b42e30a9a2a1ba..f195d15eb4785fc35ee34bf527c7ead63c0c769c 100644
--- a/tangostationcontrol/tangostationcontrol/devices/antennafield.py
+++ b/tangostationcontrol/tangostationcontrol/devices/antennafield.py
@@ -54,6 +54,7 @@ from tangostationcontrol.devices.base_device_classes.mapper import (
MappedAttribute,
AntennaToRecvMapper,
AntennaToSdpMapper,
+ RecvDeviceWalker,
)
logger = logging.getLogger()
@@ -241,6 +242,8 @@ class AntennaField(LOFARDevice):
default_value=2015.5,
)
+ # ----- HBA properties
+
HBAT_PQR_rotation_angles_deg = device_property(
doc='Rotation of each tile in the PQ plane ("horizontal") in degrees.',
dtype="DevVarFloatArray",
@@ -270,6 +273,39 @@ class AntennaField(LOFARDevice):
default_value=HBATAntennaOffsets.HBAT1_BASE_ANTENNA_OFFSETS.flatten(),
)
+ # see also https://github.com/lofar-astron/lofarimaging/blob/9672b52bb9be8f3405e6e3f85701175bdc4bf211/lofarimaging/singlestationutil.py#L43
+ HBAT_single_element_selection_GENERIC_201512 = device_property(
+ doc="Which element to select when using a single dipole per tile in the 'GENERIC_201512' strategy",
+ dtype="DevVarLongArray",
+ mandatory=False,
+ default_value=[
+ 0,
+ 10,
+ 4,
+ 3,
+ 14,
+ 0,
+ 5,
+ 5,
+ 3,
+ 13,
+ 10,
+ 3,
+ 12,
+ 2,
+ 7,
+ 15,
+ 6,
+ 14,
+ 7,
+ 5,
+ 7,
+ 9,
+ 0,
+ 15,
+ ],
+ )
+
# ----- SDP mapping
Antenna_to_SDP_Mapping = device_property(
@@ -370,6 +406,27 @@ class AntennaField(LOFARDevice):
def RECV_Devices_R(self):
return numpy.array(self.RECV_Devices)
+ @attribute(
+ access=AttrWriteType.READ,
+ dtype=((numpy.float64,),),
+ max_dim_x=3,
+ max_dim_y=3,
+ )
+ def PQR_to_ETRS_rotation_matrix_R(self):
+ return numpy.array(
+ self.PQR_to_ETRS_rotation_matrix, dtype=numpy.float64
+ ).reshape(3, 3)
+
+ @attribute(
+ access=AttrWriteType.READ,
+ dtype=(numpy.int32,),
+ max_dim_x=MAX_ANTENNA,
+ )
+ def HBAT_single_element_selection_GENERIC_201512_R(self):
+ return numpy.array(
+ self.HBAT_single_element_selection_GENERIC_201512, dtype=numpy.int32
+ )
+
Frequency_Band_RW = attribute(
doc="The selected frequency band of each polarisation of each antenna.",
dtype=((str,),),
@@ -1088,6 +1145,26 @@ class AntennaField(LOFARDevice):
return result_values.flatten()
+ @command()
+ @DebugIt()
+ @only_in_states(DEFAULT_COMMAND_STATES)
+ def RCU_DTH_on(self):
+ walker = RecvDeviceWalker(
+ self.read_attribute("Control_to_RECV_mapping_R"),
+ self.read_attribute("Antenna_Usage_Mask_R"),
+ )
+ walker.walk_receivers(self.recv_proxies, lambda recv: recv.RCU_DTH_on())
+
+ @command()
+ @DebugIt()
+ @only_in_states(DEFAULT_COMMAND_STATES)
+ def RCU_DTH_off(self):
+ walker = RecvDeviceWalker(
+ self.read_attribute("Control_to_RECV_mapping_R"),
+ self.read_attribute("Antenna_Usage_Mask_R"),
+ )
+ walker.walk_receivers(self.recv_proxies, lambda recv: recv.RCU_DTH_off())
+
# ----------
# Run server
diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
index 064bba36a3b77b9f06961eda7d5ea1267f4ada35..1af7d34c06f5a59725d1a04a483a0e542902e04d 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
@@ -6,10 +6,11 @@
"""
from enum import Enum
from typing import List, Optional, Dict
+from collections.abc import Callable
import numpy
# PyTango imports
-from tango import AttrWriteType
+from tango import AttrWriteType, DeviceProxy
from tango.server import attribute
from tangostationcontrol.common.constants import (
@@ -421,3 +422,54 @@ class AntennaToRecvMapper(AntennaMapper):
else:
value_mapper[attr] = [self._control_mapping]
return value_mapper
+
+
+class RecvDeviceWalker(object):
+ """Walks over child devices with the appropriate mask set to affect the hardware
+ under the parent's control.."""
+
+ def __init__(
+ self, control_to_recv_mapping: numpy.ndarray, antenna_usage_mask: list[bool]
+ ):
+ self.control_to_recv_mapping = control_to_recv_mapping
+ self.antenna_usage_mask = antenna_usage_mask
+
+ def recv_ant_masks(self) -> numpy.ndarray:
+ """Return the antenna mask for the control inputs of the antennas enabled in Antenna_Usage_Mask_R."""
+
+ nr_recv_devices = max(self.control_to_recv_mapping[:, 0])
+
+ recv_ant_masks = numpy.zeros((nr_recv_devices, N_rcu, N_rcu_inp), dtype=bool)
+
+ for use, (recv, rcu_input) in zip(
+ self.antenna_usage_mask, self.control_to_recv_mapping
+ ):
+ if not use:
+ continue
+ if recv <= 0:
+ continue
+
+ recv_ant_masks[recv - 1][rcu_input // N_rcu_inp][
+ rcu_input % N_rcu_inp
+ ] = True
+
+ return recv_ant_masks
+
+ def walk_receivers(
+ self,
+ recv_proxies: list[DeviceProxy],
+ visitor_func: Callable[[DeviceProxy], None],
+ ):
+ """Walk over all RECV devices with the mask set for our antennas."""
+ ant_masks = self.recv_ant_masks()
+
+ for recv_proxy, ant_mask in zip(recv_proxies, ant_masks):
+ # configure RECV to address only our antennas. Save existing mask.
+ old_mask = recv_proxy.ANT_mask_RW
+ recv_proxy.ANT_mask_RW = ant_mask
+
+ try:
+ visitor_func(recv_proxy)
+ finally:
+ # restore mask
+ recv_proxy.ANT_mask_RW = old_mask
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation.py b/tangostationcontrol/tangostationcontrol/devices/observation.py
index d195f5480962ebfe5469f35e81f9a25c617e0a21..9e35080a3028bc119d14c556f28262dcc26d6c95 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation.py
@@ -16,6 +16,7 @@ from tangostationcontrol.common.constants import (
DEFAULT_POLLING_PERIOD,
MAX_ANTENNA,
N_beamlets_ctrl,
+ N_elements,
N_point_prop,
N_pol,
)
@@ -108,6 +109,41 @@ class Observation(LOFARDevice):
def antenna_set_R(self):
return self._observation_settings.antenna_set
+ @attribute(
+ doc="Whether to add dithering to the signal to increase its linearity",
+ dtype=bool,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def dithering_enabled_R(self):
+ try:
+ return self._observation_settings.dithering.enabled or False
+ except AttributeError:
+ return False
+
+ @attribute(
+ doc="Power of dithering signal, in dBm",
+ unit="dBm",
+ dtype=numpy.float64,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def dithering_power_R(self):
+ try:
+ return self._observation_settings.dithering.power or -4.0
+ except AttributeError:
+ return -4.0
+
+ @attribute(
+ doc="Frequency of dithering signal, in Hz",
+ unit="Hz",
+ dtype=numpy.int64,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def dithering_frequency_R(self):
+ try:
+ return self._observation_settings.dithering.frequency or 102_000_000
+ except AttributeError:
+ return 102_000_000
+
@attribute(
doc="Which band filter to use for all antennas",
dtype=str,
@@ -147,16 +183,27 @@ class Observation(LOFARDevice):
)
return saps_pointing
+ @attribute(
+ doc="Beamlet index of the FPGA output, at which to start mapping the beamlets of this observation.",
+ dtype=numpy.uint64,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def first_beamlet_R(self):
+ return self._observation_settings.first_beamlet
+
@attribute(
doc="Which pointing to beamform all HBA tiles to (if any).",
dtype=(str,),
max_dim_x=N_point_prop,
)
- def tile_beam_R(self):
- if self._observation_settings.tile_beam is None:
+ def HBA_tile_beam_R(self):
+ try:
+ if self._observation_settings.HBA.tile_beam is None:
+ return None
+ except AttributeError:
return None
- pointing_direction = self._observation_settings.tile_beam
+ pointing_direction = self._observation_settings.HBA.tile_beam
return [
str(pointing_direction.direction_type),
f"{pointing_direction.angle1}rad",
@@ -164,12 +211,26 @@ class Observation(LOFARDevice):
]
@attribute(
- doc="Beamlet index of the FPGA output, at which to start mapping the beamlets of this observation.",
- dtype=numpy.uint64,
+ doc="Whether to enable the DAB filter",
+ dtype=bool,
fisallowed="is_attribute_access_allowed",
)
- def first_beamlet_R(self):
- return self._observation_settings.first_beamlet
+ def HBA_DAB_filter_R(self):
+ try:
+ return self._observation_settings.HBA.DAB_filter or False
+ except AttributeError:
+ return False
+
+ @attribute(
+ doc="Which element(s) to enable in each tile",
+ dtype=str,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def HBA_element_selection_R(self):
+ try:
+ return self._observation_settings.HBA.element_selection or "ALL"
+ except AttributeError:
+ return "ALL"
observation_settings_RW = attribute(dtype=str, access=AttrWriteType.READ_WRITE)
@@ -265,7 +326,7 @@ class Observation(LOFARDevice):
f"{util.get_ds_inst_name()}/DigitalBeam/{antennafield}"
)
- if self._has_tilebeam():
+ if self._is_HBA():
# Set a reference of TileBeam device that is correlated to this device
self.tilebeam_proxy = create_device_proxy(
f"{util.get_ds_inst_name()}/TileBeam/{antennafield}"
@@ -288,6 +349,20 @@ class Observation(LOFARDevice):
)
self.antennafield_proxy.Frequency_Band_RW = frequency_band
+ # Apply dithering configuration
+ if self.read_attribute("dithering_enabled_R"):
+ self.antennafield_proxy.RCU_DTH_freq_RW = [
+ self.read_attribute("dithering_frequency_R")
+ ] * self.antennafield_proxy.nr_antennas_R
+
+ self.antennafield_proxy.RCU_DTH_PWR_RW = [
+ self.read_attribute("dithering_power_R")
+ ] * self.antennafield_proxy.nr_antennas_R
+
+ self.antennafield_proxy.RCU_DTH_on()
+ else:
+ self.antennafield_proxy.RCU_DTH_off()
+
# Apply Beamlet configuration
self.beamlet_proxy.subband_select_RW = self._apply_saps_subbands(
self.read_attribute("saps_subband_R")
@@ -296,16 +371,33 @@ class Observation(LOFARDevice):
self.read_attribute("saps_pointing_R")
)
- # Apply Tile Beam pointing direction
- # NB: This is applied to all tiles, not just the ones in the selected
- # antenna set.
- if self._has_tilebeam():
- tile_beam = self.read_attribute("tile_beam_R")
+ if self._is_HBA():
+ # Apply Tile Beam pointing direction
+ # NB: These settings are applied to all tiles, not just the ones
+ # in the selected antenna set.
+ tile_beam = self.read_attribute("HBA_tile_beam_R")
self.tilebeam_proxy.Pointing_direction_RW = [
tuple(tile_beam)
] * self.antennafield_proxy.nr_antennas_R
+ # Apply DAB filter setting
+ DAB_filter = self.read_attribute("HBA_DAB_filter_R")
+
+ self.antennafield_proxy.RCU_DAB_filter_on_RW = [
+ DAB_filter
+ ] * self.antennafield_proxy.nr_antennas_R
+
+ # Apply HBAT element selection
+ element_selection = self.read_attribute("HBA_element_selection_R")
+
+ self.antennafield_proxy.HBAT_PWR_LNA_on_RW = self._apply_element_selection(
+ element_selection
+ )
+ self.antennafield_proxy.HBAT_PWR_on_RW = self._apply_element_selection(
+ element_selection
+ )
+
@log_exceptions()
def _stop_observation(self):
"""Tear down station resources we used."""
@@ -332,7 +424,7 @@ class Observation(LOFARDevice):
self._observation_settings = None
raise
- def _has_tilebeam(self):
+ def _is_HBA(self):
"""Return whether this observation should control a TileBeam device."""
return self._observation_settings.antenna_field.startswith("HBA")
@@ -370,6 +462,29 @@ class Observation(LOFARDevice):
"""Convert the observation id value into the correct format for Antennafield device"""
return numpy.array([observation_id] * MAX_ANTENNA, dtype=numpy.uint32)
+ def _apply_element_selection(self, element_selection: str):
+ """Convert the element selection strategy to a boolean array per element per polarisation per tile."""
+
+ nr_antennas = self.antennafield_proxy.nr_antennas_R
+
+ selection = numpy.zeros((nr_antennas, N_elements, N_pol), dtype=bool)
+
+ if element_selection == "ALL":
+ selection[:, :, :] = True
+ elif element_selection == "FIRST":
+ selection[:, 0, :] = True
+ elif element_selection == "GENERIC_201512":
+ # select a specific element for each tile
+ element_per_tile = (
+ self.antennafield_proxy.HBAT_single_element_selection_GENERIC_201512_R
+ )
+ for antenna, element in enumerate(element_per_tile):
+ selection[antenna, element, :] = True
+ else:
+ raise ValueError(f"Unsupported element selection: {element_selection}")
+
+ return selection.reshape(nr_antennas, N_elements * N_pol)
+
# ----------
# Run server
diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_observation_base.py b/tangostationcontrol/tangostationcontrol/test/devices/test_observation_base.py
index 03dc58103d247e26978d36bea41fa0e3ca8bde9e..37e863b4e4f1e11e027082872c0d9b97d56a6d8c 100644
--- a/tangostationcontrol/tangostationcontrol/test/devices/test_observation_base.py
+++ b/tangostationcontrol/tangostationcontrol/test/devices/test_observation_base.py
@@ -16,8 +16,17 @@ class TestObservationBase:
"angle1": 0.0261799, "angle2": 0, "direction_type": "J2000"
}
}],
- "tile_beam":
- { "angle1": 0.0261799, "angle2": 0, "direction_type": "J2000" },
+ "dithering": {
+ "enabled": true,
+ "power": -10.0,
+ "frequency": 123000000
+ },
+ "HBA": {
+ "tile_beam":
+ { "angle1": 0.0261799, "angle2": 0, "direction_type": "J2000" },
+ "DAB_filter": true,
+ "element_selection": "GENERIC_201512"
+ },
"first_beamlet": 0
}
"""
diff --git a/tangostationcontrol/test/configuration/_mock_requests.py b/tangostationcontrol/test/configuration/_mock_requests.py
index 314c1b40d9cfad27f73c380b53d98d4b2971cbd7..d89e75cb5c66a1939f4f6f8a3ea49be4595a2747 100644
--- a/tangostationcontrol/test/configuration/_mock_requests.py
+++ b/tangostationcontrol/test/configuration/_mock_requests.py
@@ -72,6 +72,64 @@ SAP_SCHEMA = """
}
"""
+HBA_SCHEMA = """
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "tile_beam"
+ ],
+ "properties": {
+ "tile_beam": {
+ "$ref": "pointing.json"
+ },
+ "dab_filter": {
+ "type": "boolean",
+ "default": false,
+ "description": "Enable hardware filter on DAB frequencies"
+ },
+ "element_selection": {
+ "type": "string",
+ "default": "ALL",
+ "description": "Which element(s) to enable in each tile",
+ "enum": [
+ "ALL",
+ "GENERIC_201512"
+ ]
+ }
+ }
+}
+"""
+
+DITHERING_SCHEMA = """
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "description": "Settings for adding dithering to the signal to increase its linearity",
+ "required": [
+ "enabled"
+ ],
+ "properties": {
+ "enabled": {
+ "type": "boolean",
+ "default": false
+ },
+ "power": {
+ "type": "number",
+ "description": "Power of dithering signal, in dBm",
+ "minimum": -25.0,
+ "maximum": -4.0,
+ "default": -4.0
+ },
+ "frequency": {
+ "type": "number",
+ "description": "Frequency of dithering signal, in Hz",
+ "default": 102000000
+ }
+ }
+}
+"""
+
OBSERVATION_SETTINGS_SCHEMA = """
{
"$schema": "http://json-schema.org/draft-07/schema",
@@ -89,18 +147,58 @@ OBSERVATION_SETTINGS_SCHEMA = """
"type": "number",
"minimum": 1
},
+ "start_time": {
+ "type": "string",
+ "format": "date-time",
+ "default": "1970-01-01T00:00:00"
+ },
"stop_time": {
"type": "string",
"format": "date-time"
},
+ "lead_time": {
+ "type": "number",
+ "description": "Number of seconds to start before the provided start time, to account for initialising the on-line signal chain, and for possibly negative geometrical delay compensation.",
+ "default": 2.0,
+ "minimum": 0
+ },
"antenna_field": {
- "type": "string"
+ "default": "HBA",
+ "description": "Antenna field to use",
+ "type": "string",
+ "enum": [
+ "LBA",
+ "HBA",
+ "HBA0",
+ "HBA1"
+ ]
},
"antenna_set": {
- "type": "string"
+ "default": "ALL",
+ "description": "Fields & antennas to use",
+ "type": "string",
+ "enum": [
+ "ALL",
+ "INNER",
+ "OUTER",
+ "SPARSE_EVEN",
+ "SPARSE_ODD"
+ ]
+ },
+ "dithering": {
+ "$ref": "dithering.json"
},
"filter": {
- "type": "string"
+ "type": "string",
+ "enum": [
+ "LBA_10_90",
+ "LBA_10_70",
+ "LBA_30_90",
+ "LBA_30_70",
+ "HBA_170_230",
+ "HBA_110_190",
+ "HBA_210_250"
+ ]
},
"SAPs": {
"type": "array",
@@ -109,12 +207,13 @@ OBSERVATION_SETTINGS_SCHEMA = """
"$ref": "sap.json"
}
},
- "tile_beam": {
- "$ref": "pointing.json"
- },
"first_beamlet": {
"type": "number",
+ "default": 0,
"minimum": 0
+ },
+ "HBA": {
+ "$ref": "hba.json"
}
}
}
@@ -137,6 +236,10 @@ def mocked_requests_get(*args, **kwargs):
return MockResponse(POINTING_SCHEMA, 200)
elif args[0] == "http://http-json-schemas/sap.json":
return MockResponse(SAP_SCHEMA, 200)
+ elif args[0] == "http://http-json-schemas/hba.json":
+ return MockResponse(HBA_SCHEMA, 200)
+ elif args[0] == "http://http-json-schemas/dithering.json":
+ return MockResponse(DITHERING_SCHEMA, 200)
elif args[0] == "http://http-json-schemas/observation-settings.json":
return MockResponse(OBSERVATION_SETTINGS_SCHEMA, 200)
diff --git a/tangostationcontrol/test/configuration/test_observation_settings.py b/tangostationcontrol/test/configuration/test_observation_settings.py
index 049ec2416b806181ff9149542d4edff1d2668140..4f876563126bbe5e3c670bbe15a1d211bae18a62 100644
--- a/tangostationcontrol/test/configuration/test_observation_settings.py
+++ b/tangostationcontrol/test/configuration/test_observation_settings.py
@@ -4,6 +4,7 @@
from datetime import datetime
from unittest import mock
+import json
import requests
from jsonschema.exceptions import ValidationError, RefResolutionError
@@ -19,77 +20,77 @@ class TestObservationSettings(base.TestCase):
sut = ObservationSettings.from_json(
'{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
'"antenna_field": "HBA", '
- '"antenna_set": "ALL", "filter": "filter_settings",'
+ '"antenna_set": "ALL", "filter": "HBA_110_190",'
'"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}]}'
)
self.assertEqual(sut.observation_id, 3)
self.assertEqual(sut.stop_time, datetime.fromisoformat("2012-04-23T18:25:43"))
self.assertEqual(sut.antenna_set, "ALL")
- self.assertEqual(sut.filter, "filter_settings")
+ self.assertEqual(sut.filter, "HBA_110_190")
self.assertEqual(len(sut.SAPs), 1)
sut = ObservationSettings.from_json(
'{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
'"antenna_field": "HBA", '
- '"antenna_set": "ALL", "filter": "filter_settings",'
+ '"antenna_set": "ALL", "filter": "HBA_110_190",'
'"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],'
- '"tile_beam": {"angle1":2.2, "angle2": 3.1, "direction_type":"MOON"} }'
+ '"HBA": { "tile_beam": {"angle1":2.2, "angle2": 3.1, "direction_type":"MOON"} } }'
)
- self.assertEqual(sut.tile_beam.angle1, 2.2)
- self.assertEqual(sut.tile_beam.angle2, 3.1)
- self.assertEqual(sut.tile_beam.direction_type, "MOON")
+ self.assertEqual(sut.HBA.tile_beam.angle1, 2.2)
+ self.assertEqual(sut.HBA.tile_beam.angle2, 3.1)
+ self.assertEqual(sut.HBA.tile_beam.direction_type, "MOON")
sut = ObservationSettings.from_json(
'{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
'"antenna_field": "HBA", '
- '"antenna_set": "ALL", "filter": "filter_settings",'
+ '"antenna_set": "ALL", "filter": "HBA_110_190",'
'"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],'
- '"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}'
+ '"HBA": {"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}, "first_beamlet": 2}'
)
self.assertEqual(sut.first_beamlet, 2)
def test_from_json_type_missmatch(self, _):
- for json in [
+ for json_str in [
# observation_id
- '{"observation_id": "3", "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
+ '{"observation_id": "3", "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "HBA_110_190","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "first_beamlet": 2}',
# stop_time
- '{"observation_id": 3, "stop_time": "test", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
+ '{"observation_id": 3, "stop_time": "test", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "HBA_110_190","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "first_beamlet": 2}',
# antenna_set
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": 4, "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
+ '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": 4, "filter": "HBA_110_190","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "first_beamlet": 2}',
# filter
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": 1,"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
+ '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": 1,"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "first_beamlet": 2}',
# SAPs
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": {"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}},"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [1],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": 1, "first_beamlet": 2}',
+ '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "HBA_110_190","SAPs": {"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}, "first_beamlet": 2}',
# first_beamlet
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": "2"}',
+ '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "HBA_110_190","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "first_beamlet": "2"}',
]:
- with self.assertRaises((ValidationError, ValueError), msg=f"{json}"):
- ObservationSettings.from_json(json)
+ with self.assertRaises((ValidationError, ValueError), msg=f"{json_str}"):
+ ObservationSettings.from_json(json_str)
def test_from_json_missing_fields(self, _):
- for json in [
- # observation_id
- '{"stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # stop_time
- '{"observation_id": 3, "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # antenna_field
- '{"observation_id": 3, "antenna_set": "ALL", "stop_time": "2012-04-23T18:25:43", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # antenna_set
- '{"observation_id": 3, "antenna_field": "HBA", "stop_time": "2012-04-23T18:25:43", "filter": "filter_settings","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # filter
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # SAPs
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- # SAPs (empty list)
- '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "filter_settings","SAPs": [],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}',
- ]:
- with self.assertRaises((ValidationError, ValueError), msg=f"{json}"):
- ObservationSettings.from_json(json)
+ complete_json = '{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", "antenna_field": "HBA", "antenna_set": "ALL", "filter": "HBA_110_190","SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}], "HBA": {"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}, "first_beamlet": 2}'
+
+ for field in (
+ "observation_id",
+ "stop_time",
+ "antenna_field",
+ "antenna_set",
+ "filter",
+ "SAPs",
+ ):
+ # construct a JSON string with the provided field missing
+ json_dict = json.loads(complete_json)
+ del json_dict[field]
+ json_str = json.dumps(json_dict)
+
+ # trigger validation error
+ with self.assertRaises(
+ (ValidationError, ValueError), msg=f"Omitted field {field}: {json_str}"
+ ):
+ ObservationSettings.from_json(json_str)
def test_to_json(self, _):
sut = ObservationSettings(
@@ -122,15 +123,15 @@ class TestObservationSettings(base.TestCase):
ObservationSettings.from_json(
'{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
'"antenna_field": "HBA", '
- '"antenna_set": "ALL", "filter": "filter_settings",'
+ '"antenna_set": "ALL", "filter": "HBA_110_190",'
'"SAPs": [{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}]}'
)
self.assertEqual(5, mock_get.call_count)
def test_throw_wrong_instance(self, _):
- for json in [
+ for json_str in [
'{"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}',
'{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
]:
with self.assertRaises(ValidationError):
- ObservationSettings.from_json(json)
+ ObservationSettings.from_json(json_str)
diff --git a/tangostationcontrol/test/devices/test_antennafield_device.py b/tangostationcontrol/test/devices/test_antennafield_device.py
index 12edfafefe60b49e1a0f72b03bfd96fb614d1329..0ce70d37a2b9af165372f63fc2bed2ef9e370ae7 100644
--- a/tangostationcontrol/test/devices/test_antennafield_device.py
+++ b/tangostationcontrol/test/devices/test_antennafield_device.py
@@ -30,6 +30,7 @@ from tangostationcontrol.devices.base_device_classes.mapper import (
MappingKeys,
AntennaToRecvMapper,
AntennaToSdpMapper,
+ RecvDeviceWalker,
)
logger = logging.getLogger()
@@ -1106,3 +1107,137 @@ class TestAntennafieldDevice(device_base.DeviceTestCase):
) as proxy:
expected = [True] * int(MAX_ANTENNA / 2) + [False] * int(MAX_ANTENNA / 2)
numpy.testing.assert_equal(expected, proxy.antenna_set_to_mask("INNER"))
+
+
+class TestRecvDeviceWalker(base.TestCase):
+ class MockDeviceProxy:
+ """Mock of DeviceProxy that simulates recv.ANT_mask_RW."""
+
+ @property
+ def ANT_mask_RW(self):
+ return self.ant_mask
+
+ @ANT_mask_RW.setter
+ def ANT_mask_RW(self, value):
+ self.ant_mask = value
+
+ def __init__(self, index):
+ self.visited = False
+ self.index = index
+
+ # fill with a value we don't use, to make sure
+ # there will be a difference when set
+ self.ant_mask = numpy.array([[False, True, True]] * N_rcu)
+
+ # save the value we originally use
+ self.original_ant_mask = self.ant_mask
+
+ def test_recv_masks_identity_mapping(self):
+ """Test whether a straight setup works."""
+
+ control_to_recv_mapping = numpy.array([[1, 0], [1, 1], [1, 2]])
+ antenna_usage_mask = numpy.array([True, True, True])
+
+ sut = RecvDeviceWalker(control_to_recv_mapping, antenna_usage_mask)
+
+ expected = numpy.array(
+ [[[True, True, True]] + [[False, False, False]] * (N_rcu - 1)]
+ )
+ numpy.testing.assert_equal(expected, sut.recv_ant_masks())
+
+ def test_recv_masks_antenna_usage_mask(self):
+ """Test whether the antenna_usage_mask is respected."""
+
+ control_to_recv_mapping = numpy.array([[1, 0], [1, 1], [1, 2]])
+ antenna_usage_mask = numpy.array([False, True, False])
+
+ sut = RecvDeviceWalker(control_to_recv_mapping, antenna_usage_mask)
+
+ expected = numpy.array(
+ [[[False, True, False]] + [[False, False, False]] * (N_rcu - 1)]
+ )
+ numpy.testing.assert_equal(expected, sut.recv_ant_masks())
+
+ def test_recv_masks_control_to_recv_mapping(self):
+ """Test whether control_to_recv_mapping is respected."""
+
+ control_to_recv_mapping = numpy.array([[1, 0], [2, 1], [1, 2]])
+ antenna_usage_mask = numpy.array([True, True, True])
+
+ sut = RecvDeviceWalker(control_to_recv_mapping, antenna_usage_mask)
+
+ expected = numpy.array(
+ [
+ [[True, False, True]] + [[False, False, False]] * (N_rcu - 1),
+ [[False, True, False]] + [[False, False, False]] * (N_rcu - 1),
+ ]
+ )
+ numpy.testing.assert_equal(expected, sut.recv_ant_masks())
+
+ def test_walk_receivers(self):
+ """Test walk_receivers on multiple recv_proxies."""
+
+ control_to_recv_mapping = numpy.array([[1, 0], [2, 1], [1, 2]])
+ antenna_usage_mask = numpy.array([True, True, True])
+
+ sut = RecvDeviceWalker(control_to_recv_mapping, antenna_usage_mask)
+
+ recv_proxies = [TestRecvDeviceWalker.MockDeviceProxy(n + 1) for n in range(2)]
+
+ def visitor(recv_proxy):
+ recv_proxy.visited = True
+
+ # is our mask set correctly?
+ if recv_proxy.index == 1:
+ expected = numpy.array(
+ [[True, False, True]] + [[False, False, False]] * (N_rcu - 1)
+ )
+ elif recv_proxy.index == 2:
+ expected = numpy.array(
+ [[False, True, False]] + [[False, False, False]] * (N_rcu - 1)
+ )
+
+ numpy.testing.assert_equal(expected, recv_proxy.ANT_mask_RW)
+
+ sut.walk_receivers(recv_proxies, visitor)
+
+ # make sure both recv_proxies were visited
+ self.assertTrue(recv_proxies[0].visited)
+ self.assertTrue(recv_proxies[1].visited)
+
+ # make sure both masks were restored
+ numpy.testing.assert_equal(
+ recv_proxies[0].original_ant_mask, recv_proxies[0].ant_mask
+ )
+ numpy.testing.assert_equal(
+ recv_proxies[1].original_ant_mask, recv_proxies[1].ant_mask
+ )
+
+ def test_walk_receivers_restores_mask_on_exception(self):
+ """Test whether walk_receivers() also restores the recv_proxy.ANT_mask_RW
+ if the visitor function throws."""
+ control_to_recv_mapping = numpy.array([[1, 0], [2, 1], [1, 2]])
+ antenna_usage_mask = numpy.array([True, True, True])
+
+ sut = RecvDeviceWalker(control_to_recv_mapping, antenna_usage_mask)
+
+ recv_proxies = [TestRecvDeviceWalker.MockDeviceProxy(n + 1) for n in range(2)]
+
+ class MyException(Exception):
+ """A exception noone can raise but us."""
+
+ pass
+
+ def visitor(recv_proxy):
+ raise MyException("foo")
+
+ with self.assertRaises(MyException):
+ sut.walk_receivers(recv_proxies, visitor)
+
+ # make sure no mask was disturbed
+ numpy.testing.assert_equal(
+ recv_proxies[0].original_ant_mask, recv_proxies[0].ant_mask
+ )
+ numpy.testing.assert_equal(
+ recv_proxies[1].original_ant_mask, recv_proxies[1].ant_mask
+ )