From 1ca3c5c8ca9c1944e15b06bf13bb592d5cd4c714 Mon Sep 17 00:00:00 2001
From: thijs snijder <snijder@astron.nl>
Date: Fri, 21 Oct 2022 14:40:36 +0200
Subject: [PATCH] moved observationWrapper class to lsc
---
.../observation/observation_control.py | 33 +------------------
.../default/devices/test_observation.py | 4 +--
2 files changed, 3 insertions(+), 34 deletions(-)
diff --git a/tangostationcontrol/tangostationcontrol/devices/observation/observation_control.py b/tangostationcontrol/tangostationcontrol/devices/observation/observation_control.py
index e051d7d50..83040f6c9 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation/observation_control.py
+++ b/tangostationcontrol/tangostationcontrol/devices/observation/observation_control.py
@@ -5,7 +5,7 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from json import loads, dumps
+from json import loads
import jsonschema
from jsonschema import Draft7Validator, FormatChecker
import logging
@@ -28,37 +28,6 @@ logger = logging.getLogger()
__all__ = ["ObservationControl", "main"]
-class ObservationWrapper(object):
- """ This class provides a pythonic interface to the ObservationControl and Observation devices on a station. """
-
- def __init__(self, specification: dict, host: str = "databaseds.tangonet:10000"):
-
- # create device proxies to station at "host"
- self._observation_control_dev = DeviceProxy(f"tango://{host}/STAT/ObservationControl/1")
-
- # convert specification dict to json
- self._specification = dumps(specification)
-
- # get the ID
- self._id = int(specification['observation_id'])
-
- def observation_id(self) -> int:
- return self._id
-
- def start(self):
- self._observation_control_dev.start_observation(self._specification)
-
- def abort(self):
- self._observation_control_dev.stop_observation(self._id)
-
- def observation_proxy(self):
- # return a proxy to the correct Observation device on the station ("STAT/Observation/{observation_id}")
- return DeviceProxy(f"STAT/Observation/{self._id}")
-
- def is_running(self):
- # return whether the observation is still running (ObservationControl.is_observation_running command)
- return self._observation_control_dev.is_observation_running(self._id)
-
@device_logging_to_python()
class ObservationControl(lofar_device):
""" Observation Control Device Server for LOFAR2.0
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
index f13804a09..def1be7cd 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
@@ -11,7 +11,7 @@ from tangostationcontrol.test.devices.test_observation_base import TestObservati
from tangostationcontrol.integration_test import base
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.devices.observation.observation_control import ObservationWrapper
+from lofar_station_client.observation.observation import Observation
from os import environ
from json import loads
@@ -32,7 +32,7 @@ class TestObservation(base.IntegrationTestCase):
specification_dict = loads(TestObservationBase.VALID_JSON)
# create an observation class using the dict and as host just get it using a util function
- observation = ObservationWrapper(specification=specification_dict, host=environ["TANGO_HOST"])
+ observation = Observation(specification=specification_dict, host=environ["TANGO_HOST"])
# Assert the observation is running after starting it
observation.start()
--
GitLab