1# Copyright 2018 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5"""Helper class for managing charging the DUT with Servo v4.""" 6 7import logging 8import time 9 10from autotest_lib.client.common_lib import error 11from autotest_lib.client.common_lib.cros import retry 12 13# Base delay time in seconds for Servo role change and PD negotiation. 14_DELAY_SEC = 0.1 15# Total delay time in minutes for Servo role change and PD negotiation. 16_TIMEOUT_MIN = 0.3 17# Exponential backoff for Servo role change and PD negotiation. 18_BACKOFF = 2 19# Number of attempts to recover Servo v4. 20_RETRYS = 3 21# Seconds to wait after resetting the role on a recovery attempt 22# before trying to set it to the intended role again. 23_RECOVERY_WAIT_SEC = 1 24# Delay to wait before polling whether the role as been changed successfully. 25_ROLE_SETTLING_DELAY_SEC = 1 26 27 28def _invert_role(role): 29 """Helper to invert the role. 30 31 @param role: role to invert 32 33 @returns: 34 'src' if |role| is 'snk' 35 'snk' if |role| is 'src' 36 """ 37 return 'src' if role == 'snk' else 'snk' 38 39class ServoV4ChargeManager(object): 40 """A helper class for managing charging the DUT with Servo v4.""" 41 42 def __init__(self, host, servo): 43 """Check for correct Servo setup. 44 45 Make sure that Servo is v4 and can manage charging. Make sure that DUT 46 responds to Servo charging commands. Restore Servo v4 power role after 47 sanity check. 48 49 @param host: CrosHost object representing the DUT. 50 @param servo: a proxy for servod. 51 """ 52 super(ServoV4ChargeManager, self).__init__() 53 self._host = host 54 self._servo = servo 55 if not self._servo.supports_built_in_pd_control(): 56 raise error.TestNAError('Servo setup does not support PD control. ' 57 'Check logs for details.') 58 59 self._original_role = self._servo.get('servo_v4_role') 60 if self._original_role == 'snk': 61 self.start_charging() 62 self.stop_charging() 63 elif self._original_role == 'src': 64 self.stop_charging() 65 self.start_charging() 66 else: 67 raise error.TestNAError('Unrecognized Servo v4 power role: %s.' % 68 self._original_role) 69 70 # TODO(b/129882930): once both sides are stable, remove the _retry_wrapper 71 # wrappers as they aren't needed anymore. The current motivation for the 72 # retry loop in the autotest framework is to have a 'stable' library i.e. 73 # retries but also a mechanism and and easy to remove bridge once the bug 74 # is fixed, and we don't require the bandaid anymore. 75 76 def _retry_wrapper(self, role, verify): 77 """Try up to |_RETRYS| times to set the v4 to |role|. 78 79 @param role: string 'src' or 'snk'. If 'src' connect DUT to AC power; if 80 'snk' disconnect DUT from AC power. 81 @param verify: bool to verify that charging started/stopped. 82 83 @returns: number of retries needed for success 84 """ 85 for retry in range(_RETRYS): 86 try: 87 self._change_role(role, verify) 88 return retry 89 except error.TestError as e: 90 if retry < _RETRYS - 1: 91 # Ensure this retry loop and logging isn't run on the 92 # last iteration. 93 logging.warning('Failed to set to %s %d times. %s ' 94 'Trying to cycle through %s to ' 95 'recover.', role, retry + 1, str(e), 96 _invert_role(role)) 97 # Cycle through the other state before retrying. Do not 98 # verify as this is strictly a recovery mechanism - sleep 99 # instead. 100 self._change_role(_invert_role(role), verify=False) 101 time.sleep(_RECOVERY_WAIT_SEC) 102 logging.error('Giving up on %s.', role) 103 raise e 104 105 def stop_charging(self, verify=True): 106 """Cut off AC power supply to DUT with Servo. 107 108 @param verify: whether to verify that charging stopped. 109 110 @returns: number of retries needed for success 111 """ 112 return self._retry_wrapper('snk', verify) 113 114 def start_charging(self, verify=True): 115 """Connect AC power supply to DUT with Servo. 116 117 @param verify: whether to verify that charging started. 118 119 @returns: number of retries needed for success 120 """ 121 return self._retry_wrapper('src', verify) 122 123 def restore_original_setting(self, verify=True): 124 """Restore Servo to original charging setting. 125 126 @param verify: whether to verify that original role was restored. 127 """ 128 self._retry_wrapper(self._original_role, verify) 129 130 def _change_role(self, role, verify=True): 131 """Change Servo PD role and check if DUT responded accordingly. 132 133 @param role: string 'src' or 'snk'. If 'src' connect DUT to AC power; if 134 'snk' disconnect DUT from AC power. 135 @param verify: bool to verify that charging started/stopped. 136 137 @raises error.TestError: if the role did not change successfully. 138 """ 139 self._servo.set_nocheck('servo_v4_role', role) 140 # Sometimes the role reverts quickly. Add a short delay to let the new 141 # role stabilize. 142 time.sleep(_ROLE_SETTLING_DELAY_SEC) 143 144 if not verify: 145 return 146 @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, 147 delay_sec=_DELAY_SEC, backoff=_BACKOFF) 148 def check_servo_role(role): 149 """Check if servo role is as expected, if not, retry.""" 150 if self._servo.get('servo_v4_role') != role: 151 raise error.TestError('Servo v4 failed to set its PD role to ' 152 '%s.' % role) 153 check_servo_role(role) 154 155 connected = True if role == 'src' else False 156 157 @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, 158 delay_sec=_DELAY_SEC, backoff=_BACKOFF) 159 def check_ac_connected(connected): 160 """Check if the EC believes an AC charger is connected.""" 161 if not self._servo.has_control('charger_connected'): 162 # TODO(coconutruben): remove this check once labs have the 163 # latest hdctools with the required control. 164 logging.warn('Could not verify %r control as the ' 165 'control is not available on servod.', 166 'charger_connected') 167 return 168 ec_opinion = self._servo.get('charger_connected') 169 if ec_opinion != connected: 170 str_lookup = {True: 'connected', False: 'disconnected'} 171 msg = ('EC thinks charger is %s but it should be %s.' 172 % (str_lookup[ec_opinion], 173 str_lookup[connected])) 174 raise error.TestError(msg) 175 176 check_ac_connected(connected) 177 178 @retry.retry(error.TestError, timeout_min=_TIMEOUT_MIN, 179 delay_sec=_DELAY_SEC, backoff=_BACKOFF) 180 def check_host_ac(connected): 181 """Check if DUT AC power is as expected, if not, retry.""" 182 if self._host.is_ac_connected() != connected: 183 intent = 'connect' if connected else 'disconnect' 184 raise error.TestError('DUT failed to %s AC power.'% intent) 185 # TODO(b:143467862): Replace this try/except with a servo.has_control() 186 # test once Sarien servo overlay correctly says that Sarien does not 187 # have this control. 188 try: 189 power_state = self._servo.get('ec_system_powerstate') 190 except error.TestFail: 191 logging.warn('Could not verify that the DUT observes power as the ' 192 '%r control is not available on servod.', 193 'ec_system_powerstate') 194 power_state = None 195 if power_state == 'S0': 196 # If the DUT has been charging in S3/S5/G3, cannot verify. 197 check_host_ac(connected) 198