| # Copyright 2016 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 re |
| import logging |
| import time |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.server.cros.servo import pd_console |
| |
| |
| class PDDevice(object): |
| """Base clase for all PD devices |
| |
| This class provides a set of APIs for expected Type C PD required actions |
| in TypeC FAFT tests. The base class is specific for Type C devices that |
| do not have any console access. |
| |
| """ |
| |
| def is_src(self): |
| """Checks if the port is connected as a source |
| |
| """ |
| raise NotImplementedError( |
| 'is_src should be implemented in derived class') |
| |
| def is_snk(self): |
| """Checks if the port is connected as a sink |
| |
| @returns None |
| """ |
| raise NotImplementedError( |
| 'is_snk should be implemented in derived class') |
| |
| def is_connected(self): |
| """Checks if the port is connected |
| |
| @returns True if in a connected state, False otherwise |
| """ |
| return self.is_src() or self.is_snk() |
| |
| def is_disconnected(self): |
| """Checks if the port is disconnected |
| |
| """ |
| raise NotImplementedError( |
| 'is_disconnected should be implemented in derived class') |
| |
| def is_ufp(self): |
| """Checks if data role is UFP |
| |
| """ |
| raise NotImplementedError( |
| 'is_ufp should be implemented in derived class') |
| |
| def is_dfp(self): |
| """Checks if data role is DFP |
| |
| """ |
| raise NotImplementedError( |
| 'is_dfp should be implemented in derived class') |
| |
| def is_drp(self): |
| """Checks if dual role mode is supported |
| |
| """ |
| raise NotImplementedError( |
| 'is_drp should be implemented in derived class') |
| |
| def dr_swap(self): |
| """Attempts a data role swap |
| |
| """ |
| raise NotImplementedError( |
| 'dr_swap should be implemented in derived class') |
| |
| def pr_swap(self): |
| """Attempts a power role swap |
| |
| """ |
| raise NotImplementedError( |
| 'pr_swap should be implemented in derived class') |
| |
| def vbus_request(self, voltage): |
| """Requests a specific VBUS voltage from SRC |
| |
| @param voltage: requested voltage level (5, 12, 20) in volts |
| """ |
| raise NotImplementedError( |
| 'vbus_request should be implemented in derived class') |
| |
| def soft_reset(self): |
| """Initates a PD soft reset sequence |
| |
| """ |
| raise NotImplementedError( |
| 'soft_reset should be implemented in derived class') |
| |
| def hard_reset(self): |
| """Initates a PD hard reset sequence |
| |
| """ |
| raise NotImplementedError( |
| 'hard_reset should be implemented in derived class') |
| |
| def drp_set(self, mode): |
| """Sets dualrole mode |
| |
| @param mode: desired dual role setting (on, off, snk, src) |
| """ |
| raise NotImplementedError( |
| 'drp_set should be implemented in derived class') |
| |
| def drp_disconnect_connect(self, disc_time_sec): |
| """Force PD disconnect/connect via drp settings |
| |
| @param disc_time_sec: Time in seconds between disconnect and reconnect |
| """ |
| raise NotImplementedError( |
| 'drp_disconnect_reconnect should be implemented in \ |
| derived class') |
| |
| def cc_disconnect_connect(self, disc_time_sec): |
| """Force PD disconnect/connect using Plankton fw command |
| |
| @param disc_time_sec: Time in seconds between disconnect and reconnect |
| """ |
| raise NotImplementedError( |
| 'cc_disconnect_reconnect should be implemented in \ |
| derived class') |
| |
| |
| class PDConsoleDevice(PDDevice): |
| """Class for PD devices that have console access |
| |
| This class contains methods for common PD actions for any PD device which |
| has UART console access. It inherits the PD device base class. In addition, |
| it stores both the UART console and port for the PD device. |
| """ |
| |
| def __init__(self, console, port): |
| """Initialization method |
| |
| @param console: UART console object |
| @param port: USB PD port number |
| """ |
| # Save UART console |
| self.console = console |
| # Instantiate PD utilities used by methods in this class |
| self.utils = pd_console.PDConsoleUtils(console) |
| # Save the PD port number for this device |
| self.port = port |
| # Not a Plankton device |
| self.is_plankton = False |
| |
| def is_src(self): |
| """Checks if the port is connected as a source |
| |
| @returns True if connected as SRC, False otherwise |
| """ |
| state = self.utils.get_pd_state(self.port) |
| return bool(state == self.utils.SRC_CONNECT) |
| |
| def is_snk(self): |
| """Checks if the port is connected as a sink |
| |
| @returns True if connected as SNK, False otherwise |
| """ |
| state = self.utils.get_pd_state(self.port) |
| return bool(state == self.utils.SNK_CONNECT) |
| |
| def is_connected(self): |
| """Checks if the port is connected |
| |
| @returns True if in a connected state, False otherwise |
| """ |
| state = self.utils.get_pd_state(self.port) |
| return bool(state == self.utils.SNK_CONNECT or |
| state == self.utils.SRC_CONNECT) |
| |
| def is_disconnected(self): |
| """Checks if the port is disconnected |
| |
| @returns True if in a disconnected state, False otherwise |
| """ |
| state = self.utils.get_pd_state(self.port) |
| return bool(state == self.utils.SRC_DISC or |
| state == self.utils.SNK_DISC) |
| |
| def is_drp(self): |
| """Checks if dual role mode is supported |
| |
| @returns True if dual role mode is 'on', False otherwise |
| """ |
| return self.utils.is_pd_dual_role_enabled() |
| |
| def drp_disconnect_connect(self, disc_time_sec): |
| """Disconnect/reconnect using drp mode settings |
| |
| A PD console device doesn't have an explicit connect/disconnect |
| command. Instead, the dualrole mode setting is used to force |
| disconnects in devices which support this feature. To disconnect, |
| force the dualrole mode to be the opposite role of the current |
| connected state. |
| |
| @param disc_time_sec: time in seconds to wait to reconnect |
| |
| @returns True if device disconnects, then returns to a connected |
| state. False if either step fails. |
| """ |
| # Dualrole mode must be supported |
| if self.is_drp() is False: |
| logging.warn('Device not DRP capable, unabled to force disconnect') |
| return False |
| # Force state will be the opposite of current connect state |
| if self.is_src(): |
| drp_mode = 'snk' |
| swap_state = self.utils.SNK_CONNECT |
| else: |
| drp_mode = 'src' |
| swap_state = self.utils.SRC_CONNECT |
| # Force disconnect |
| self.drp_set(drp_mode) |
| # Wait for disconnect time |
| time.sleep(disc_time_sec) |
| # Verify that the device is disconnected |
| disconnect = self.is_disconnected() |
| |
| # If the other device is dualrole, then forcing dualrole mode will |
| # only cause the disconnect to appear momentarily and reconnect |
| # in the power role forced by the drp_set() call. For this case, |
| # the role swap verifies that a disconnect/connect sequence occurred. |
| if disconnect == False: |
| time.sleep(self.utils.CONNECT_TIME) |
| # Connected, verify if power role swap has ocurred |
| if swap_state == self.utils.get_pd_state(self.port): |
| # Restore default dualrole mode |
| self.drp_set('on') |
| # Restore orignal power role |
| connect = self.pr_swap() |
| if connect == False: |
| logging.warn('DRP on both devices, 2nd power swap failed') |
| return connect |
| |
| # Restore default dualrole mode |
| self.drp_set('on') |
| # Allow enough time for protocol state machine |
| time.sleep(self.utils.CONNECT_TIME) |
| # Check if connected |
| connect = self.is_connected() |
| logging.info('Disconnect = %r, Connect = %r', disconnect, connect) |
| return bool(disconnect and connect) |
| |
| def drp_set(self, mode): |
| """Sets dualrole mode |
| |
| @param mode: desired dual role setting (on, off, snk, src) |
| |
| @returns True is set was successful, False otherwise |
| """ |
| # Set desired dualrole mode |
| self.utils.set_pd_dualrole(mode) |
| # Get the expected output |
| resp = self.utils.dualrole_resp[self.utils.dual_index[mode]] |
| # Get current setting |
| current = self.utils.get_pd_dualrole() |
| # Verify that setting is correct |
| return bool(resp == current) |
| |
| def try_src(self, enable): |
| """Enables/Disables Try.SRC PD protocol setting |
| |
| @param enable: True to enable, False to disable |
| |
| @returns True is setting was successful, False if feature not |
| supported by the device, or not set as desired. |
| """ |
| # Create Try.SRC pd command |
| cmd = 'pd trysrc %d' % int(enable) |
| # Try.SRC on/off is output, if supported feature |
| regex = ['Try\.SRC\s([\w]+)|(Parameter)'] |
| m = self.utils.send_pd_command_get_output(cmd, regex) |
| # Determine if Try.SRC feature is supported |
| trysrc = re.search('Try\.SRC\s([\w]+)', m[0][0]) |
| if not trysrc: |
| logging.warn('Try.SRC not supported on this PD device') |
| return False |
| # TrySRC is supported on this PD device, verify setting. |
| logging.info('Try.SRC mode = %s', trysrc.group(1)) |
| if enable: |
| val = 'on' |
| else: |
| val = 'off' |
| return bool(val == m[0][1]) |
| |
| def soft_reset(self): |
| """Initates a PD soft reset sequence |
| |
| To verify that a soft reset sequence was initiated, the |
| reply message is checked to verify that the reset command |
| was acknowledged by its port pair. The connect state should |
| be same as it was prior to issuing the reset command. |
| |
| @returns True if the port pair acknowledges the the reset message |
| and if following the command, the device returns to the same |
| connected state. False otherwise. |
| """ |
| RESET_DELAY = 0.5 |
| cmd = 'pd %d soft' % self.port |
| state_before = self.utils.get_pd_state(self.port) |
| reply = self.utils.send_pd_command_get_reply_msg(cmd) |
| if reply != self.utils.PD_CONTROL_MSG_DICT['Accept']: |
| return False |
| time.sleep(RESET_DELAY) |
| state_after = self.utils.get_pd_state(self.port) |
| return state_before == state_after |
| |
| def hard_reset(self): |
| """Initates a PD hard reset sequence |
| |
| To verify that a hard reset sequence was initiated, the |
| console ouput is scanned for HARD RST TX. In addition, the connect |
| state should be same as it was prior to issuing the reset command. |
| |
| @returns True if the port pair acknowledges that hard reset was |
| initiated and if following the command, the device returns to the same |
| connected state. False otherwise. |
| """ |
| RESET_DELAY = 1.0 |
| cmd = 'pd %d hard' % self.port |
| state_before = self.utils.get_pd_state(self.port) |
| self.utils.enable_pd_console_debug() |
| try: |
| self.utils.send_pd_command_get_output(cmd, ['.*(HARD\sRST\sTX)']) |
| except error.TestFail: |
| logging.warn('HARD RST TX not found') |
| return False |
| finally: |
| self.utils.disable_pd_console_debug() |
| |
| time.sleep(RESET_DELAY) |
| state_after = self.utils.get_pd_state(self.port) |
| return state_before == state_after |
| |
| def pr_swap(self): |
| """Attempts a power role swap |
| |
| In order to attempt a power role swap the device must be |
| connected and support dualrole mode. Once these two criteria |
| are checked a power role command is issued. Following a delay |
| to allow for a reconnection the new power role is checked |
| against the power role prior to issuing the command. |
| |
| @returns True if the device has swapped power roles, False otherwise. |
| """ |
| # Get starting state |
| if not self.is_drp() and not self.drp_set('on'): |
| logging.warn('Dualrole Mode not enabled!') |
| return False |
| if self.is_connected() == False: |
| logging.warn('PD contract not established!') |
| return False |
| current_pr = self.utils.get_pd_state(self.port) |
| swap_cmd = 'pd %d swap power' % self.port |
| self.utils.send_pd_command(swap_cmd) |
| time.sleep(self.utils.CONNECT_TIME) |
| new_pr = self.utils.get_pd_state(self.port) |
| logging.info('Power swap: %s -> %s', current_pr, new_pr) |
| if self.is_connected() == False: |
| logging.warn('Device not connected following PR swap attempt.') |
| return False |
| return current_pr != new_pr |
| |
| |
| class PDPlanktonDevice(PDConsoleDevice): |
| """Class for PD Plankton devices |
| |
| This class contains methods for PD funtions which are unique to the |
| Plankton Type C factory testing board. It inherits all the methods |
| for PD console devices. |
| """ |
| |
| def __init__(self, console, port): |
| """Initialization method |
| |
| @param console: UART console for this device |
| @param port: USB PD port number |
| """ |
| # Instantiate the PD console object |
| super(PDPlanktonDevice, self).__init__(console, 0) |
| # Indicate this is Plankton device |
| self.is_plankton = True |
| |
| def _toggle_plankton_drp(self): |
| """Issue 'usbc_action drp' Plankton command |
| |
| @returns value of drp_enable in Plankton FW |
| """ |
| drp_cmd = 'usbc_action drp' |
| drp_re = ['DRP\s=\s(\d)'] |
| # Send DRP toggle command to Plankton and get value of 'drp_enable' |
| m = self.utils.send_pd_command_get_output(drp_cmd, drp_re) |
| return int(m[0][1]) |
| |
| def _enable_plankton_drp(self): |
| """Enable DRP mode on Plankton |
| |
| DRP mode can only be toggled and is not able to be explicitly |
| enabled/disabled via the console. Therefore, this method will |
| toggle DRP mode until the console reply indicates that this |
| mode is enabled. The toggle happens a maximum of two times |
| in case this is called when it's already enabled. |
| |
| @returns True when DRP mode is enabled, False if not successful |
| """ |
| for attempt in xrange(2): |
| if self._toggle_plankton_drp() == True: |
| logging.info('Plankton DRP mode enabled') |
| return True |
| logging.error('Plankton DRP mode set failure') |
| return False |
| |
| def _verify_state_sequence(self, states_list, console_log): |
| """Compare PD state transitions to expected values |
| |
| @param states_list: list of expected PD state transitions |
| @param console_log: console output which contains state names |
| |
| @returns True if the sequence matches, False otherwise |
| """ |
| # For each state in the expected state transiton table, build |
| # the regexp and search for it in the state transition log. |
| for state in states_list: |
| state_regx = r'C{0}\s+[\w]+:\s({1})'.format(self.port, |
| state) |
| if re.search(state_regx, console_log) is None: |
| return False |
| return True |
| |
| def cc_disconnect_connect(self, disc_time_sec): |
| """Disconnect/reconnect using Plankton |
| |
| Plankton supports a feature which simulates a USB Type C disconnect |
| and reconnect. |
| |
| @param disc_time_sec: Time in seconds for disconnect period. |
| """ |
| DISC_DELAY = 100 |
| disc_cmd = 'fake_disconnect %d %d' % (DISC_DELAY, |
| disc_time_sec * 1000) |
| self.utils.send_pd_command(disc_cmd) |
| |
| def drp_disconnect_connect(self, disc_time_sec): |
| """Disconnect/reconnect using Plankton |
| |
| Utilize Plankton disconnect/connect utility and verify |
| that both disconnect and reconnect actions were successful. |
| |
| @param disc_time_sec: Time in seconds for disconnect period. |
| |
| @returns True if device disconnects, then returns to a connected |
| state. False if either step fails. |
| """ |
| self.cc_disconnect_connect(disc_time_sec) |
| time.sleep(disc_time_sec / 2) |
| disconnect = self.is_disconnected() |
| time.sleep(disc_time_sec / 2 + self.utils.CONNECT_TIME) |
| connect = self.is_connected() |
| return disconnect and connect |
| |
| def drp_set(self, mode): |
| """Sets dualrole mode |
| |
| @param mode: desired dual role setting (on, off, snk, src) |
| |
| @returns True if dualrole mode matches the requested value or |
| is successfully set to that value. False, otherwise. |
| """ |
| # Get correct dualrole console response |
| resp = self.utils.dualrole_resp[self.utils.dual_index[mode]] |
| # Get current value of dualrole |
| drp = self.utils.get_pd_dualrole() |
| if drp == resp: |
| return True |
| |
| if mode == 'on': |
| # Setting dpr_enable on Plankton will set dualrole mode to on |
| return self._enable_plankton_drp() |
| else: |
| # If desired setting is other than 'on', need to ensure that |
| # drp mode on Plankton is disabled. |
| if resp == 'on': |
| # This will turn off drp_enable flag and set dualmode to 'off' |
| return self._toggle_plankton_drp() |
| # With drp_enable flag off, can set to desired setting |
| return self.utils.set_pd_dualrole(mode) |
| |
| def _reset(self, cmd, states_list): |
| """Initates a PD reset sequence |
| |
| Plankton device has state names available on the console. When |
| a soft reset is issued the console log is extracted and then |
| compared against the expected state transisitons. |
| |
| @param cmd: reset type (soft or hard) |
| @param states_list: list of expected PD state transitions |
| |
| @returns True if state transitions match, False otherwise |
| """ |
| # Want to grab all output until either SRC_READY or SNK_READY |
| reply_exp = ['(.*)(C\d)\s+[\w]+:\s([\w]+_READY)'] |
| m = self.utils.send_pd_command_get_output(cmd, reply_exp) |
| return self._verify_state_sequence(states_list, m[0][0]) |
| |
| def soft_reset(self): |
| """Initates a PD soft reset sequence |
| |
| @returns True if state transitions match, False otherwise |
| """ |
| snk_reset_states = [ |
| 'SOFT_RESET', |
| 'SNK_DISCOVERY', |
| 'SNK_REQUESTED', |
| 'SNK_TRANSITION', |
| 'SNK_READY' |
| ] |
| |
| src_reset_states = [ |
| 'SOFT_RESET', |
| 'SRC_DISCOVERY', |
| 'SRC_NEGOCIATE', |
| 'SRC_ACCEPTED', |
| 'SRC_POWERED', |
| 'SRC_TRANSITION', |
| 'SRC_READY' |
| ] |
| |
| if self.is_src(): |
| states_list = src_reset_states |
| elif self.is_snk(): |
| states_list = snk_reset_states |
| else: |
| raise error.TestFail('Port Pair not in a connected state') |
| |
| cmd = 'pd %d soft' % self.port |
| return self._reset(cmd, states_list) |
| |
| def hard_reset(self): |
| """Initates a PD hard reset sequence |
| |
| @returns True if state transitions match, False otherwise |
| """ |
| snk_reset_states = [ |
| 'HARD_RESET_SEND', |
| 'HARD_RESET_EXECUTE', |
| 'SNK_HARD_RESET_RECOVER', |
| 'SNK_DISCOVERY', |
| 'SNK_REQUESTED', |
| 'SNK_TRANSITION', |
| 'SNK_READY' |
| ] |
| |
| src_reset_states = [ |
| 'HARD_RESET_SEND', |
| 'HARD_RESET_EXECUTE', |
| 'SRC_HARD_RESET_RECOVER', |
| 'SRC_DISCOVERY', |
| 'SRC_NEGOCIATE', |
| 'SRC_ACCEPTED', |
| 'SRC_POWERED', |
| 'SRC_TRANSITION', |
| 'SRC_READY' |
| ] |
| |
| if self.is_src(): |
| states_list = src_reset_states |
| elif self.is_snk(): |
| states_list = snk_reset_states |
| else: |
| raise error.TestFail('Port Pair not in a connected state') |
| |
| cmd = 'pd %d hard' % self.port |
| return self._reset(cmd, states_list) |
| |
| |
| class PDPortPartner(object): |
| """Methods used to instantiate PD device objects |
| |
| This class is initalized with a list of servo consoles. It |
| contains methods to determine if USB PD devices are accessible |
| via the consoles and attempts to determine USB PD port partners. |
| A PD device is USB PD port specific, a single console may access |
| multiple PD devices. |
| |
| """ |
| |
| def __init__(self, consoles): |
| """Initialization method |
| |
| @param consoles: list of servo consoles |
| """ |
| self.consoles = consoles |
| |
| def _send_pd_state(self, port, console): |
| """Tests if PD device exists on a given port number |
| |
| @param port: USB PD port number to try |
| @param console: servo UART console |
| |
| @returns True if 'pd <port> state' command gives a valid |
| response, False otherwise |
| """ |
| cmd = 'pd %d state' % port |
| regex = r'(Port C\d)|(Parameter)' |
| m = console.send_command_get_output(cmd, [regex]) |
| # If PD port exists, then output will be Port C0 or C1 |
| regex = r'Port C{0}'.format(port) |
| if re.search(regex, m[0][0]): |
| return True |
| return False |
| |
| def _find_num_pd_ports(self, console): |
| """Determine number of PD ports for a given console |
| |
| @param console: uart console accssed via servo |
| |
| @returns: number of PD ports accessible via console |
| """ |
| MAX_PORTS = 2 |
| num_ports = 0 |
| for port in xrange(MAX_PORTS): |
| if self._send_pd_state(port, console): |
| num_ports += 1 |
| return num_ports |
| |
| def _is_pd_console(self, console): |
| """Check if pd option exists in console |
| |
| @param console: uart console accssed via servo |
| |
| @returns: True if 'pd' is found, False otherwise |
| """ |
| try: |
| m = console.send_command_get_output('help', [r'(pd)\s+']) |
| return True |
| except error.TestFail: |
| return False |
| |
| def _is_plankton_console(self, console): |
| """Check for Plankton console |
| |
| This method looks for a console command option 'usbc_action' which |
| is unique to Plankton PD devices. |
| |
| @param console: uart console accssed via servo |
| |
| @returns True if usbc_action command is present, False otherwise |
| """ |
| try: |
| m = console.send_command_get_output('help', [r'(usbc_action)']) |
| return True |
| except error.TestFail: |
| return False |
| |
| def _check_port_pair(self, dev_pair): |
| """Check if two PD devices could be connected |
| |
| If two USB PD devices are connected, then they should be in |
| either the SRC_READY or SNK_READY states and have opposite |
| power roles. In addition, they must be on different servo |
| consoles. |
| |
| @param: list of two possible PD port parters |
| |
| @returns True if not the same console and both PD devices |
| are a plausible pair based only on their PD states. |
| """ |
| # Don't test if on the same servo console |
| if dev_pair[0].console == dev_pair[1].console: |
| logging.info('PD Devices are on same platform -> cant be a pair') |
| return False |
| # Must be SRC <--> SNK or SNK <--> SRC |
| return bool((dev_pair[0].is_src() and dev_pair[1].is_snk()) or |
| (dev_pair[0].is_snk() and dev_pair[1].is_src())) |
| |
| def _verify_plankton_connection(self, dev_pair): |
| """Verify DUT to Plankton PD connection |
| |
| This method checks for a Plankton PD connection for the |
| given port by first verifying if a PD connection is present. |
| If found, then it uses a Plankton feature to force a PD disconnect. |
| If the port is no longer in the connected state, and following |
| a delay, is found to be back in the connected state, then |
| a DUT pd to Plankton connection is verified. |
| |
| @param dev_pair: list of two PD devices |
| |
| @returns True if DUT to Plankton pd connection is verified |
| """ |
| DISC_CHECK_TIME = .5 |
| DISC_WAIT_TIME = 2 |
| CONNECT_TIME = 4 |
| |
| if not self._check_port_pair(dev_pair): |
| return False |
| |
| for index in xrange(len(dev_pair)): |
| try: |
| # Force PD disconnect |
| dev_pair[index].cc_disconnect_connect(DISC_WAIT_TIME) |
| time.sleep(DISC_CHECK_TIME) |
| # Verify that both devices are now disconnected |
| if (dev_pair[0].is_disconnected() and |
| dev_pair[1].is_disconnected()): |
| # Allow enough time for reconnection |
| time.sleep(DISC_WAIT_TIME + CONNECT_TIME) |
| if self._check_port_pair(dev_pair): |
| # Have verifed a pd disconnect/reconnect sequence |
| logging.info('Plankton <-> DUT pair found') |
| return True |
| else: |
| # Delay to allow |
| time.sleep(DISC_WAIT_TIME + CONNECT_TIME) |
| except NotImplementedError: |
| logging.info('dev %d is not Plankton', index) |
| return False |
| |
| def identify_pd_devices(self): |
| """Instantiate PD devices present in test setup |
| |
| @returns list of 2 PD devices if a DUT <-> Plankton found. If |
| not found, then returns an empty list. |
| """ |
| devices = [] |
| # For each possible uart console, check to see if a PD console |
| # is present and determine the number of PD ports. |
| for console in self.consoles: |
| if self._is_pd_console(console): |
| is_plank = self._is_plankton_console(console) |
| num_ports = self._find_num_pd_ports(console) |
| # For each PD port that can be accessed via the console, |
| # instantiate either PDConsole or PDPlankton device. |
| for port in xrange(num_ports): |
| if is_plank: |
| logging.info('Plankton PD Device on port %d', port) |
| devices.append(PDPlanktonDevice(console, port)) |
| else: |
| devices.append(PDConsoleDevice(console, port)) |
| logging.info('Console PD Device on port %d', port) |
| |
| # Determine PD port partners in the list of PD devices. Note, that |
| # there can be PD devices which are not accessible via a uart console, |
| # but are connected to a PD port which is accessible. |
| test_pair = [] |
| for deva in devices: |
| for dev_idx in range(devices.index(deva) + 1, len(devices)): |
| devb = devices[dev_idx] |
| pair = [deva, devb] |
| if self._verify_plankton_connection(pair): |
| test_pair = pair |
| devices.remove(deva) |
| devices.remove(devb) |
| return test_pair |
| |
