diff --git a/tangostationcontrol/integration_test/default/devices/test_device_rcu2h.py b/tangostationcontrol/integration_test/default/devices/test_device_rcu2h.py
deleted file mode 100644
index 35b3b30c5ffecc78d39bf327cfeee4f0f133284e..0000000000000000000000000000000000000000
--- a/tangostationcontrol/integration_test/default/devices/test_device_rcu2h.py
+++ /dev/null
@@ -1,9 +0,0 @@
-# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
-# SPDX-License-Identifier: Apache-2.0
-
-from integration_test.default.devices.base import AbstractTestBases
-
-
-class TestDeviceRECV(AbstractTestBases.TestDeviceBase):
- def setUp(self):
- super().setUp("STAT/RECVH/H0")
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_recvh.py b/tangostationcontrol/integration_test/default/devices/test_device_recvh.py
new file mode 100644
index 0000000000000000000000000000000000000000..ba2122c5d3be90776819fb0ec9e5dcfd59b31498
--- /dev/null
+++ b/tangostationcontrol/integration_test/default/devices/test_device_recvh.py
@@ -0,0 +1,37 @@
+# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
+# SPDX-License-Identifier: Apache-2.0
+
+import time
+
+import numpy
+
+from integration_test.default.devices.base import AbstractTestBases
+
+
+class TestDeviceRECV(AbstractTestBases.TestDeviceBase):
+ HBAT_DELAY_UPDATE_INTERVAL = 0.1
+
+ def setUp(self):
+ super().setUp("STAT/RECVH/H0")
+
+ def test_update_HBAT_bf_delay_steps(self):
+ """Verify HBAT_bf_delay_steps_stage propagates to HBAT_bf_delay_steps."""
+
+ # update faster to speed up tests
+ self.proxy.put_property(
+ {"HBAT_BF_delay_steps_update_interval": self.HBAT_DELAY_UPDATE_INTERVAL}
+ )
+ self.proxy.off()
+ self.proxy.boot()
+
+ # write new value to HBAT_BF_delay_steps_stage_RW
+ delay_steps = numpy.ones(
+ self.proxy.HBAT_BF_delay_steps_stage_RW.shape, dtype=numpy.uint64
+ )
+ self.proxy.HBAT_BF_delay_steps_stage_RW = delay_steps
+
+ # make sure HBAT_BF_delay_steps_RW is updated
+ time.sleep(self.HBAT_DELAY_UPDATE_INTERVAL * 2)
+
+ # verify whether it did
+ numpy.testing.assert_equal(self.proxy.HBAT_BF_delay_steps_RW, delay_steps)
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_rcu2l.py b/tangostationcontrol/integration_test/default/devices/test_device_recvl.py
similarity index 100%
rename from tangostationcontrol/integration_test/default/devices/test_device_rcu2l.py
rename to tangostationcontrol/integration_test/default/devices/test_device_recvl.py
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py b/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
index c4be173ce21673437db27e37444eff6e2ff999f8..991b3b47997504710f30de65fa719dd092bc9e00 100644
--- a/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
+++ b/tangostationcontrol/integration_test/default/devices/test_device_tilebeam.py
@@ -93,7 +93,7 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
# Verify attribute is present (all zeros if never used before)
delays_r1 = numpy.array(
- self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_RW").value
+ self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_stage_RW").value
)
self.assertIsNotNone(delays_r1)
@@ -103,7 +103,7 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
# Verify writing operation does not lead to errors
self.proxy.set_pointing(self.POINTING_DIRECTION) # write values to RECV
delays_r2 = numpy.array(
- self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_RW").value
+ self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_stage_RW").value
)
self.assertIsNotNone(delays_r2)
@@ -121,7 +121,7 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
)
calculated_HBAT_delay_steps = numpy.array(
- self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_RW").value
+ self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_stage_RW").value
)
expected_HBAT_delay_steps = numpy.array(
@@ -142,9 +142,9 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
self.proxy.set_pointing(["AZELGEO", "0rad", "0rad"] * CS001_TILES)
# obtain delays of the X polarisation of all the elements of the first tile
- north_beam_delay_steps = antennafield_proxy.HBAT_BF_delay_steps_RW[0].reshape(
- 4, 4, 2
- )[:, :, 0]
+ north_beam_delay_steps = antennafield_proxy.HBAT_BF_delay_steps_stage_RW[
+ 0
+ ].reshape(4, 4, 2)[:, :, 0]
# delays must differ under rotation, or our test will give a false positive
self.assertNotEqual(
@@ -157,7 +157,7 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
self.proxy.set_pointing(["AZELGEO", f"{angle}rad", "0rad"] * CS001_TILES)
# obtain delays of the X polarisation of all the elements of the first tile
- angled_beam_delay_steps = antennafield_proxy.HBAT_BF_delay_steps_RW[
+ angled_beam_delay_steps = antennafield_proxy.HBAT_BF_delay_steps_stage_RW[
0
].reshape(4, 4, 2)[:, :, 0]
@@ -188,7 +188,7 @@ class TestDeviceTileBeam(AbstractTestBases.TestDeviceBase):
self.proxy.set_pointing_for_specific_time(json_string)
calculated_HBAT_delay_steps = numpy.array(
- self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_RW").value
+ self.antennafield_proxy.read_attribute("HBAT_BF_delay_steps_stage_RW").value
) # dims (CS001_TILES, 32)
# Check all delay steps are zero with small margin
diff --git a/tangostationcontrol/tangostationcontrol/beam/managers/_tilebeam.py b/tangostationcontrol/tangostationcontrol/beam/managers/_tilebeam.py
index 64383773efcef5622c774fbeab3102653b71fa8a..650d6069cebb3f5e84871d136e56fe98868099ab 100644
--- a/tangostationcontrol/tangostationcontrol/beam/managers/_tilebeam.py
+++ b/tangostationcontrol/tangostationcontrol/beam/managers/_tilebeam.py
@@ -33,6 +33,7 @@ class TileBeamManager(AbstractBeamManager):
# Labels for DurationMetric
self.metric_labels = device_labels(device)
+ @DurationMetric()
def delays(self, pointing_direction: numpy.array, timestamp: datetime.datetime):
"""
Calculate the delays (in seconds) based on the pointing list and the timestamp.
@@ -138,11 +139,14 @@ class TileBeamManager(AbstractBeamManager):
(len(recv_proxies), MAX_ANTENNA, N_elements * N_pol)
)
+ # Stage the new delays for periodic writes that combine
+ # delays for other antenna fields, and only applies when the
+ # value changes.
for idx, recv_proxy in enumerate(recv_proxies):
self.device.atomic_read_modify_write_attribute(
mapped_values[idx],
recv_proxy,
- "HBAT_bf_delay_steps_RW",
+ "HBAT_bf_delay_steps_stage_RW",
cast_type=numpy.int64,
)
diff --git a/tangostationcontrol/tangostationcontrol/clients/opcua_client.py b/tangostationcontrol/tangostationcontrol/clients/opcua_client.py
index 542b947b5c36fcd5f8a4b80da4704e14c7f2f6ec..e11225004eb763fc80c9140fbb025edc7618acc3 100644
--- a/tangostationcontrol/tangostationcontrol/clients/opcua_client.py
+++ b/tangostationcontrol/tangostationcontrol/clients/opcua_client.py
@@ -4,6 +4,7 @@
import asyncio
import logging
import socket
+import textwrap
from typing import Dict, List
import asyncua
@@ -479,7 +480,9 @@ class ProtocolAttribute:
value = value.tolist() if type(value) == numpy.ndarray else value
if self.log_writes:
- logger.info(f"OPC-UA write: {self.name} := {str(value)[:80]}")
+ logger.info(
+ f"OPC-UA write: {self.name} := {textwrap.shorten(str(value), 150)}"
+ )
try:
await self.node.set_data_value(
diff --git a/tangostationcontrol/tangostationcontrol/common/asyncio.py b/tangostationcontrol/tangostationcontrol/common/asyncio.py
index e5c0c40e7b591af7847114779e59ff4d006b35c3..f716bc83a9fc6cb2e01fa57d97c7c111539aa93a 100644
--- a/tangostationcontrol/tangostationcontrol/common/asyncio.py
+++ b/tangostationcontrol/tangostationcontrol/common/asyncio.py
@@ -65,7 +65,12 @@ class EventLoopThread:
class PeriodicTask:
"""Provide a periodic call to a coroutine."""
- def __init__(self, event_loop, func: Callable[[], None], interval: float = 1.0):
+ def __init__(
+ self,
+ event_loop,
+ func: Callable[[], None],
+ interval: float = 1.0,
+ ):
self.event_loop = event_loop
self.func = func
self.interval = interval
diff --git a/tangostationcontrol/tangostationcontrol/devices/antennafield/afh.py b/tangostationcontrol/tangostationcontrol/devices/antennafield/afh.py
index 79652c4b72e59c43acdc29459caa0b06a31f6a06..cdd4368f7b31ae68ef7a980c93e9b63c87a8e402 100644
--- a/tangostationcontrol/tangostationcontrol/devices/antennafield/afh.py
+++ b/tangostationcontrol/tangostationcontrol/devices/antennafield/afh.py
@@ -146,6 +146,13 @@ class AFH(AF):
max_dim_y=MAX_ANTENNA,
access=AttrWriteType.READ_WRITE,
)
+ HBAT_BF_delay_steps_stage_RW = MappedAttribute(
+ "HBAT_BF_delay_steps_stage_RW",
+ dtype=((numpy.int64,),),
+ max_dim_x=N_elements * N_pol,
+ max_dim_y=MAX_ANTENNA,
+ access=AttrWriteType.READ_WRITE,
+ )
HBAT_LED_on_R = MappedAttribute(
"HBAT_LED_on_R",
dtype=((bool,),),
diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/beam_device.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/beam_device.py
index 5122e67ec89fe71cb107ccd106903f3f8600469b..63523df6228635fdc0bb13d513303da6f37bc3a5 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/beam_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/beam_device.py
@@ -78,7 +78,7 @@ class BeamDevice(AsyncDevice):
dtype="DevFloat",
doc="Beam weights updating interval time [seconds]",
mandatory=False,
- default_value=10.0,
+ default_value=1.0,
)
Beam_tracking_application_offset = device_property(
diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/lofar_device.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/lofar_device.py
index 57381221c60aa81fa225dd99e971ac1efc607dbc..e5d1248daf69da168813556dcd75225454676719 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/lofar_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/lofar_device.py
@@ -283,7 +283,7 @@ class LOFARDevice(Device):
)
poll_thread_running_R = attribute(
- doc="Whether the attributes are being polled." "",
+ doc="Whether the attributes are being polled.",
dtype=bool,
fget=lambda self: (
self.event_loop_thread
diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
index 1872e1f09b7de82eb73e5add59ed57af978045eb..5c0b17347564ea35796fbba8a70383d1d068f614 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
@@ -283,6 +283,7 @@ class AntennaToRecvMapper(ABC, AntennaMapper):
"ANT_mask_RW": value_map_ant_bool,
"HBAT_BF_delay_steps_R": value_map_ant_32_int,
"HBAT_BF_delay_steps_RW": value_map_ant_32_int,
+ "HBAT_BF_delay_steps_stage_RW": value_map_ant_32_int,
"HBAT_LED_on_R": value_map_ant_32_bool,
"HBAT_LED_on_RW": value_map_ant_32_bool,
"HBAT_PWR_LNA_on_R": value_map_ant_32_bool,
diff --git a/tangostationcontrol/tangostationcontrol/devices/recv/recvh.py b/tangostationcontrol/tangostationcontrol/devices/recv/recvh.py
index e8c43bb59fa9ad14ec83ed518949ba5caf7042bb..2f2f68d1f7eb8d23c4a97b3ee10137d65121c4ef 100644
--- a/tangostationcontrol/tangostationcontrol/devices/recv/recvh.py
+++ b/tangostationcontrol/tangostationcontrol/devices/recv/recvh.py
@@ -4,26 +4,36 @@
""" Receiver Unit High Device Server for LOFAR2.0
"""
+import logging
+
import numpy
from attribute_wrapper.attribute_wrapper import AttributeWrapper
from tango import AttrWriteType, DevVarFloatArray
# PyTango imports
-from tango.server import command, device_property
+from tango.server import command, device_property, attribute
# Additional import
+from tangostationcontrol.common.asyncio import PeriodicTask
from tangostationcontrol.common.constants import (
N_rcu,
N_elements,
N_pol,
N_rcu_inp,
+ DEFAULT_POLLING_PERIOD_MS,
+)
+from tangostationcontrol.common.device_decorators import DurationMetric
+from tangostationcontrol.common.lofar_logging import (
+ device_logging_to_python,
+ exception_to_str,
)
-from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.devices.base_device_classes.recv_device import RECVDevice
from tangostationcontrol.metrics import device_metrics
__all__ = ["RECVH"]
+logger = logging.getLogger()
+
@device_logging_to_python()
@device_metrics(
@@ -44,6 +54,13 @@ class RECVH(RECVDevice):
# Device Properties
# -----------------
+ HBAT_BF_delay_steps_update_interval = device_property(
+ dtype="DevFloat",
+ doc="Interval with which to update HBAT_BF_delay_steps_RW [seconds]",
+ mandatory=False,
+ default_value=10.0,
+ )
+
# ----- Default settings
HBAT_PWR_on_RW_default = device_property(
@@ -118,6 +135,21 @@ class RECVH(RECVDevice):
# ----------
# Attributes
# ----------
+
+ bf_delay_steps_thread_running_R = attribute(
+ doc="Whether the HBAT_BF_delay_steps_RW is being updated.",
+ dtype=bool,
+ fget=lambda self: (
+ self.event_loop_thread
+ and self.event_loop_thread.is_running()
+ and self.bf_delay_steps_update_task
+ and self.bf_delay_steps_update_task.is_running()
+ ),
+ # Tango needs to poll this, as otherwise this attribute will never
+ # be exposed as "False" as the event thread must run to do so.
+ polling_period=DEFAULT_POLLING_PERIOD_MS,
+ )
+
RCU_firmware_version_R = AttributeWrapper(
comms_annotation=["RCU_firmware_version_R"],
datatype=numpy.int64,
@@ -146,16 +178,37 @@ class RECVH(RECVDevice):
# The 32 delays deconstruct as delays[polarisation][dipole],
# and each delay is the number of 'delay steps' to apply (0.5ns for HBAT1).
HBAT_BF_delay_steps_R = AttributeWrapper(
- comms_annotation={"path": ["HBAT_BF_delay_steps_R"], "log_writes": False},
+ comms_annotation=["HBAT_BF_delay_steps_R"],
datatype=numpy.int64,
dims=(N_rcu * N_rcu_inp, N_elements, N_pol),
)
HBAT_BF_delay_steps_RW = AttributeWrapper(
- comms_annotation={"path": ["HBAT_BF_delay_steps_RW"], "log_writes": False},
+ doc="HBAT beamformer delay steps. The hardware supports writing this value only once every 10s. This device will write the value of HBAT_BF_delay_steps_stage_RW every HBAT_BF_delay_steps_update_interval seconds.",
+ comms_annotation=["HBAT_BF_delay_steps_RW"],
datatype=numpy.int64,
dims=(N_rcu * N_rcu_inp, N_elements, N_pol),
access=AttrWriteType.READ_WRITE,
)
+
+ # Multiple TileBeams want to read-modify-write HBAT_BF_delay_steps to
+ # interleave the delays for their respective antenna fields. Yet we are only
+ # allowed to write HBAT_BF_delay_steps_RW every 10s. This "stage" attribute
+ # allows collecting changes, which are periodically applied through
+ # the PeriodicTask "_update_bf_delay_steps".
+ @attribute(
+ doc="HBAT beamformer delay steps staged for writing at the next interval.",
+ dtype=((numpy.int64,),),
+ max_dim_y=N_rcu * N_rcu_inp,
+ max_dim_x=N_elements * N_pol,
+ access=AttrWriteType.READ_WRITE,
+ )
+ def HBAT_BF_delay_steps_stage_RW(self):
+ return self._HBAT_BF_delay_steps_stage_RW
+
+ @HBAT_BF_delay_steps_stage_RW.write
+ def HBAT_BF_delay_steps_stage_RW(self, value):
+ self._HBAT_BF_delay_steps_stage_RW = value
+
HBAT_LED_on_R = AttributeWrapper(
comms_annotation=["HBAT_LED_on_R"],
datatype=bool,
@@ -190,9 +243,53 @@ class RECVH(RECVDevice):
access=AttrWriteType.READ_WRITE,
)
+ @DurationMetric()
+ async def _update_bf_delay_steps(self):
+ try:
+ staged_value = await self.async_read_attribute(
+ "HBAT_BF_delay_steps_stage_RW"
+ )
+ last_written_value = await self.async_read_attribute(
+ "HBAT_BF_delay_steps_RW"
+ )
+
+ if (staged_value != last_written_value).all():
+ logger.debug(
+ "Writing HBAT_BF_delay_steps_RW := HBAT_BF_delay_steps_stage_RW after detecting changes"
+ )
+ self.proxy.write_attribute("HBAT_BF_delay_steps_RW", staged_value)
+ except Exception as ex:
+ logger.error(
+ f"Failed to update HBAT_BF_delay_steps_RW: {exception_to_str(ex)}"
+ )
+
# --------
# overloaded functions
# --------
+ def __init__(self, cl, name):
+ self.bf_delay_steps_update_task = None
+ self._HBAT_BF_delay_steps_stage_RW = numpy.zeros(
+ (N_rcu * N_rcu_inp, N_elements * N_pol), dtype=numpy.int64
+ )
+
+ super().__init__(cl, name)
+
+ def configure_for_on(self):
+ super().configure_for_on()
+
+ self.bf_delay_steps_update_task = PeriodicTask(
+ self.event_loop_thread.event_loop,
+ self._update_bf_delay_steps,
+ self.HBAT_BF_delay_steps_update_interval,
+ )
+
+ def configure_for_off(self):
+ # stop writing delays (gracefully, as it is fast)
+ if self.bf_delay_steps_update_task:
+ self.bf_delay_steps_update_task.stop()
+
+ super().configure_for_off()
+
def properties_changed(self):
super().properties_changed()
diff --git a/tangostationcontrol/test/beam/managers/test_tilebeam_manager.py b/tangostationcontrol/test/beam/managers/test_tilebeam_manager.py
index be4ae13a9601df4735cbd1483d62c54a0d22e79f..af4d8738327b9d556f830efa654a2596fe7088db 100644
--- a/tangostationcontrol/test/beam/managers/test_tilebeam_manager.py
+++ b/tangostationcontrol/test/beam/managers/test_tilebeam_manager.py
@@ -19,11 +19,17 @@ from tangostationcontrol.common.constants import (
class TestTileBeamManager(base.TestCase):
"""Test class for TileBeam manager"""
- def test_delays(self):
+ @patch.object(device_decorators, "get_current_device")
+ def test_delays(self, mock_get_current_device):
"""Verify delays are retrieved with correct dimensions"""
_dt = datetime.datetime.now()
+
device_mock = MagicMock()
device_mock.get_name = MagicMock(return_value="domain/family/member")
+ device_mock.metric_labels = {"foo": "bar"}
+
+ mock_get_current_device.return_value = device_mock
+
sut = TileBeamManager(device_mock)
sut.nr_tiles = DEFAULT_N_HBA_TILES
diff --git a/tangostationcontrol/test/beam/test_delays.py b/tangostationcontrol/test/beam/test_delays.py
index 809aaa21714ba0284849087b3eacee5b95847d6e..4169f4d53b76d429a5a9efc03d4a7403dcae1e4b 100644
--- a/tangostationcontrol/test/beam/test_delays.py
+++ b/tangostationcontrol/test/beam/test_delays.py
@@ -3,6 +3,7 @@
import datetime
import logging
+import statistics
import time
import casacore
@@ -11,7 +12,7 @@ import numpy
import numpy.testing
import threading
-from tangostationcontrol.beam.delays import Delays
+from tangostationcontrol.beam.delays import Delays, threading as delays_threading
from tangostationcontrol.common.constants import MAX_ANTENNA, N_beamlets_ctrl
from test import base
@@ -170,67 +171,52 @@ class TestDelays(base.TestCase):
self.assertAlmostEqual(0.1, delays[0], 6, f"delays[0] = {delays[0]}")
- def test_delays_bulk(self):
- d = Delays([0, 0, 0])
+
+class TestDelaysBulk(base.TestCase):
+ def setUp(self):
+ self.d = Delays([0, 0, 0])
timestamp = datetime.datetime(
2022, 3, 1, 0, 0, 0
) # timestamp does not actually matter, but casacore doesn't know that.
- d.set_measure_time(timestamp)
+ self.d.set_measure_time(timestamp)
# generate different positions and directions
- positions = numpy.array([[i, 2, 3] for i in range(5)])
- directions = numpy.array(
+ self.positions = numpy.array([[i, 2, 3] for i in range(MAX_ANTENNA)])
+ self.directions = numpy.array(
[
["J2000", f"{i*numpy.pi/180}rad", f"{i*numpy.pi/180}rad"]
- for i in range(90)
+ for i in range(N_beamlets_ctrl)
]
)
- bulk_result = d.delays_bulk(directions, positions)
+ def test_delays_bulk(self):
+ bulk_result = self.d.delays_bulk(self.directions, self.positions)
# verify parallellisation along direction axis
- for i, single_dir in enumerate(directions):
- single_dir_result = d.delays_bulk([single_dir], positions)
+ for i, single_dir in enumerate(self.directions):
+ single_dir_result = self.d.delays_bulk([single_dir], self.positions)
numpy.testing.assert_almost_equal(
single_dir_result[:, 0], bulk_result[:, i], 4
)
# verify parallellisation along position axis
- for i, single_pos in enumerate(positions):
- single_pos_result = d.delays_bulk(directions, [single_pos])
+ for i, single_pos in enumerate(self.positions):
+ single_pos_result = self.d.delays_bulk(self.directions, [single_pos])
numpy.testing.assert_almost_equal(
single_pos_result[0, :], bulk_result[i, :], 4
)
def test_delays_bulk_speed(self):
- d = Delays([0, 0, 0])
- timestamp = datetime.datetime(
- 2022, 3, 1, 0, 0, 0
- ) # timestamp does not actually matter, but casacore doesn't know that.
- d.set_measure_time(timestamp)
-
- positions = numpy.array([[1, 2, 3]] * MAX_ANTENNA)
- directions = numpy.array([["J2000", "0rad", "0rad"]] * N_beamlets_ctrl)
-
count = 10
before = time.monotonic_ns()
for _ in range(count):
- _ = d.delays_bulk(directions, positions)
+ _ = self.d.delays_bulk(self.directions, self.positions)
after = time.monotonic_ns()
logging.error(
f"delays bulk averaged {(after - before) / count / 1e6} ms to convert 488 directions for 96 antennas."
)
def test_delays_bulk_parallel_speed(self):
- d = Delays([0, 0, 0])
- timestamp = datetime.datetime(
- 2022, 3, 1, 0, 0, 0
- ) # timestamp does not actually matter, but casacore doesn't know that.
- d.set_measure_time(timestamp)
-
- positions = numpy.array([[1, 2, 3]] * MAX_ANTENNA)
- directions = numpy.array([["J2000", "0rad", "0rad"]] * N_beamlets_ctrl)
-
count = 10
nr_threads = 3
duration_results_ms: list[float] = []
@@ -238,11 +224,22 @@ class TestDelays(base.TestCase):
def run_delays_bulk():
before = time.monotonic_ns()
for _ in range(count):
- _ = d.delays_bulk(directions, positions)
+ _ = self.d.delays_bulk(self.directions, self.positions)
after = time.monotonic_ns()
duration_results_ms.append((after - before) / count / 1e6)
+ # measure single-threaded performance first
+ for _ in range(nr_threads):
+ run_delays_bulk()
+ single_thread_execution_time = statistics.median(duration_results_ms)
+
+ logging.error(
+ f"delays bulk single-threaded cost averages at {single_thread_execution_time} ms per call to convert 488 directions for 96 antennas {count} times."
+ )
+
+ # compare against multi-threaded performance
+ duration_results_ms = []
threads = [threading.Thread(target=run_delays_bulk) for _ in range(nr_threads)]
[t.start() for t in threads]
[t.join() for t in threads]
@@ -251,9 +248,25 @@ class TestDelays(base.TestCase):
f"delays bulk multi-threaded averages are {[int(d) for d in duration_results_ms]} ms per call to convert 488 directions for 96 antennas {count} times using {nr_threads} threads."
)
- MAX_CALL_DURATION_MS = 150
-
# as we have a real time system and sufficient cores, we care
# most about the worst case for a thread. We assume the worst
# case thread is slowed down by all the work of the other threads.
- self.assertLess(max(duration_results_ms) / nr_threads, MAX_CALL_DURATION_MS)
+ self.assertLess(
+ max(duration_results_ms) / nr_threads, single_thread_execution_time * 1.25
+ )
+
+ # report the performance if we remove the compute lock around casacore, to detect
+ # on which systems it matters.
+
+ with mock.patch.object(delays_threading, "Lock") as m_lock:
+ # compare against multi-threaded performance
+ duration_results_ms = []
+ threads = [
+ threading.Thread(target=run_delays_bulk) for _ in range(nr_threads)
+ ]
+ [t.start() for t in threads]
+ [t.join() for t in threads]
+
+ logging.error(
+ f"delays bulk multi-threaded averages WITHOUT LOCK are {[int(d) for d in duration_results_ms]} ms per call to convert 488 directions for 96 antennas {count} times using {nr_threads} threads."
+ )