Fixed bug in dec2bin (added regression test) (d340bcb6) · Commits · INAF / DISCOS

Common/Servers/DewarPositioner/src/DewarPositioner/positioner.py

+10 −3

Original line number	Diff line number	Diff line
		@@ -738,11 +738,13 @@ class Status(object):
		>>> Status.dec2bin(16, 6)
		'010000'
		>>> Status.dec2bin(16, 5)
		'010000'
		'10000'
		>>> Status.dec2bin(16, 10)
		'0000010000'
		>>> Status.dec2bin(0, 6)
		'000000'
		>>> Status.dec2bin(32, 6)
		'100000'
		"""
		if n == 0:
		value = '0'
		@@ -750,7 +752,12 @@ class Status(object):
		value = Status.dec2bin(n//2) + str(n%2)

		if nbits:
		return value if len(value) >= nbits else '0'*(nbits - len(value)) + value
		if len(value) > nbits:
		return value[-nbits:]
		elif len(value) == nbits:
		return value
		else:
		return '0'*(nbits - len(value)) + value
		else:
		return value

Common/Servers/DewarPositioner/test/functional/test_rewind.py

0 → 100644

+310 −0

Original line number	Diff line number	Diff line
		from __future__ import with_statement
		import unittest2
		import time
		import math
		from DewarPositioner.posgenerator import PosGenerator
		from Antenna import ANT_HORIZONTAL, ANT_NORTH, ANT_SOUTH
		from Management import MNG_TRACK

		from Acspy.Clients.SimpleClient import PySimpleClient
		from ComponentErrors import ComponentErrorsEx


		class StartUpdatingTest(unittest2.TestCase):
		"""Test the DewarPositioner.startUpdating() end-to-end method"""
		def setUp(self):
		self.client = PySimpleClient()
		self.device = self.client.getComponent('RECEIVERS/SRTKBandDerotator')
		self.antenna = self.client.getComponent('ANTENNA/BossSimulator')
		self.positioner = self.client.getComponent('RECEIVERS/DewarPositioner')
		self.positioner.setup('KKG')
		self.positioner.setConfiguration('custom')
		self.lat = 0.689 # 39.5 degrees

		def tearDown(self):
		if self.positioner.isReady():
		self.positioner.stopUpdating()
		time.sleep(1)
		self.positioner.park()
		time.sleep(0.5)
		self.client.releaseComponent('RECEIVERS/DewarPositioner')
		self.client.releaseComponent('ANTENNA/BossSimulator')
		self.device = self.client.releaseComponent('RECEIVERS/SRTKBandDerotator')


		def test_not_positive_number_of_steps(self):
		"""Raise ComponentErrorsEx when the number of steps is not positive"""
		with self.assertRaisesRegexp(ComponentErrorsEx, 'steps must be positive'):
		self.positioner.setRewindingMode('MANUAL')
		az, el, target = self.prepare_negative_oor()
		# after prepare_negative_oor(), startUpdating() will cause oor
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertTrue(self.positioner.isRewindingRequired())
		self.positioner.rewind(0)

		def test_number_of_steps_oor(self):
		"""Raise ComponentErrors when the number of steps is out of range"""
		with self.assertRaisesRegexp(ComponentErrorsEx, 'cannot rewind the derotator'):
		self.positioner.setRewindingMode('MANUAL')
		az, el, target = self.prepare_negative_oor()
		# after prepare_negative_oor(), startUpdating() will cause oor
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertTrue(self.positioner.isRewindingRequired())
		self.positioner.rewind(4)

		def test_tracking_when_rewinding_required(self):
		"""Verify the tracking flag is False when a rewinding is required"""
		self.positioner.setRewindingMode('MANUAL')
		az, el, target = self.prepare_negative_oor()
		# after prepare_negative_oor(), startUpdating() will cause oor
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertEqual(self.positioner.isRewindingRequired(), True)
		self.assertEqual(self.positioner.isRewinding(), False)
		self.assertEqual(self.positioner.isTracking(), False)

		def test_updating_and_position_after_manual_rewind(self):
		"""Verify position and that it stills updating after a manual rewind"""
		self.positioner.setRewindingMode('MANUAL')
		az, el, target = self.prepare_negative_oor()
		# after prepare_negative_oor(), startUpdating() will cause oor
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		steps = 2
		self.positioner.rewind(steps)
		self.assertTrue(self.positioner.isUpdating())
		expected = target + steps * self.positioner.getRewindingStep()
		self.assertAlmostEqual(self.positioner.getPosition(), expected, delta=0.1)

		def test_rewind_offset_properly_cleared(self):
		"""Verify the startUpdating() clears previous rewinding offsets.

		For instance, in the following case:

		- startUpdating() out of range -> rewind
		- new startUpdating() with (Pis + parallactic) in range

		In the first startUpdating() we want the position to be::

		Pis + Pip + Pid + rewinding_offset

		In the second startUpdating() the position must be::

		Pis + Pip + Pid
		"""

		# First step: we choose (az, el) in order to have a parallactic angle
		# out of range. That means the derotator has to rewind
		az, el = 1.2217, 0.6109 # 70 degrees, 35 degrees
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # -63
		min_limit = self.device.getMinLimit() # -85.77 degrees for the K Band
		Pis = -30
		# Final angle -93, lower than the min limit (-85.77)
		self.positioner.setPosition(Pis)
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertLess(target, min_limit)
		# If sector is SOUTH, we expect a rewind of 60 degrees: -93 + 60 -> -33
		expected = target + 60
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertAlmostEqual(expected, self.device.getActPosition(), delta=0.1)

		self.positioner.stopUpdating()

		# Second step: we do not call stopUpdating(), and we choose (az, el)
		# in order to have a parallactic angle in range
		az = math.radians(120)
		el = math.radians(45)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # -42.78
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertGreater(target, min_limit)
		# We do not expect a rewind
		expected = target
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertAlmostEqual(expected, self.device.getActPosition(), delta=0.1)

		def test_keep_updating_after_auto_rewind(self):
		"""Verify the dewar positioner stills updating after the rewind"""
		Pis = 120
		self.positioner.setPosition(Pis)

		# First step: we choose (az, el) in order to have a parallactic angle
		# of a just one or two degrees, in order to stay in range. i.e. we
		# can set the following value to start the updating:
		#
		# >>> az = math.radians(181)
		# >>> el = math.radians(45)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# 0.77546054825239319
		#
		# and at this point we set the following in order to go out of range:
		#
		# >>> az = math.radians(190)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# 7.7330295207687838
		az = math.radians(181)
		el = math.radians(45)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # 0.77

		max_limit = self.device.getMaxLimit() # 125.23 degrees for the K Band
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertLess(target, max_limit)
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertAlmostEqual(target, self.device.getActPosition(), delta=0.1)
		self.assertTrue(self.positioner.isUpdating())
		self.assertTrue(self.positioner.isTracking())

		# Second step: we set the az, in order to have a parallactic angle
		# that causes an out of range
		# in order to have a parallactic angle in range
		az = math.radians(190)
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # 7.73
		target = Pis + parallactic
		self.assertGreater(target, max_limit)
		expected = target - 180
		# We expect a rewind
		time.sleep(0.5)
		self.assertAlmostEqual(expected, self.device.getActPosition(), delta=0.1)
		# Verify it keeps the system updating the position
		self.assertFalse(self.positioner.isRewinding())
		self.assertTrue(self.positioner.isUpdating())
		self.assertTrue(self.positioner.isTracking())

		def test_rewinding_angle_sector_NORTH(self):
		"""The rewinding angle depends of the antenna sector.

		For instance, this is the case: we set a Pis to 120 degrees and we are
		following the source in sector NORTH, so the derotator is going to the
		lower limit (CCW). At this point we set an antenna offset that causes
		the derotator movement slighly up to its higher limit. In that case to
		guarantee the maximum movement space we have to rewind but going close
		to the higher limit, not the lower one. If this does not appen, so if
		after the rewind we set an antenna position that causes a Pdp of -90
		degrees, the derotator tries to go below the lower limit, and so
		performs a new rewind.
		"""
		Pis = 120
		self.positioner.setPosition(Pis)

		# First step: we choose (az, el) in order to have a parallactic angle
		# of a just one or two degrees, in order to stay in range. i.e. we
		# can set the following value to start the updating:
		#
		# >>> az = math.radians(181)
		# >>> el = math.radians(45)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# 0.77546054825239319
		#
		# and at this point we set the following in order to go out of range:
		#
		# >>> az = math.radians(190)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# 7.7330295207687838
		az = math.radians(181)
		el = math.radians(45)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # 0.77

		max_limit = self.device.getMaxLimit() # 125.23 degrees for the K Band
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertLess(target, max_limit)
		self.positioner.startUpdating(MNG_TRACK, ANT_NORTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertAlmostEqual(target, self.device.getActPosition(), delta=0.1)
		self.assertTrue(self.positioner.isUpdating())
		self.assertTrue(self.positioner.isTracking())

		# Second step: we set the az, in order to have a parallactic angle
		# that causes an out of range
		az = math.radians(190)
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		# We expect a rewind
		time.sleep(0.5)
		# At this point we want the new position to be closer to the max_limit
		# than the min_limit
		actual_pos = self.positioner.getPosition()
		right_distance = abs(actual_pos - self.positioner.getMaxLimit())
		left_distance = abs(actual_pos - self.positioner.getMinLimit())
		self.assertGreater(left_distance, right_distance)

		def test_rewinding_angle_sector_SOUTH(self):
		"""The rewinding angle depends of the antenna sector.

		Look at the test_rewinding_angle_sector_NORTH docstring.
		"""
		Pis = -80
		self.positioner.setPosition(Pis)

		# First step: we choose (az, el) in order to have a parallactic angle
		# of a just one or two degrees, in order to stay in range. i.e. we
		# can set the following value to start the updating:
		#
		# >>> az = math.radians(179)
		# >>> el = math.radians(45)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# -0.77546054825239319
		#
		# and at this point we set the following in order to go out of range:
		#
		# >>> az = math.radians(170)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# -7.7330295207687838
		az = math.radians(179)
		el = math.radians(45)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # -0.77

		min_limit = self.device.getMinLimit() # -85.77 degrees for the K Band
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertGreater(target, min_limit)
		self.positioner.startUpdating(MNG_TRACK, ANT_SOUTH, az, el, 0, 0)
		time.sleep(0.5)
		self.assertAlmostEqual(target, self.device.getActPosition(), delta=0.1)
		self.assertTrue(self.positioner.isUpdating())
		self.assertTrue(self.positioner.isTracking())

		# Second step: we set the az, in order to have a parallactic angle
		# that causes an out of range
		az = math.radians(170)
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		# We expect a rewind
		time.sleep(0.5)
		# At this point we want the new position to be closer to the min_limit
		# than the max_limit
		actual_pos = self.positioner.getPosition()
		right_distance = abs(actual_pos - self.positioner.getMaxLimit())
		left_distance = abs(actual_pos - self.positioner.getMinLimit())
		self.assertGreater(right_distance, left_distance)

		def prepare_negative_oor(self):
		"""Set the derotator position an antenna az-el in order to
		cause an out of range when startUpdating() is called.
		"""
		Pis = -80
		self.positioner.setPosition(Pis)
		# We choose (az, el) in order to have a parallactic angle out of range
		#
		# >>> az = math.radians(170)
		# >>> el = math.radians(45)
		# >>> PosGenerator.getParallacticAngle(self.lat, az, el)
		# -7.7330295207687838
		az = math.radians(170)
		el = math.radians(45)
		parallactic = PosGenerator.getParallacticAngle(self.lat, az, el) # -0.77
		min_limit = self.device.getMinLimit() # -85.77 degrees for the K Band
		self.antenna.setOffsets(az, el, ANT_HORIZONTAL)
		target = Pis + parallactic
		self.assertLess(target, min_limit)
		return az, el, target


		if __name__ == '__main__':
		unittest2.main()

Common/Servers/DewarPositioner/test/pyunit/positioner/test_dec2bin.py

0 → 100644

+15 −0

Original line number	Diff line number	Diff line
		import unittest2
		from DewarPositioner.positioner import Status

		class Dec2BinTest(unittest2.TestCase):

		def test_return_the_required_length(self):
		"""The lenght of the string must be as specified"""
		for i in range(1000):
		self.assertEqual(len(Status.dec2bin(i, 6)), 6)

		def test_expected_value(self):
		self.assertEqual(Status.dec2bin(0, 7), '0'*7)

		if __name__ == '__main__':
		unittest2.main()

Common/Servers/DewarPositioner/test/submodules/cdbconf/test_cdbconf.py

deleted100644 → 0

+0 −101

Original line number	Diff line number	Diff line
		import unittest2
		from DewarPositioner.cdbconf import CDBConf
		from ComponentErrors import ValidationErrorEx
		from Acspy.Clients.SimpleClient import PySimpleClient
		from Acspy.Util import ACSCorba


		class CDBConfTest(unittest2.TestCase):
		def setUp(self):
		self.conf = CDBConf()
		self.dal = ACSCorba.cdb()
		# self.conf.setup('KKG')

		def test_getAttribute(self):
		"""Test the getAttribute() method"""
		# It must raise an exception in case of wrong attribute name
		self.assertRaises(ValidationErrorEx, self.conf.getAttribute, 'foo')
		# It must get all the attributes listed in CDBConf.componentAttributes
		dao = self.dal.get_DAO_Servant(self.conf.componentPath)
		for name in self.conf.componentAttributes:
		value = dao.get_field_data(name)
		self.assertEqual(value, self.conf.getAttribute(name))

		def test_setConfiguration(self):
		"""Test the setConfiguration() method"""
		# Vefify the configurationCode and path are not set before the setup
		self.assertEqual(self.conf.configurationCode, 'none')
		self.assertEqual(self.conf.configurationPath, '')
		# Vefify it raises an exception in case a setup is required
		self.assertRaises(ValidationErrorEx, self.conf.setConfiguration, 'BSC')
		# After the setup, it must raise an exception in case of wrong configuration
		setupCode = 'KKG'
		self.conf.setup(setupCode)
		self.assertRaises(ValidationErrorEx, self.conf.setConfiguration, 'foo')
		# Verify it sets properly the configuration
		confCode = 'BSC'
		self.conf.setConfiguration(confCode)
		self.assertEqual(self.conf.configurationCode, confCode)
		self.assertRegexpMatches(self.conf.configurationPath, confCode)
		self.assertTrue(any(self.conf.UpdatingPosition))
		confCode = 'FIXED'
		self.conf.setConfiguration(confCode)
		self.assertEqual(self.conf.configurationCode, confCode)
		self.assertRegexpMatches(self.conf.configurationPath, confCode)
		self.assertTrue(any(self.conf.UpdatingPosition))


		def test_getUpdatingConfiguration(self):
		"""Test the getUpdatingConfiguration() method"""
		# It raises an exception in case a setConfiguration() is required
		self.assertRaises(ValidationErrorEx, self.conf.getUpdatingConfiguration, '')
		# Once configured, it raises an exception in case of wrong axis name
		self.conf.setup('KKG')
		self.conf.setConfiguration('BSC')
		self.assertRaises(ValidationErrorEx, self.conf.getUpdatingConfiguration, 'foo')
		# Get the position
		self.assertEqual(
		self.conf.getUpdatingConfiguration('JUST_FOR_TEST'),
		{'initialPosition': '-9.0', 'functionName': 'gpa'})
		self.conf.setConfiguration('FIXED')
		self.assertEqual(
		self.conf.getUpdatingConfiguration('JUST_FOR_TEST'),
		{'initialPosition': '0', 'functionName': 'none'})


		def test_updateInitialPositions(self):
		"""Test the updateInitialPositions() method"""
		self.conf.setup('KKG')
		self.conf.setConfiguration('BSC')
		# Get the position
		self.assertEqual(
		self.conf.getUpdatingConfiguration('JUST_FOR_TEST'),
		{'initialPosition': '-9.0', 'functionName': 'gpa'})
		self.conf.updateInitialPositions(-100)
		self.assertEqual(
		self.conf.getUpdatingConfiguration('JUST_FOR_TEST'),
		{'initialPosition': '-100.00', 'functionName': 'gpa'})


		def test_setup(self):
		"""Test the setup method"""
		# Vefify the Mapping attribute is not set before the setup
		self.assertRaises(ValidationErrorEx, self.conf.getAttribute, 'DerotatorName')
		# The default paths and codes must be empty strings
		self.assertEqual(self.conf.setupCode, '')
		self.assertEqual(self.conf.setupPath, '')
		self.assertEqual(self.conf.mappingPath, '')
		setupCode = 'KKG'
		self.conf.setup(setupCode)
		# Paths and codes must contain the setup code
		self.assertEqual(self.conf.setupCode, setupCode)
		self.assertRegexpMatches(self.conf.setupPath, setupCode)
		self.assertRegexpMatches(self.conf.mappingPath, setupCode)
		# It must get all the attributes listed in CDBConf.mappingAttributes
		dao = self.dal.get_DAO_Servant(self.conf.mappingPath)
		for name in self.conf.mappingAttributes:
		value = dao.get_field_data(name)
		self.assertEqual(value, self.conf.getAttribute(name))

		if __name__ == '__main__':
		unittest2.main()

Common/Servers/DewarPositioner/test/submodules/devios/test_status.py

deleted100644 → 0

+0 −34

Original line number	Diff line number	Diff line
		import unittest2
		import mocker
		from DewarPositioner.devios import StatusDevIO


		class StatusDevIOTest(unittest2.TestCase):

		def setUp(self):
		self.m = mocker.Mocker()
		self.device = self.m.mock()

		def tearDown(self):
		self.m.restore()
		self.m.verify()

		def test_singlebit(self):
		"""Test the result is as excpected"""
		self.device.getStatus()
		self.m.result('100000')
		self.m.replay()
		devio = StatusDevIO(self.device)
		self.assertEqual(devio.read(), 2**5)

		def test_severalbits(self):
		"""Test the result is as excpected"""
		self.device.getStatus()
		self.m.result('111001')
		self.m.replay()
		devio = StatusDevIO(self.device)
		self.assertEqual(devio.read(), 25 + 24 + 23 + 20)


		if __name__ == '__main__':
		unittest2.main()