From eaf164c784deb1dc6b80b838f9dd93d1196d7086 Mon Sep 17 00:00:00 2001
From: Hannes Feldt <feldt@astron.nl>
Date: Wed, 9 Nov 2022 09:50:10 +0000
Subject: [PATCH] L2SS-1019: Create and use obeservation control JSON schema
---
.gitignore | 1 +
docker-compose/device-observation-control.yml | 2 +
docker-compose/schemas.yml | 31 ++
docker-compose/schemas/Dockerfile | 2 +
.../definitions/observation-settings.json | 56 +++
.../schemas/definitions/pointing.json | 44 +++
docker-compose/schemas/definitions/sap.json | 20 ++
sbin/run_integration_test.sh | 26 +-
tangostationcontrol/requirements.txt | 2 +-
.../tangostationcontrol/common/__init__.py | 10 +
.../common/observation_controller.py | 245 +++++++++++++
.../configuration/__init__.py | 12 +
.../configuration/_json_parser.py | 31 ++
.../configuration/configuration_base.py | 106 ++++++
.../configuration/observation_settings.py | 46 +++
.../configuration/pointing.py | 28 ++
.../tangostationcontrol/configuration/sap.py | 25 ++
.../devices/observation.py | 138 +++----
.../devices/observation_control.py | 339 +++---------------
.../devices/test_device_observation.py | 4 +-
.../test_device_observation_control.py | 36 +-
.../common/test_observation_controller.py | 98 +++++
.../test/configuration}/__init__.py | 0
.../test/configuration/_mock_requests.py | 148 ++++++++
.../test_observation_settings.py | 100 ++++++
.../test/configuration/test_pointing.py | 53 +++
.../test/configuration/test_sap_settings.py | 65 ++++
27 files changed, 1273 insertions(+), 395 deletions(-)
create mode 100644 docker-compose/schemas.yml
create mode 100644 docker-compose/schemas/Dockerfile
create mode 100644 docker-compose/schemas/definitions/observation-settings.json
create mode 100644 docker-compose/schemas/definitions/pointing.json
create mode 100644 docker-compose/schemas/definitions/sap.json
create mode 100644 tangostationcontrol/tangostationcontrol/common/observation_controller.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/__init__.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/pointing.py
create mode 100644 tangostationcontrol/tangostationcontrol/configuration/sap.py
create mode 100644 tangostationcontrol/tangostationcontrol/test/common/test_observation_controller.py
rename tangostationcontrol/{ => tangostationcontrol/test/configuration}/__init__.py (100%)
create mode 100644 tangostationcontrol/tangostationcontrol/test/configuration/_mock_requests.py
create mode 100644 tangostationcontrol/tangostationcontrol/test/configuration/test_observation_settings.py
create mode 100644 tangostationcontrol/tangostationcontrol/test/configuration/test_pointing.py
create mode 100644 tangostationcontrol/tangostationcontrol/test/configuration/test_sap_settings.py
diff --git a/.gitignore b/.gitignore
index 247edda4a..69b645a3c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -23,6 +23,7 @@ tangostationcontrol/docs/build
**/coverage.xml
**/.coverage
+**/.coverage*
**/.ipynb_checkpoints
**/pending_log_messages.db
**/.eggs
diff --git a/docker-compose/device-observation-control.yml b/docker-compose/device-observation-control.yml
index 4a3567049..c71f31656 100644
--- a/docker-compose/device-observation-control.yml
+++ b/docker-compose/device-observation-control.yml
@@ -56,3 +56,5 @@ services:
restart: unless-stopped
stop_signal: SIGINT # request a graceful shutdown of Tango
stop_grace_period: 2s
+ depends_on:
+ - schemas
diff --git a/docker-compose/schemas.yml b/docker-compose/schemas.yml
new file mode 100644
index 000000000..dc4cb4aa8
--- /dev/null
+++ b/docker-compose/schemas.yml
@@ -0,0 +1,31 @@
+#
+# Docker compose file that launches a LOFAR2.0 station's
+# ObservationControl device. It also runs the dynamically
+# created Observation devices.
+#
+# Defines:
+# - schemas: LOFAR2.0 station
+#
+version: '2.1'
+
+services:
+ schemas:
+ build:
+ context: schemas
+ container_name: ${CONTAINER_NAME_PREFIX}schemas
+ networks:
+ - control
+ # set the hostname, otherwise duplicate device registrations result every
+ # time the hostname changes as the container is restarted.
+ hostname: schemas
+ environment:
+ - NGINX_HOST=schemas
+ - NGINX_PORT=80
+ ports:
+ - 9999:80
+ logging:
+ driver: "json-file"
+ options:
+ max-size: "100m"
+ max-file: "10"
+ restart: unless-stopped
\ No newline at end of file
diff --git a/docker-compose/schemas/Dockerfile b/docker-compose/schemas/Dockerfile
new file mode 100644
index 000000000..5d6aad5cf
--- /dev/null
+++ b/docker-compose/schemas/Dockerfile
@@ -0,0 +1,2 @@
+FROM nginx
+COPY definitions /usr/share/nginx/html
\ No newline at end of file
diff --git a/docker-compose/schemas/definitions/observation-settings.json b/docker-compose/schemas/definitions/observation-settings.json
new file mode 100644
index 000000000..baccdd069
--- /dev/null
+++ b/docker-compose/schemas/definitions/observation-settings.json
@@ -0,0 +1,56 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "observation_id",
+ "stop_time",
+ "antenna_mask",
+ "filter",
+ "SAPs"
+ ],
+ "properties": {
+ "observation_id": {
+ "type": "number",
+ "minimum": 1
+ },
+ "stop_time": {
+ "type": "string",
+ "format": "date-time"
+ },
+ "antenna_mask": {
+ "type": "array",
+ "uniqueItems": true,
+ "minItems": 1,
+ "items": {
+ "type": "number"
+ }
+ },
+ "filter": {
+ "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",
+ "minItems": 1,
+ "items": {
+ "$ref": "sap.json"
+ }
+ },
+ "tile_beam": {
+ "$ref": "pointing.json"
+ },
+ "first_beamlet": {
+ "type": "number",
+ "default": 0,
+ "minimum": 0
+ }
+ }
+}
\ No newline at end of file
diff --git a/docker-compose/schemas/definitions/pointing.json b/docker-compose/schemas/definitions/pointing.json
new file mode 100644
index 000000000..ac3ddc141
--- /dev/null
+++ b/docker-compose/schemas/definitions/pointing.json
@@ -0,0 +1,44 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "angle1",
+ "angle2",
+ "direction_type"
+ ],
+ "properties": {
+ "angle1": {
+ "default": 0.6624317181687094,
+ "description": "First angle (e.g. RA)",
+ "title": "Angle 1",
+ "type": "number"
+ },
+ "angle2": {
+ "default": 1.5579526427549426,
+ "description": "Second angle (e.g. DEC)",
+ "title": "Angle 2",
+ "type": "number"
+ },
+ "direction_type": {
+ "default": "J2000",
+ "description": "",
+ "enum": [
+ "J2000",
+ "AZELGEO",
+ "LMN",
+ "SUN",
+ "MOON",
+ "MERCURY",
+ "VENUS",
+ "MARS",
+ "JUPITER",
+ "SATURN",
+ "URANUS",
+ "NEPTUNE",
+ "PLUTO"
+ ],
+ "title": "Reference frame",
+ "type": "string"
+ }
+ }
+}
\ No newline at end of file
diff --git a/docker-compose/schemas/definitions/sap.json b/docker-compose/schemas/definitions/sap.json
new file mode 100644
index 000000000..50e945e87
--- /dev/null
+++ b/docker-compose/schemas/definitions/sap.json
@@ -0,0 +1,20 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "subbands",
+ "pointing"
+ ],
+ "properties": {
+ "subbands": {
+ "type": "array",
+ "minItems": 1,
+ "items": {
+ "type": "number"
+ }
+ },
+ "pointing": {
+ "$ref": "pointing.json"
+ }
+ }
+}
\ No newline at end of file
diff --git a/sbin/run_integration_test.sh b/sbin/run_integration_test.sh
index 57e7e4d2c..f78d11a6b 100755
--- a/sbin/run_integration_test.sh
+++ b/sbin/run_integration_test.sh
@@ -27,7 +27,7 @@ function integration_test {
echo "Updating config ${config} ..."
bash "${LOFAR20_DIR}"/sbin/update_ConfigDb.sh "${config}"
done
- if [ ! -z "${2+x}" ]; then
+ if [ -n "${2+x}" ]; then
# shellcheck disable=SC2145
echo "make restart ${restarts[@]} ..."
make restart "${restarts[@]}"
@@ -76,28 +76,27 @@ sleep 1 # dsconfig container must be up and running...
# shellcheck disable=SC2016
echo '/usr/local/bin/wait-for-it.sh ${TANGO_HOST} --strict --timeout=300 -- true' | make run dsconfig bash -
-# Devices list is used to explitly word split when supplied to commands, must
-# disable shellcheck SC2086 for each case.
-DEVICES="device-boot device-apsct device-ccd device-apspu device-sdp device-recv 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"
+DEVICES=(device-boot device-apsct device-ccd device-apspu device-sdp device-recv 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)
-SIMULATORS="sdptr-sim recv-sim unb2-sim apsct-sim apspu-sim ccd-sim"
+SIMULATORS=(sdptr-sim recv-sim unb2-sim apsct-sim apspu-sim ccd-sim)
# Build only the required images, please do not build everything that makes CI
# take really long to finish, especially grafana / jupyter / prometheus.
# jupyter is physically large > 2.5gb and overlayfs is really slow.
# shellcheck disable=SC2086
-make build $DEVICES $SIMULATORS
+make build "${DEVICES[@]}" "${SIMULATORS[@]}"
# make build elk integration-test # L2SS-970: elk temporarily disabled
-make build logstash integration-test
+make build logstash integration-test schemas
make build archiver-timescale hdbppts-cm hdbppts-es
# Start and stop sequence
# shellcheck disable=SC2086
-make stop $DEVICES $SIMULATORS hdbppts-es hdbppts-cm archiver-timescale
+make stop schemas
+make stop "${DEVICES[@]}" "${SIMULATORS[@]}" hdbppts-es hdbppts-cm archiver-timescale
make stop device-docker # this one does not test well in docker-in-docker
# make stop elk # L2SS-970: elk temporarily disabled
-make stop logstash
+make stop logstash schemas
# Run dummy integration test to install pytango in tox virtualenv without
# the memory pressure of the ELK stack.
@@ -108,7 +107,7 @@ make stop logstash
integration_test dummy
# make start elk # L2SS-970: elk temporarily disabled
-make start logstash
+make start logstash schemas
# Update the dsconfig
# Do not remove `bash`, otherwise statement ignored by gitlab ci shell!
@@ -120,19 +119,18 @@ bash "${LOFAR20_DIR}"/sbin/update_ConfigDb.sh "${LOFAR20_DIR}"/CDB/stations/dumm
cd "$LOFAR20_DIR/docker-compose" || exit 1
# shellcheck disable=SC2086
-make start $SIMULATORS
+make start "${SIMULATORS[@]}"
# Give the simulators time to start
sleep 5
# shellcheck disable=SC2086
-make start $DEVICES
+make start "${DEVICES[@]}"
# Archive devices: archive-timescale first
make start archiver-timescale
# Wait for archiver and devices to restart
-# shellcheck disable=SC2086
-make await archiver-timescale $DEVICES
+make await archiver-timescale "${DEVICES[@]}"
# Give archiver-timescale time to start
# shellcheck disable=SC2016
diff --git a/tangostationcontrol/requirements.txt b/tangostationcontrol/requirements.txt
index 8d6285449..8a21857b9 100644
--- a/tangostationcontrol/requirements.txt
+++ b/tangostationcontrol/requirements.txt
@@ -12,7 +12,7 @@ psycopg2-binary >= 2.9.2 # LGPL
sqlalchemy >= 1.4.26 # MIT
pysnmp >= 0.1.7 # BSD
h5py >= 3.1.0 # BSD
-jsonschema >= 3.2.0 # MIT
+jsonschema >= 4.0.0 # MIT
psutil >= 5.8.0 # BSD
docker >= 5.0.3 # Apache 2
python-logstash-async >= 2.3.0 # MIT
diff --git a/tangostationcontrol/tangostationcontrol/common/__init__.py b/tangostationcontrol/tangostationcontrol/common/__init__.py
index e69de29bb..3708ddd2a 100644
--- a/tangostationcontrol/tangostationcontrol/common/__init__.py
+++ b/tangostationcontrol/tangostationcontrol/common/__init__.py
@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from .observation_controller import ObservationController
+
+__all__ = ['ObservationController', ]
diff --git a/tangostationcontrol/tangostationcontrol/common/observation_controller.py b/tangostationcontrol/tangostationcontrol/common/observation_controller.py
new file mode 100644
index 000000000..d77bfc9ea
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/common/observation_controller.py
@@ -0,0 +1,245 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+import logging
+import time
+from datetime import datetime
+
+from tango import DevFailed, DevState, Except, Util, EventType, DeviceProxy
+
+from tangostationcontrol.common.lofar_logging import log_exceptions
+from tangostationcontrol.configuration import ObservationSettings
+
+logger = logging.getLogger()
+
+
+class RunningObservation(object):
+ @property
+ def observation_id(self) -> int:
+ return self._parameters.observation_id
+
+ @property
+ def class_name(self) -> str:
+ from tangostationcontrol.devices.observation import Observation
+ return Observation.__name__
+
+ @property
+ def device_name(self) -> str:
+ return f"{self._tango_domain}/{self.class_name}/{self.observation_id}"
+
+ # Name for the Observation.observation_running subscription
+ @property
+ def attribute_name(self) -> str:
+ return f"{self.device_name}/observation_running_R"
+
+ def __init__(self, tango_domain, parameters: ObservationSettings):
+ self._device_proxy: DeviceProxy | None = None
+ self._event_id: int | None = None
+ self._parameters: ObservationSettings = parameters
+ self._tango_domain: str = tango_domain
+
+ # The pyTango.Util class is a singleton and every DS can only
+ # have one instance of it.
+ self._tango_util: Util = Util.instance()
+
+ def create_tango_device(self):
+ logger.info(f"Create device: {self.device_name}")
+ try:
+ # Create the Observation device and instantiate it.
+ self._tango_util.create_device(self.class_name, f"{self.device_name}")
+ except DevFailed as ex:
+ logger.exception(ex)
+ if ex.args[0].desc == f"The device {self.device_name.lower()} is already defined in the database":
+ # and self.is_observation_running(self.observation_id) is False:
+ self._tango_util.delete_device(self.class_name, self.device_name)
+ error_string = f"Cannot create the Observation device {self.device_name} because it is already present in the Database but it is not running. Try to re-run the start_observation command"
+ logger.exception(error_string)
+ Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
+ else:
+ error_string = f"Cannot create the Observation device instance {self.device_name} for ID={self.observation_id}. This means that the observation did not start."
+ logger.exception(error_string)
+ Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
+
+ def create_device_proxy(self):
+ # Instantiate a dynamic Tango Device "Observation".
+ self._device_proxy = DeviceProxy(self.device_name)
+
+ # Configure the dynamic device its attribute for the observation
+ # parameters.
+ self._device_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()
+
+ # The call to On will actually tell the Observation device to
+ # become fully active.
+ self._device_proxy.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)
+
+ # Right. Now subscribe to periodic events.
+ self._event_id = self._device_proxy.subscribe_event(self.attribute_name.split('/')[-1],
+ EventType.PERIODIC_EVENT,
+ cb)
+ logger.info(f"Successfully started an observation with ID={self.observation_id}.")
+
+ def shutdown(self):
+ # Check if the device has not terminated itself in the meanwhile.
+ try:
+ self._device_proxy.ping()
+ except DevFailed:
+ logger.warning(
+ f"The device for the Observation with ID={self.observation_id} has unexpectedly already disappeared. It is advised to check the logs up to 10s prior to this message to see what happened.")
+ else:
+ # Unsubscribe from the subscribed event.
+ event_id = self._event_id
+ self._device_proxy.unsubscribe_event(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(f"Successfully stopped the observation with ID={self.observation_id}")
+ else:
+ logger.warning(
+ f"Could not shut down the Observation device ( {self.device_name} ) for observation ID={self.observation_id}. This means that there is a chance for a memory leak. Will continue anyway and forcefully delete the Observation object.")
+
+ # 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(
+ f"Something went wrong when the device {self.device_name} 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.")
+
+
+class ObservationController(object):
+ @property
+ def running_observations(self) -> [int]:
+ return list(self._running_observations.keys())
+
+ 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
+
+ @log_exceptions()
+ def observation_running_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.
+ """
+ if event.err:
+ # Something is fishy with this event.
+ logger.warning(
+ f"The Observation device {event.device} 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={event}")
+ return
+
+ # Get the Observation ID from the sending device.
+ obs_id = event.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(
+ f"Received an observation_running event for the observation with ID={obs_id}. 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.")
+ return
+
+ 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(
+ f"Received an observation_running event for the observation with ID={obs_id}. 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.")
+ return
+
+ def start_observation(self, settings: ObservationSettings):
+ # 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__)
+
+ obs = RunningObservation(self._tango_domain, settings)
+ obs.create_tango_device()
+
+ try:
+ obs.create_device_proxy()
+ 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
+
+ try:
+ obs.subscribe(self.observation_running_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__)
+
+ def stop_observation(self, obs_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):
+ for obs_id in self._running_observations.copy().keys():
+ self.stop_observation(obs_id)
diff --git a/tangostationcontrol/tangostationcontrol/configuration/__init__.py b/tangostationcontrol/tangostationcontrol/configuration/__init__.py
new file mode 100644
index 000000000..2b1c62bb5
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/__init__.py
@@ -0,0 +1,12 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from .observation_settings import ObservationSettings
+from .pointing import Pointing
+from .sap import Sap
+
+__all__ = ['ObservationSettings', 'Pointing', 'Sap', ]
diff --git a/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py b/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
new file mode 100644
index 000000000..92ecaa609
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/_json_parser.py
@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+from typing import Type
+
+from jsonschema.exceptions import ValidationError
+
+
+def _from_json_hook_t(primary: Type):
+ from tangostationcontrol.configuration import Pointing, Sap, ObservationSettings
+ def actual_hook(json_dct):
+ primary_ex = None
+ for t in [Pointing, Sap, ObservationSettings]:
+ try:
+ t.get_validator().validate(json_dct)
+ except ValidationError as ex:
+ if t is primary:
+ primary_ex = ex
+ pass
+ else:
+ return t.to_object(json_dct)
+ if primary_ex:
+ raise primary_ex
+ return None
+
+ return actual_hook
diff --git a/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py b/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
new file mode 100644
index 000000000..7820eefb7
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/configuration_base.py
@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+import json
+import re
+import time
+from abc import ABC, abstractmethod
+from typing import TypeVar, Type
+
+import jsonschema
+import requests
+from jsonschema import Draft7Validator, FormatChecker, ValidationError
+from jsonschema.validators import RefResolver
+
+from tangostationcontrol.configuration._json_parser import _from_json_hook_t
+
+T = TypeVar('T')
+
+
+def _fetch_url(url):
+ attempt_nr = 1
+ while True:
+ try:
+ response = requests.get(url)
+ response.raise_for_status()
+ return response.json()
+ except requests.exceptions.RequestException as e:
+ time.sleep(2) # retry after a little sleep
+ if attempt_nr >= 5:
+ raise e
+ else:
+ attempt_nr += 1
+
+
+class RetryHttpRefResolver(RefResolver):
+
+ def resolve_remote(self, uri):
+ result = _fetch_url(uri)
+
+ if self.cache_remote:
+ self.store[uri] = result
+ return result
+
+
+def _is_object(_, instance):
+ return isinstance(instance, dict) or issubclass(type(instance), _ConfigurationBase)
+
+
+jsonschema.validators.Draft7Validator.TYPE_CHECKER = Draft7Validator.TYPE_CHECKER.redefine(
+ "object", _is_object,
+)
+
+
+class _ConfigurationBase(ABC):
+ BASE_URL = "http://schemas/"
+
+ @staticmethod
+ def _class_to_url(cls_name):
+ return re.sub(r'(?<!^)(?=[A-Z])', '-', cls_name).lower()
+
+ @classmethod
+ def get_validator(cls):
+ name = cls.__name__
+ url = f"{_ConfigurationBase.BASE_URL}{_ConfigurationBase._class_to_url(name)}.json"
+ resolver = RetryHttpRefResolver(base_uri=_ConfigurationBase.BASE_URL, referrer=url)
+ _, resolved = resolver.resolve(url)
+ return Draft7Validator(resolved, format_checker=FormatChecker(), resolver=resolver)
+
+ @abstractmethod
+ def __iter__(self):
+ pass
+
+ def __str__(self):
+ return json.dumps(dict(self), ensure_ascii=False)
+
+ def __repr__(self):
+ return self.__str__()
+
+ # required for jsonschema validation
+ def __getitem__(self, item):
+ return getattr(self, item)
+
+ # required for jsonschema validation
+ def __contains__(self, item):
+ return hasattr(self, item) and getattr(self, item) is not None
+
+ def to_json(self):
+ return self.__str__()
+
+ @staticmethod
+ @abstractmethod
+ def to_object(json_dct) -> T:
+ pass
+
+ @classmethod
+ def from_json(cls: Type[T], data: str) -> T:
+ s = json.loads(data, object_hook=_from_json_hook_t(cls))
+ if not isinstance(s, cls):
+ raise ValidationError(f"Unexpected type: expected <{cls.__class__.__name__}>, got <{type(s).__name__}>")
+ return s
diff --git a/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
new file mode 100644
index 000000000..65c024b05
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/observation_settings.py
@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from datetime import datetime
+from typing import Sequence
+
+from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
+from tangostationcontrol.configuration.pointing import Pointing
+from tangostationcontrol.configuration.sap import Sap
+
+
+class ObservationSettings(_ConfigurationBase):
+ def __init__(self, observation_id: int, stop_time: datetime, antenna_mask: Sequence[int], filter: str,
+ SAPs: Sequence[Sap],
+ tile_beam: Pointing = None, first_beamlet: int = 0):
+ self.observation_id = observation_id
+ self.stop_time = stop_time
+ self.antenna_mask = antenna_mask
+ self.filter = filter
+ self.SAPs = SAPs
+ self.tile_beam = tile_beam
+ self.first_beamlet = first_beamlet
+
+ def __iter__(self):
+ yield from {
+ "observation_id": self.observation_id,
+ "stop_time" : self.stop_time.isoformat(),
+ "antenna_mask" : self.antenna_mask,
+ "filter" : self.filter,
+ "SAPs" : [dict(s) for s in self.SAPs]
+ }.items()
+ if self.tile_beam:
+ yield "tile_beam", dict(self.tile_beam)
+ yield "first_beamlet", self.first_beamlet
+
+ @staticmethod
+ def to_object(json_dct) -> 'ObservationSettings':
+ return ObservationSettings(json_dct['observation_id'], datetime.fromisoformat(json_dct['stop_time']),
+ json_dct['antenna_mask'],
+ 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)
diff --git a/tangostationcontrol/tangostationcontrol/configuration/pointing.py b/tangostationcontrol/tangostationcontrol/configuration/pointing.py
new file mode 100644
index 000000000..8e7933f6f
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/pointing.py
@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
+
+
+class Pointing(_ConfigurationBase):
+ VALIDATOR = None
+
+ def __init__(self, angle1=0.6624317181687094, angle2=1.5579526427549426, direction_type="J2000"):
+ self.angle1 = angle1
+ self.angle2 = angle2
+ self.direction_type = direction_type
+
+ def __iter__(self):
+ yield from {
+ "angle1" : self.angle1,
+ "angle2" : self.angle2,
+ "direction_type": self.direction_type
+ }.items()
+
+ @staticmethod
+ def to_object(json_dct) -> 'Pointing':
+ return Pointing(json_dct['angle1'], json_dct['angle2'], json_dct['direction_type'])
diff --git a/tangostationcontrol/tangostationcontrol/configuration/sap.py b/tangostationcontrol/tangostationcontrol/configuration/sap.py
new file mode 100644
index 000000000..d74e12a88
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/configuration/sap.py
@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR2.0 Station Control project.
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tangostationcontrol.configuration.configuration_base import _ConfigurationBase
+from tangostationcontrol.configuration.pointing import Pointing
+
+
+class Sap(_ConfigurationBase):
+ def __init__(self, subbands: [int], pointing: Pointing):
+ self.subbands = subbands
+ self.pointing = pointing
+
+ def __iter__(self):
+ yield from {
+ "subbands": self.subbands,
+ "pointing": dict(self.pointing)
+ }.items()
+
+ @staticmethod
+ def to_object(json_dct) -> 'Sap':
+ return Sap(json_dct['subbands'], json_dct['pointing'])
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation.py b/tangostationcontrol/tangostationcontrol/devices/observation.py
index cae2bd238..f4746628e 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation.py
@@ -4,27 +4,26 @@
#
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
+import logging
+from time import time
+from typing import Optional
+import numpy
+from jsonschema.exceptions import ValidationError
# PyTango imports
-from tango import AttrWriteType, DeviceProxy, DevState, DevSource, Util
+from tango import AttrWriteType, DeviceProxy, DevState, DevSource, Util, Except
from tango.server import attribute
-import numpy
-
+from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD, MAX_ANTENNA, N_beamlets_ctrl, N_point_prop
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python
from tangostationcontrol.common.lofar_logging import log_exceptions
-from tangostationcontrol.common.constants import DEFAULT_POLLING_PERIOD, MAX_ANTENNA, N_beamlets_ctrl, N_point_prop
+from tangostationcontrol.configuration import ObservationSettings
from tangostationcontrol.devices.device_decorators import fault_on_error
-from tangostationcontrol.devices.device_decorators import only_when_on
from tangostationcontrol.devices.device_decorators import only_in_states
+from tangostationcontrol.devices.device_decorators import only_when_on
from tangostationcontrol.devices.lofar_device import lofar_device
-from datetime import datetime
-from json import loads
-from time import time
-import logging
-
logger = logging.getLogger()
__all__ = ["Observation", "main"]
@@ -40,60 +39,58 @@ class Observation(lofar_device):
The lifecycle of instances of this device is controlled by ObservationControl
"""
-
-
# Attributes
- observation_running_R = attribute(dtype=numpy.float64, access=AttrWriteType.READ, polling_period=DEFAULT_POLLING_PERIOD, period=DEFAULT_POLLING_PERIOD,
+ observation_running_R = attribute(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_mask_R = attribute(dtype=(numpy.int64,), max_dim_x=MAX_ANTENNA, access=AttrWriteType.READ)
filter_R = attribute(dtype=numpy.str, access=AttrWriteType.READ)
- saps_subband_R = attribute(dtype=((numpy.uint32,),), max_dim_x=N_beamlets_ctrl, max_dim_y=N_beamlets_ctrl, access=AttrWriteType.READ)
- saps_pointing_R = attribute(dtype=((numpy.str,),), max_dim_x=N_point_prop, max_dim_y=N_beamlets_ctrl, access=AttrWriteType.READ)
+ saps_subband_R = attribute(dtype=((numpy.uint32,),), max_dim_x=N_beamlets_ctrl, max_dim_y=N_beamlets_ctrl,
+ access=AttrWriteType.READ)
+ saps_pointing_R = attribute(dtype=((numpy.str,),), max_dim_x=N_point_prop, max_dim_y=N_beamlets_ctrl,
+ access=AttrWriteType.READ)
tile_beam_R = attribute(dtype=(numpy.str,), max_dim_x=N_point_prop, access=AttrWriteType.READ)
first_beamlet_R = attribute(dtype=numpy.int64, access=AttrWriteType.READ)
observation_settings_RW = attribute(dtype=str, access=AttrWriteType.READ_WRITE)
+ def __init__(self, cl, name):
+ super().__init__(cl, name)
+ self.recv_proxy: Optional[DeviceProxy] = None
+ self.antennafield_proxy: Optional[DeviceProxy] = None
+ self.beamlet_proxy: Optional[DeviceProxy] = None
+ self.digitalbeam_proxy: Optional[DeviceProxy] = None
+ self.tilebeam_proxy: Optional[DeviceProxy] = None
+ self._observation_settings: Optional[ObservationSettings] = None
+ self._num_inputs: int = 0
+
def init_device(self):
"""Setup some class member variables for observation state"""
-
super().init_device()
- self._observation_settings = loads("{}")
- self._observation_id = -1
- self._stop_time = datetime.now()
-
- self._num_inputs = 0
def configure_for_initialise(self):
"""Load the JSON from the attribute and configure member variables"""
super().configure_for_initialise()
+ if self._observation_settings is None:
+ Except.throw_exception("IllegalCommand", "Device can not be initialized without configuration", __name__)
+
# 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.
- parameters = loads(self._observation_settings)
-
- self._observation_id = parameters["observation_id"]
- self._stop_time = datetime.fromisoformat(parameters["stop_time"])
- self._antenna_mask = parameters["antenna_mask"]
- self._filter = parameters["filter"]
- self._num_saps = len(parameters["SAPs"])
- self._saps_subband = [ parameters["SAPs"][i]['subbands'] for i in range(0, self._num_saps)]
- self._saps_pointing = self._build_saps_pointing(parameters)
- self._tile_beam = self._build_tilebeam_pointing(parameters)
- self._first_beamlet = parameters["first_beamlet"]
+ self._num_saps = len(self._observation_settings.SAPs)
+ self._saps_pointing = self._build_saps_pointing(self._observation_settings)
# 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
+ # 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 = DeviceProxy(
- f"{util.get_ds_inst_name()}/AntennaField/1")
+ self.antennafield_proxy = DeviceProxy(f"{util.get_ds_inst_name()}/AntennaField/1")
self.antennafield_proxy.set_source(DevSource.DEV)
# Set a reference of RECV device that is correlated to this device
@@ -116,16 +113,16 @@ class Observation(lofar_device):
self._num_inputs = self.digitalbeam_proxy.antenna_select_RW.shape[0]
logger.info(
- f"The observation with ID={self._observation_id} is "
+ f"The observation with ID={self._observation_settings.observation_id} is "
"configured. It will begin as soon as On() is called and it is"
- f"supposed to stop at {self._stop_time}")
+ f"supposed to stop at {self._observation_settings.stop_time}")
def configure_for_off(self):
"""Indicate the observation has stopped"""
super().configure_for_off()
- logger.info(f"Stopped the observation with ID={self._observation_id}.")
+ logger.info(f"Stopped the observation with ID={self._observation_settings.observation_id}.")
def configure_for_on(self):
"""Indicate the observation has started"""
@@ -133,9 +130,9 @@ class Observation(lofar_device):
super().configure_for_on()
# Apply Antenna Mask and Filter
- ANT_mask, RCU_band_select = self._apply_antennafield_settings(self.read_antenna_mask_R(), self.read_filter_R())
- self.antennafield_proxy.ANT_mask_RW = ANT_mask
- self.antennafield_proxy.RCU_band_select_RW = RCU_band_select
+ ant_mask, rcu_band_select = self._apply_antennafield_settings(self.read_antenna_mask_R(), self.read_filter_R())
+ self.antennafield_proxy.ANT_mask_RW = ant_mask
+ self.antennafield_proxy.RCU_band_select_RW = rcu_band_select
# Apply Beamlet configuration
self.beamlet_proxy.subband_select_RW = self._apply_saps_subbands(self.read_saps_subband_R())
@@ -143,44 +140,46 @@ class Observation(lofar_device):
self.digitalbeam_proxy.antenna_select_RW = self._apply_saps_antenna_select(self.read_antenna_mask_R())
# Apply Tile Beam pointing direction
- self.tilebeam_proxy.Pointing_direction_RW = [tuple(self.read_tile_beam_R())] * self.antennafield_proxy.nr_antennas_R
+ tile_beam = self.read_tile_beam_R()
+ if tile_beam is not None:
+ self.tilebeam_proxy.Pointing_direction_RW = [tuple(tile_beam)] * self.antennafield_proxy.nr_antennas_R
- logger.info(f"Started the observation with ID={self._observation_id}.")
+ logger.info(f"Started the observation with ID={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_id
+ 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._stop_time.timestamp()
+ return self._observation_settings.stop_time.timestamp()
@only_in_states([DevState.STANDBY, DevState.ON])
@fault_on_error()
@log_exceptions()
def read_antenna_mask_R(self):
"""Return the antenna_mask_R attribute."""
- return self._antenna_mask
+ return self._observation_settings.antenna_mask
@only_in_states([DevState.STANDBY, DevState.ON])
@fault_on_error()
@log_exceptions()
def read_filter_R(self):
"""Return the filter_R attribute."""
- return self._filter
+ 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 self._saps_subband
+ return [sap.subbands for sap in self._observation_settings.SAPs]
@only_in_states([DevState.STANDBY, DevState.ON])
@fault_on_error()
@@ -194,27 +193,35 @@ class Observation(lofar_device):
@log_exceptions()
def read_tile_beam_R(self):
"""Return the tile_beam_R attribute."""
- return self._tile_beam
+ 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._first_beamlet
+ return self._observation_settings.first_beamlet
@fault_on_error()
@log_exceptions()
def read_observation_settings_RW(self):
"""Return current observation_parameters string"""
- return self._observation_settings
+ return None if self._observation_settings is None else self._observation_settings.to_json()
@only_in_states([DevState.OFF])
@fault_on_error()
@log_exceptions()
def write_observation_settings_RW(self, parameters: str):
"""No validation on configuring parameters as task of control device"""
- self._observation_settings = parameters
+ try:
+ self._observation_settings = ObservationSettings.from_json(parameters)
+ except ValidationError as e:
+ self._observation_settings = None
+ #Except.throw_exception("IllegalCommand", e.message, __name__)
@only_when_on()
@fault_on_error()
@@ -225,20 +232,17 @@ class Observation(lofar_device):
# value
return time()
- def _build_saps_pointing(self, parameters:dict):
+ def _build_saps_pointing(self, parameters: ObservationSettings):
""" Build the sap pointing array preserving the correct order from JSON """
saps_pointing = []
for i in range(0, self._num_saps):
- pointing_direction = parameters["SAPs"][i]['pointing']
- saps_pointing.insert(i,(pointing_direction['direction_type'], f"{pointing_direction['angle1']}deg", f"{pointing_direction['angle2']}deg"))
+ pointing_direction = parameters.SAPs[i].pointing
+ saps_pointing.insert(i, (
+ pointing_direction.direction_type, f"{pointing_direction.angle1}deg",
+ f"{pointing_direction.angle2}deg"))
return saps_pointing
- def _build_tilebeam_pointing(self, parameters:dict):
- """ Build the sap pointing array preserving the correct order from JSON """
- pointing_direction = parameters["tile_beam"]
- return [str(pointing_direction['direction_type']), f"{pointing_direction['angle1']}deg", f"{pointing_direction['angle2']}deg"]
-
- def _apply_antennafield_settings(self, antenna_mask:list, filter_name:str):
+ def _apply_antennafield_settings(self, antenna_mask: list, filter_name: str):
""" Convert an array of antenna indexes into a boolean mask array and
retrieve the RCU band from filter name, returning the correct format for
AntennaField device
@@ -251,7 +255,7 @@ class Observation(lofar_device):
RCU_band_select_RW[a] = rcu_band
return numpy.array(ANT_mask_RW), numpy.array(RCU_band_select_RW)
- def _apply_saps_subbands(self, sap_subbands:list):
+ 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)
@@ -260,23 +264,25 @@ class Observation(lofar_device):
subband_select[first_beamlet:len(sap_subbands)] = sap_subbands
return subband_select
- def _apply_saps_pointing(self, sap_pointing:list):
+ 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
+ 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)
# Insert pointing values starting from the first beamlet
pointing_direction[first_beamlet:len(sap_pointing)] = sap_pointing
return tuple(pointing_direction)
- def _apply_saps_antenna_select(self, antenna_mask:list):
+ def _apply_saps_antenna_select(self, antenna_mask: list):
""" Convert an array of antenna indexes into a boolean select array"""
antenna_select = numpy.array([[False] * N_beamlets_ctrl] * self._num_inputs)
first_beamlet = numpy.array(self.read_first_beamlet_R(), dtype=numpy.int64)
for a in antenna_mask:
for i in range(first_beamlet, N_beamlets_ctrl):
- antenna_select[a,i] = True
+ antenna_select[a, i] = True
return antenna_select
+
# ----------
# Run server
# ----------
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation_control.py b/tangostationcontrol/tangostationcontrol/devices/observation_control.py
index 2b7b6690b..f02571f86 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation_control.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation_control.py
@@ -5,20 +5,16 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from json import loads
-import jsonschema
-from jsonschema import Draft7Validator, FormatChecker
import logging
-import time
-from datetime import datetime
import numpy
-from tango import Except, DevFailed, DevState, AttrWriteType, DebugIt, DeviceProxy, Util, DevBoolean, DevString
+from tango import Except, DevState, AttrWriteType, DebugIt, Util, DevBoolean, DevString
from tango.server import Device, command, attribute
-from tango import EventType
from tangostationcontrol.common.entrypoint import entry
from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
+from tangostationcontrol.common import ObservationController
+from tangostationcontrol.configuration import ObservationSettings
from tangostationcontrol.devices.device_decorators import only_when_on, fault_on_error
from tangostationcontrol.devices.lofar_device import lofar_device
from tangostationcontrol.devices.observation import Observation
@@ -31,14 +27,19 @@ __all__ = ["ObservationControl", "main"]
@device_logging_to_python()
class ObservationControl(lofar_device):
""" Observation Control Device Server for LOFAR2.0
- The ObservationControl Tango device controls the instantiation of a Tango Dynamic Device from the Observation class.  ObservationControl then keeps a record of the Observation devices and if they are still alive.
+ The ObservationControl Tango device controls the instantiation of a Tango Dynamic Device from the Observation class.
+ ObservationControl then keeps a record of the Observation devices and if they are still alive.
- At the end of an observation ObservationControl checks if the respective Observation device has stopped its execution and releases it. If the Observation device has not stopped its execution yet, it is attempted to forcefully stop the execution of the Observation device.  Then the Observation device is removed from the list of running observations.
+ At the end of an observation ObservationControl checks if the respective Observation device has stopped its
+ execution and releases it. If the Observation device has not stopped its execution yet, it is attempted to
+ forcefully stop the execution of the Observation device.
- The Observation devices are responsible for the "real" execution of an observation. They get references to the hardware devices that are needed to set values in the relevant Control Points. The Observation device performs only a check if enough parameters are available to perform the set-up.
+ The Observation devices are responsible for the "real" execution of an observation. They get references to the
+ hardware devices that are needed to set values in the relevant Control Points. The Observation device performs only
+ a check if enough parameters are available to perform the set-up.
Essentially this is what happens:
- Somebody calls ObservationControl.start_observation(parameters). Then ObservationControl will perform:
+ Somebody calls ObservationControl.start_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
@@ -48,11 +49,12 @@ class ObservationControl(lofar_device):
- Subscribe to the Observation.running MP's periodic event
- Register the observation in the dict self.running_observations[ID]
- The Observation updates the MP every second with the current time
- - The callback gets called periodically. It checks if MP value > stop (stored in the dict under the obs IDS.  By this it can determine if the observation is done.
+ - The callback gets called periodically. It checks if MP value > stop (stored in the dict under the obs IDS.
+ By this it can determine if the observation is done.
- if MP value > observation end
- Remove observation ID from running obs dict
- Unsubscribe from the MP's event
- - Call off()Â
+ - Call off()
- Remove the device from the Tango DB which will make the device disappear
This should in broad strokes pretty much cover any type of observation.
@@ -68,85 +70,34 @@ class ObservationControl(lofar_device):
- is_observation_running(obs_id) -> bool
MPs
- - array[int] running_observations
- string version
"""
- # JSON Schema
- OBSERVATION_SETTINGS_SCHEMA = {
- "$schema": "http://json-schema.org/draft-07/schema",
- "type": "object",
- "required": [],
- "properties": {
- "observation_id": {"type": "number", "minimum": 1},
- "stop_time": {"type": "string", "format": "date-time"},
- "antenna_mask": {"type": "array"},
- "filter": {"type": "string"},
- "SAPs": {"type": "array",
- "minItems": 1,
- "items": {
- "type": "object",
- "properties": {
- "subbands": {"type": "array"},
- "pointing": {"type": "object",
- "properties": {
- "angle1" : {"type": "number"},
- "angle2" : {"type": "number"},
- "direction_type": {"type": "string"}
- }
- }
- }
- }
- },
- "tile_beam": {"type": "object",
- "properties": {
- "angle1" : {"type": "number"},
- "angle2" : {"type": "number"},
- "direction_type": {"type": "string"}
- }
- },
- "first_beamlet": {"type": "number", "minimum": 0}
- },
- }
- VALIDATOR = Draft7Validator(OBSERVATION_SETTINGS_SCHEMA, format_checker=FormatChecker())
# Attributes
- running_observations_R = attribute(dtype = (numpy.int64, ), access = AttrWriteType.READ)
+ running_observations_R = attribute(dtype=(numpy.int64,), access=AttrWriteType.READ)
+
+ def __init__(self, cl, name):
+ super().__init__(cl, name)
+
+ # The top level tango domain is the left-most part of a
+ # device's name.
+ self.myTangoDomain: str = self.get_name().split('/')[0]
+
+ self._observation_controller: ObservationController = ObservationController(self.myTangoDomain)
# Core functions
@log_exceptions()
@DebugIt()
def init_device(self):
- """ Keep all the observation objects for currently running
- observations in this dict. The key is the observation idea
- and the value is a dict of the parameters.
- The value dict contains at least these key/value pairs:
- ["proxy"]: tango.DeviceProxy - the DeviceProxy for the Observation object
- ["event_id"]: int
- ["parameters"]: {parameters as passed to the start_observation call}
- ["stop_time"]: timestamp when the observation is supposed to stop.
- ["device_name"]: name of the device instance in Tango DB
- ["class_name"]: name of the device's class, needed to create an instance
- on the fly.
- """
Device.init_device(self)
self.set_state(DevState.OFF)
- self.running_observations = {}
-
- # The pyTango.Util class is a singleton and every DS can only
- # have one instance of it.
- self.tango_util = Util.instance()
-
# Increase the number of polling threads for this device server.
- self.tango_util.set_polling_threads_pool_size(10)
-
- # The top level tango domain is the left-most part of a
- # device's name.
- self.myTangoDomain = self.get_name().split('/')[0]
+ Util.instance().set_polling_threads_pool_size(10)
# Lifecycle functions
def configure_for_initialise(self):
- self.running_observations.clear()
+ self._observation_controller = ObservationController(self.myTangoDomain)
def configure_for_off(self):
self.stop_all_observations()
@@ -155,173 +106,16 @@ class ObservationControl(lofar_device):
@fault_on_error()
@log_exceptions()
def read_running_observations_R(self):
- obs = [ key for key in self.running_observations ]
- return obs
-
- @log_exceptions()
- def observation_running_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.
- """
- if event.err:
- # Something is fishy with this event.
- logger.warning(f"The Observation device {event.device} 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={event}")
- return
-
- # Get the Observation ID from the sending device.
- obs_id = event.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(f"Received an observation_running event for the observation with ID={obs_id}. 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.")
- return
-
- 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(f"Received an observation_running event for the observation with ID={obs_id}. 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.")
- return
-
- @only_when_on()
- @log_exceptions()
- def check_and_convert_parameters(self, parameters: DevString = None) -> dict:
- """
- The parameters are passed as JSON in a char[].
- Conveniently json.loads() creates a dict from the input.
- :param parameters: The parameters as JSON
- :type parameters: DevVarUCharArray
- :return: None
- """
- # Convert the input parameter to a dict.
- parameter_dict = loads(parameters)
-
- logger.debug("incoming parameter_array = %s, parameter_dict = %s", parameters, parameter_dict)
-
- # Parameter check, do not execute an observation in case
- # the parameters are not sufficiently defined.
- obs_id = int(parameter_dict["observation_id"])
- stop_datetime = datetime.fromisoformat(parameter_dict["stop_time"])
- try:
- self.VALIDATOR.validate(parameter_dict)
- except jsonschema.exceptions.ValidationError as error:
- Except.throw_exception("IllegalCommand", error, __name__)
- # Check further properties that cannot be validated through a JSON schema
- if stop_datetime <= datetime.now():
- error = f"Cannot start an observation with ID={obs_id} because the parameter stop_time parameter value=\"{stop_datetime}\" is invalid. Set a stop_time parameter later in time than the start time."
- Except.throw_exception("IllegalCommand", error, __name__)
- return parameter_dict
+ return self._observation_controller.running_observations
# API
- @command(dtype_in = DevString)
+ @command(dtype_in=DevString)
@only_when_on()
@log_exceptions()
def start_observation(self, parameters: DevString = None):
- # Store everything about the observation in this dict. I store this
- # dict at the end in self.running_observations.
- observation = {"parameters": self.check_and_convert_parameters(parameters)}
- observation_id = observation['parameters']['observation_id']
-
- # The class name of the Observation class is needed to create and
- # delete the device.
- class_name = Observation.__name__
- observation["class_name"] = class_name
-
- # Generate the Tango DB name for the Observation device.
- device_name = f"{self.myTangoDomain}/{class_name}/{observation_id}"
- observation["device_name"] = device_name
-
- try:
- # Create the Observation device and instantiate it.
- self.tango_util.create_device(class_name, device_name)
- except DevFailed as ex:
- if ex.args[0].desc == f"The device {device_name.lower()} is already defined in the database" and self.is_observation_running(observation_id) is False :
- self.tango_util.delete_device(class_name, device_name)
- error_string = f"Cannot create the Observation device {device_name} because it is already present in the Database but it is not running. Try to re-run the start_observation command"
- logger.exception(error_string)
- Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
- else:
- error_string = f"Cannot create the Observation device instance {device_name} for ID={observation_id}. This means that the observation did not start."
- logger.exception(error_string)
- Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
-
- try:
- # Instantiate a dynamic Tango Device "Observation".
- device_proxy = DeviceProxy(device_name)
- observation["device_proxy"] = device_proxy
-
- # Configure the dynamic device its attribute for the observation
- # parameters.
- device_proxy.observation_settings_RW = parameters
-
- # Take the Observation device through the motions. Pass the
- # entire JSON set of parameters so that it can pull from it what it
- # needs.
- device_proxy.Initialise()
-
- # The call to On will actually tell the Observation device to
- # become fully active.
- device_proxy.On()
- except DevFailed as ex:
- # Remove the device again.
- self.tango_util.delete_device(class_name, device_name)
- error_string = f"Cannot access the Observation device instance for observation ID={observation_id} with device class name={class_name} and device instance name={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__)
-
- try:
- # Subscribe to the obs.observation_running MP
- #
- # Generate the name for the Observation.observation_running
- # MP.
- attribute_name = f"{device_name}/observation_running_R"
- observation["attribute_name"] = attribute_name
-
- # 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.
- device_proxy.poll_attribute(attribute_name.split('/')[-1], 1000)
-
- # Note: I update the running_observations dict already here because
- # the addition of an event listener immediately triggers that
- # event. And since the call back checks if the obs_id is in the dict
- # this triggers an error message if the ID is not already known.
- # There is no harm in copying the dict twice.
- self.running_observations[observation_id] = observation
-
- # Right. Now subscribe to periodic events.
- event_id = device_proxy.subscribe_event(attribute_name.split('/')[-1], EventType.PERIODIC_EVENT, self.observation_running_callback)
- observation["event_id"] = event_id
-
- # Finally update the self.running_observation dict's entry of this
- # observation with the complete set of info.
- self.running_observations[observation_id] = observation
- logger.info(f"Successfully started an observation with ID={observation_id}.")
- except DevFailed as ex:
- self.tango_util.delete_device(class_name, 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, observation_id, Observation.__name__, device_name)
- Except.re_throw_exception(ex, "DevFailed", error_string, __name__)
-
- @command(dtype_in = numpy.int64)
+ self._observation_controller.start_observation(ObservationSettings.from_json(parameters))
+
+ @command(dtype_in=numpy.int64)
@only_when_on()
@log_exceptions()
def stop_observation(self, obs_id: numpy.int64 = 0):
@@ -331,54 +125,13 @@ class ObservationControl(lofar_device):
# 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.is_observation_running(obs_id) is False:
+ 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(f"Stopping the observation with ID={obs_id}.")
- # Fetch the obs data and remove it from the dict of
- # currently running observations.
- observation = self.running_observations.pop(obs_id)
- device_proxy = observation.pop("device_proxy")
-
- # Check if the device has not terminated itself in the meanwhile.
- try:
- device_proxy.ping()
- except DevFailed:
- logger.warning(f"The device for the Observation with ID={obs_id} has unexpectedly already disappeared. It is advised to check the logs up to 10s prior to this message to see what happened.")
- else:
- # Unsubscribe from the subscribed event.
- event_id = observation.pop("event_id")
- device_proxy.unsubscribe_event(event_id)
-
- # Tell the Observation device to stop the running
- # observation. This is a synchronous call and the clean-up
- # does not take long.
- 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 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(f"Successfully stopped the observation with ID={obs_id}")
- else:
- logger.warning(f"Could not shut down the Observation device ( {observation['device_name']} ) for observation ID={obs_id}. This means that there is a chance for a memory leak. Will continue anyway and forcefully delete the Observation object.")
-
- # Finally remove the device object from the Tango DB.
- try:
- self.tango_util.delete_device(observation["class_name"], observation["device_name"])
- except DevFailed:
- logger.warning(f"Something went wrong when the device {observation['device_name']} 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._observation_controller.stop_observation(obs_id)
@command()
@only_when_on()
@@ -386,14 +139,13 @@ class ObservationControl(lofar_device):
def stop_all_observations(self):
# Make a copy of the running_observations dict. This
# should prevent race conditions.
- if self.is_any_observation_running():
- # Make certain that the dict does not get modified
- # while I am iterating over it.
- active_obs = self.running_observations.copy()
- for obs_id in active_obs.keys():
- self.stop_observation(obs_id)
-
- @command(dtype_in = numpy.int64, dtype_out = DevBoolean)
+ 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()
+
+ @command(dtype_in=numpy.int64, dtype_out=DevBoolean)
@only_when_on()
@log_exceptions()
def is_observation_running(self, obs_id: numpy.int64 = -1) -> DevBoolean:
@@ -402,14 +154,13 @@ class ObservationControl(lofar_device):
# 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__)
- observation = self.running_observations.get(obs_id)
- return observation is not None
+ return self._observation_controller.is_observation_running(obs_id)
- @command(dtype_out = DevBoolean)
+ @command(dtype_out=DevBoolean)
@only_when_on()
@log_exceptions()
def is_any_observation_running(self) -> DevBoolean:
- return len(self.running_observations) > 0
+ return self._observation_controller.is_any_observation_running()
# ----------
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
index 32b4f5841..d4f0e729c 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation.py
@@ -106,8 +106,8 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
"""Initialize an observation with _invalid_ JSON"""
self.proxy.off()
- self.proxy.observation_settings_RW = "{}"
with self.assertRaises(DevFailed):
+ self.proxy.observation_settings_RW = "{}"
self.proxy.Initialise()
self.assertEqual(DevState.FAULT, self.proxy.state())
@@ -125,6 +125,7 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
def test_attribute_match(self):
"""Test that JSON data is exposed to attributes"""
+ #failing
data = loads(self.VALID_JSON)
stop_timestamp = datetime.fromisoformat(data["stop_time"]).timestamp()
observation_id = data["observation_id"]
@@ -209,6 +210,7 @@ class TestDeviceObservation(AbstractTestBases.TestDeviceBase):
self.assertListEqual(digitalbeam_proxy.antenna_select_RW.tolist()[10], [False] * N_beamlets_ctrl)
def test_apply_tilebeam(self):
+ #failing
"""Test that attribute tilebeam is correctly applied"""
tilebeam_proxy = self.setup_tilebeam_proxy()
pointing_direction = [("J2000","0deg","0deg")] * DEFAULT_N_HBA_TILES
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
index 1e6c7d72a..4b5489505 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_observation_control.py
@@ -7,31 +7,30 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from tango import DevState
-from tango import DevFailed
-
-import numpy
import json
from datetime import datetime
from datetime import timedelta
-from tangostationcontrol.test.devices.test_observation_base import TestObservationBase
+import numpy
+from tango import DevFailed
+from tango import DevState
+
+from tangostationcontrol.common.constants import DEFAULT_N_HBA_TILES
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+from tangostationcontrol.test.devices.test_observation_base import TestObservationBase
from .base import AbstractTestBases
-from tangostationcontrol.common.constants import DEFAULT_N_HBA_TILES
class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
-
ANTENNA_TO_SDP_MAPPING = [
- "0", "0", "0", "1", "0", "2", "0", "3", "0", "4", "0", "5",
- "1", "0", "1", "1", "1", "2", "1", "3", "1", "4", "1", "5",
- "2", "0", "2", "1", "2", "2", "2", "3", "2", "4", "2", "5",
- "3", "0", "3", "1", "3", "2", "3", "3", "3", "4", "3", "5",
- "4", "0", "4", "1", "4", "2", "4", "3", "4", "4", "4", "5",
- "5", "0", "5", "1", "5", "2", "5", "3", "5", "4", "5", "5",
- "6", "0", "6", "1", "6", "2", "6", "3", "6", "4", "6", "5",
- "7", "0", "7", "1", "7", "2", "7", "3", "7", "4", "7", "5",
+ "0", "0", "0", "1", "0", "2", "0", "3", "0", "4", "0", "5",
+ "1", "0", "1", "1", "1", "2", "1", "3", "1", "4", "1", "5",
+ "2", "0", "2", "1", "2", "2", "2", "3", "2", "4", "2", "5",
+ "3", "0", "3", "1", "3", "2", "3", "3", "3", "4", "3", "5",
+ "4", "0", "4", "1", "4", "2", "4", "3", "4", "4", "4", "5",
+ "5", "0", "5", "1", "5", "2", "5", "3", "5", "4", "5", "5",
+ "6", "0", "6", "1", "6", "2", "6", "3", "6", "4", "6", "5",
+ "7", "0", "7", "1", "7", "2", "7", "3", "7", "4", "7", "5",
]
def setUp(self):
@@ -62,10 +61,10 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
def setup_antennafield_proxy(self):
# setup AntennaField
antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
- control_mapping = [[1,i] for i in range(DEFAULT_N_HBA_TILES)]
+ control_mapping = [[1, i] for i in range(DEFAULT_N_HBA_TILES)]
antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
- "Power_to_RECV_mapping": numpy.array(control_mapping).flatten(),
- "Antenna_to_SDP_Mapping": self.ANTENNA_TO_SDP_MAPPING})
+ "Power_to_RECV_mapping": numpy.array(control_mapping).flatten(),
+ "Antenna_to_SDP_Mapping": self.ANTENNA_TO_SDP_MAPPING})
antennafield_proxy.off()
antennafield_proxy.warm_boot()
antennafield_proxy.set_defaults()
@@ -123,7 +122,6 @@ class TestObservationControlDevice(AbstractTestBases.TestDeviceBase):
parameters = json.loads(self.VALID_JSON)
parameters['observation_id'] = -1
-
self.on_device_assert(self.proxy)
self.assertRaises(
DevFailed, self.proxy.start_observation, json.dumps(parameters))
diff --git a/tangostationcontrol/tangostationcontrol/test/common/test_observation_controller.py b/tangostationcontrol/tangostationcontrol/test/common/test_observation_controller.py
new file mode 100644
index 000000000..d9ddbbdcc
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/common/test_observation_controller.py
@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+import importlib
+import sys
+from datetime import datetime
+from unittest import mock
+from unittest.mock import Mock
+
+from tango import DevState
+
+from tangostationcontrol.common import ObservationController
+from tangostationcontrol.common.observation_controller import RunningObservation
+from tangostationcontrol.configuration import ObservationSettings, Pointing, Sap
+from tangostationcontrol.test import base
+
+
+@mock.patch('tango.Util.instance')
+class TestObservationController(base.TestCase):
+ def test_is_any_observation_running(self, _):
+ sut = ObservationController("DMR")
+ self.assertFalse(sut.is_any_observation_running())
+ sut._running_observations[1] = {}
+ self.assertTrue(sut.is_any_observation_running())
+
+ def test_is_observation_running(self, _):
+ sut = ObservationController("DMR")
+ self.assertFalse(sut.is_observation_running(2))
+ sut._running_observations[2] = {}
+ self.assertTrue(sut.is_observation_running(2))
+
+ def test_stop_all_observations_no_running(self, _):
+ sut = ObservationController("DMR")
+ sut.stop_all_observations()
+
+
+@mock.patch('tango.Util.instance')
+class TestRunningObservation(base.TestCase):
+ SETTINGS = ObservationSettings(5, datetime.fromisoformat("2022-10-26T11:35:54.704150"), [3, 2, 1],
+ "filter settings",
+ [Sap([3, 2], Pointing(1.2, 2.1, "LMN")),
+ Sap([1], Pointing(3.3, 4.4, "MOON"))])
+
+ def test_properties(self, _):
+ sut = RunningObservation("DMR", TestRunningObservation.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)
+
+ def test_create_tango_device(self, tu_mock):
+ sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ sut.create_tango_device()
+
+ @mock.patch('tango.DeviceProxy')
+ def test_create_device_proxy(self, dp_mock, tu_mock):
+ importlib.reload(sys.modules[RunningObservation.__module__])
+ sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ sut.create_device_proxy()
+
+ self.assertEqual(dp_mock.return_value.observation_settings_RW, TestRunningObservation.SETTINGS.to_json())
+ dp_mock.return_value.Initialise.assert_called()
+ dp_mock.return_value.On.assert_called()
+
+ def test_subscribe(self, _):
+ def dummy():
+ pass
+
+ sut = RunningObservation("DMR", TestRunningObservation.SETTINGS)
+ dp_mock = Mock()
+ sut._device_proxy = dp_mock
+ sut.subscribe(dummy)
+
+ 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)
+
+ dp_mock = Mock()
+ dp_mock.state.return_value = DevState.OFF
+
+ sut._device_proxy = dp_mock
+
+ sut.shutdown()
+
+ 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/__init__.py b/tangostationcontrol/tangostationcontrol/test/configuration/__init__.py
similarity index 100%
rename from tangostationcontrol/__init__.py
rename to tangostationcontrol/tangostationcontrol/test/configuration/__init__.py
diff --git a/tangostationcontrol/tangostationcontrol/test/configuration/_mock_requests.py b/tangostationcontrol/tangostationcontrol/test/configuration/_mock_requests.py
new file mode 100644
index 000000000..ae9720e1b
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/configuration/_mock_requests.py
@@ -0,0 +1,148 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+import json
+
+POINTING_SCHEMA = '''
+{
+ "type": "object",
+ "required": [
+ "angle1",
+ "angle2",
+ "direction_type"
+ ],
+ "properties": {
+ "angle1": {
+ "default": 0.6624317181687094,
+ "description": "First angle (e.g. RA)",
+ "title": "Angle 1",
+ "type": "number"
+ },
+ "angle2": {
+ "default": 1.5579526427549426,
+ "description": "Second angle (e.g. DEC)",
+ "title": "Angle 2",
+ "type": "number"
+ },
+ "direction_type": {
+ "default": "J2000",
+ "description": "",
+ "enum": [
+ "J2000",
+ "AZELGEO",
+ "LMN",
+ "SUN",
+ "MOON",
+ "MERCURY",
+ "VENUS",
+ "MARS",
+ "JUPITER",
+ "SATURN",
+ "URANUS",
+ "NEPTUNE",
+ "PLUTO"
+ ],
+ "title": "Reference frame",
+ "type": "string"
+ }
+ }
+}
+'''
+
+SAP_SCHEMA = '''
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "subbands",
+ "pointing"
+ ],
+ "properties": {
+ "subbands": {
+ "type": "array",
+ "minItems": 1,
+ "items": {
+ "type": "number"
+ }
+ },
+ "pointing": {
+ "$ref": "pointing.json"
+ }
+ }
+}
+'''
+
+OBSERVATION_SETTINGS_SCHEMA = '''
+{
+ "$schema": "http://json-schema.org/draft-07/schema",
+ "type": "object",
+ "required": [
+ "observation_id",
+ "stop_time",
+ "antenna_mask",
+ "filter",
+ "SAPs"
+ ],
+ "properties": {
+ "observation_id": {
+ "type": "number",
+ "minimum": 1
+ },
+ "stop_time": {
+ "type": "string",
+ "format": "date-time"
+ },
+ "antenna_mask": {
+ "type": "array",
+ "minItems": 1,
+ "items": {
+ "type": "number"
+ }
+ },
+ "filter": {
+ "type": "string"
+ },
+ "SAPs": {
+ "type": "array",
+ "minItems": 1,
+ "items": {
+ "$ref": "sap.json"
+ }
+ },
+ "tile_beam": {
+ "$ref": "pointing.json"
+ },
+ "first_beamlet": {
+ "type": "number",
+ "minimum": 0
+ }
+ }
+}
+'''
+
+
+def mocked_requests_get(*args, **kwargs):
+ class MockResponse:
+ def __init__(self, text, status_code):
+ self.text = text
+ self.status_code = status_code
+
+ def raise_for_status(self):
+ pass
+
+ def json(self):
+ return json.loads(self.text)
+
+ if args[0] == 'http://schemas/pointing.json':
+ return MockResponse(POINTING_SCHEMA, 200)
+ elif args[0] == 'http://schemas/sap.json':
+ return MockResponse(SAP_SCHEMA, 200)
+ elif args[0] == 'http://schemas/observation-settings.json':
+ return MockResponse(OBSERVATION_SETTINGS_SCHEMA, 200)
+
+ return MockResponse(None, 404)
diff --git a/tangostationcontrol/tangostationcontrol/test/configuration/test_observation_settings.py b/tangostationcontrol/tangostationcontrol/test/configuration/test_observation_settings.py
new file mode 100644
index 000000000..718595392
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/configuration/test_observation_settings.py
@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+from datetime import datetime
+from unittest import mock
+
+import requests
+from jsonschema.exceptions import ValidationError, RefResolutionError
+
+from tangostationcontrol.configuration import Pointing, ObservationSettings, Sap
+from tangostationcontrol.test import base
+from tangostationcontrol.test.configuration._mock_requests import mocked_requests_get
+
+
+@mock.patch('requests.get', side_effect=mocked_requests_get)
+class TestObservationSettings(base.TestCase):
+ def test_from_json(self, _):
+ sut = ObservationSettings.from_json('{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
+ '"antenna_mask": [3, 2, 1], "filter": "filter_settings",'
+ '"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_mask, [3, 2, 1])
+ self.assertEqual(sut.filter, "filter_settings")
+ self.assertEqual(len(sut.SAPs), 1)
+
+ sut = ObservationSettings.from_json('{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
+ '"antenna_mask": [3, 2, 1], "filter": "filter_settings",'
+ '"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"} }')
+
+ self.assertEqual(sut.tile_beam.angle1, 2.2)
+ self.assertEqual(sut.tile_beam.angle2, 3.1)
+ self.assertEqual(sut.tile_beam.direction_type, "MOON")
+
+ sut = ObservationSettings.from_json('{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
+ '"antenna_mask": [3, 2, 1], "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}')
+
+ self.assertEqual(sut.first_beamlet, 2)
+
+ def test_from_json_type_missmatch(self, _):
+ for json in [
+ '{"observation_id": "3", "stop_time": "2012-04-23T18:25:43", "antenna_mask": [3, 2, 1], "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_mask": [3, 2, 1], "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_mask": ["3", 2, 1], "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_mask": "3", "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_mask": [3, 2, 1], "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_mask": [3, 2, 1], "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_mask": [3, 2, 1], "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_mask": [3, 2, 1], "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_mask": [3, 2, 1], "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"}'
+ ]:
+ with self.assertRaises((ValidationError, ValueError)):
+ ObservationSettings.from_json(json)
+
+ def test_from_json_missing_fields(self, _):
+ for json in [
+ '{"stop_time": "2012-04-23T18:25:43", "antenna_mask": [3, 2, 1], "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, "antenna_mask": [3, 2, 1], "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", "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_mask": [], "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_mask": [3, 2, 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_mask": [3, 2, 1], "filter": "filter_settings","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_mask": [3, 2, 1], "filter": "filter_settings","SAPs": [],"tile_beam": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}, "first_beamlet": 2}'
+ ]:
+ with self.assertRaises((ValidationError, ValueError)):
+ ObservationSettings.from_json(json)
+
+ def test_to_json(self, _):
+ sut = ObservationSettings(5, datetime.fromisoformat("2022-10-26T11:35:54.704150"), [3, 2, 1], "filter settings",
+ [Sap([3, 2], Pointing(1.2, 2.1, "LMN")),
+ Sap([1], Pointing(3.3, 4.4, "MOON"))])
+ self.assertEqual(sut.to_json(), '{"observation_id": 5, "stop_time": "2022-10-26T11:35:54.704150", ' \
+ '"antenna_mask": [3, 2, 1], "filter": "filter settings", "SAPs": ' \
+ '[{"subbands": [3, 2], "pointing": {"angle1": 1.2, "angle2": 2.1, ' \
+ '"direction_type": "LMN"}}, {"subbands": [1], "pointing": {"angle1": 3.3, ' \
+ '"angle2": 4.4, "direction_type": "MOON"}}], "first_beamlet": 0}')
+
+ def test_throw_exception_if_schema_not_available(self, mock_get):
+ ObservationSettings.VALIDATOR = None
+ mock_get.side_effect = requests.exceptions.Timeout
+ with self.assertRaises(RefResolutionError):
+ ObservationSettings.from_json('{"observation_id": 3, "stop_time": "2012-04-23T18:25:43", '
+ '"antenna_mask": [3, 2, 1], "filter": "filter_settings",'
+ '"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 ['{"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)
diff --git a/tangostationcontrol/tangostationcontrol/test/configuration/test_pointing.py b/tangostationcontrol/tangostationcontrol/test/configuration/test_pointing.py
new file mode 100644
index 000000000..7f1e3edfc
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/configuration/test_pointing.py
@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+from unittest import mock
+
+import requests
+from jsonschema.exceptions import ValidationError, RefResolutionError
+
+from tangostationcontrol.configuration import Pointing
+from tangostationcontrol.test import base
+from tangostationcontrol.test.configuration._mock_requests import mocked_requests_get
+
+
+@mock.patch('requests.get', side_effect=mocked_requests_get)
+class TestPointing(base.TestCase):
+ def test_from_json(self, _):
+ ps = Pointing.from_json('{"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}')
+
+ self.assertEqual(1.2, ps.angle1)
+ self.assertEqual(2.1, ps.angle2)
+ self.assertEqual('LMN', ps.direction_type)
+
+ def test_from_json_type_missmatch(self, _):
+ for json in ['{"angle1":"1.2", "angle2": 2.1, "direction_type":"LMN"}',
+ '{"angle1":1.2, "angle2": "2.1", "direction_type":"LMN"}',
+ '{"angle1":1.2, "angle2": 2.1, "direction_type":"ABC"}'
+ ]:
+ with self.assertRaises(ValidationError):
+ Pointing.from_json(json)
+
+ def test_from_json_missing_fields(self, _):
+ for json in ['{"angle2": 2.1, "direction_type":"LMN"}',
+ '{"angle1":1.2, "direction_type":"LMN"}',
+ '{"angle1":1.2, "angle2": 2.1}'
+ ]:
+ with self.assertRaises(ValidationError):
+ Pointing.from_json(json)
+
+ def test_to_json(self, _):
+ ps = Pointing(1.3, 2.3, "URANUS")
+ self.assertEqual(ps.to_json(), '{"angle1": 1.3, "angle2": 2.3, "direction_type": "URANUS"}')
+
+ def test_throw_exception_if_schema_not_available(self, mock_get):
+ Pointing.VALIDATOR = None
+ mock_get.side_effect = requests.exceptions.Timeout
+ with self.assertRaises(RefResolutionError):
+ Pointing.from_json('{"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}')
+ self.assertEqual(5, mock_get.call_count)
diff --git a/tangostationcontrol/tangostationcontrol/test/configuration/test_sap_settings.py b/tangostationcontrol/tangostationcontrol/test/configuration/test_sap_settings.py
new file mode 100644
index 000000000..f457e963b
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/test/configuration/test_sap_settings.py
@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+from unittest import mock
+
+import requests
+from jsonschema.exceptions import ValidationError, RefResolutionError
+
+from tangostationcontrol.configuration import Pointing, Sap
+from tangostationcontrol.test import base
+from tangostationcontrol.test.configuration._mock_requests import mocked_requests_get
+
+
+@mock.patch('requests.get', side_effect=mocked_requests_get)
+class TestSapSettings(base.TestCase):
+ def test_from_json(self, _):
+ sap = Sap.from_json(
+ '{"subbands": [3, 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}')
+
+ self.assertEqual(sap.subbands, [3, 2, 1])
+ self.assertEqual(sap.pointing.angle1, 1.2)
+ self.assertEqual(sap.pointing.angle2, 2.1)
+ self.assertEqual(sap.pointing.direction_type, 'LMN')
+
+ def test_from_json_type_missmatch(self, _):
+ for json in ['{"subbands": ["3", 2, 1], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": "3", "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": 3, "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": 3, "pointing": {"angle1":"1.2", "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": "3", "pointing": "test"}',
+ '{"angle1":"1.2", "angle2": 2.1, "direction_type":"LMN"}'
+ ]:
+ with self.assertRaises(ValidationError):
+ Sap.from_json(json)
+
+ def test_from_json_missing_fields(self, _):
+ for json in ['{"subbands": [], "pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"pointing": {"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": [1], "pointing": {"angle2": 2.1, "direction_type":"LMN"}}',
+ '{"subbands": [1]}'
+ ]:
+ with self.assertRaises(ValidationError):
+ Sap.from_json(json)
+
+ def test_to_json(self, _):
+ sut = Sap([3, 2, 1], Pointing(1.3, 2.3, "URANUS"))
+ self.assertEqual(sut.to_json(),
+ '{"subbands": [3, 2, 1], "pointing": {"angle1": 1.3, "angle2": 2.3, "direction_type": "URANUS"}}')
+
+ def test_throw_exception_if_schema_not_available(self, mock_get):
+ Sap.VALIDATOR = None
+ mock_get.side_effect = requests.exceptions.Timeout
+ with self.assertRaises(RefResolutionError):
+ Sap.from_json(
+ '{"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, _):
+ with self.assertRaises(ValidationError):
+ Sap.from_json('{"angle1":1.2, "angle2": 2.1, "direction_type":"LMN"}')
--
GitLab