xref: /aosp_15_r20/external/autotest/server/cros/servo/pd_console.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright 2015 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.

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import re
import logging
import six
from six.moves import range
import time

from autotest_lib.client.common_lib import error


class PDConsoleUtils(object):
    """Base clase for all PD console utils

    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 console access.

    """
    def __init__(self, console):
        """Console can be either usbpd, ec, or pdtester UART

        This object with then be used by the class which creates
        the PDConsoleUtils class to send/receive commands to UART
        """
        # save console for UART access functions
        self.console = console

    def send_pd_command(self, cmd):
        """Send command to PD console UART

        @param cmd: pd command string
        """
        self.console.send_command(cmd)

    def send_pd_command_get_output(self, cmd, regexp, debug_on=True):
        """Send command to PD console, wait for response

        @param cmd: pd command string
        @param regexp: regular expression for desired output
        """
        # Enable PD console debug mode to show control messages
        if debug_on:
            self.enable_pd_console_debug()
        output = self.console.send_command_get_output(cmd, regexp)
        if debug_on:
            self.disable_pd_console_debug()
        return output

    def send_pd_command_get_reply_msg(self, cmd):
        """Send PD protocol msg, get PD control msg reply

        The PD console debug mode is enabled prior to sending
        a pd protocol message. This allows the
        control message reply to be extracted. The debug mode
        is disabled prior to exiting.

        @param cmd: pd command to issue to the UART console

        @returns: PD control header message
        """
        m = self.send_pd_command_get_output(cmd, ['RECV\s([\w]+)\W'])
        ctrl_msg = int(m[0][1], 16) & self.PD_CONTROL_MSG_MASK
        return ctrl_msg

    def verify_pd_console(self):
        """Verify that PD commands exist on UART console

        Send 'help' command to UART console

        @returns: True if 'pd' is found, False if not
        """

        l = self.console.send_command_get_output('help', ['(pd)\s+([\w]+)'])
        if l[0][1] == 'pd':
            return True
        else:
            return False

    def get_pd_version(self):
        """Get the version of the PD stack

        @returns: version of PD stack, one of (1, 2)
        """
        # Match a number or an error ("Wrong number of params")
        matches = self.console.send_command_get_output('pd version',
                                                       [r'\s+(\d+|Wrong.*)'])
        if matches:
            result = matches[0][1]
            if result[0].isdigit():
                return int(result)
        return 1

    def execute_pd_state_cmd(self, port):
        """Get PD state for specified channel

        pd 0/1 state command gives produces 5 fields. The full response
        line is captured and then parsed to extract each field to fill
        the dict containing port, polarity, role, pd_state, and flags.

        @param port: Type C PD port 0 or 1

        @returns: A dict with the 5 fields listed above
        @raises: TestFail if any field not found
        """
        cmd = 'pd'
        subcmd = 'state'
        pd_cmd = cmd +" " + str(port) + " " + subcmd
        time.sleep(self.CURRENT_STATE_PROBE_DELAY)
        # Two FW versions for this command, get full line.
        m = self.send_pd_command_get_output(pd_cmd, ['(Port.*) - (Role:.*)\n'],
                                            debug_on=False)

        # Extract desired values from result string
        state_result = {}
        pd_state_dict = self.PD_STATE_DICT

        for key, regexp in six.iteritems(pd_state_dict):
            value = re.search(regexp, m[0][0])
            if value:
                state_result[key] = value.group(1)
            else:
                raise error.TestFail('pd %d state: %r value not found' %
                                     (port, key))

        return state_result

    def get_pd_state(self, port):
        """Get the current PD state

        """
        raise NotImplementedError(
            'should be implemented in derived class')

    def get_pd_port(self, port):
        """Get the current PD port

        @param port: Type C PD port 0/1
        @returns: current pd state
        """
        pd_dict = self.execute_pd_state_cmd(port)
        return pd_dict['port']

    def get_pd_role(self, port):
        """Get the current PD power role (source or sink)

        @param port: Type C PD port 0/1
        @returns: current pd state
        """
        pd_dict = self.execute_pd_state_cmd(port)
        return pd_dict['role']

    def get_pd_flags(self, port):
        """Get the current PD flags

        @param port: Type C PD port 0/1
        @returns: current pd state
        """
        pd_dict = self.execute_pd_state_cmd(port)
        return pd_dict['flags']

    def get_pd_dualrole(self, port):
        """Get the current PD dualrole setting

        @param port: Type C PD port 0/1
        @returns: current PD dualrole setting, one of (on, off, snk, src)
        """

        if self.per_port_dualrole_setting is True:
            cmd = 'pd %d dualrole' % port
        elif self.per_port_dualrole_setting is False:
            cmd = 'pd dualrole'
        else:
            try:
                self.per_port_dualrole_setting = True
                return self.get_pd_dualrole(port)
            except:
                self.per_port_dualrole_setting = False
                return self.get_pd_dualrole(port)

        dualrole_values = self.DUALROLE_VALUES

        m = self.send_pd_command_get_output(
                cmd, ['dual-role toggling:\s+([\w ]+)[\r\n]'], debug_on=False)
        # Find the index according to the output of "pd dualrole" command
        dual_index = self.DUALROLE_CMD_RESULTS.index(m[0][1])
        # Map to a string which is the output of this method
        return dualrole_values[dual_index]

    def set_pd_dualrole(self, port, value):
        """Set pd dualrole

        """
        raise NotImplementedError(
            'should be implemented in derived class')

    def query_pd_connection(self):
        """Determine if PD connection is present

        Try the 'pd 0/1 state' command and see if it's in either
        expected state of a connection. Record the port number
        that has an active connection

        @returns: dict with params port, connect, and state
        """
        status = {}
        port = 0;
        status['connect'] = False
        status['port'] = port
        state = self.get_pd_state(port)
        # Check port 0 first

        if self.is_pd_connected(port):
            status['connect'] = True
            status['role'] = state
        else:
            port = 1
            status['port'] = port
            state = self.get_pd_state(port)
            logging.info('CHECK PORT 1: %s', state)
            # Check port 1
            if self.is_pd_connected(port):
                status['connect'] = True
                status['role'] = state

        return status

    def swap_power_role(self, port):
        """Attempt a power role swap

        This method attempts to execute a power role swap. A check
        is made to ensure that dualrole mode is enabled and that
        a PD contract is currently established. If both checks pass,
        then the power role swap command is issued. After a delay,
        if a PD contract is established and the current state does
        not equal the starting state, then it was successful.

        @param port: pd port number

        @returns: True if power swap is successful, False otherwise.
        """
        # Get starting state
        if self.is_pd_dual_role_enabled(port) == False:
            logging.info('Dualrole Mode not enabled!')
            return False
        if self.is_pd_connected(port) == False:
            logging.info('PD contract not established!')
            return False
        current_pr = self.get_pd_state(port)
        swap_cmd = 'pd %d swap power' % port
        self.send_pd_command(swap_cmd)
        time.sleep(self.CONNECT_TIME)
        new_pr = self.get_pd_state(port)
        logging.info('Power swap: %s -> %s', current_pr, new_pr)
        if self.is_pd_connected(port) == False:
            return False
        return bool(current_pr != new_pr)

    def disable_pd_console_debug(self):
        """Turn off PD console debug

        """
        cmd = 'pd dump 0'
        self.send_pd_command(cmd)

    def enable_pd_console_debug(self):
        """Enable PD console debug level 1

        """
        cmd = 'pd dump 2'
        self.send_pd_command(cmd)

    def is_pd_flag_set(self, port, key):
        """Test a bit in PD protocol state flags

        The flag word contains various PD protocol state information.
        This method allows for a specific flag to be tested.

        @param port: Port which has the active PD connection
        @param key: dict key to retrieve the flag bit mapping

        @returns True if the bit to be tested is set
        """
        pd_flags = self.get_pd_flags(port)
        return bool(self.PD_STATE_FLAGS_DICT[key] & int(pd_flags, 16))

    def is_pd_connected(self, port):
        """Check if a PD connection is active

        @param port: port to be used for pd console commands

        @returns True if port is in connected state
        """
        return self.is_src_connected(port) or self.is_snk_connected(port)

    def is_pd_dual_role_enabled(self, port):
        """Check if a PD device is in dualrole mode

        @param port: Type C PD port 0/1

        @returns True is dualrole mode is active, false otherwise
        """
        drp = self.get_pd_dualrole(port)
        return drp == 'on'

    def is_src_connected(self, port, state=None):
        """Checks if the port is connected as a source

        @param port: Type C PD port 0/1
        @param state: the state to check (None to get current state)

        @returns True if connected as SRC, False otherwise
        """
        if state is None:
            state = self.get_pd_state(port)
        return state in self.get_src_connect_states()

    def is_snk_connected(self, port, state=None):
        """Checks if the port is connected as a sink

        @param port: Type C PD port 0/1
        @param state: the state to check (None to get current state)

        @returns True if connected as SNK, False otherwise
        """
        if state is None:
            state = self.get_pd_state(port)
        return state in self.get_snk_connect_states()

    def is_disconnected(self, port, state=None):
        """Checks if the port is disconnected

        @param port: Type C PD port 0/1
        @param state: the state to check (None to get current state)

        @return True if disconnected
        """
        if state is None:
            state = self.get_pd_state(port)
        return state in self.get_disconnected_states()

    def get_src_connect_states(self):
        """Returns the name of the SRC state

        """
        raise NotImplementedError(
            'should be implemented in derived class')

    def get_snk_connect_states(self):
        """Returns the name of the SNK state

        """
        raise NotImplementedError(
            'should be implemented in derived class')

    def get_disconnected_states(self):
        """Returns the names of the disconnected states

        """
        return self.DISCONNECTED_STATES

    def is_snk_discovery_state(self, port):
        """Returns true if in snk discovery state, else false

        @param port: Type C PD port 0/1

        @return: True if in SNK Discovery state
        """
        raise NotImplementedError(
            'should be implemented in derived class')

class TCPMv1ConsoleUtils(PDConsoleUtils):
    """ Provides a set of methods common to USB PD TCPMv1 FAFT tests

    Each instance of this class is associated with a particular
    servo UART console. USB PD tests will typically use the console
    command 'pd' and its subcommands to control/monitor Type C PD
    connections. The servo object used for UART operations is
    passed in and stored when this object is created.

    """
    SRC_CONNECT = ('SRC_READY',)
    SNK_CONNECT = ('SNK_READY',)
    SRC_DISC = 'SRC_DISCONNECTED'
    SNK_DISC = 'SNK_DISCONNECTED'
    SNK_DISCOVERY = 'SNK_DISCOVERY'
    DRP_AUTO_TOGGLE = 'DRP_AUTO_TOGGLE'
    DISCONNECTED_STATES = (SRC_DISC, SNK_DISC, DRP_AUTO_TOGGLE)

    PD_MAX_PORTS = 2
    CONNECT_TIME = 4

    CURRENT_STATE_PROBE_DELAY = 2
    DUALROLE_QUERY_DELAY = 1
    # Dualrole input/output values of methods in this class.
    DUALROLE_VALUES = ['on', 'off', 'snk', 'src']
    # Strings passing to the console command "pd dualrole"
    DUALROLE_CMD_ARGS = ['on', 'off', 'sink', 'source']
    # Strings returned from the console command "pd dualrole"
    DUALROLE_CMD_RESULTS = ['on', 'off', 'force sink', 'force source']

    # Some old firmware uses a single dualrole setting for all ports; while
    # some new firmware uses a per port dualrole setting. This flag will be
    # initialized to True or False.
    # TODO: Remove this flag when the old setting phases out
    per_port_dualrole_setting = None

    # Dictionary for 'pd 0/1 state' parsing
    PD_STATE_DICT = {
        'port': 'Port\s+([\w]+)',
        'role': 'Role:\s+([\w]+-[\w]+)',
        'pd_state': 'State:\s+([\w()]+)',
        'flags': 'Flags:\s+([\w]+)',
        'polarity': '(CC\d)'
    }

    # Regex to match PD state name; work for both old and new formats
    RE_PD_STATE = r"(\d+)?\(?([\w_]+)?\)?"
    # Copied from ec repo: common/usb_pd_protocol.c
    PD_STATE_NAMES = [
        "DISABLED",                   # index: 0
        "SUSPENDED",
        "SNK_DISCONNECTED",
        "SNK_DISCONNECTED_DEBOUNCE",
        "SNK_HARD_RESET_RECOVER",
        "SNK_DISCOVERY",              # index: 5
        "SNK_REQUESTED",
        "SNK_TRANSITION",
        "SNK_READY",
        "SNK_SWAP_INIT",
        "SNK_SWAP_SNK_DISABLE",       # index: 10
        "SNK_SWAP_SRC_DISABLE",
        "SNK_SWAP_STANDBY",
        "SNK_SWAP_COMPLETE",
        "SRC_DISCONNECTED",
        "SRC_DISCONNECTED_DEBOUNCE",  # index: 15
        "SRC_HARD_RESET_RECOVER",
        "SRC_STARTUP",
        "SRC_DISCOVERY",
        "SRC_NEGOCIATE",
        "SRC_ACCEPTED",               # index: 20
        "SRC_POWERED",
        "SRC_TRANSITION",
        "SRC_READY",
        "SRC_GET_SNK_CAP",
        "DR_SWAP",                    # index: 25
        "SRC_SWAP_INIT",
        "SRC_SWAP_SNK_DISABLE",
        "SRC_SWAP_SRC_DISABLE",
        "SRC_SWAP_STANDBY",
        "VCONN_SWAP_SEND",            # index: 30
        "VCONN_SWAP_INIT",
        "VCONN_SWAP_READY",
        "SOFT_RESET",
        "HARD_RESET_SEND",
        "HARD_RESET_EXECUTE",         # index: 35
        "BIST_RX",
        "BIST_TX",
        "DRP_AUTO_TOGGLE",
    ]

    # Dictionary for PD control message types
    PD_CONTROL_MSG_MASK = 0x1f
    PD_CONTROL_MSG_DICT = {
        'GoodCRC': 1,
        'GotoMin': 2,
        'Accept': 3,
        'Reject': 4,
        'Ping': 5,
        'PS_RDY': 6,
        'Get_Source_Cap': 7,
        'Get_Sink_Cap': 8,
        'DR_Swap': 9,
        'PR_Swap': 10,
        'VCONN_Swap': 11,
        'Wait': 12,
        'Soft_Reset': 13
    }

    # Dictionary for PD firmware state flags
    PD_STATE_FLAGS_DICT = {
        'power_swap': 1 << 1,
        'data_swap': 1 << 2,
        'data_swap_active': 1 << 3,
        'vconn_on': 1 << 12
    }

    def _normalize_pd_state(self, state):
        """Normalize the PD state name which handles both old and new formats.

        The old format is like: "SNK_READY"
        The new format is like: "8()" if debug_level == 0, or
                                "8(SNK_READY)" if debug_level > 0

        This method will convert the new format to the old one.

        @param state: The raw PD state text

        @returns: The normalized PD state name
        @raises: TestFail if unexpected PD state format
        """
        m = re.match(self.RE_PD_STATE, state)
        if m and any(m.groups()):
            state_index, state_name = m.groups()
            if state_index is None:
                # The old format: return the name
                return state_name
            # The new format: map the index to a name
            mapped_name = self.PD_STATE_NAMES[int(state_index)]
            if state_name is not None:
                assert mapped_name == state_name
            return mapped_name
        else:
            raise error.TestFail('Unexpected PD state format: %s' % state)

    def get_pd_state(self, port):
        """Get the current PD state

        @param port: Type C PD port 0/1
        @returns: current pd state
        """

        pd_dict = self.execute_pd_state_cmd(port)
        return self._normalize_pd_state(pd_dict['pd_state'])

    def set_pd_dualrole(self, port, value):
        """Set pd dualrole

        It can be set to either:
        1. on
        2. off
        3. snk (force sink mode)
        4. src (force source mode)
        After setting, the current value is read to confirm that it
        was set properly.

        @param port: Type C PD port 0/1
        @param value: One of the 4 options listed
        """
        dualrole_values = self.DUALROLE_VALUES
        # If the dualrole setting is not initialized, call the get method to
        # initialize it.
        if self.per_port_dualrole_setting is None:
            self.get_pd_dualrole(port)

        # Get string required for console command
        dual_index = dualrole_values.index(value)
        # Create console command
        if self.per_port_dualrole_setting is True:
            cmd = 'pd %d dualrole %s' % (port, self.DUALROLE_CMD_ARGS[dual_index])
        elif self.per_port_dualrole_setting is False:
            cmd = 'pd dualrole %s' % (self.DUALROLE_CMD_ARGS[dual_index])
        else:
            raise error.TestFail("dualrole error")

        self.console.send_command(cmd)
        time.sleep(self.DUALROLE_QUERY_DELAY)
        # Get current setting to verify that command was successful
        dual = self.get_pd_dualrole(port)
        # If it doesn't match, then raise error
        if dual != value:
            raise error.TestFail("dualrole error: " + value + " != " + dual)

    def get_src_connect_states(self):
        """Returns the name of the SRC state

        """
        return self.SRC_CONNECT

    def get_snk_connect_states(self):
        """Returns the name of the SRC state

        """
        return self.SNK_CONNECT

    def is_snk_discovery_state(self, port):
        """Returns true if in snk discovery state, else false

        @param port: Type C PD port 0/1

        @return: True if in SNK Discovery state
        """
        state = self.get_pd_state(port)
        return state == self.SNK_DISCOVERY

class TCPMv2ConsoleUtils(PDConsoleUtils):
    """ Provides a set of methods common to USB PD TCPMv1 FAFT tests

    Each instance of this class is associated with a particular
    servo UART console. USB PD tests will typically use the console
    command 'pd' and its subcommands to control/monitor Type C PD
    connections. The servo object used for UART operations is
    passed in and stored when this object is created.

    """
    SRC_CONNECT = ('Attached.SRC', 'UnorientedDebugAccessory.SRC')
    SNK_CONNECT = ('Attached.SNK', 'DebugAccessory.SNK')
    SRC_DISC = 'Unattached.SRC'
    SNK_DISC = 'Unattached.SNK'
    DRP_AUTO_TOGGLE = 'DRPAutoToggle'
    LOW_POWER_MODE = 'LowPowerMode'
    SNK_DISCOVERY = 'PE_SNK_DISCOVERY'
    DISCONNECTED_STATES = (SRC_DISC, SNK_DISC, DRP_AUTO_TOGGLE, LOW_POWER_MODE)

    PD_MAX_PORTS = 2
    CONNECT_TIME = 4

    CURRENT_STATE_PROBE_DELAY = 2
    DUALROLE_QUERY_DELAY = 1
    # Dualrole input/output values of methods in this class.
    DUALROLE_VALUES = ['on', 'off', 'sink', 'source']
    # Strings passing to the console command "pd dualrole"
    DUALROLE_CMD_ARGS = ['on', 'off', 'sink', 'source']
    # Strings returned from the console command "pd dualrole"
    DUALROLE_CMD_RESULTS = ['on', 'off', 'force sink', 'force source']

    # Some old firmware uses a single dualrole setting for all ports; while
    # some new firmware uses a per port dualrole setting. This flag will be
    # initialized to True or False.
    # TODO: Remove this flag when the old setting phases out
    per_port_dualrole_setting = None

    # Dictionary for 'pd 0/1 state' parsing
    PD_STATE_DICT = {
        'port': 'Port\s+([\w]+)',
        'role': 'Role:\s+([\w]+-[\w]+)',
        'pd_state': 'TC State:\s+([\w().]+)',
        'flags': 'Flags:\s+([\w]+)',
        'pe_state': 'PE State:\s+(\w*)',
        'polarity': '(CC\d)'
    }

    # Regex to match PD state name; work for both old and new formats
    RE_PD_STATE = r"(\d+)?\(?([\w_]+)?\)?"

    # Dictionary for PD control message types
    PD_CONTROL_MSG_MASK = 0x1f
    PD_CONTROL_MSG_DICT = {
        'GoodCRC': 1,
        'GotoMin': 2,
        'Accept': 3,
        'Reject': 4,
        'Ping': 5,
        'PS_RDY': 6,
        'Get_Source_Cap': 7,
        'Get_Sink_Cap': 8,
        'DR_Swap': 9,
        'PR_Swap': 10,
        'VCONN_Swap': 11,
        'Wait': 12,
        'Soft_Reset': 13
    }

    # Dictionary for PD firmware state flags
    PD_STATE_FLAGS_DICT = {
        'power_swap': 1 << 1,
        'data_swap': 1 << 2,
        'data_swap_active': 1 << 3,
        'vconn_on': 1 << 12
    }

    def get_pe_state(self, port):
        """Get the current Policy Engine state

        @param port: Type C PD port 0/1
        @returns: current pe state
        """

        pd_dict = self.execute_pd_state_cmd(port)
        return pd_dict['pe_state']

    def get_pd_state(self, port):
        """Get the current PD state

        @param port: Type C PD port 0/1
        @returns: current pd state
        """

        pd_dict = self.execute_pd_state_cmd(port)
        return pd_dict['pd_state']

    def set_pd_dualrole(self, port, value):
        """Set pd dualrole

        It can be set to either:
        1. on
        2. off
        3. snk (force sink mode)
        4. src (force source mode)
        After setting, the current value is read to confirm that it
        was set properly.

        @param port: Type C PD port 0/1
        @param value: One of the 4 options listed
        """
        dualrole_values = self.DUALROLE_VALUES

        if value == 'src':
            value = 'source'
        elif value == 'snk':
            value = 'sink'

        # Get string required for console command
        dual_index = dualrole_values.index(value)
        # Create console command
        cmd = 'pd %d dualrole %s' % (port, self.DUALROLE_CMD_ARGS[dual_index])
        self.console.send_command(cmd)
        time.sleep(self.DUALROLE_QUERY_DELAY)
        # Get current setting to verify that command was successful
        dual = self.get_pd_dualrole(port)
        # If it doesn't match, then raise error
        if dual != value:
            raise error.TestFail("dualrole error: " + value + " != " + dual)

    def get_src_connect_states(self):
        """Returns the name of the SRC states

        @returns: List of connected source state names
        """
        return self.SRC_CONNECT

    def get_snk_connect_states(self):
        """Returns the name of the SRC states

        @returns: List of connected sink state names
        """
        return self.SNK_CONNECT

    def is_snk_discovery_state(self, port):
        """Returns true if in snk discovery state, else false

        @param port: Type C PD port 0/1

        @return: True if in SNK Discovery state
        """
        state = self.get_pe_state(port)
        return state == self.SNK_DISCOVERY

class PDConnectionUtils(PDConsoleUtils):
    """Provides a set of methods common to USB PD FAFT tests

    This class is used for PD utility methods that require access
    to both PDTester and DUT PD consoles.

    """

    def __init__(self, dut_console, pdtester_console):
        """
        @param dut_console: PD console object for DUT
        @param pdtester_console: PD console object for PDTester
        """
        # save console for DUT PD UART access functions
        self.dut_console = dut_console
        # save console for PDTester UART access functions
        self.pdtester_console = pdtester_console

    def _verify_pdtester_connection(self, port):
        """Verify DUT to PDTester PD connection

        This method checks for a PDTester PD connection for the
        given port by first verifying if a PD connection is present.
        If found, then it uses a PDTester 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 PDTester connection is verified.

        @param port: DUT pd port to test

        @returns True if DUT to PDTester pd connection is verified
        """
        DISCONNECT_CHECK_TIME = 2
        DISCONNECT_TIME_SEC = 10
        # pdtester console command to force PD disconnect
        disc_cmd = 'fakedisconnect 100 %d' % (DISCONNECT_TIME_SEC * 1000)
        # Only check for PDTester if DUT has active PD connection
        if self.dut_console.is_pd_connected(port):
            # Attempt to force PD disconnection
            self.pdtester_console.send_pd_command(disc_cmd)
            time.sleep(DISCONNECT_CHECK_TIME)
            # Verify that DUT PD port is no longer connected
            if self.dut_console.is_pd_connected(port) == False:
                # Wait for disconnect timer and give time to reconnect
                time.sleep(self.dut_console.CONNECT_TIME + DISCONNECT_TIME_SEC)
                if self.dut_console.is_pd_connected(port):
                    logging.info('PDTester connection verified on port %d',
                                 port)
                    return True
            else:
                # Could have disconnected other port, allow it to reconnect
                # before exiting.
                time.sleep(self.dut_console.CONNECT_TIME + DISCONNECT_TIME_SEC)
        return False

    def find_dut_to_pdtester_connection(self):
        """Find the PD port which is connected to PDTester

        @returns DUT pd port number if found, None otherwise
        """
        for port in range(self.dut_console.PD_MAX_PORTS):
            # Check for DUT to PDTester connection on port
            if self._verify_pdtester_connection(port):
                # PDTester PD connection found so exit
                return port
        return None

def create_pd_console_utils(console):
    """Factory that detects the proper PDConsole Utils to use for DUT

    @param console: DUT PD console

    @returns: An instance of TCPMv1ConsoleUtils or TCPMv2ConsoleUtils
    """
    pd_console_utils = {
        1: TCPMv1ConsoleUtils,
        2: TCPMv2ConsoleUtils,
    }

    version = PDConsoleUtils(console).get_pd_version()
    logging.debug('%s is TCPM v%s', console, version)
    cls = pd_console_utils[version]
    return cls(console)