From eff2b053b0c898f5b11ab99cc4bb4a573c6495df Mon Sep 17 00:00:00 2001
From: Jan David Mol <mol@astron.nl>
Date: Fri, 25 Aug 2023 12:00:39 +0000
Subject: [PATCH] L2SS-1477: Add observation start time, simplified observation
classes.
---
CDB/LOFAR_ConfigDb.json | 10 -
CDB/integrations/multiobs_ConfigDb.json | 2 +-
CDB/stations/common.json | 9 -
CDB/stations/testenv_cs001.json | 13 +-
CDB/test_environment_ConfigDb.json | 22 --
docker-compose/device-observation.yml | 53 ---
.../definitions/observation-settings.json | 11 +
sbin/run_integration_test.sh | 2 +-
tangostationcontrol/docs/source/observing.rst | 33 +-
.../configDB/test_environment_ConfigDb.json | 2 +-
.../test_device_configuration.py | 10 +-
.../devices/test_device_observation.py | 25 +-
.../test_device_observation_control.py | 71 ++--
.../common/observation_controller.py | 281 ++++++--------
.../configuration/observation_settings.py | 16 +-
.../devices/observation.py | 365 ++++++++++--------
.../devices/observation_control.py | 121 +++---
.../common/test_observation_controller.py | 70 ++--
.../test_observation_settings.py | 4 +-
19 files changed, 556 insertions(+), 564 deletions(-)
delete mode 100644 CDB/test_environment_ConfigDb.json
delete mode 100644 docker-compose/device-observation.yml
diff --git a/CDB/LOFAR_ConfigDb.json b/CDB/LOFAR_ConfigDb.json
index f82fcb5d1..97de683bb 100644
--- a/CDB/LOFAR_ConfigDb.json
+++ b/CDB/LOFAR_ConfigDb.json
@@ -96,24 +96,14 @@
}
}
},
- "Observation": {
- "STAT": {
- "Observation": {
- "STAT/Observation/0": {
- }
- }
- }
- },
"ObservationControl": {
"STAT": {
"ObservationControl": {
"STAT/ObservationControl/1": {
"properties": {
"Control_Children": [
- "STAT/Observation/0"
],
"Power_Children": [
- "STAT/Observation/0"
]
}
}
diff --git a/CDB/integrations/multiobs_ConfigDb.json b/CDB/integrations/multiobs_ConfigDb.json
index e3cf4855d..ed52f2202 100644
--- a/CDB/integrations/multiobs_ConfigDb.json
+++ b/CDB/integrations/multiobs_ConfigDb.json
@@ -1,6 +1,6 @@
{
"servers": {
- "Observation": {
+ "ObservationControl": {
"STAT": {
"Observation": {
"STAT/Observation/1": {},
diff --git a/CDB/stations/common.json b/CDB/stations/common.json
index 9d77c0392..45768cf24 100644
--- a/CDB/stations/common.json
+++ b/CDB/stations/common.json
@@ -85,23 +85,14 @@
}
}
},
- "Observation": {
- "STAT": {
- "Observation": {
- "STAT/Observation/0": {}
- }
- }
- },
"ObservationControl": {
"STAT": {
"ObservationControl": {
"STAT/ObservationControl/1": {
"properties": {
"Power_Children": [
- "STAT/Observation/0"
],
"Control_Children": [
- "STAT/Observation/0"
]
}
}
diff --git a/CDB/stations/testenv_cs001.json b/CDB/stations/testenv_cs001.json
index cb0c1bc74..0cfb260ea 100644
--- a/CDB/stations/testenv_cs001.json
+++ b/CDB/stations/testenv_cs001.json
@@ -1120,24 +1120,19 @@
}
}
},
- "Observation": {
- "STAT": {
- "Observation": {
- "STAT/Observation/1": {}
- }
- }
- },
"ObservationControl": {
"STAT": {
+ "Observation": {
+ "STAT/Observation/1": {
+ }
+ },
"ObservationControl": {
"STAT/ObservationControl/1": {
"properties": {
"Power_Children": [
- "STAT/Observation/0",
"STAT/Observation/1"
],
"Control_Children": [
- "STAT/Observation/0",
"STAT/Observation/1"
]
}
diff --git a/CDB/test_environment_ConfigDb.json b/CDB/test_environment_ConfigDb.json
deleted file mode 100644
index 49216e563..000000000
--- a/CDB/test_environment_ConfigDb.json
+++ /dev/null
@@ -1,22 +0,0 @@
-{
- "servers": {
- "TileBeam": {
- "STAT": {
- "TileBeam": {
- "STAT/TileBeam/HBA": {
- "properties": {
- "Tracking_enabled_RW_default": [ "True" ]
- }
- }
- }
- }
- },
- "Observation": {
- "STAT": {
- "Observation": {
- "STAT/Observation/1": {}
- }
- }
- }
- }
-}
diff --git a/docker-compose/device-observation.yml b/docker-compose/device-observation.yml
deleted file mode 100644
index d016f2071..000000000
--- a/docker-compose/device-observation.yml
+++ /dev/null
@@ -1,53 +0,0 @@
-# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
-# SPDX-License-Identifier: Apache-2.0
-#
-# Docker compose file that launches a LOFAR2.0 station's
-# Observation device.
-#
-# Defines:
-# - device-observation: LOFAR2.0 station Observation
-#
-# Requires:
-# - lofar-device-base.yml
-#
-version: '2.1'
-
-services:
- device-observation:
- image: ${LOCAL_DOCKER_REGISTRY_HOST}/${LOCAL_DOCKER_REGISTRY_USER}/lofar-device-base:${TAG}
- hostname: device-observation
- container_name: device-observation
- logging:
- driver: "json-file"
- options:
- max-size: "100m"
- max-file: "10"
- networks:
- - control
- ports:
- - "5718:5718" # unique port for this DS
- - "5818:5818" # ZeroMQ event port
- - "5918:5918" # ZeroMQ heartbeat port
- extra_hosts:
- - "host.docker.internal:host-gateway"
- volumes:
- - ..:/opt/lofar/tango:rw
- environment:
- - TANGO_HOST=${TANGO_HOST}
- - TANGO_ZMQ_EVENT_PORT=5818
- - TANGO_ZMQ_HEARTBEAT_PORT=5918
- healthcheck:
- test: nc -z -v device-observation 5718
- interval: 1m
- timeout: 30s
- retries: 3
- start_period: 30s
- working_dir: /opt/lofar/tango
- entrypoint:
- - bin/start-ds.sh
- # configure CORBA to _listen_ on 0:port, but tell others we're _reachable_ through ${HOSTNAME}:port, since CORBA
- # can't know about our Docker port forwarding
- - l2ss-observation Observation STAT -v -ORBendPoint giop:tcp:0:5718 -ORBendPointPublish giop:tcp:${HOSTNAME}:5718
- restart: on-failure
- stop_signal: SIGINT # request a graceful shutdown of Tango
- stop_grace_period: 2s
diff --git a/docker-compose/http-json-schemas/definitions/observation-settings.json b/docker-compose/http-json-schemas/definitions/observation-settings.json
index 2162b6007..50a6590f3 100644
--- a/docker-compose/http-json-schemas/definitions/observation-settings.json
+++ b/docker-compose/http-json-schemas/definitions/observation-settings.json
@@ -14,10 +14,21 @@
"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": {
"default": "HBA",
"description": "Antenna field to use",
diff --git a/sbin/run_integration_test.sh b/sbin/run_integration_test.sh
index 62a878b30..2a908c163 100755
--- a/sbin/run_integration_test.sh
+++ b/sbin/run_integration_test.sh
@@ -89,7 +89,7 @@ echo '/usr/local/bin/wait-for-it.sh ${TANGO_HOST} --strict --timeout=300 -- true
# Devices list is used to explitly word split when supplied to commands, must
# disable shellcheck SC2086 for each case.
-DEVICES=(device-station-manager device-boot device-aps device-apsct device-ccd device-ec device-apspu device-sdpfirmware device-sdp device-recvh device-recvl device-bst device-sst device-unb2 device-xst device-beamlet device-digitalbeam device-tilebeam device-psoc device-pcon device-antennafield device-temperature-manager device-observation device-observation-control device-configuration device-calibration)
+DEVICES=(device-station-manager device-boot device-aps device-apsct device-ccd device-ec device-apspu device-sdpfirmware device-sdp device-recvh device-recvl device-bst device-sst device-unb2 device-xst device-beamlet device-digitalbeam device-tilebeam device-psoc device-pcon device-antennafield device-temperature-manager device-observation-control device-configuration device-calibration)
SIMULATORS=(sdptr-sim recvh-sim recvl-sim unb2-sim apsct-sim apspu-sim ccd-sim ec-sim)
diff --git a/tangostationcontrol/docs/source/observing.rst b/tangostationcontrol/docs/source/observing.rst
index 7a871f1ab..a207accf0 100644
--- a/tangostationcontrol/docs/source/observing.rst
+++ b/tangostationcontrol/docs/source/observing.rst
@@ -10,7 +10,9 @@ To observe with a station, you must construct the observation's specifications,
observation_spec = {
"observation_id": 12345,
- "stop_time": "2106-02-07T00:00:00",
+ "start_time": "2106-02-07T00:00:00",
+ "stop_time": "2106-02-07T01:00:00",
+ "antenna_field": "HBA",
"antenna_set": "ALL",
"filter": "HBA_210_250",
"SAPs": [{
@@ -22,7 +24,7 @@ To observe with a station, you must construct the observation's specifications,
import json
obs_control = DeviceProxy("STAT/ObservationControl/1")
- obs_control.start_observation(json.dumps(observation_spec))
+ obs_control.add_observation(json.dumps(observation_spec))
The above specification contains the following parameters:
@@ -31,7 +33,11 @@ The above specification contains the following parameters:
+====================+=========================================================================================+
| ``observation_id`` | User-specified unique reference to this observation. |
+--------------------+-----------------------------------------------------------------------------------------+
-| ``stop_time`` | Automatically stop observing when this timestamp is reached. |
+| ``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). |
+--------------------+-----------------------------------------------------------------------------------------+
@@ -42,7 +48,7 @@ The above specification contains the following parameters:
| ``tile_beam`` | Pointing to track with the HBA tiles (optional). |
+--------------------+-----------------------------------------------------------------------------------------+
-This will configure the antenna field ``HBA`` as follows:
+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,
@@ -61,30 +67,39 @@ The effect of the observations can be observed through the following means, all
Life cycle
````````````````````````
-The observation will start the moment it enters the system, and stop at the specified time. Each antenna field can have one observation running at a time.
+The ObservationControl device will start each Observation when its start time is reached or past, and will stop it at the specified stop time. You can also force this to happen::
+
+ obs_control = DeviceProxy("STAT/ObservationControl/1")
+ obs_control.start_observation_now(12345) # starts observation 12345 now, regardless of its specified start time
+ obs_control.stop_observation_now(12345) # stops observation 12345 now, regardless of its specified stop time
Managing observation(s)
-------------------------
To manage running observations, we can interact with ObservationControl::
+
+ >>> # Check which observations are known (running or yet to run)
+ >>> obs_control.observations_R
+ array([12345])
+
>>> # Check which observations are running
>>> obs_control.running_observations_R
array([12345])
>>> # Stop a running observation
- >>> obs_control.stop_observation(12345)
+ >>> obs_control.stop_observation_now(12345)
>>> # Stop all running observations
- >>> obs_control.stop_all_observations()
+ >>> obs_control.stop_all_observations_now()
Alternatively, we can inspect a running observation more closely. Each observation is represented by its own device: ``STAT/Observation/$id``, so if observation 12345 has been started, we can do the following::
observation = DeviceProxy("STAT/Observation/12345")
-This device exposes several attributes:
+This device exposes its settings as individual attributes, as well as:
-:observation_running_R: Ever-increasing value as long as the observation is running. Allows one to check whether monitoring has become stale.
+:alive_R: Ever-increasing value as long as the observation is running. Allows one to check whether monitoring has become stale.
:type: ``int``
diff --git a/tangostationcontrol/integration_test/configuration/configDB/test_environment_ConfigDb.json b/tangostationcontrol/integration_test/configuration/configDB/test_environment_ConfigDb.json
index 49216e563..17ce2a7f5 100644
--- a/tangostationcontrol/integration_test/configuration/configDB/test_environment_ConfigDb.json
+++ b/tangostationcontrol/integration_test/configuration/configDB/test_environment_ConfigDb.json
@@ -11,7 +11,7 @@
}
}
},
- "Observation": {
+ "ObservationControl": {
"STAT": {
"Observation": {
"STAT/Observation/1": {}
diff --git a/tangostationcontrol/integration_test/configuration/test_device_configuration.py b/tangostationcontrol/integration_test/configuration/test_device_configuration.py
index ef6bdc38c..78c3044b9 100644
--- a/tangostationcontrol/integration_test/configuration/test_device_configuration.py
+++ b/tangostationcontrol/integration_test/configuration/test_device_configuration.py
@@ -33,7 +33,7 @@ class TestDeviceConfiguration(AbstractTestBases.TestDeviceBase):
}
}
},
- "Observation": {
+ "ObservationControl": {
"STAT": {
"Observation": {
"STAT/Observation/11": {}
@@ -149,7 +149,7 @@ class TestDeviceConfiguration(AbstractTestBases.TestDeviceBase):
) # configuration device
self.assertFalse(
"stat/observation/11"
- in dbdata["servers"]["observation"]["stat"]["observation"],
+ in dbdata["servers"]["observationcontrol"]["stat"]["observation"],
msg=f"{dbdata}",
) # observation device
self.assertTrue(
@@ -184,7 +184,9 @@ class TestDeviceConfiguration(AbstractTestBases.TestDeviceBase):
# Test whether new device has been added
self.assertTrue(
"stat/observation/11"
- in updated_dbdata["servers"]["observation"]["stat"]["observation"],
+ in updated_dbdata["servers"]["observationcontrol"]["stat"][
+ "observation"
+ ],
msg=f"{updated_dbdata}",
) # observation device
# Test whether old device has been deleted
@@ -240,7 +242,7 @@ class TestDeviceConfiguration(AbstractTestBases.TestDeviceBase):
# Test whether new device has been added
self.assertTrue(
"stat/observation/11"
- in updated_dbdata["servers"]["observation"]["stat"]["observation"],
+ in updated_dbdata["servers"]["observationcontrol"]["stat"]["observation"],
msg=f"{updated_dbdata}",
) # observation device
# Test whether old device has NOT been deleted
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_observation.py b/tangostationcontrol/integration_test/default/devices/test_device_observation.py
index 634b0f222..b36350447 100644
--- a/tangostationcontrol/integration_test/default/devices/test_device_observation.py
+++ b/tangostationcontrol/integration_test/default/devices/test_device_observation.py
@@ -178,15 +178,29 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
self.proxy.Initialise()
self.assertEqual(DevState.STANDBY, self.proxy.state())
- def test_init_invalid(self):
+ def test_init_no_settings(self):
"""Initialize an observation with _invalid_ JSON"""
self.proxy.off()
+
+ # Cannot start without valid settings
with self.assertRaises(DevFailed):
- self.proxy.observation_settings_RW = "{}"
self.proxy.Initialise()
+
+ # Since initialisation did not succeed, we're still in OFF
self.assertEqual(DevState.FAULT, self.proxy.state())
+ def test_init_invalid(self):
+ """Initialize an observation with _invalid_ JSON"""
+
+ self.proxy.off()
+
+ # Cannot write invalid settings
+ with self.assertRaises(DevFailed):
+ self.proxy.observation_settings_RW = "{}"
+
+ self.assertEqual(DevState.OFF, self.proxy.state())
+
def test_prohibit_rewriting_settings(self):
"""Test that changing observation settings is disallowed once init"""
@@ -205,9 +219,8 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
stop_timestamp = datetime.fromisoformat(data["stop_time"]).timestamp()
observation_id = data["observation_id"]
antenna_set = data["antenna_set"]
- filter = data["filter"]
- num_saps = len(data["SAPs"])
- saps_subband = [data["SAPs"][i]["subbands"] for i in range(0, num_saps)]
+ filter_ = data["filter"]
+ saps_subband = data["SAPs"][0]["subbands"]
pointing_direction = data["SAPs"][0]["pointing"]
saps_pointing = [
(
@@ -232,7 +245,7 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
self.assertEqual(stop_timestamp, self.proxy.stop_time_R)
self.assertEqual(observation_id, self.proxy.observation_id_R)
self.assertEqual(antenna_set, self.proxy.antenna_set_R)
- self.assertEqual(filter, self.proxy.filter_R)
+ 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))
diff --git a/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py b/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
index 1e53d4647..b97cdf03f 100644
--- a/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
+++ b/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
@@ -169,9 +169,7 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
parameters["observation_id"] = -1
self.on_device_assert(self.proxy)
- self.assertRaises(
- DevFailed, self.proxy.start_observation, json.dumps(parameters)
- )
+ self.assertRaises(DevFailed, self.proxy.add_observation, json.dumps(parameters))
def test_check_and_convert_parameters_invalid_time(self):
"""Test invalid parameter detection"""
@@ -180,29 +178,44 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
parameters["stop_time"] = (datetime.now() - timedelta(seconds=1)).isoformat()
self.on_device_assert(self.proxy)
- self.assertRaises(
- DevFailed, self.proxy.start_observation, json.dumps(parameters)
- )
+ self.assertRaises(DevFailed, self.proxy.add_observation, json.dumps(parameters))
def test_check_and_convert_parameters_invalid_empty(self):
"""Test empty parameter detection"""
self.on_device_assert(self.proxy)
- self.assertRaises(DevFailed, self.proxy.start_observation, "{}")
+ self.assertRaises(DevFailed, self.proxy.add_observation, "{}")
- def test_start_observation(self):
- """Test starting an observation"""
+ def test_add_observation_now(self):
+ """Test starting an observation now"""
self.on_device_assert(self.proxy)
- self.proxy.start_observation(self.VALID_JSON)
+ self.proxy.add_observation(self.VALID_JSON)
self.assertTrue(self.proxy.is_any_observation_running())
self.assertTrue(self.proxy.is_observation_running(12345))
- self.proxy.stop_observation(12345)
+ self.proxy.stop_observation_now(12345)
+
+ def test_add_observation_future(self):
+ """Test starting an observation in the future"""
+
+ self.on_device_assert(self.proxy)
+
+ parameters = json.loads(self.VALID_JSON)
+ parameters["start_time"] = (datetime.now() + timedelta(days=1)).isoformat()
+ parameters["stop_time"] = (datetime.now() + timedelta(days=2)).isoformat()
+ self.proxy.add_observation(json.dumps(parameters))
+
+ self.assertIn(12345, self.proxy.observations_R)
+ self.assertNotIn(12345, self.proxy.running_observations_R)
+ self.assertFalse(self.proxy.is_any_observation_running())
+ self.assertFalse(self.proxy.is_observation_running(12345))
+
+ self.proxy.stop_observation_now(12345)
- def test_start_observation_multiple(self):
+ def test_add_observation_multiple(self):
"""Test starting multiple observations"""
second_observation_json = json.loads(self.VALID_JSON)
@@ -210,49 +223,49 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
self.on_device_assert(self.proxy)
- self.proxy.start_observation(self.VALID_JSON)
- self.proxy.start_observation(json.dumps(second_observation_json))
+ self.proxy.add_observation(self.VALID_JSON)
+ self.proxy.add_observation(json.dumps(second_observation_json))
self.assertTrue(self.proxy.is_any_observation_running())
self.assertTrue(self.proxy.is_observation_running(12345))
self.assertTrue(self.proxy.is_observation_running(54321))
- self.proxy.stop_observation(12345)
- self.proxy.stop_observation(54321)
+ self.proxy.stop_observation_now(12345)
+ self.proxy.stop_observation_now(54321)
def test_stop_observation_invalid_id(self):
"""Test stop_observation exceptions for invalid ids"""
self.on_device_assert(self.proxy)
- self.assertRaises(DevFailed, self.proxy.stop_observation, -1)
+ self.assertRaises(DevFailed, self.proxy.stop_observation_now, -1)
def test_stop_observation_invalid_running(self):
"""Test stop_observation exceptions for not running"""
self.on_device_assert(self.proxy)
- self.assertRaises(DevFailed, self.proxy.stop_observation, 2)
+ self.assertRaises(DevFailed, self.proxy.stop_observation_now, 2)
- def test_is_any_observation_running_after_stop_all_observations(self):
+ def test_is_any_observation_running_after_stop_all_observations_now(self):
"""Test whether is_any_observation_running conforms when we start & stop an observation"""
self.on_device_assert(self.proxy)
- self.proxy.start_observation(self.VALID_JSON)
- self.proxy.stop_all_observations()
+ self.proxy.add_observation(self.VALID_JSON)
+ self.proxy.stop_all_observations_now()
# Test false
self.assertFalse(self.proxy.is_any_observation_running())
- def test_start_stop_observation(self):
+ def test_start_stop_observation_now(self):
"""Test starting and stopping an observation"""
self.on_device_assert(self.proxy)
# uses ID 12345
- self.proxy.start_observation(self.VALID_JSON)
- self.proxy.stop_observation(12345)
+ self.proxy.add_observation(self.VALID_JSON)
+ self.proxy.stop_observation_now(12345)
# Test false
self.assertFalse(self.proxy.is_observation_running(12345))
@@ -266,9 +279,9 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
self.on_device_assert(self.proxy)
# uses ID 12345
- self.proxy.start_observation(self.VALID_JSON)
- self.proxy.start_observation(json.dumps(second_observation_json))
- self.proxy.stop_all_observations()
+ self.proxy.add_observation(self.VALID_JSON)
+ self.proxy.add_observation(json.dumps(second_observation_json))
+ self.proxy.stop_all_observations_now()
# Test false
self.assertFalse(self.proxy.is_observation_running(12345))
@@ -279,6 +292,4 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
parameters = json.loads(self.VALID_JSON)
parameters["antenna_set"] = "ZZZ"
self.on_device_assert(self.proxy)
- self.assertRaises(
- DevFailed, self.proxy.start_observation, json.dumps(parameters)
- )
+ self.assertRaises(DevFailed, self.proxy.add_observation, json.dumps(parameters))
diff --git a/tangostationcontrol/tangostationcontrol/common/observation_controller.py b/tangostationcontrol/tangostationcontrol/common/observation_controller.py
index 9843316f2..360e35351 100644
--- a/tangostationcontrol/tangostationcontrol/common/observation_controller.py
+++ b/tangostationcontrol/tangostationcontrol/common/observation_controller.py
@@ -4,7 +4,6 @@
import logging
import time
from datetime import datetime
-from typing import List
from tango import DevFailed, DevState, Except, Util, EventType, DeviceProxy
from tangostationcontrol.common.lofar_logging import log_exceptions
@@ -14,7 +13,11 @@ from tangostationcontrol.configuration import ObservationSettings
logger = logging.getLogger()
-class RunningObservation(object):
+class Observation(object):
+ @property
+ def proxy(self) -> DeviceProxy:
+ return self._device_proxy
+
@property
def observation_id(self) -> int:
return self._parameters.observation_id
@@ -32,7 +35,7 @@ class RunningObservation(object):
# Name for the Observation.observation_running subscription
@property
def attribute_name(self) -> str:
- return f"{self.device_name}/observation_running_R"
+ return f"{self.device_name}/alive_R"
def __init__(self, tango_domain, parameters: ObservationSettings):
self._device_proxy: DeviceProxy | None = None
@@ -44,7 +47,7 @@ class RunningObservation(object):
# have one instance of it.
self._tango_util: Util = Util.instance()
- def create_tango_device(self):
+ def create_observation_device(self):
"""Instatiate an Observation Device"""
logger.info("Create device: %s", self.device_name)
try:
@@ -70,28 +73,41 @@ class RunningObservation(object):
logger.exception(error_string)
Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
- def create_device_proxy(self):
+ def destroy_observation_device(self):
+ try:
+ self._tango_util.delete_device(self.class_name, self.device_name)
+ except DevFailed:
+ logger.exception(
+ f"Could not delete device {self.device_name} of class {self.class_name} from Tango DB."
+ )
+
+ def initialise_observation(self):
# Instantiate a dynamic Tango Device "Observation".
self._device_proxy = create_device_proxy(self.device_name)
+ # Initialise generic properties
+ self.proxy.put_property({"Control_Children": [], "Power_Children": []})
+
# Configure the dynamic device its attribute for the observation
# parameters.
- self._device_proxy.observation_settings_RW = self._parameters.to_json()
+ self.proxy.observation_settings_RW = self._parameters.to_json()
# Take the Observation device through the motions. Pass the
# entire JSON set of parameters so that it can pull from it what it
# needs.
- self._device_proxy.Initialise()
+ self.proxy.Initialise()
- # The call to On will actually tell the Observation device to
- # become fully active.
- self._device_proxy.On()
+ def start(self):
+ self.proxy.On()
+
+ def is_running(self):
+ return self.proxy and self.proxy.state() == DevState.ON
def subscribe(self, cb):
# Turn on the polling for the attribute.
# Note that this is not automatically done despite the attribute
# having the right polling values set in the ctor.
- self._device_proxy.poll_attribute(self.attribute_name.split("/")[-1], 1000)
+ self.proxy.poll_attribute(self.attribute_name.split("/")[-1], 1000)
# Right. Now subscribe to periodic events.
self._event_id = self._device_proxy.subscribe_event(
@@ -101,92 +117,47 @@ class RunningObservation(object):
"Successfully started an observation with ID=%s.", self.observation_id
)
- def shutdown(self):
+ def stop(self):
# Check if the device has not terminated itself in the meanwhile.
try:
- self._device_proxy.ping()
+ self.proxy.ping()
except DevFailed:
- logger.warning(
- "The device for the Observation with ID=%s \
- has unexpectedly already disappeared. It is advised to check \
- the logs up to 10s prior to this message to see what happened.",
- {self.observation_id},
+ logger.error(
+ f"Observation device for ID={self.observation_id} unexpectedly disappeared."
)
else:
# Unsubscribe from the subscribed event.
- event_id = self._event_id
- self._device_proxy.unsubscribe_event(event_id)
+ self.proxy.unsubscribe_event(self._event_id)
# Tell the Observation device to stop the running
# observation. This is a synchronous call and the clean-up
# does not take long.
- self._device_proxy.Off()
-
- # Wait for 1s for the Observation device to go to
- # DevState.OFF. Force shutdown if observation.state() is
- # not OFF.
- remaining_wait_time = 1.0
- sleep_time = 0.1
- stopped = False
- while remaining_wait_time > 0.0:
- if self._device_proxy.state() is DevState.OFF:
- stopped = True
- break
- time.sleep(sleep_time)
- remaining_wait_time = remaining_wait_time - sleep_time
- # Check if the observation object is really in OFF state.
- if stopped:
- logger.info(
- "Successfully stopped the observation with ID=%s",
- {self.observation_id},
- )
- else:
- logger.warning(
- "Could not shut down the Observation device ( %s ) \
- for observation ID=%s. This means that there is a \
- chance for a memory leak. Will continue anyway and forcefully delete \
- the Observation object.",
- {self.device_name},
- {self.observation_id},
- )
+ self.proxy.Off()
# Finally remove the device object from the Tango DB.
- try:
- self._tango_util.delete_device(self.class_name, self.device_name)
- except DevFailed:
- logger.warning(
- "Something went wrong when the device %s \
- was removed from the Tango DB. There is nothing that can be done \
- about this here at this moment but you should check the Tango DB yourself.",
- {self.device_name},
- )
+ self.destroy_observation_device()
-class ObservationController(object):
- @property
- def running_observations(self) -> List[int]:
- return list(self._running_observations.keys())
+class ObservationController(dict[str, Observation]):
+ """A dictionary of observations. Actively manages the observation state transtions
+ (start, stop)."""
def __init__(self, tango_domain: str):
self._tango_util = Util.instance()
self._tango_domain = tango_domain
- self._running_observations: dict[int, RunningObservation] = {}
- def is_any_observation_running(self):
- return len(self._running_observations) > 0
-
- def is_observation_running(self, obs_id):
- observation = self._running_observations.get(obs_id)
- return observation is not None
+ @property
+ def running_observations(self) -> list[str]:
+ return [obs_id for obs_id, obs in self.items() if obs.is_running()]
@log_exceptions()
- def observation_running_callback(self, event):
+ def observation_callback(self, event):
"""
- This callback checks if a running observation is still
- supposed to run. If this function finds out that the
- observation is not supposed to run any more, then
- self.stop_observation(obs_id) is called which takes care of the
- clean up.
+ This callback checks and manages the state transitions
+ for each observation.
+
+ It starts observations at their specified start_time,
+ and stops & removes them at their specified stop_time.
"""
if event.err:
# Something is fishy with this event.
@@ -194,108 +165,98 @@ class ObservationController(object):
"The Observation device %s sent an event but the event \
signals an error. It is advised to check the logs for any indication \
that something went wrong in that device. Event data=%s",
- {event.device},
- {event},
+ event.device,
+ event,
)
return
+ # update the state of this observation, if needed
+ self._update_observation_state(event.device)
+
+ def _update_observation_state(self, device: DeviceProxy):
+ """Start or stop the observation managed by the given Observation
+ device."""
+
# Get the Observation ID from the sending device.
- obs_id = event.device.observation_id_R
+ obs_id = device.observation_id_R
- # Check if the observation is still supposed to run.
- running_obs = self._running_observations.copy()
- if not running_obs:
- # No obs is running???
- logger.warning(
- "Received an observation_running event for the observation with ID=%s. \
- According to the records in ObservationControl, this observation is \
- not supposed to run. Please check previous logs, especially around the time \
- an observation with this ID was started. Will continue and ignore this event.",
- {obs_id},
- )
- return
+ # Get the start/stop times from the sending device
+ obs_start_time = device.start_time_R
+ obs_stop_time = device.stop_time_R
- if obs_id in running_obs:
- # Get the Observation's stop_time from the Observation device.
- obs_stop_time = event.device.stop_time_R
- current_obs_time = event.attr_value.value
- # I expect that current_obs_time is always
- # smaller than the stop time that I read from my
- # records.
- delta_t = obs_stop_time - current_obs_time
- if delta_t < 0.0:
- # The observation has not finished yet and is
- # more than 1.0 seconds past its scheduled stop
- # time. Tell the observation to finish and clean up.
- obs = running_obs[obs_id]
- self.stop_observation(obs_id)
- else:
- # The observation that we are trying to process is not part
- # of the running_obs dictionary
- logger.warning(
- "Received an observation_running event for the observation with ID=%s. \
- According to the records in ObservationControl, this observation is \
- not supposed to run. Please check previous logs, especially around \
- the time an observation with this ID was started. \
- Will continue and ignore this event.",
- {obs_id},
- )
- return
+ # Get how much earlier we have to start
+ obs_lead_time = device.lead_time_R
+
+ # Obtain the current time ONCE to avoid race conditions
+ now = time.time()
+
+ # Manage state transitions
+ if now > obs_stop_time:
+ # Stop observation
+ self.stop_observation_now(obs_id)
+ elif (
+ now >= obs_start_time - obs_lead_time and device.state() == DevState.STANDBY
+ ):
+ # Start observation
+ self.start_observation(obs_id)
- def start_observation(self, settings: ObservationSettings):
+ def add_observation(self, settings: ObservationSettings):
"""Create a new Observation Device and start an observation"""
# Check further properties that cannot be validated through a JSON schema
if settings.stop_time <= datetime.now():
- error = f'Cannot start an observation with ID={settings.observation_id} \
- because the parameter stop_time parameter value="{settings.stop_time}" \
- is invalid. Set a stop_time parameter later in time than the start time.'
- Except.throw_exception("IllegalCommand", error, __name__)
+ raise ValueError(
+ f"Cannot start observation {settings.observation_id} because it is already past its stop time {settings.stop_time}"
+ )
- obs = RunningObservation(self._tango_domain, settings)
- obs.create_tango_device()
+ obs = Observation(self._tango_domain, settings)
+ obs.create_observation_device()
try:
- obs.create_device_proxy()
+ obs.initialise_observation()
except DevFailed as ex:
# Remove the device again.
- self._tango_util.delete_device(obs.class_name, obs.device_name)
- error_string = f"Cannot access the Observation device instance for observation \
- ID={obs.observation_id} with device class name={obs.class_name} and \
- device instance name={obs.device_name}. \
- This means that the observation likely did not start \
- but certainly cannot be controlled and/or forcefully be stopped."
- logger.exception(error_string)
- Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
-
- # Finally update the self.running_observation dict's entry of this
- # observation with the complete set of info.
- self._running_observations[obs.observation_id] = obs
+ obs.destroy_observation_device()
+
+ raise Exception(
+ "Failed to initialise observation {settings.observation_id}"
+ ) from ex
+
+ # Register this observation now that is has been succesfully created
+ self[obs.observation_id] = obs
+ # Keep pinging it to manage its state transitions
try:
- obs.subscribe(self.observation_running_callback)
+ obs.subscribe(self.observation_callback)
except DevFailed as ex:
- self._tango_util.delete_device(obs.class_name, obs.device_name)
- error_string = "Cannot access the Observation device instance for observation ID=%s \
- with device class name=%s and device instance name=%s. This means that the observation \
- cannot be controlled and/or forcefully be stopped."
- logger.exception(
- error_string, obs.observation_id, obs.class_name, obs.device_name
- )
- Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
+ # Remove the device again.
+ obs.destroy_observation_device()
- def stop_observation(self, obs_id):
+ raise Exception(
+ "Cannot subscribe to attribute. Cancelled observation {settings.observation_id}."
+ ) from ex
+
+ # Make sure the current state is accurate
+ self._update_observation_state(obs.proxy)
+
+ def start_observation(self, obs_id):
+ """Start the observation with the given ID"""
+ try:
+ observation = self[obs_id]
+ except KeyError as _exc:
+ raise Exception(f"Unknown observation: {obs_id}")
+
+ observation.start()
+
+ def stop_observation_now(self, obs_id):
"""Stop the observation with the given ID"""
- if self.is_observation_running(obs_id) is False:
- error = f"Cannot stop an observation with ID={obs_id}, \
- because the observation is not running."
- Except.throw_exception("IllegalCommand", error, __name__)
-
- # Fetch the obs data and remove it from the dict of
- # currently running observations.
- observation = self._running_observations.pop(obs_id)
- observation.shutdown()
-
- def stop_all_observations(self):
- """Stop all running observations"""
- for obs_id in self._running_observations.copy().keys():
- self.stop_observation(obs_id)
+ try:
+ observation = self.pop(obs_id)
+ except KeyError as _exc:
+ raise Exception(f"Unknown observation: {obs_id}")
+
+ observation.stop()
+
+ def stop_all_observations_now(self):
+ """Stop all observations (running or to be run)"""
+ for obs_id in list(self): # draw a copy as we modify the list
+ self.stop_observation_now(obs_id)
diff --git a/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
index 4e8c4e5df..b8f49369f 100644
--- a/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
+++ b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
@@ -13,15 +13,18 @@ class ObservationSettings(_ConfigurationBase):
def __init__(
self,
observation_id: int,
+ start_time: datetime | None,
stop_time: datetime,
antenna_field: str,
antenna_set: str,
filter: str,
SAPs: Sequence[Sap],
- tile_beam: Pointing = None,
+ tile_beam: Pointing | None = None,
first_beamlet: int = 0,
+ lead_time: float | None = None,
):
self.observation_id = observation_id
+ self.start_time = start_time
self.stop_time = stop_time
self.antenna_field = antenna_field
self.antenna_set = antenna_set
@@ -29,10 +32,13 @@ class ObservationSettings(_ConfigurationBase):
self.SAPs = SAPs
self.tile_beam = tile_beam
self.first_beamlet = first_beamlet
+ self.lead_time = lead_time
def __iter__(self):
+ yield "observation_id", self.observation_id
+ if self.start_time:
+ yield "start_time", self.start_time.isoformat()
yield from {
- "observation_id": self.observation_id,
"stop_time": self.stop_time.isoformat(),
"antenna_field": self.antenna_field,
"antenna_set": self.antenna_set,
@@ -42,11 +48,16 @@ class ObservationSettings(_ConfigurationBase):
if self.tile_beam:
yield "tile_beam", dict(self.tile_beam)
yield "first_beamlet", self.first_beamlet
+ if self.lead_time is not None:
+ yield "lead_time", self.lead_time
@staticmethod
def to_object(json_dct) -> "ObservationSettings":
return ObservationSettings(
json_dct["observation_id"],
+ datetime.fromisoformat(json_dct["start_time"])
+ if "start_time" in json_dct
+ else None,
datetime.fromisoformat(json_dct["stop_time"]),
json_dct["antenna_field"],
json_dct["antenna_set"],
@@ -54,4 +65,5 @@ class ObservationSettings(_ConfigurationBase):
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,
)
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation.py b/tangostationcontrol/tangostationcontrol/devices/observation.py
index af767e7cc..471f0a1fc 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation.py
@@ -1,6 +1,7 @@
# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
+from itertools import chain
import logging
from time import time
from typing import Optional
@@ -9,7 +10,7 @@ import numpy
from jsonschema.exceptions import ValidationError
# PyTango imports
-from tango import AttrWriteType, DeviceProxy, DevState, Util, Except
+from tango import AttrWriteType, DeviceProxy, DevState, Util
from tango.server import attribute
from tangostationcontrol.common.constants import (
DEFAULT_POLLING_PERIOD,
@@ -25,7 +26,6 @@ from tangostationcontrol.common.proxy import create_device_proxy
from tangostationcontrol.configuration import ObservationSettings
from tangostationcontrol.devices.device_decorators import fault_on_error
from tangostationcontrol.devices.device_decorators import only_in_states
-from tangostationcontrol.devices.device_decorators import only_when_on
from tangostationcontrol.devices.base_device_classes.lofar_device import LOFARDevice
logger = logging.getLogger()
@@ -40,38 +40,136 @@ class Observation(LOFARDevice):
specific observation. It will, if necessary keep tabs on HW MPs to signal
issues that are not caught by MPs being outside their nominal range.
- The lifecycle of instances of this device is controlled by ObservationControl
+ The lifecycle of instances of this device is controlled by ObservationControl.
+
+ Settings are written to observation_settings_RW, which can only be done
+ in the OFF state.
"""
# Attributes
- observation_running_R = attribute(
+ @attribute(
+ doc="Return a value that changes over time, to check if the device is alive.",
dtype=numpy.float64,
- access=AttrWriteType.READ,
polling_period=DEFAULT_POLLING_PERIOD,
period=DEFAULT_POLLING_PERIOD,
rel_change="1.0",
)
- observation_id_R = attribute(dtype=numpy.int64, access=AttrWriteType.READ)
- stop_time_R = attribute(dtype=numpy.float64, access=AttrWriteType.READ)
- antenna_field_R = attribute(dtype=str, access=AttrWriteType.READ)
- antenna_set_R = attribute(dtype=str, access=AttrWriteType.READ)
- filter_R = attribute(dtype=str, access=AttrWriteType.READ)
- saps_subband_R = attribute(
- dtype=((numpy.uint32,),),
+ def alive_R(self):
+ return time()
+
+ @attribute(
+ doc="Observation ID",
+ dtype=numpy.int64,
+ )
+ def observation_id_R(self):
+ return self._observation_settings.observation_id
+
+ @attribute(
+ doc="Observation start time (seconds since 1970), or 0 if immediate.",
+ unit="s",
+ dtype=numpy.float64,
+ )
+ def start_time_R(self):
+ return (
+ self._observation_settings.start_time.timestamp()
+ if self._observation_settings.start_time
+ else 0.0
+ )
+
+ @attribute(
+ doc="Observation stop time (seconds since 1970).",
+ unit="s",
+ dtype=numpy.float64,
+ )
+ def stop_time_R(self):
+ return self._observation_settings.stop_time.timestamp()
+
+ @attribute(
+ doc="Seconds to be on sky before the observation start time, to fill buffers and allow for negative geometric delay compensation downstream.",
+ unit="s",
+ dtype=numpy.float64,
+ )
+ def lead_time_R(self):
+ return self._observation_settings.lead_time or 0.0
+
+ @attribute(
+ doc="Which antenna field this Observation configures",
+ dtype=str,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def antenna_field_R(self):
+ return self._observation_settings.antenna_field
+
+ @attribute(
+ doc="Which antenna set this Observation configures",
+ dtype=str,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def antenna_set_R(self):
+ return self._observation_settings.antenna_set
+
+ @attribute(
+ doc="Which band filter to use for all antennas",
+ dtype=str,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def filter_R(self):
+ return self._observation_settings.filter
+
+ @attribute(
+ doc="Which subbands to beamform for each beamlet.",
+ dtype=(numpy.uint32,),
max_dim_x=N_beamlets_ctrl,
- max_dim_y=N_beamlets_ctrl,
- access=AttrWriteType.READ,
+ fisallowed="is_attribute_access_allowed",
)
- saps_pointing_R = attribute(
+ def saps_subband_R(self):
+ return numpy.array(
+ list(chain(*[sap.subbands for sap in self._observation_settings.SAPs])),
+ dtype=numpy.uint32,
+ )
+
+ @attribute(
+ doc="Which pointing to beamform towards for each beamlet.",
dtype=((str,),),
max_dim_x=N_point_prop,
max_dim_y=N_beamlets_ctrl,
- access=AttrWriteType.READ,
)
- tile_beam_R = attribute(
- dtype=(str,), max_dim_x=N_point_prop, access=AttrWriteType.READ
+ def saps_pointing_R(self):
+ saps_pointing = []
+ for sap in self._observation_settings.SAPs:
+ for _ in sap.subbands:
+ saps_pointing.append(
+ (
+ sap.pointing.direction_type,
+ f"{sap.pointing.angle1}deg",
+ f"{sap.pointing.angle2}deg",
+ ),
+ )
+ return saps_pointing
+
+ @attribute(
+ doc="Which pointing to beamform all HBA tiles to (if any).",
+ dtype=(str,),
+ max_dim_x=N_point_prop,
)
- first_beamlet_R = attribute(dtype=numpy.int64, access=AttrWriteType.READ)
+ def tile_beam_R(self):
+ if self._observation_settings.tile_beam is None:
+ return None
+
+ pointing_direction = self._observation_settings.tile_beam
+ return [
+ str(pointing_direction.direction_type),
+ f"{pointing_direction.angle1}deg",
+ f"{pointing_direction.angle2}deg",
+ ]
+
+ @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
observation_settings_RW = attribute(dtype=str, access=AttrWriteType.READ_WRITE)
@@ -92,30 +190,67 @@ class Observation(LOFARDevice):
def configure_for_initialise(self):
"""Load the JSON from the attribute and configure member variables"""
+ if self._observation_settings is None:
+ raise RuntimeError("Device can not be initialized without configuration")
+
super().configure_for_initialise()
- if self._observation_settings is None:
- Except.throw_exception(
- "IllegalCommand",
- "Device can not be initialized without configuration",
- __name__,
- )
+ logger.info(
+ f"Initialising observation with ID={self._observation_settings.observation_id} "
+ f"with settings: {self._observation_settings.to_json()}"
+ )
+
+ self._prepare_observation()
+
+ logger.info(
+ f"The observation with ID={self._observation_settings.observation_id} "
+ f"is initialised to run between {self._observation_settings.start_time} "
+ f"and {self._observation_settings.stop_time}."
+ )
+
+ def configure_for_on(self):
+ """Indicate the observation has started"""
+
+ super().configure_for_on()
+
+ self._start_observation()
+
+ logger.info(
+ f"Started the observation with ID={self._observation_settings.observation_id}."
+ )
+
+ def configure_for_off(self):
+ """Indicate the observation has stopped"""
+
+ super().configure_for_off()
+
+ self._stop_observation()
+
+ logger.info(
+ "Stopped the observation with ID=%s.",
+ {
+ self._observation_settings.observation_id
+ if self._observation_settings
+ else None
+ },
+ )
+
+ @log_exceptions()
+ def _prepare_observation(self):
+ """Setup proxies and other preparations that can be done ahead of
+ the observation start time without changing any station
+ settings."""
# ObservationControl takes already good care of checking that the
# parameters are in order and sufficient. It is therefore unnecessary
# at the moment to check the parameters here again.
# This could change when the parameter check becomes depending on
# certain aspects that only an Observation device can know.
- self._num_saps = len(self._observation_settings.SAPs)
- self._saps_pointing = self._build_saps_pointing(self._observation_settings)
antennafield = self._observation_settings.antenna_field
- # Set a reference of AntennaField device that is correlated to this device
util = Util.instance()
- # TODO(Stefano): set a proper policy for the devices instance number
- # It cannot be inherited from the Observation instance number
- # (i.e. Observation_id)
+
self.antennafield_proxy = create_device_proxy(
f"{util.get_ds_inst_name()}/AntennaField/{antennafield}"
)
@@ -130,140 +265,52 @@ class Observation(LOFARDevice):
f"{util.get_ds_inst_name()}/DigitalBeam/{antennafield}"
)
- if self._observation_settings.tile_beam:
+ if self._has_tilebeam():
# 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}"
)
- logger.info(
- "The observation with ID=%s is configured."
- "It will begin as soon as On() is called and it is supposed to stop at %s.",
- self._observation_settings.observation_id,
- self._observation_settings.stop_time,
- )
-
- def configure_for_off(self):
- """Indicate the observation has stopped"""
-
- super().configure_for_off()
-
- logger.info(
- "Stopped the observation with ID=%s.",
- {
- self._observation_settings.observation_id
- if self._observation_settings
- else None
- },
- )
-
- def configure_for_on(self):
- """Indicate the observation has started"""
-
- super().configure_for_on()
+ @log_exceptions()
+ def _start_observation(self):
+ """Configure the station for this observation."""
# Apply ObservationID
self.antennafield_proxy.FPGA_sdp_info_observation_id_RW = (
- self._apply_observation_id(self._observation_settings.observation_id)
+ self._apply_observation_id(self.read_attribute("observation_id_R"))
)
# Apply Antenna Set
- self.digitalbeam_proxy.Antenna_Set_RW = self._observation_settings.antenna_set
+ self.digitalbeam_proxy.Antenna_Set_RW = self.read_attribute("antenna_set_R")
- (frequency_band,) = self._apply_antennafield_settings(self.read_filter_R())
+ (frequency_band,) = self._apply_antennafield_settings(
+ self.read_attribute("filter_R")
+ )
self.antennafield_proxy.Frequency_Band_RW = frequency_band
# Apply Beamlet configuration
self.beamlet_proxy.subband_select_RW = self._apply_saps_subbands(
- self.read_saps_subband_R()
+ self.read_attribute("saps_subband_R")
)
self.digitalbeam_proxy.Pointing_direction_RW = self._apply_saps_pointing(
- self.read_saps_pointing_R()
+ 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.
- tile_beam = self.read_tile_beam_R()
- if tile_beam is not None:
+ if self._has_tilebeam():
+ tile_beam = self.read_attribute("tile_beam_R")
+
self.tilebeam_proxy.Pointing_direction_RW = [
tuple(tile_beam)
] * self.antennafield_proxy.nr_antennas_R
- logger.info(
- "Started the observation with ID=%s.",
- {self._observation_settings.observation_id},
- )
-
- @only_when_on()
- @fault_on_error()
- @log_exceptions()
- def read_observation_id_R(self):
- """Return the observation_id_R attribute."""
- return self._observation_settings.observation_id
-
- @only_when_on()
- @fault_on_error()
@log_exceptions()
- def read_stop_time_R(self):
- """Return the stop_time_R attribute."""
- return self._observation_settings.stop_time.timestamp()
+ def _stop_observation(self):
+ """Tear down station resources we used."""
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_antenna_field_R(self):
- """Return the antenna_field_R attribute."""
- return self._observation_settings.antenna_field
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_antenna_set_R(self):
- """Return the antenna_set_R attribute."""
- return self._observation_settings.antenna_set
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_filter_R(self):
- """Return the filter_R attribute."""
- return self._observation_settings.filter
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_saps_subband_R(self):
- """Return the saps_subband_R attribute."""
- return [sap.subbands for sap in self._observation_settings.SAPs]
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_saps_pointing_R(self):
- """Return the saps_pointing_R attribute."""
- return self._saps_pointing
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_tile_beam_R(self):
- """Return the tile_beam_R attribute."""
- if self._observation_settings.tile_beam is None:
- return None
- pointing_direction = self._observation_settings.tile_beam
- return [
- str(pointing_direction.direction_type),
- f"{pointing_direction.angle1}deg",
- f"{pointing_direction.angle2}deg",
- ]
-
- @only_in_states([DevState.STANDBY, DevState.ON])
- @fault_on_error()
- @log_exceptions()
- def read_first_beamlet_R(self):
- """Return the first_beamlet_R attribute."""
- return self._observation_settings.first_beamlet
+ pass
@fault_on_error()
@log_exceptions()
@@ -284,30 +331,11 @@ class Observation(LOFARDevice):
self._observation_settings = ObservationSettings.from_json(parameters)
except ValidationError:
self._observation_settings = None
- # Except.throw_exception("IllegalCommand", e.message, __name__)
+ raise
- @only_when_on()
- @fault_on_error()
- @log_exceptions()
- def read_observation_running_R(self):
- """Return the observation_running_R attribute."""
- # TODO(Corne): Actually keep track of the running time and perform proper
- # value
- return time()
-
- def _build_saps_pointing(self, parameters: ObservationSettings):
- """Build the sap pointing array preserving the correct order from JSON"""
- saps_pointing = []
- for sap in parameters.SAPs:
- for _ in sap.subbands:
- saps_pointing.append(
- (
- sap.pointing.direction_type,
- f"{sap.pointing.angle1}deg",
- f"{sap.pointing.angle2}deg",
- ),
- )
- return saps_pointing
+ def _has_tilebeam(self):
+ """Return whether this observation should control a TileBeam device."""
+ return self._observation_settings.antenna_field.startswith("HBA")
def _apply_antennafield_settings(self, filter_name: str):
"""Retrieve the RCU band from filter name, returning the correct format for
@@ -320,20 +348,23 @@ class Observation(LOFARDevice):
def _apply_saps_subbands(self, sap_subbands: list):
"""Convert an array of subbands into the correct format for Beamlet device"""
subband_select = self.beamlet_proxy.subband_select_RW
- first_beamlet = numpy.array(self.read_first_beamlet_R(), dtype=numpy.int64)
+ first_beamlet = self.read_attribute("first_beamlet_R")
+
# Insert subband values starting from the first beamlet
- sap_subbands = numpy.array(sap_subbands).flatten()
- subband_select[first_beamlet : len(sap_subbands)] = sap_subbands
+ subband_select[first_beamlet : first_beamlet + len(sap_subbands)] = sap_subbands
+
return subband_select
def _apply_saps_pointing(self, sap_pointing: list):
"""Convert an array of string directions into the correct format for DigitalBeam device"""
- pointing_direction = list(
- self.digitalbeam_proxy.Pointing_direction_RW
- ) # convert to list to allows item assignment
- first_beamlet = numpy.array(self.read_first_beamlet_R(), dtype=numpy.int64)
+ pointing_direction = list(self.digitalbeam_proxy.Pointing_direction_RW)
+ first_beamlet = self.read_attribute("first_beamlet_R")
+
# Insert pointing values starting from the first beamlet
- pointing_direction[first_beamlet : len(sap_pointing)] = sap_pointing
+ pointing_direction[
+ first_beamlet : first_beamlet + len(sap_pointing)
+ ] = sap_pointing
+
return tuple(pointing_direction)
def _apply_observation_id(self, observation_id: numpy.int64):
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation_control.py b/tangostationcontrol/tangostationcontrol/devices/observation_control.py
index 3cb53961a..acf35a664 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation_control.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation_control.py
@@ -5,9 +5,7 @@ import logging
import numpy
from tango import (
- Except,
DevState,
- AttrWriteType,
DebugIt,
Util,
DevBoolean,
@@ -21,7 +19,7 @@ from tangostationcontrol.common.lofar_logging import (
log_exceptions,
)
from tangostationcontrol.configuration import ObservationSettings
-from tangostationcontrol.devices.device_decorators import only_when_on, fault_on_error
+from tangostationcontrol.devices.device_decorators import only_when_on
from tangostationcontrol.devices.base_device_classes.lofar_device import LOFARDevice
from tangostationcontrol.devices.observation import Observation
@@ -48,13 +46,14 @@ class ObservationControl(LOFARDevice):
are available to perform the set-up.
Essentially this is what happens:
- Somebody calls ObservationControl.start_observation(parameters).
+ Somebody calls ObservationControl.add_observation(parameters).
Then ObservationControl will perform:
- Creates a new instance of an Observation device in the Tango DB
- Call Initialise(parameters)
- Wait for initialise to return
- Check status()
- If status() is NOT STANDBY, abort with an exception
+ - Wait for start_time - lead_time
- Call On()
- Subscribe to the Observation.running MP's periodic event
- Register the observation in the dict self.running_observations[ID]
@@ -74,9 +73,10 @@ class ObservationControl(LOFARDevice):
Functions
- Normal lifecycle funcs: initialise, on, off
- - start_observation(parameters)
- - stop_observation(ID)
- - stop_all_observations()
+ - add_observation(parameters)
+ - start_observation_now(ID)
+ - stop_observation_now(ID)
+ - stop_all_observations_now()
- running_observations() -> dict
- is_observation_running(obs_id) -> bool
@@ -85,9 +85,23 @@ class ObservationControl(LOFARDevice):
"""
# Attributes
- running_observations_R = attribute(
- dtype=(numpy.int64,), max_dim_x=1000, access=AttrWriteType.READ
+ @attribute(
+ doc="List of registered observations.",
+ dtype=(numpy.int64,),
+ max_dim_x=1000,
+ fisallowed="is_attribute_access_allowed",
)
+ def observations_R(self):
+ return list(self._observation_controller)
+
+ @attribute(
+ doc="List of registered observations that are running.",
+ dtype=(numpy.int64,),
+ max_dim_x=1000,
+ fisallowed="is_attribute_access_allowed",
+ )
+ def running_observations_R(self):
+ return self._observation_controller.running_observations
def __init__(self, cl, name):
# Super must be called after variable assignment due to executing init_device!
@@ -109,79 +123,79 @@ class ObservationControl(LOFARDevice):
self.set_state(DevState.OFF)
# Increase the number of polling threads for this device server.
+ # NB: This server hosts both ObservationControl and Observation
+ # devices.
Util.instance().set_polling_threads_pool_size(10)
# Lifecycle functions
- def configure_for_initialise(self):
- self._observation_controller = ObservationController(self.myTangoDomain)
-
+ @log_exceptions()
def configure_for_off(self):
- self.stop_all_observations()
+ super().configure_for_off()
- @only_when_on()
- @fault_on_error()
- @log_exceptions()
- def read_running_observations_R(self):
- return self._observation_controller.running_observations
+ self.stop_all_observations_now()
# API
@command(dtype_in=DevString)
@only_when_on()
@log_exceptions()
def start_observation(self, parameters: DevString = None):
- self._observation_controller.start_observation(
+ """Deprecated. For backward compatibility with old lofar-station-clients."""
+
+ self.add_observation(parameters)
+
+ @command(dtype_in=numpy.int64)
+ @only_when_on()
+ @log_exceptions()
+ def stop_observation(self, obs_id: numpy.int64):
+ """Deprecated. For backward compatibility with old lofar-station-clients."""
+
+ self.stop_observation_now(obs_id)
+
+ @command(dtype_in=DevString)
+ @only_when_on()
+ @log_exceptions()
+ def add_observation(self, parameters: DevString = None):
+ """Add an observation. Start/stop it at the specified start/stop times."""
+ self._observation_controller.add_observation(
ObservationSettings.from_json(parameters)
)
+ @command(dtype_in=DevString)
+ @only_when_on()
+ @log_exceptions()
+ def start_observation_now(self, obs_id: numpy.int64 = 0):
+ """Force an observation to start now."""
+ logger.info("Force starting the observation with ID=%s.", obs_id)
+
+ self._observation_controller.start_observation(obs_id)
+
@command(dtype_in=numpy.int64)
@only_when_on()
@log_exceptions()
- def stop_observation(self, obs_id: numpy.int64 = 0):
- # Parameter check, do not execute an observation in case
- # the parameters are not sufficient.
- if obs_id < 1:
- # Do not execute
- error = f"Cannot stop an observation with ID={obs_id}, \
- because the observation ID is invalid."
- Except.throw_exception("IllegalCommand", error, __name__)
- elif self._observation_controller.is_observation_running(obs_id) is False:
- error = f"Cannot stop an observation with ID={obs_id}, \
- because the observation is not running."
- Except.throw_exception("IllegalCommand", error, __name__)
-
- logger.info("Stopping the observation with ID=%s.", obs_id)
-
- self._observation_controller.stop_observation(obs_id)
+ def stop_observation_now(self, obs_id: numpy.int64):
+ """Force an observation to stop now."""
+ logger.info("Force stopping the observation with ID=%s.", obs_id)
+
+ self._observation_controller.stop_observation_now(obs_id)
@command()
@only_when_on()
@log_exceptions()
- def stop_all_observations(self):
- # Make a copy of the running_observations dict. This
- # should prevent race conditions.
- if not self.is_any_observation_running():
- return
- # Make certain that the dict does not get modified
- # while I am iterating over it.
- self._observation_controller.stop_all_observations()
+ def stop_all_observations_now(self):
+ """Force all observations to stop now."""
+ self._observation_controller.stop_all_observations_now()
@command(dtype_in=numpy.int64, dtype_out=DevBoolean)
@only_when_on()
@log_exceptions()
- def is_observation_running(self, obs_id: numpy.int64 = -1) -> DevBoolean:
- # Parameter check, do not execute if obs_id is invalid
- if obs_id < 1:
- # Do not execute
- error = f"Cannot check if an observation with ID={obs_id} is running, \
- because the observation ID is invalid"
- Except.throw_exception("IllegalCommand", error, __name__)
- return self._observation_controller.is_observation_running(obs_id)
+ def is_observation_running(self, obs_id: numpy.int64) -> DevBoolean:
+ return obs_id in self._observation_controller.running_observations
@command(dtype_out=DevBoolean)
@only_when_on()
@log_exceptions()
def is_any_observation_running(self) -> DevBoolean:
- return self._observation_controller.is_any_observation_running()
+ return len(self._observation_controller.running_observations) > 0
# ----------
@@ -189,4 +203,7 @@ class ObservationControl(LOFARDevice):
# ----------
def main(**kwargs):
"""Main function of the ObservationControl module."""
+
+ # The ObservationControl device spawns Observation devices, so we manage
+ # both in this DeviceServer.
return entry((ObservationControl, Observation), verbose=True, **kwargs)
diff --git a/tangostationcontrol/test/common/test_observation_controller.py b/tangostationcontrol/test/common/test_observation_controller.py
index 3aa1091ab..a9d31c01c 100644
--- a/tangostationcontrol/test/common/test_observation_controller.py
+++ b/tangostationcontrol/test/common/test_observation_controller.py
@@ -10,39 +10,48 @@ from unittest.mock import Mock
from tango import DevState
from tangostationcontrol.common import ObservationController
-from tangostationcontrol.common.observation_controller import RunningObservation
+from tangostationcontrol.common.observation_controller import Observation
from tangostationcontrol.common.proxy import create_device_proxy
from tangostationcontrol.configuration import ObservationSettings, Pointing, Sap
from test import base
+class MockObservation(object):
+ def __init__(self, running):
+ self._running = running
+
+ def is_running(self):
+ return self._running
+
+
@mock.patch("tango.Util.instance")
class TestObservationController(base.TestCase):
"""Test Observation Controller main operations"""
- def test_is_any_observation_running(self, _):
+ def test_observations_running(self, _):
sut = ObservationController("DMR")
- self.assertFalse(sut.is_any_observation_running())
- sut._running_observations[1] = {}
- self.assertTrue(sut.is_any_observation_running())
+ self.assertListEqual([], sut.running_observations)
+ sut[1] = MockObservation(True)
+ self.assertListEqual([1], sut.running_observations)
- def test_is_observation_running(self, _):
+ def test_observations_not_running(self, _):
sut = ObservationController("DMR")
- self.assertFalse(sut.is_observation_running(2))
- sut._running_observations[2] = {}
- self.assertTrue(sut.is_observation_running(2))
+ self.assertListEqual([], sut.running_observations)
+ sut[2] = MockObservation(False)
+ self.assertListEqual([], sut.running_observations)
- def test_stop_all_observations_no_running(self, _):
+ def test_stop_all_observations_now_no_running(self, _):
sut = ObservationController("DMR")
- sut.stop_all_observations()
+ sut.stop_all_observations_now()
@mock.patch("tango.Util.instance")
-class TestRunningObservation(base.TestCase):
+class TestObservation(base.TestCase):
SETTINGS = ObservationSettings(
5,
datetime.fromisoformat("2022-10-26T11:35:54.704150"),
+ datetime.fromisoformat("2022-10-27T11:35:54.704150"),
"HBA",
"ALL",
"filter settings",
@@ -50,34 +59,42 @@ class TestRunningObservation(base.TestCase):
)
def test_properties(self, _):
- sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ sut = Observation("DMR", TestObservation.SETTINGS)
self.assertEqual(5, sut.observation_id)
self.assertEqual("Observation", sut.class_name)
self.assertEqual("DMR/Observation/5", sut.device_name)
- self.assertEqual("DMR/Observation/5/observation_running_R", sut.attribute_name)
+ self.assertEqual("DMR/Observation/5/alive_R", sut.attribute_name)
- def test_create_tango_device(self, tu_mock):
- sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
- sut.create_tango_device()
+ def test_create_observation_device(self, tu_mock):
+ sut = Observation("DMR", TestObservation.SETTINGS)
+ sut.create_observation_device()
@mock.patch("tango.DeviceProxy")
- def test_create_device_proxy(self, dp_mock, tu_mock):
+ def test_initialise_observation(self, dp_mock, tu_mock):
importlib.reload(sys.modules[create_device_proxy.__module__])
- sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
- sut.create_device_proxy()
+ sut = Observation("DMR", TestObservation.SETTINGS)
+ sut.initialise_observation()
self.assertEqual(
dp_mock.return_value.observation_settings_RW,
- TestRunningObservation.SETTINGS.to_json(),
+ TestObservation.SETTINGS.to_json(),
)
dp_mock.return_value.Initialise.assert_called()
+
+ @mock.patch("tango.DeviceProxy")
+ def test_start(self, dp_mock, tu_mock):
+ importlib.reload(sys.modules[create_device_proxy.__module__])
+ sut = Observation("DMR", TestObservation.SETTINGS)
+ sut.initialise_observation()
+ sut.start()
+
dp_mock.return_value.On.assert_called()
def test_subscribe(self, _):
def dummy():
pass
- sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ sut = Observation("DMR", TestObservation.SETTINGS)
dp_mock = Mock()
sut._device_proxy = dp_mock
sut.subscribe(dummy)
@@ -85,20 +102,19 @@ class TestRunningObservation(base.TestCase):
dp_mock.poll_attribute.assert_called()
dp_mock.subscribe_event.assert_called()
- def test_shutdown(self, tu_mock):
- importlib.reload(sys.modules[RunningObservation.__module__])
- sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ def test_stop(self, tu_mock):
+ importlib.reload(sys.modules[Observation.__module__])
+ sut = Observation("DMR", TestObservation.SETTINGS)
dp_mock = Mock()
dp_mock.state.return_value = DevState.OFF
sut._device_proxy = dp_mock
- sut.shutdown()
+ sut.stop()
dp_mock.ping.assert_called()
dp_mock.unsubscribe_event.assert_called()
dp_mock.Off.assert_called()
- dp_mock.state.assert_called()
tu_mock.return_value.delete_device.assert_called()
diff --git a/tangostationcontrol/test/configuration/test_observation_settings.py b/tangostationcontrol/test/configuration/test_observation_settings.py
index 7f67d02e1..049ec2416 100644
--- a/tangostationcontrol/test/configuration/test_observation_settings.py
+++ b/tangostationcontrol/test/configuration/test_observation_settings.py
@@ -95,6 +95,7 @@ class TestObservationSettings(base.TestCase):
sut = ObservationSettings(
5,
datetime.fromisoformat("2022-10-26T11:35:54.704150"),
+ datetime.fromisoformat("2022-10-27T11:35:54.704150"),
"HBA",
"ALL",
"filter settings",
@@ -105,7 +106,8 @@ class TestObservationSettings(base.TestCase):
)
self.assertEqual(
sut.to_json(),
- '{"observation_id": 5, "stop_time": "2022-10-26T11:35:54.704150", '
+ '{"observation_id": 5, "start_time": "2022-10-26T11:35:54.704150", '
+ '"stop_time": "2022-10-27T11:35:54.704150", '
'"antenna_field": "HBA", '
'"antenna_set": "ALL", "filter": "filter settings", "SAPs": '
'[{"subbands": [3, 2], "pointing": {"angle1": 1.2, "angle2": 2.1, '
--
GitLab