CT-57: Use lmcbaseclasses package version 0.6.0. (b1663657) · Commits · Gianluca Marotta / csp-lmc-old

csp-lmc-common/csp_lmc_common/CspMaster.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -102,12 +102,16 @@ class CspMaster(SKAMaster):

		:return: None
		"""
		self.logger.info("_se_scm_change_event_cb init")
		dev_name = evt.device.dev_name()
		self.logger.info("_se_scm_change_event_cb dev_name:{}".format(dev_name))
		self.logger.info("_se_scm_change_event_cb se_fqdn:{}".format(self._se_fqdn))

		if not evt.err:
		try:
		if dev_name in self._se_fqdn:
		if evt.attr_value.name.lower() == "state":
		self.logger.debug("{}: received event on {} value {}".format(dev_name,
		self.logger.info("{}: received event on {} value {}".format(dev_name,
		evt.attr_value.name,
		evt.attr_value.value))
		self._se_state[dev_name] = evt.attr_value.value
		@@ -348,12 +352,14 @@ class CspMaster(SKAMaster):
		# In this way the CspMaster is able to detect a change in the admin value.


		self.logger.info("Subscribing for adminMode")
		ev_id = device_proxy.subscribe_event("adminMode",
		EventType.CHANGE_EVENT,
		self._se_scm_change_event_cb,
		stateless=True)
		self._se_event_id[fqdn]['adminMode'] = ev_id

		self.logger.info("Subscribing for State")
		ev_id = device_proxy.subscribe_event("State",
		EventType.CHANGE_EVENT,
		self._se_scm_change_event_cb,

csp-lmc-common/csp_lmc_common/CspSubarray.py

+31 −27

Original line number	Diff line number	Diff line
		@@ -31,8 +31,8 @@ from tango.server import run
		from tango.server import Device
		from tango.server import attribute, command
		from tango.server import class_property, device_property
		from tango import AttrQuality, DispLevel, DevState
		from tango import AttrWriteType, PipeWriteType
		from tango import AttrQuality, EventType, DevState
		from tango import AttrWriteType, DeviceProxy
		# Additional import
		# PROTECTED REGION ID(CspSubarray.additionnal_import) ENABLED START #
		from ska.base import SKASubarray, SKASubarrayStateModel
		@@ -332,11 +332,22 @@ class CspSubarray(SKASubarray):
		class OnCommand(SKASubarray.OnCommand):

		def do(self):
		device = self.target
		result_code = ResultCode.OK
		try:
		for fqdn in device._sc_subarray_fqdn:
		device._sc_subarray_proxies[fqdn].On()
		(result_code, message) = device._sc_subarray_proxies[fqdn].On()
		if result_code == ResultCode.FAILED:
		if fqdn != device.CbfSubarray:
		device._health_state = HealthState.DEGRADED
		else:
		return (ResultCode.FAILED, message)
		message = "On command completed OK"
		self.logger.info(message)
		return (ResultCode.OK, message)
		return (result_code, message)
		except tango.DevFailed as tango_err:
		message = str(tango_err.args[0].desc)
		return (ResultCode.FAILED, message)

		'''
		class AbortCommand(SKASubarray.AbortCommand):
		@@ -501,8 +512,10 @@ class CspSubarray(SKASubarray):
		self.logger.info(log_msg)
		# update CSP sub-array SCM
		if evt.attr_value.name.lower() in ["state", "healthstate", "adminmode"]:
		self.logger.info("call to _update_subarray_state()")
		self._update_subarray_state()
		if evt.attr_value.name.lower() == "obsstate":
		self.logger.info("call to _update_subarray_obs_state()")
		self._update_subarray_obs_state()
		except tango.DevFailed as df:
		self.logger.error(str(df.args[0].desc))
		@@ -635,12 +648,6 @@ class CspSubarray(SKASubarray):
		# CSP can work. The state of PSS and PST sub-elements only contributes
		# to determine the CSP health state.
		self.set_state(self._sc_subarray_state[self.CbfSubarray])
		if self._sc_subarray_admin_mode[self.CbfSubarray] not in [AdminMode.ONLINE,
		AdminMode.MAINTENANCE]:
		self.set_state(tango.DevState.DISABLE)
		if self._admin_mode not in [AdminMode.ONLINE,
		AdminMode.MAINTENANCE]:
		self.set_state[tango.DevState.DISABLE]

		def _update_subarray_health_state(self):
		"""
		@@ -736,10 +743,9 @@ class CspSubarray(SKASubarray):
		return
		# read the sub-componet adminMode (memorized) attribute from
		# the CSP.LMC TANGO DB.
		self.logger.info("_connect_to_subarray_subcomponent fqdn: {}".format(fqdn))
		attribute_properties = self._csp_tango_db.get_device_attribute_property(fqdn,
		{'adminMode': ['__value']})
		self.logger.info("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		self.logger.debug("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		try:
		admin_mode_memorized = attribute_properties['adminMode']['__value']
		self._sc_subarray_admin_mode[fqdn] = int(admin_mode_memorized[0])
		@@ -748,7 +754,7 @@ class CspSubarray(SKASubarray):
		try:
		log_msg = "Trying connection to " + str(fqdn) + " device"
		self.logger.info(log_msg)
		device_proxy = tango.DeviceProxy(fqdn)
		device_proxy = DeviceProxy(fqdn)
		# Note: The DeviceProxy is initialized even if the sub-component
		# device is not running (but defined into the TANGO DB! If not defined in the
		# TANGO DB a exception is throw).
		@@ -758,38 +764,36 @@ class CspSubarray(SKASubarray):
		self._sc_subarray_fqdn.append(fqdn)
		self._sc_subarray_proxies[fqdn] = device_proxy

		self.logger.info("_sc_subarray_fqdn: {}".format(self._sc_subarray_fqdn))
		self.logger.info("_sc_subarray_proxies: {}".format(self._sc_subarray_proxies))
		# subscription of SCM attributes (State, healthState and adminMode).
		# Note: subscription is performed also for devices not ONLINE or MAINTENANCE.
		# In this way the CspMaster is able to detect a change in the admin value.

		ev_id = device_proxy.subscribe_event("adminMode",
		tango.EventType.CHANGE_EVENT,
		EventType.CHANGE_EVENT,
		self._sc_scm_change_event_cb,
		stateless=True)
		self._sc_subarray_event_id[fqdn]['adminMode'] = ev_id

		ev_id = device_proxy.subscribe_event("State",
		tango.EventType.CHANGE_EVENT,
		EventType.CHANGE_EVENT,
		self._sc_scm_change_event_cb,
		stateless=True)
		self._sc_subarray_event_id[fqdn]['state'] = ev_id

		ev_id = device_proxy.subscribe_event("healthState",
		tango.EventType.CHANGE_EVENT,
		EventType.CHANGE_EVENT,
		self._sc_scm_change_event_cb,
		stateless=True)
		self._sc_subarray_event_id[fqdn]['healthState'] = ev_id

		ev_id = device_proxy.subscribe_event("obsState",
		tango.EventType.CHANGE_EVENT,
		EventType.CHANGE_EVENT,
		self._sc_scm_change_event_cb,
		stateless=True)
		self._sc_subarray_event_id[fqdn]['obsState'] = ev_id

		ev_id = device_proxy.subscribe_event("obsMode",
		tango.EventType.CHANGE_EVENT,
		EventType.CHANGE_EVENT,
		self._sc_scm_change_event_cb,
		stateless=True)
		self._sc_subarray_event_id[fqdn]['obsMode'] = ev_id
		@@ -1636,10 +1640,10 @@ class CspSubarray(SKASubarray):
		doc="List of devices that completed the task",
		)

		cbfOutputLink = attribute(name="cbfOutputLink",
		label="cbfOutputLink",
		forwarded=True
		)
		#cbfOutputLink = attribute(name="cbfOutputLink",
		# label="cbfOutputLink",
		# forwarded=True
		#)
		#pssOutputLink = attribute(name="pssOutputLink",
		# label="cbfOutputLink",
		# forwarded=True

csp-lmc-common/setup.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -38,7 +38,7 @@ setup(
		install_requires = [
		'pytango >=9.3.1',
		'future',
		'lmcbaseclasses > 0.5.1',
		'lmcbaseclasses==0.6.0'
		],
		setup_requires=[
		'pytest-runner',

csp-lmc-common/tests/unit/cspmaster_unit_test.py

+4 −3

Original line number	Diff line number	Diff line
		@@ -30,7 +30,8 @@ def test_cspmaster_state_after_init():
		}

		with fake_tango_system(device_under_test, initial_dut_properties=dut_properties, proxies_to_mock=proxies_to_mock) as tango_context:
		dummy_event = create_dummy_event(cbf_master_fqdn)
		dummy_event = create_dummy_event(cbf_master_fqdn, DevState.STANDBY)
		print(event_subscription_map)
		event_subscription_map[cbf_master_state_attr](dummy_event)
		# next check is not passed because the CspMaster device is in ALARM
		# for the missing forwarded attributes configuration
		@@ -40,11 +41,11 @@ def test_cspmaster_state_after_init():
		def mock_event_dev_name(device_name):
		return device_name

		def create_dummy_event(cbf_master_fqdn):
		def create_dummy_event(cbf_master_fqdn, tango_state_value):
		fake_event = Mock()
		fake_event.err = False
		fake_event.attr_name = f"{cbf_master_fqdn}/state"
		fake_event.attr_value.value = DevState.STANDBY
		fake_event.attr_value.value = tango_state_value
		fake_event.attr_value.name = 'State'
		fake_event.device.name = cbf_master_fqdn
		fake_event.device.dev_name.side_effect=(lambda args, *kwargs: mock_event_dev_name(cbf_master_fqdn))