Merge branch 'CT-206' into 'master' (16de7c3c) · Commits · Gianluca Marotta / csp-lmc-old

csp-lmc-common/HISTORY

+3 −0

Original line number	Diff line number	Diff line
		0.6.10
		- use of ska-log-transaction via transaction_id decorator.

		0.6.9:
		- use lmcbaseclasses 0.6.5
		0.6.8:

csp-lmc-common/csp_lmc_common/CspSubarray.py

+9 −1

Original line number	Diff line number	Diff line
		@@ -40,8 +40,8 @@ from ska.base import SKASubarray, SKASubarrayStateModel
		from ska.base.commands import ActionCommand, ResultCode
		from ska.base.faults import CapabilityValidationError
		from ska.base.control_model import HealthState, AdminMode, ObsState, ObsMode
		from .utils.decorators import AdminModeCheck, ObsStateCheck, SubarrayRejectCmd
		from .utils.cspcommons import CmdExecState
		from .utils.decorators import transaction_id
		from . import release
		# PROTECTED REGION END # // CspSubarray.additionnal_import

		@@ -371,8 +371,16 @@ class CspSubarray(SKASubarray):
		self.logger.info(message)
		return (ResultCode.OK, message)

		class AssignResourcesCommand(SKASubarray.AssignResourcesCommand):

		@transaction_id
		def do(self,argin):
		return super().do(argin)
		self.logger.warning("Assign Resource Command not yet implemented in CSP Subarray. This is an instance of the lmcbaseclasses")

		class ConfigureCommand(SKASubarray.ConfigureCommand):

		@transaction_id
		def do(self, argin):
		# checks on State, adminMode and obsState values are performed inside the
		# python decorators

csp-lmc-common/csp_lmc_common/release.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -10,7 +10,7 @@
		"""Release information for Python Package"""

		name = """csp-lmc-common"""
		version = "0.6.9"
		version = "0.6.10"
		version_info = version.split(".")
		description = """SKA CSP.LMC Common Software"""
		author = "INAF-OAA"

csp-lmc-common/csp_lmc_common/utils/decorators.py

+36 −47

Original line number	Diff line number	Diff line
		import sys
		import os

		import tango
		from .cspcommons import CmdExecState
		from ska.base.control_model import AdminMode,ObsState
		import functools
		import json
		from ska.log_transactions import transaction
		from ska.base.control_model import AdminMode,ObsState
		from .cspcommons import CmdExecState

		def transaction_id(func):
		"""
		Add a transaction id to the input of the decorated method.
		The input of the decorated method is a json string.

		"""
		@functools.wraps(func)
		def wrap(args, *kwargs):

		# Argument of the decorated method can be either positional or keyword.
		# Following code manage both cases.
		# A keyword argument occurs when applied to "do" method of SKA Base Command Classes.
		# argument must be single.
		# note that args[0] is the "self" object of the class of the decorated method.
		if kwargs:
		keys=list(kwargs.keys())
		argin = kwargs[keys[0]]
		elif len(args)>1:
		argin = args[1]
		else:
		raise Exception('No argument are passed to the transaction ID decorator')
		parameters =json.loads(argin)
		obj = args[0]

		#note: obj.name is the Command Class Name.
		with transaction(obj.name, parameters,logger=obj.logger) as transaction_id:

		parameters['transaction_id'] = transaction_id

		argin = json.dumps(parameters)
		return func(obj,argin)
		return wrap

		class AdminModeCheck(object):
		"""
		@@ -47,50 +80,6 @@ class AdminModeCheck(object):
		return f(args, *kwargs)
		return admin_mode_check


		VALID_OBS_STATE = {'reset': [ObsState.ABORTED],
		'configscan': [ObsState.IDLE, ObsState.READY],
		'scan': [ObsState.READY],
		'endscan': [ObsState.SCANNING],
		'gotoidle': [ObsState.READY, ObsState.IDLE],
		'addresources': [ObsState.IDLE],
		'removeresources': [ObsState.IDLE]
		}


		class ObsStateCheck(object):
		"""
		Class designed to be a decorator for the CspMaster methods.
		It checks the obsMode attribute value
		"""

		def __init__(self, args=False, kwargs=False):
		self._args = args
		self._kwargs = kwargs

		def __call__(self, f):
		@functools.wraps(f)
		def obs_state_check(args, *kwargs):
		# get the device instance
		dev_instance = args[0]
		cmd_type = self._args

		dev_instance.logger.info("device obs_state: {}".format(dev_instance._obs_state))
		# Check the obsState attribute value: valid values for the command to
		# execute are defined by VALID_OBS_STATE dictionary
		if dev_instance._obs_state not in VALID_OBS_STATE[cmd_type]:
		msg_args = (f.__name__, ObsState(dev_instance._obs_state).name)
		err_msg = ("{} command can't be issued when the"
		" obsState is {} ".format(*msg_args))

		tango.Except.throw_exception("Command not executable",
		err_msg,
		"ObsStateCheck decorator",
		tango.ErrSeverity.ERR)
		return f(args, *kwargs)
		return obs_state_check


		class CmdInputArgsCheck(object):
		"""
		Class designed to be a decorator for the CspMaster methods.

csp-lmc-common/csp_lmc_common/utils/test_utils.py

0 → 100644

+67 −0

Original line number	Diff line number	Diff line
		import time
		import logging
		import numpy
		from tango import DeviceProxy, DevState

		LOGGER = logging.getLogger(__name__)


		class Timeout:
		def __init__(self, duration):
		self.endtime = time.time() + duration

		def has_expired(self):
		return time.time() > self.endtime


		class Probe:
		def __init__(self, proxy, attr_name, expected_state, message):
		"""
		"""
		self.proxy = proxy
		self.attr_name = attr_name
		self.expected_state = expected_state
		self.message = message
		self.current_state = DevState.DISABLE

		def get_attribute(self):
		return self.attr_name

		def sample(self):
		"""
		extract the state of client and store it
		"""
		device_attr = self.proxy.read_attribute(self.attr_name)
		self.current_state = device_attr.value
		#LOGGER.info("attr_name: {} current_state:{}".format(self.attr_name, self.current_state))

		def is_satisfied(self):
		"""
		Check if the state satisfies this test condition
		"""

		if isinstance(self.current_state, numpy.ndarray):
		return (self.expected_state == self.current_state).all()
		return self.expected_state == self.current_state


		class Poller:
		def __init__(self, timeout, interval):
		self.timeout = timeout
		self.interval = interval

		def check(self, probe: Probe):
		"""
		Repeatedly check if the probe is satisfied.
		Assert false when the timeout expires.
		"""
		timer = Timeout(self.timeout)
		probe.sample()
		while not probe.is_satisfied():
		if timer.has_expired():
		LOGGER.debug("Check Timeout on:{}".format(probe.get_attribute()))
		assert False, probe.message
		time.sleep(self.interval)
		probe.sample()
		LOGGER.debug("Check success on: {}".format(probe.get_attribute()))
		assert True