CT-60: Work on Configure command and related tests. (96be990c) · Commits · Gianluca Marotta / csp-lmc-old

csp-lmc-common/csp_lmc_common/CspSubarray.py

+116 −228

File changed.

Preview size limit exceeded, changes collapsed.

csp-lmc-mid/csp_lmc_mid/MidCspSubarrayBase.py

+121 −84

Original line number	Diff line number	Diff line
		@@ -95,7 +95,7 @@ class MidCspSubarrayBase(CspSubarray):
		# NOTE: To remove when CBF Subarray implements the correct state
		def do(self):
		self.logger.info("dev_State:{}".format(self.state_model.dev_state))
		super().do()
		#super().do()
		return (ResultCode.OK, "On command completed OK")

		class AddReceptorsCommand(SKASubarray.AssignResourcesCommand):
		@@ -136,7 +136,7 @@ class MidCspSubarrayBase(CspSubarray):
		log_msg = "No available receptor to add to subarray {}".format(
		device.SubID)
		self.logger.warning(log_msg)
		return (ResultCode.OK, "AddREsources OK")
		return (ResultCode.OK, "AddResources OK")
		receptor_membership = csp_master_proxy.receptorMembership
		except tango.DevFailed as df:
		msg = "Failure in getting receptors information:" + \
		@@ -292,7 +292,7 @@ class MidCspSubarrayBase(CspSubarray):
		try:
		if len(device):
		receptors = device._get_cbfsubarray_assigned_receptors()
		return device._remove_receptors_cmd.do(receptors[:])
		return device._removereceptors_cmd_obj.do(receptors[:])
		return (ResultCode.OK, "No receptor to remove")
		except tango.DevFailed as df:
		log_msg = ("RemoveAllReceptors failure. Reason: {} "
		@@ -301,10 +301,116 @@ class MidCspSubarrayBase(CspSubarray):
		self.logger.error(log_msg)
		return (ResultCode.FAILED, log_msg)

		class ConfigureCommand(CspSubarray.ConfigureCommand):
		def do(self,argin):
		self.logger.info("Call Mid CspSubarray Configure")
		try:
		super().do(argin)
		except tango.DevFailed as tango_err:
		log_msg = ("Configure Command failure. Reason: {} "
		"Desc: {}".format(tango_err.args[0].reason,
		tango_err.args[0].desc))
		self.logger.error(log_msg)
		return (ResultCode.FAILED, log_msg)

		def validate_scan_configuration(self, argin):
		"""
		Overwritten method.
		Validate the MID scan configuration file.
		Currently it only copies the received configuration because it does not
		still exist any "cbf" block inside the JSON script.
		:param json_config: the input JSON formatted string with the configuration
		for a scan
		"""

		'''
		TODO:
		- look for the sub-element entries
		- add eventual information to the sub-element blocks (for example for PSS add the
		addresses of the PSS pipelines or the IDs of the SearchBeams)
		- set the observing mode for each sub-element sub-array/PSTBeam
		json_config = {}
		try:
		json_config = json.loads(argin)
		except json.JSONDecodeError as e: # argument not a valid JSON object
		# this is a fatal error
		msg = ("Scan configuration object is not a valid JSON object."
		"Aborting configuration:{}".format(str(e)))
		self.logger.error(msg)
		tango.Except.throw_exception("Command failed",
		msg,
		"ConfigureScan execution",
		tango.ErrSeverity.ERR)
		if "cbf" in json_config:
		if self._sc_subarray_state[self.CbfSubarray] != tango.DevState.ON:
		pass
		# throw exception
		self._sc_subarray_obs_mode[self.CbfSubarray] = ObsMode.CORRELATION
		self._sc_scan_configuration[self.CbfSubarray] = json.dumps(json_config["cbf"])
		self._sc_subarray_assigned_fqdn.append(self.CbfSubarray)
		if "pss" in json_config:
		if self._sc_subarray_state[self.PssSubarray] != tango.DevState.ON:
		pass
		# throw exception
		#self._sc_subarray_obs_mode[self.PssSubarray] = ??
		self._sc_subarray_assigned_fqdn.append(self.PssSubarray)
		pss_json = json_config["pss"]
		if "searchbeams" not in self._sc_scan_configuration[self.PssSubarray]:
		pss_json["searchbeams"] = self._assigned_search_beams
		self._sc_scan_configuration[self.PssSubarray] = json.dumps(pss_json)

		if "pst" in json_config:
		....
		'''
		self.logger.info("Mid CspSubarray validate_scan")
		device = self.target
		device._sc_subarray_assigned_fqdn.clear()
		json_dict = {}
		try:
		#CspSubarray.validate_scan_configuration(self, argin)
		super().validate_scan_configuration(argin)
		json_dict = json.loads(argin)
		except json.JSONDecodeError as e: # argument not a valid JSON object
		# this is a fatal error
		msg = ("Scan configuration object is not a valid JSON object."
		"Aborting configuration:{}".format(str(e)))
		self.logger.error(msg)
		tango.Except.throw_exception("Command failed",
		msg,
		"ConfigureScan execution",
		tango.ErrSeverity.ERR)
		# TODO:
		# validate some entries of json_dict
		self.logger.debug("Validate scan configuration for MID CSP")
		#self._sc_subarray_obs_mode[self.CbfSubarray] = ObsMode.IMAGING
		#self._sc_scan_configuration[self.CbfSubarray] = json_config["cbf"]
		device._sc_subarray_scan_configuration[device.CbfSubarray] = argin
		device._sc_subarray_assigned_fqdn.append(device.CbfSubarray)
		self.logger.debug("Validate scan: {}".format(device._sc_subarray_assigned_fqdn))

		def __len__(self):
		assigned_receptors = self._get_cbfsubarray_assigned_receptors()
		return len(assigned_receptors)

		def update_subarray_state(self):
		"""
		Class protected method.
		Retrieve the State attribute values of the CSP sub-elements and aggregate
		them to build up the CSP global state.

		:param: None

		:return: None
		"""
		self.logger.info("MidCspSubarray")
		CspSubarray.update_subarray_state(self)
		if self.get_state() == DevState.ON:
		if self._sc_subarray_obs_state[self.CbfSubarray] == ObsState.IDLE:
		self._addreceptors_cmd_obj.succeeded()
		if self.get_state() == DevState.ON:
		if self._sc_subarray_obs_state[self.CbfSubarray] == ObsState.EMPTY:
		self._removereceptors_cmd_obj.succeeded()
		self.logger.info("MidCsp subarray state: {} obsState: {}".format(self.get_state(), self.state_model.obs_state))

		def _get_cbfsubarray_assigned_receptors(self):
		"""
		@@ -389,7 +495,7 @@ class MidCspSubarrayBase(CspSubarray):
		self._cmd_execution_state[cmd_name] = CmdExecState.IDLE
		self.logger.info("AddReceptors end!")
		#self.AddReceptorsCommand.succeeded()
		self._add_receptors_cmd.succeeded()
		self._addreceptors_cmd_obj.succeeded()

		# PROTECTED REGION END # // MidCspSubarrayBase.private_methods

		@@ -467,88 +573,18 @@ class MidCspSubarrayBase(CspSubarray):
		# reset the command exeuction state
		self._sc_subarray_cmd_exec_state[device][cmd_name] = CmdExecState.IDLE
		self._cmd_execution_state[cmd_name] = CmdExecState.IDLE
		self._remove_receptors_cmd.succeeded()
		self._removereceptors_cmd_obj.succeeded()
		self.logger.info("CspSubarray monitor_remove end")

		def validate_scan_configuration(self, argin):
		"""
		Overwritten method.
		Validate the MID scan configuration file.
		Currently it only copies the received configuration because it does not
		still exist any "cbf" block inside the JSON script.

		TODO:
		- look for the sub-element entries
		- add eventual information to the sub-element blocks (for example for PSS add the
		addresses of the PSS pipelines or the IDs of the SearchBeams)
		- set the observing mode for each sub-element sub-array/PSTBeam
		json_config = {}
		try:
		json_config = json.loads(argin)
		except json.JSONDecodeError as e: # argument not a valid JSON object
		# this is a fatal error
		msg = ("Scan configuration object is not a valid JSON object."
		"Aborting configuration:{}".format(str(e)))
		self.logger.error(msg)
		tango.Except.throw_exception("Command failed",
		msg,
		"ConfigureScan execution",
		tango.ErrSeverity.ERR)
		if "cbf" in json_config:
		if self._sc_subarray_state[self.CbfSubarray] != tango.DevState.ON:
		pass
		# throw exception
		self._sc_subarray_obs_mode[self.CbfSubarray] = ObsMode.CORRELATION
		self._sc_scan_configuration[self.CbfSubarray] = json.dumps(json_config["cbf"])
		self._sc_subarray_assigned_fqdn.append(self.CbfSubarray)
		if "pss" in json_config:
		if self._sc_subarray_state[self.PssSubarray] != tango.DevState.ON:
		pass
		# throw exception
		#self._sc_subarray_obs_mode[self.PssSubarray] = ??
		self._sc_subarray_assigned_fqdn.append(self.PssSubarray)
		pss_json = json_config["pss"]
		if "searchbeams" not in self._sc_scan_configuration[self.PssSubarray]:
		pss_json["searchbeams"] = self._assigned_search_beams
		self._sc_scan_configuration[self.PssSubarray] = json.dumps(pss_json)

		if "pst" in json_config:
		....
		:param json_config: the input JSON formatted string with the configuration
		for a scan

		"""
		self._sc_subarray_assigned_fqdn.clear()
		json_dict = {}
		try:
		CspSubarray.validate_scan_configuration(self, argin)
		json_dict = json.loads(argin)
		except json.JSONDecodeError as e: # argument not a valid JSON object
		# this is a fatal error
		msg = ("Scan configuration object is not a valid JSON object."
		"Aborting configuration:{}".format(str(e)))
		self.logger.error(msg)
		tango.Except.throw_exception("Command failed",
		msg,
		"ConfigureScan execution",
		tango.ErrSeverity.ERR)
		# TODO:
		# validate some entries of json_dict
		self.logger.debug("Validate scan configuration for MID CSP")
		#self._sc_subarray_obs_mode[self.CbfSubarray] = ObsMode.IMAGING
		#self._sc_scan_configuration[self.CbfSubarray] = json_config["cbf"]
		self._sc_subarray_scan_configuration[self.CbfSubarray] = argin
		self._sc_subarray_assigned_fqdn.append(self.CbfSubarray)
		self.logger.debug("Validate scan: {}".format(self._sc_subarray_assigned_fqdn))


		@command(
		dtype_in='DevString',
		doc_in="A Json-encoded string with the scan configuration.",
		)
		@DebugIt()
		def Configure(self, argin):
		CspSubarray.Configure(self, argin)
		#@command(
		# dtype_in='DevString',
		# doc_in="A Json-encoded string with the scan configuration.",
		#)
		#@DebugIt()
		#def Configure(self, argin):
		# CspSubarray.Configure(self, argin)

		def init_command_objects(self):
		"""
		@@ -557,8 +593,9 @@ class MidCspSubarrayBase(CspSubarray):
		super().init_command_objects()
		self.logger.info("MID call to init_commands")
		args = (self, self.state_model, self.logger)
		self._add_receptors_cmd = self.AddReceptorsCommand(*args)
		self._remove_receptors_cmd = self.RemoveReceptorsCommand(*args)
		self._addreceptors_cmd_obj = self.AddReceptorsCommand(*args)
		self._removereceptors_cmd_obj = self.RemoveReceptorsCommand(*args)
		self._removereceptors_cmd_obj = self.RemoveReceptorsCommand(*args)
		self.register_command_object("AddReceptors", self.AddReceptorsCommand(*args))
		self.register_command_object("RemoveReceptors", self.RemoveReceptorsCommand(*args))
		self.register_command_object("RemoveAllReceptors", self.RemoveAllReceptorsCommand(*args))

csp-lmc-mid/tests/MidCspSubarray_test.py

+35 −10

Original line number	Diff line number	Diff line
		@@ -17,11 +17,12 @@ import random
		import numpy as np
		import logging
		import unittest
		import pytest

		# Tango imports
		import tango
		from tango import DevState
		import pytest
		from assertpy import assert_that

		# Path
		file_path = os.path.dirname(os.path.abspath(__file__))
		@@ -77,6 +78,10 @@ class TestCspSubarray(TestBase):
		obs_state = self.midcsp_subarray01.obsState
		LOGGER.info("setup_subarray obs_state:{}".format(obs_state))
		if subarray_state == DevState.ON:
		if obs_state == ObsState.FAULT:
		self.midcsp_subarray01.ObsReset()
		prober_obs_state = Probe(self.midcsp_subarray01, "obsState", ObsState.IDLE, f"CSP Subarray not IDLE")
		Poller(4, 0.2).check(prober_obs_state)
		if obs_state == ObsState.EMPTY:
		return
		else:
		@@ -308,25 +313,45 @@ class TestCspSubarray(TestBase):
		assigned_receptors = self.midcsp_subarray01.assignedReceptors
		# check the array is empty (any() in this case returns False)
		assert not assigned_receptors.any()
		'''
		def test_configure_WHEN_subarray_is_in_invalid_state(self, midcsp_subarray01):

		def test_configure_WHEN_subarray_is_in_wrong_state(self):
		"""
		Test that the Configure() command fails if the Subarray
		state is not ON
		"""
		# check the subarray state is Off (the previous test has removed
		# all the receptors from the subarray, so its state should be OFF
		obs_state = midcsp_subarray01.obsState
		assert obs_state == ObsState.EMPTY, f"CSP Subarray is not EMPTY"
		self._setup_subarray()
		obs_state = self.midcsp_subarray01.obsState
		state = self.midcsp_subarray01.State()
		LOGGER.info("CSP State: {}-{}".format(state, obs_state))
		filename = os.path.join(data_pkg_path, "test_ConfigureScan_basic.json")
		f = open(file_path + "/test_ConfigureScan_basic.json")
		with pytest.raises(tango.DevFailed) as df:
		midcsp_subarray01.Configure(f.read().replace("\n", ""))
		self.midcsp_subarray01.Configure(f.read().replace("\n", ""))
		if df:
		err_msg = str(df.value.args[0].desc)
		LOGGER.error(err_msg)
		obs_state = midcsp_subarray01.obsState
		assert obs_state == ObsState.OBSFAULT, f"CSP Subarray obsState is not OBSFAULT"
		obs_state = self.midcsp_subarray01.obsState
		assert obs_state == ObsState.EMPTY, f"CSP Subarray obsState is not EMPTY"

		def test_configure_WHITH_wrong_configuration(self):
		"""
		Test that the Configure() command fails if the Subarray
		state is not ON
		"""
		self._setup_subarray()
		self._assign_receptors()
		obs_state = self.midcsp_subarray01.obsState
		state = self.midcsp_subarray01.State()
		LOGGER.info("CSP State: {}-{}".format(state, obs_state))
		f = open(file_path + "/acceptance_tests/test_ConfigureScan_without_configID.json")
		self.midcsp_subarray01.Configure(f.read().replace("\n", ""))
		# check
		obs_state = self.midcsp_subarray01.obsState
		#assert obs_state == ObsState.IDLE, f"CSP Subarray obsState has wrong value ({obs_state})"
		assert_that(obs_state).described_as("CSP Subarray obsState has wrong value ({obs_state}").is_equal_to(ObsState.FAULT)
		#assert obs_state == ObsState.FAULT, f"CSP Subarray obsState is not EMPTY"

		'''

		def test_configureScan(self, midcsp_subarray01, midcsp_master):
		"""