server/cros/network/attenuator_controller.py - mirrors/cros/chromiumos/third_party/autotest - Git at Google

 # Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 import logging

 from autotest_lib.client.common_lib import error
 from autotest_lib.server.cros.network import attenuator
 from autotest_lib.server.cros.network import attenuator_hosts

 from chromite.lib import timeout_util

 HOST_TO_FIXED_ATTENUATIONS = attenuator_hosts.HOST_FIXED_ATTENUATIONS


 class AttenuatorController(object):
     """Represents a minicircuits variable attenuator.

     This device is used to vary the attenuation between a router and a client.
     This allows us to measure throughput as a function of signal strength and
     test some roaming situations.  The throughput vs signal strength tests
     are referred to rate vs range (RvR) tests in places.

     """

     @property
     def supported_attenuators(self):
         """@return iterable of int attenuators supported on this host."""
         return self._fixed_attenuations.keys()


     def __init__(self, hostname):
         """Construct a AttenuatorController.

         @param hostname: Hostname representing minicircuits attenuator.

         """
         self.hostname = hostname
         super(AttenuatorController, self).__init__()
         if hostname not in HOST_TO_FIXED_ATTENUATIONS.keys():
             raise error.TestError('Unexpected RvR host name %r.' % hostname)
         self._fixed_attenuations = HOST_TO_FIXED_ATTENUATIONS[hostname]
         num_atten = len(self.supported_attenuators)

         self._attenuator = attenuator.Attenuator(hostname, num_atten)
         self.set_variable_attenuation(0)


     def _approximate_frequency(self, attenuator_num, freq):
         """Finds an approximate frequency to freq.

         In case freq is not present in self._fixed_attenuations, we use a value
         from a nearby channel as an approximation.

         @param attenuator_num: attenuator in question on the remote host.  Each
                 attenuator has a different fixed path loss per frequency.
         @param freq: int frequency in MHz.
         @returns int approximate frequency from self._fixed_attenuations.

         """
         old_offset = None
         approx_freq = None
         for defined_freq in self._fixed_attenuations[attenuator_num].keys():
             new_offset = abs(defined_freq - freq)
             if old_offset is None or new_offset < old_offset:
                 old_offset = new_offset
                 approx_freq = defined_freq

         logging.debug('Approximating attenuation for frequency %d with '
                       'constants for frequency %d.', freq, approx_freq)
         return approx_freq


     def close(self):
         """Close variable attenuator connection."""
         self._attenuator.close()


     def set_total_attenuation(self, atten_db, frequency_mhz,
                               attenuator_num=None):
         """Set the total attenuation on one or all attenuators.

         @param atten_db: int level of attenuation in dB.  This must be
                 higher than the fixed attenuation level of the affected
                 attenuators.
         @param frequency_mhz: int frequency for which to calculate the
                 total attenuation.  The fixed component of attenuation
                 varies with frequency.
         @param attenuator_num: int attenuator to change, or None to
                 set all variable attenuators.

         """
         affected_attenuators = self.supported_attenuators
         if attenuator_num is not None:
             affected_attenuators = [attenuator_num]
         for atten in affected_attenuators:
             freq_to_fixed_loss = self._fixed_attenuations[atten]
             approx_freq = self._approximate_frequency(atten,
                                                       frequency_mhz)
             variable_atten_db = atten_db - freq_to_fixed_loss[approx_freq]
             self.set_variable_attenuation(variable_atten_db,
                                           attenuator_num=atten)


     def set_variable_attenuation(self, atten_db, attenuator_num=None):
         """Set the variable attenuation on one or all attenuators.

         @param atten_db: int non-negative level of attenuation in dB.
         @param attenuator_num: int attenuator to change, or None to
                 set all variable attenuators.

         """
         affected_attenuators = self.supported_attenuators
         if attenuator_num is not None:
             affected_attenuators = [attenuator_num]
         for atten in affected_attenuators:
             try:
                 self._attenuator.set_atten(atten, atten_db)
                 if int(self._attenuator.get_atten(atten)) != atten_db:
                     raise error.TestError('Attenuation did not set as expected '
                                           'on attenuator %d' % atten)
             except error.TestError:
                 self._attenuator.reopen(self.hostname)
                 self._attenuator.set_atten(atten, atten_db)
                 if int(self._attenuator.get_atten(atten)) != atten_db:
                     raise error.TestError('Attenuation did not set as expected '
                                           'on attenuator %d' % atten)
             logging.info('%ddb attenuation set successfully on attenautor %d',
                          atten_db, atten)


     def get_minimal_total_attenuation(self):
         """Get attenuator's maximum fixed attenuation value.

         This is pulled from the current attenuator's lines and becomes the
         minimal total attenuation when stepping through attenuation levels.

         @return maximum starting attenuation value

         """
         max_atten = 0
         for atten_num in self._fixed_attenuations.iterkeys():
             atten_values = self._fixed_attenuations[atten_num].values()
             max_atten = max(max(atten_values), max_atten)
         return max_atten


     def set_signal_level(self, client_context, requested_sig_level,
             min_sig_level_allowed=-85, tolerance_percent=3, timeout=240):
         """Set wifi signal to desired level by changing attenuation.

         @param client_context: Client context object.
         @param requested_sig_level: Negative int value in dBm for wifi signal
                 level to be set.
         @param min_sig_level_allowed: Minimum signal level allowed; this is to
                 ensure that we don't set a signal that is too weak and DUT can
                 not associate.
         @param tolerance_percent: Percentage to be used to calculate the desired
                 range for the wifi signal level.
         """
         atten_db = 0
         starting_sig_level = client_context.wifi_signal_level
         if not starting_sig_level:
             raise error.TestError("No signal detected.")
         if not (min_sig_level_allowed <= requested_sig_level <=
                 starting_sig_level):
             raise error.TestError("Requested signal level (%d) is either "
                                   "higher than current signal level (%r) with "
                                   "0db attenuation or lower than minimum "
                                   "signal level (%d) allowed." %
                                   (requested_sig_level,
                                   starting_sig_level,
                                   min_sig_level_allowed))

         try:
             with timeout_util.Timeout(timeout):
                 while True:
                     client_context.reassociate(timeout_seconds=1)
                     current_sig_level = client_context.wifi_signal_level
                     logging.info("Current signal level %r", current_sig_level)
                     if not current_sig_level:
                         raise error.TestError("No signal detected.")
                     if self.signal_in_range(requested_sig_level,
                             current_sig_level, tolerance_percent):
                         logging.info("Signal level set to %r.",
                                      current_sig_level)
                         break
                     if current_sig_level > requested_sig_level:
                         self.set_variable_attenuation(atten_db)
                         atten_db +=1
                     if current_sig_level < requested_sig_level:
                         self.set_variable_attenuation(atten_db)
                         atten_db -= 1
         except (timeout_util.TimeoutError, error.TestError,
                 error.TestFail) as e:
             raise error.TestError("Not able to set wifi signal to requested "
                                   "level. \n%s" % e)


     def signal_in_range(self, req_sig_level, curr_sig_level, tolerance_percent):
         """Check if wifi signal is within the threshold of requested signal.

         @param req_sig_level: Negative int value in dBm for wifi signal
                 level to be set.
         @param curr_sig_level: Current wifi signal level seen by the DUT.
         @param tolerance_percent: Percentage to be used to calculate the desired
                 range for the wifi signal level.

         @returns True if wifi signal is in the desired range.
         """
         min_sig = req_sig_level + (req_sig_level * tolerance_percent / 100)
         max_sig = req_sig_level - (req_sig_level * tolerance_percent / 100)
         if min_sig <= curr_sig_level <= max_sig:
             return True
         return False
	# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	import logging

	from autotest_lib.client.common_lib import error
	from autotest_lib.server.cros.network import attenuator
	from autotest_lib.server.cros.network import attenuator_hosts

	from chromite.lib import timeout_util

	HOST_TO_FIXED_ATTENUATIONS = attenuator_hosts.HOST_FIXED_ATTENUATIONS


	class AttenuatorController(object):
	"""Represents a minicircuits variable attenuator.

	This device is used to vary the attenuation between a router and a client.
	This allows us to measure throughput as a function of signal strength and
	test some roaming situations. The throughput vs signal strength tests
	are referred to rate vs range (RvR) tests in places.

	"""

	@property
	def supported_attenuators(self):
	"""@return iterable of int attenuators supported on this host."""
	return self._fixed_attenuations.keys()


	def __init__(self, hostname):
	"""Construct a AttenuatorController.

	@param hostname: Hostname representing minicircuits attenuator.

	"""
	self.hostname = hostname
	super(AttenuatorController, self).__init__()
	if hostname not in HOST_TO_FIXED_ATTENUATIONS.keys():
	raise error.TestError('Unexpected RvR host name %r.' % hostname)
	self._fixed_attenuations = HOST_TO_FIXED_ATTENUATIONS[hostname]
	num_atten = len(self.supported_attenuators)

	self._attenuator = attenuator.Attenuator(hostname, num_atten)
	self.set_variable_attenuation(0)


	def _approximate_frequency(self, attenuator_num, freq):
	"""Finds an approximate frequency to freq.

	In case freq is not present in self._fixed_attenuations, we use a value
	from a nearby channel as an approximation.

	@param attenuator_num: attenuator in question on the remote host. Each
	attenuator has a different fixed path loss per frequency.
	@param freq: int frequency in MHz.
	@returns int approximate frequency from self._fixed_attenuations.

	"""
	old_offset = None
	approx_freq = None
	for defined_freq in self._fixed_attenuations[attenuator_num].keys():
	new_offset = abs(defined_freq - freq)
	if old_offset is None or new_offset < old_offset:
	old_offset = new_offset
	approx_freq = defined_freq

	logging.debug('Approximating attenuation for frequency %d with '
	'constants for frequency %d.', freq, approx_freq)
	return approx_freq


	def close(self):
	"""Close variable attenuator connection."""
	self._attenuator.close()


	def set_total_attenuation(self, atten_db, frequency_mhz,
	attenuator_num=None):
	"""Set the total attenuation on one or all attenuators.

	@param atten_db: int level of attenuation in dB. This must be
	higher than the fixed attenuation level of the affected
	attenuators.
	@param frequency_mhz: int frequency for which to calculate the
	total attenuation. The fixed component of attenuation
	varies with frequency.
	@param attenuator_num: int attenuator to change, or None to
	set all variable attenuators.

	"""
	affected_attenuators = self.supported_attenuators
	if attenuator_num is not None:
	affected_attenuators = [attenuator_num]
	for atten in affected_attenuators:
	freq_to_fixed_loss = self._fixed_attenuations[atten]
	approx_freq = self._approximate_frequency(atten,
	frequency_mhz)
	variable_atten_db = atten_db - freq_to_fixed_loss[approx_freq]
	self.set_variable_attenuation(variable_atten_db,
	attenuator_num=atten)


	def set_variable_attenuation(self, atten_db, attenuator_num=None):
	"""Set the variable attenuation on one or all attenuators.

	@param atten_db: int non-negative level of attenuation in dB.
	@param attenuator_num: int attenuator to change, or None to
	set all variable attenuators.

	"""
	affected_attenuators = self.supported_attenuators
	if attenuator_num is not None:
	affected_attenuators = [attenuator_num]
	for atten in affected_attenuators:
	try:
	self._attenuator.set_atten(atten, atten_db)
	if int(self._attenuator.get_atten(atten)) != atten_db:
	raise error.TestError('Attenuation did not set as expected '
	'on attenuator %d' % atten)
	except error.TestError:
	self._attenuator.reopen(self.hostname)
	self._attenuator.set_atten(atten, atten_db)
	if int(self._attenuator.get_atten(atten)) != atten_db:
	raise error.TestError('Attenuation did not set as expected '
	'on attenuator %d' % atten)
	logging.info('%ddb attenuation set successfully on attenautor %d',
	atten_db, atten)


	def get_minimal_total_attenuation(self):
	"""Get attenuator's maximum fixed attenuation value.

	This is pulled from the current attenuator's lines and becomes the
	minimal total attenuation when stepping through attenuation levels.

	@return maximum starting attenuation value

	"""
	max_atten = 0
	for atten_num in self._fixed_attenuations.iterkeys():
	atten_values = self._fixed_attenuations[atten_num].values()
	max_atten = max(max(atten_values), max_atten)
	return max_atten


	def set_signal_level(self, client_context, requested_sig_level,
	min_sig_level_allowed=-85, tolerance_percent=3, timeout=240):
	"""Set wifi signal to desired level by changing attenuation.

	@param client_context: Client context object.
	@param requested_sig_level: Negative int value in dBm for wifi signal
	level to be set.
	@param min_sig_level_allowed: Minimum signal level allowed; this is to
	ensure that we don't set a signal that is too weak and DUT can
	not associate.
	@param tolerance_percent: Percentage to be used to calculate the desired
	range for the wifi signal level.
	"""
	atten_db = 0
	starting_sig_level = client_context.wifi_signal_level
	if not starting_sig_level:
	raise error.TestError("No signal detected.")
	if not (min_sig_level_allowed <= requested_sig_level <=
	starting_sig_level):
	raise error.TestError("Requested signal level (%d) is either "
	"higher than current signal level (%r) with "
	"0db attenuation or lower than minimum "
	"signal level (%d) allowed." %
	(requested_sig_level,
	starting_sig_level,
	min_sig_level_allowed))

	try:
	with timeout_util.Timeout(timeout):
	while True:
	client_context.reassociate(timeout_seconds=1)
	current_sig_level = client_context.wifi_signal_level
	logging.info("Current signal level %r", current_sig_level)
	if not current_sig_level:
	raise error.TestError("No signal detected.")
	if self.signal_in_range(requested_sig_level,
	current_sig_level, tolerance_percent):
	logging.info("Signal level set to %r.",
	current_sig_level)
	break
	if current_sig_level > requested_sig_level:
	self.set_variable_attenuation(atten_db)
	atten_db +=1
	if current_sig_level < requested_sig_level:
	self.set_variable_attenuation(atten_db)
	atten_db -= 1
	except (timeout_util.TimeoutError, error.TestError,
	error.TestFail) as e:
	raise error.TestError("Not able to set wifi signal to requested "
	"level. \n%s" % e)


	def signal_in_range(self, req_sig_level, curr_sig_level, tolerance_percent):
	"""Check if wifi signal is within the threshold of requested signal.

	@param req_sig_level: Negative int value in dBm for wifi signal
	level to be set.
	@param curr_sig_level: Current wifi signal level seen by the DUT.
	@param tolerance_percent: Percentage to be used to calculate the desired
	range for the wifi signal level.

	@returns True if wifi signal is in the desired range.
	"""
	min_sig = req_sig_level + (req_sig_level * tolerance_percent / 100)
	max_sig = req_sig_level - (req_sig_level * tolerance_percent / 100)
	if min_sig <= curr_sig_level <= max_sig:
	return True
	return False