| # Copyright 2018 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. |
| |
| """Helper class for managing charging the DUT with Servo v4.""" |
| |
| import logging |
| import time |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import retry |
| |
| # Base delay time in seconds for Servo role change and PD negotiation. |
| _DELAY_SEC = 0.1 |
| # Total delay time in minutes for Servo role change and PD negotiation. |
| _TIMEOUT_MIN = 0.3 |
| # Exponential backoff for Servo role change and PD negotiation. |
| _BACKOFF = 2 |
| # Number of attempts to recover Servo v4. |
| _RETRYS = 3 |
| # Seconds to wait after resetting the role on a recovery attempt |
| # before trying to set it to the intended role again. |
| _RECOVERY_WAIT_SEC = 1 |
| # Delay to wait before polling whether the role as been changed successfully. |
| _ROLE_SETTLING_DELAY_SEC = 1 |
| |
| # Threshold (in millivolt) under which the test assumes that no charger |
| # is attached on the v4 charger side. |
| _CHG_ATTACHED_MIN_VOLTAGE_MV = 1000 |
| |
| def _invert_role(role): |
| """Helper to invert the role. |
| |
| @param role: role to invert |
| |
| @returns: |
| 'src' if |role| is 'snk' |
| 'snk' if |role| is 'src' |
| """ |
| return 'src' if role == 'snk' else 'snk' |
| |
| class ServoV4ChargeManager(object): |
| """A helper class for managing charging the DUT with Servo v4.""" |
| |
| def __init__(self, host, servo): |
| """Check for correct Servo setup. |
| |
| Make sure that Servo is v4 and can manage charging. Make sure that DUT |
| responds to Servo charging commands. Restore Servo v4 power role after |
| sanity check. |
| |
| @param host: CrosHost object representing the DUT. |
| @param servo: a proxy for servod. |
| """ |
| super(ServoV4ChargeManager, self).__init__() |
| self._host = host |
| self._servo = servo |
| self._original_role = self._servo.get('servo_v4_role') |
| |
| self._verify_v4() |
| if self._original_role == 'snk': |
| self.start_charging() |
| self.stop_charging() |
| elif self._original_role == 'src': |
| self.stop_charging() |
| self.start_charging() |
| else: |
| raise error.TestNAError('Unrecognized Servo v4 power role: %s.' % |
| self._original_role) |
| # Configure the INAs to read out power numbers later. |
| self._servo.set_nocheck('ppdut5_cfg_reg', 'regular_power') |
| self._servo.set_nocheck('ppchg5_cfg_reg', 'regular_power') |
| |
| # TODO(b/129882930): once both sides are stable, remove the _retry_wrapper |
| # wrappers as they aren't needed anymore. The current motivation for the |
| # retry loop in the autotest framework is to have a 'stable' library i.e. |
| # retries but also a mechanism and and easy to remove bridge once the bug |
| # is fixed, and we don't require the bandaid anymore. |
| |
| def _retry_wrapper(self, role, verify): |
| """Try up to |_RETRYS| times to set the v4 to |role|. |
| |
| @param role: string 'src' or 'snk'. If 'src' connect DUT to AC power; if |
| 'snk' disconnect DUT from AC power. |
| @param verify: bool to verify that charging started/stopped. |
| |
| @returns: number of retries needed for success |
| """ |
| for retry in range(_RETRYS): |
| try: |
| self._change_role(role, verify) |
| return retry |
| except error.TestError as e: |
| if retry < _RETRYS - 1: |
| # Ensure this retry loop and logging isn't run on the |
| # last iteration. |
| logging.warning('Failed to set to %s %d times. %s ' |
| 'Trying to cycle through %s to ' |
| 'recover.', role, retry + 1, str(e), |
| _invert_role(role)) |
| # Cycle through the other state before retrying. Do not |
| # verify as this is strictly a recovery mechanism - sleep |
| # instead. |
| self._change_role(_invert_role(role), verify=False) |
| time.sleep(_RECOVERY_WAIT_SEC) |
| logging.error('Giving up on %s.', role) |
| raise e |
| |
| def stop_charging(self, verify=True): |
| """Cut off AC power supply to DUT with Servo. |
| |
| @param verify: whether to verify that charging stopped. |
| |
| @returns: number of retries needed for success |
| """ |
| return self._retry_wrapper('snk', verify) |
| |
| def start_charging(self, verify=True): |
| """Connect AC power supply to DUT with Servo. |
| |
| @param verify: whether to verify that charging started. |
| |
| @returns: number of retries needed for success |
| """ |
| return self._retry_wrapper('src', verify) |
| |
| def restore_original_setting(self, verify=True): |
| """Restore Servo to original charging setting. |
| |
| @param verify: whether to verify that original role was restored. |
| """ |
| self._retry_wrapper(self._original_role, verify) |
| |
| def _verify_v4(self): |
| """Verify that Servo is Servo v4.""" |
| if self._servo.get_servo_version().startswith('servo_v4'): |
| servo_v4_version = self._servo.get('servo_v4_version') |
| logging.info('Servo v4 version: %s', servo_v4_version) |
| else: |
| raise error.TestNAError('This test needs to run with Servo v4. ' |
| 'Test skipped.') |
| if self._servo.get('servo_v4_type') != 'type-c': |
| raise error.TestNAError('Please use a type-c v4 to be able to ' |
| 'toggle PD.') |
| if self._servo.get('ppchg5_mv') < _CHG_ATTACHED_MIN_VOLTAGE_MV: |
| raise error.TestNAError('There seems to be no voltage on the ' |
| 'charger side. This test cannot run ' |
| 'without a PD charger attached to the v4.') |
| |
| def _change_role(self, role, verify=True): |
| """Change Servo PD role and check if DUT responded accordingly. |
| |
| @param role: string 'src' or 'snk'. If 'src' connect DUT to AC power; if |
| 'snk' disconnect DUT from AC power. |
| @param verify: bool to verify that charging started/stopped. |
| |
| @raises error.TestError: if the role did not change successfully. |
| """ |
| self._servo.set_nocheck('servo_v4_role', role) |
| # Sometimes the role reverts quickly. Add a short delay to let the new |
| # role stabilize. |
| time.sleep(_ROLE_SETTLING_DELAY_SEC) |
| |
| if not verify: |
| return |
| @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, |
| delay_sec=_DELAY_SEC, backoff=_BACKOFF) |
| def check_servo_role(role): |
| """Check if servo role is as expected, if not, retry.""" |
| if self._servo.get('servo_v4_role') != role: |
| raise error.TestError('Servo v4 failed to set its PD role to ' |
| '%s.' % role) |
| check_servo_role(role) |
| |
| connected = True if role == 'src' else False |
| |
| @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, |
| delay_sec=_DELAY_SEC, backoff=_BACKOFF) |
| def check_ac_connected(connected): |
| """Check if the EC believes an AC charger is connected.""" |
| if not self._servo.has_control('charger_connected'): |
| # TODO(coconutruben): remove this check once labs have the |
| # latest hdctools with the required control. |
| logging.warn('Could not verify %r control as the ' |
| 'control is not available on servod.', |
| 'charger_connected') |
| return |
| ec_opinion = self._servo.get('charger_connected') |
| if ec_opinion != connected: |
| str_lookup = {True: 'connected', False: 'disconnected'} |
| msg = ('EC thinks charger is %s but it should be %s.' |
| % (str_lookup[ec_opinion], |
| str_lookup[connected])) |
| raise error.TestError(msg) |
| |
| check_ac_connected(connected) |
| |
| @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, |
| delay_sec=_DELAY_SEC, backoff=_BACKOFF) |
| def check_host_ac(connected): |
| """Check if DUT AC power is as expected, if not, retry.""" |
| if self._host.is_ac_connected() != connected: |
| intent = 'connect' if connected else 'disconnect' |
| raise error.TestError('DUT failed to %s AC power.'% intent) |
| # TODO(b:143467862): Replace this try/except with a servo.has_control() |
| # test once Sarien servo overlay correctly says that Sarien does not |
| # have this control. |
| try: |
| power_state = self._servo.get('ec_system_powerstate') |
| except error.TestFail: |
| logging.warn('Could not verify that the DUT observes power as the ' |
| '%r control is not available on servod.', |
| 'ec_system_powerstate') |
| power_state = None |
| if power_state == 'S0': |
| # If the DUT has been charging in S3/S5/G3, cannot verify. |
| check_host_ac(connected) |
