diff --git a/tangostationcontrol/tangostationcontrol/devices/observation_control.py b/tangostationcontrol/tangostationcontrol/devices/observation_control.py
index c4d6e64ace208180f93b2a0831d02be00471e7da..565074572fa0cbc4f0a6d71ffdec0ac66881c7a1 100644
--- a/tangostationcontrol/tangostationcontrol/devices/observation_control.py
+++ b/tangostationcontrol/tangostationcontrol/devices/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
+from json import loads, dumps
import jsonschema
from jsonschema import Draft7Validator, FormatChecker
import logging
@@ -31,15 +31,16 @@ __all__ = ["ObservationControl", "main"]
class observation_wrapper(object):
""" This class provides a pythonic interface to the ObservationControl and Observation devices on a station. """
- def __init__(self, specification: str, host: str = "localhost:10000"):
+ 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")
- specification_dict = loads(specification)
- self._specification = specification
+ # convert specification dict to json
+ self._specification = dumps(specification)
- specification_dict = loads(specification)
- self._id = int(specification_dict['observation_id'])
+ # get the ID
+ self._id = int(specification['observation_id'])
def observation_id(self) -> int:
return self._id
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
index 1f3f8208557dbd5cfa7a7611e65ae0424cf14102..fdca6f917d67a29e07274ac3d847d98d107849df 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_observation.py
@@ -7,107 +7,25 @@ from tangostationcontrol.devices.observation_control import observation_wrapper
from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
import numpy
+from json import loads
+
from tango import DevState, ApiUtil
class TestObservation(base.IntegrationTestCase):
- NUM_TILES = 48
- NUM_BEAMLETS_CTRL = 488
- NUM_INPUTS = 96
- INPUT_TO_ANTENNA_MAPPING = [
- "0", "1", "2", "3", "4", "5",
- "6", "7", "8", "9", "10", "11",
- "12", "13", "14", "15", "16", "17",
- "18", "19", "20", "21", "22", "23",
- "24", "25", "26", "27", "28", "29",
- "30", "31", "32", "33", "34", "35",
- "36", "37", "38", "39", "40", "41",
- "42", "43", "44", "45", "46", "47",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1",
- "-1", "-1", "-1", "-1", "-1", "-1"
- ]
-
def setUp(self):
- self.recv_proxy = self.setup_recv_proxy()
- self.antennafield_proxy = self.setup_antennafield_proxy()
- self.beamlet_proxy = self.setup_beamlet_proxy()
- self.sdp_proxy = self.setup_sdp_proxy()
- self.digitalbeam_proxy = self.setup_digitalbeam_proxy()
- self.tilebeam_proxy = self.setup_tilebeam_proxy()
- self.observation_control_proxy = self.setup_ObservationControl_proxy()
-
-
- def setup_ObservationControl_proxy(self):
- # setup ObservationControl
- observation_control_proxy = TestDeviceProxy("STAT/ObservationControl/1")
- observation_control_proxy.off()
- observation_control_proxy.warm_boot()
- return observation_control_proxy
-
- def setup_recv_proxy(self):
- # setup RECV
- recv_proxy = TestDeviceProxy("STAT/RECV/1")
- recv_proxy.off()
- recv_proxy.warm_boot()
- recv_proxy.set_defaults()
- return recv_proxy
-
- def setup_sdp_proxy(self):
- # setup SDP
- sdp_proxy = TestDeviceProxy("STAT/SDP/1")
- sdp_proxy.off()
- sdp_proxy.warm_boot()
- return sdp_proxy
-
- def setup_antennafield_proxy(self):
- # setup AntennaField
- antennafield_proxy = TestDeviceProxy("STAT/AntennaField/1")
- control_mapping = [[1, i] for i in range(self.NUM_TILES)]
- antennafield_proxy.put_property({"RECV_devices": ["STAT/RECV/1"],
- "Power_to_RECV_mapping": numpy.array(control_mapping).flatten()})
- antennafield_proxy.off()
- antennafield_proxy.warm_boot()
- antennafield_proxy.set_defaults()
- return antennafield_proxy
-
- def setup_beamlet_proxy(self):
- # setup Digitalbeam
- beamlet_proxy = TestDeviceProxy("STAT/Beamlet/1")
- beamlet_proxy.off()
- beamlet_proxy.warm_boot()
- beamlet_proxy.set_defaults()
- return beamlet_proxy
-
- def setup_digitalbeam_proxy(self):
- # setup Digitalbeam
- digitalbeam_proxy = TestDeviceProxy("STAT/DigitalBeam/1")
- digitalbeam_proxy.put_property({"Input_to_Antenna_Mapping": numpy.array(self.INPUT_TO_ANTENNA_MAPPING).flatten()})
- digitalbeam_proxy.off()
- digitalbeam_proxy.warm_boot()
- digitalbeam_proxy.set_defaults()
- return digitalbeam_proxy
-
- def setup_tilebeam_proxy(self):
- # Setup Tilebeam
- tilebeam_proxy = TestDeviceProxy("STAT/TileBeam/1")
- tilebeam_proxy.off()
- tilebeam_proxy.warm_boot()
- tilebeam_proxy.set_defaults()
- return tilebeam_proxy
-
+ self.observation_control_proxy = TestDeviceProxy("STAT/ObservationControl/1")
+ self.observation_control_proxy.off()
+ self.observation_control_proxy.warm_boot()
def test_observation_wrapper(self):
"""Test of the observation_wrapper class basic functionality"""
- host = ApiUtil.get_env_var("TANGO_HOST")
+ # convert the JSON specificiation to a dict for this class
+ specification_dict = loads(TestObservationBase.VALID_JSON)
- observation = observation_wrapper(specification=TestObservationBase.VALID_JSON, host=host)
+ # create an observation class using the dict and as host just get it using a util function
+ observation = observation_wrapper(specification=specification_dict, host=ApiUtil.get_env_var("TANGO_HOST"))
# Assert the observation is running after starting it
observation.start()