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

# Copyright (c) 2012 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 ast
import logging
import re
import time
from xml.parsers import expat

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import ec
from autotest_lib.server.cros.servo import servo

# Hostevent codes, copied from:
#     ec/include/ec_commands.h
HOSTEVENT_LID_CLOSED        = 0x00000001
HOSTEVENT_LID_OPEN          = 0x00000002
HOSTEVENT_POWER_BUTTON      = 0x00000004
HOSTEVENT_AC_CONNECTED      = 0x00000008
HOSTEVENT_AC_DISCONNECTED   = 0x00000010
HOSTEVENT_BATTERY_LOW       = 0x00000020
HOSTEVENT_BATTERY_CRITICAL  = 0x00000040
HOSTEVENT_BATTERY           = 0x00000080
HOSTEVENT_THERMAL_THRESHOLD = 0x00000100
HOSTEVENT_THERMAL_OVERLOAD  = 0x00000200
HOSTEVENT_THERMAL           = 0x00000400
HOSTEVENT_USB_CHARGER       = 0x00000800
HOSTEVENT_KEY_PRESSED       = 0x00001000
HOSTEVENT_INTERFACE_READY   = 0x00002000
# Keyboard recovery combo has been pressed
HOSTEVENT_KEYBOARD_RECOVERY = 0x00004000
# Shutdown due to thermal overload
HOSTEVENT_THERMAL_SHUTDOWN  = 0x00008000
# Shutdown due to battery level too low
HOSTEVENT_BATTERY_SHUTDOWN  = 0x00010000
HOSTEVENT_INVALID           = 0x80000000

# Time to wait after sending keypress commands.
KEYPRESS_RECOVERY_TIME = 0.5

# Wakemask types, copied from:
#     ec/include/ec_commands.h
EC_HOST_EVENT_MAIN = 0
EC_HOST_EVENT_B = 1
EC_HOST_EVENT_SCI_MASK = 2
EC_HOST_EVENT_SMI_MASK = 3
EC_HOST_EVENT_ALWAYS_REPORT_MASK = 4
EC_HOST_EVENT_ACTIVE_WAKE_MASK = 5
EC_HOST_EVENT_LAZY_WAKE_MASK_S0IX = 6
EC_HOST_EVENT_LAZY_WAKE_MASK_S3 = 7
EC_HOST_EVENT_LAZY_WAKE_MASK_S5 = 8


class ChromeConsole(object):
    """Manages control of a Chrome console.

    We control the Chrome console via the UART of a Servo board. Chrome console
    provides many interfaces to set and get its behavior via console commands.
    This class is to abstract these interfaces.
    """

    CMD = "_cmd"
    REGEXP = "_regexp"
    MULTICMD = "_multicmd"

    # EC Features
    # Quoted from 'enum ec_feature_code' in platform/ec/include/ec_commands.h.
    EC_FEATURE = {
        'EC_FEATURE_LIMITED'                            : 0,
        'EC_FEATURE_FLASH'                              : 1,
        'EC_FEATURE_PWM_FAN'                            : 2,
        'EC_FEATURE_PWM_KEYB'                           : 3,
        'EC_FEATURE_LIGHTBAR'                           : 4,
        'EC_FEATURE_LED'                                : 5,
        'EC_FEATURE_MOTION_SENSE'                       : 6,
        'EC_FEATURE_KEYB'                               : 7,
        'EC_FEATURE_PSTORE'                             : 8,
        'EC_FEATURE_PORT80'                             : 9,
        'EC_FEATURE_THERMAL'                            : 10,
        'EC_FEATURE_BKLIGHT_SWITCH'                     : 11,
        'EC_FEATURE_WIFI_SWITCH'                        : 12,
        'EC_FEATURE_HOST_EVENTS'                        : 13,
        'EC_FEATURE_GPIO'                               : 14,
        'EC_FEATURE_I2C'                                : 15,
        'EC_FEATURE_CHARGER'                            : 16,
        'EC_FEATURE_BATTERY'                            : 17,
        'EC_FEATURE_SMART_BATTERY'                      : 18,
        'EC_FEATURE_HANG_DETECT'                        : 19,
        'EC_FEATURE_PMU'                                : 20,
        'EC_FEATURE_SUB_MCU'                            : 21,
        'EC_FEATURE_USB_PD'                             : 22,
        'EC_FEATURE_USB_MUX'                            : 23,
        'EC_FEATURE_MOTION_SENSE_FIFO'                  : 24,
        'EC_FEATURE_VSTORE'                             : 25,
        'EC_FEATURE_USBC_SS_MUX_VIRTUAL'                : 26,
        'EC_FEATURE_RTC'                                : 27,
        'EC_FEATURE_FINGERPRINT'                        : 28,
        'EC_FEATURE_TOUCHPAD'                           : 29,
        'EC_FEATURE_RWSIG'                              : 30,
        'EC_FEATURE_DEVICE_EVENT'                       : 31,
        'EC_FEATURE_UNIFIED_WAKE_MASKS'                 : 32,
        'EC_FEATURE_HOST_EVENT64'                       : 33,
        'EC_FEATURE_EXEC_IN_RAM'                        : 34,
        'EC_FEATURE_CEC'                                : 35,
        'EC_FEATURE_MOTION_SENSE_TIGHT_TIMESTAMPS'      : 36,
        'EC_FEATURE_REFINED_TABLET_MODE_HYSTERESIS'     : 37,
        'EC_FEATURE_EFS2'                               : 38,
        'EC_FEATURE_SCP'                                : 39,
        'EC_FEATURE_ISH'                                : 40,
    }

    def __init__(self, servo, name):
        """Initialize and keep the servo object.

        Args:
          servo: A Servo object.
          name: The console name.
        """
        self.name = name
        self.uart_cmd = self.name + self.CMD
        self.uart_regexp = self.name + self.REGEXP
        self.uart_multicmd = self.name + self.MULTICMD

        self._servo = servo

    def __repr__(self):
        """Return a string representation: <ChromeConsole 'foo_uart'>"""
        return "<%s %r>" % (self.__class__.__name__, self.name)

    def set_uart_regexp(self, regexp):
        self._servo.set(self.uart_regexp, regexp)

    def clear_uart_regex(self):
        """Clear uart_regexp"""
        self.set_uart_regexp('None')

    def send_command(self, commands):
        """Send command through UART.

        This function opens UART pty when called, and then command is sent
        through UART.

        Args:
          commands: The commands to send, either a list or a string.
        """
        self.clear_uart_regex()
        if isinstance(commands, list):
            try:
                self._servo.set_nocheck(self.uart_multicmd, ';'.join(commands))
            except servo.ControlUnavailableError:
                logging.warning('The servod is too old that uart_multicmd not '
                                'supported. Use uart_cmd instead.')
                for command in commands:
                    self._servo.set_nocheck(self.uart_cmd, command)
        else:
            self._servo.set_nocheck(self.uart_cmd, commands)
        self.clear_uart_regex()

    def has_command(self, command):
        """Check whether EC console supports |command|.

        Args:
          command: Command to look for.

        Returns:
          True: If the |command| exist on the EC image of the device.
          False: If the |command| does not exist on the EC image of the device.
        """
        result = None
        try:
            # Throws error.TestFail (on timeout) if it cannot find a line with
            # 'command' in the output. Thus return False in that case.
            result = self.send_command_get_output('help', [command])
        except error.TestFail:
            return False
        return result is not None

    def send_command_get_output(self, command, regexp_list, retries=1):
        """Send command through UART and wait for response.

        This function waits for response message matching regular expressions.

        Args:
          command: The command sent.
          regexp_list: List of regular expressions used to match response
            message. Note, list must be ordered.

        Returns:
          List of tuples, each of which contains the entire matched string and
          all the subgroups of the match. None if not matched.
          For example:
            response of the given command:
              High temp: 37.2
              Low temp: 36.4
            regexp_list:
              ['High temp: (\d+)\.(\d+)', 'Low temp: (\d+)\.(\d+)']
            returns:
              [('High temp: 37.2', '37', '2'), ('Low temp: 36.4', '36', '4')]

        Raises:
          error.TestError: An error when the given regexp_list is not valid.
        """
        if not isinstance(regexp_list, list):
            raise error.TestError('Arugment regexp_list is not a list: %s' %
                                  str(regexp_list))

        while retries > 0:
            retries -= 1
            try:
                self.set_uart_regexp(str(regexp_list))
                self._servo.set_nocheck(self.uart_cmd, command)
                return ast.literal_eval(self._servo.get(self.uart_cmd))
            except (servo.UnresponsiveConsoleError,
                    servo.ResponsiveConsoleError, expat.ExpatError) as e:
                if retries <= 0:
                    raise
                logging.warning('Failed to send EC cmd. %s', e)
            finally:
                self.clear_uart_regex()


    def is_dfp(self, port=0):
        """This function checks if EC is DFP

        Args:
          port: Port of EC to check

        Returns:
          True: if EC is DFP
          False: if EC is not DFP
        """
        is_dfp = None
        ret = None
        try:
            ret = self.send_command_get_output("pd %d state" % port,
                                               ["DFP.*Flag"])
            is_dfp = True
        except Exception as e:
            is_dfp = False

        # For TCPMv1, after disconnecting a device the data state remains
        # the same, so even when pd state shows DPF, make sure the device is
        # not disconnected
        if is_dfp:
            if "DRP_AUTO_TOGGLE" in ret[0] or "DISCONNECTED" in ret[0]:
                is_dfp = False

        return is_dfp


class ChromeEC(ChromeConsole):
    """Manages control of a Chrome EC.

    We control the Chrome EC via the UART of a Servo board. Chrome EC
    provides many interfaces to set and get its behavior via console commands.
    This class is to abstract these interfaces.
    """

    # The dict to cache the battery information
    BATTERY_INFO = {}

    def __init__(self, servo, name="ec_uart"):
        super(ChromeEC, self).__init__(servo, name)

    def __repr__(self):
        """Return a string representation of the object: <ChromeEC 'ec_uart'>"""
        return "<%s %r>" % (self.__class__.__name__, self.name)

    def key_down(self, keyname):
        """Simulate pressing a key.

        Args:
          keyname: Key name, one of the keys of KEYMATRIX.
        """
        self.send_command('kbpress %d %d 1' %
                (ec.KEYMATRIX[keyname][1], ec.KEYMATRIX[keyname][0]))


    def key_up(self, keyname):
        """Simulate releasing a key.

        Args:
          keyname: Key name, one of the keys of KEYMATRIX.
        """
        self.send_command('kbpress %d %d 0' %
                (ec.KEYMATRIX[keyname][1], ec.KEYMATRIX[keyname][0]))


    def key_press(self, keyname):
        """Press and then release a key.

        Args:
          keyname: Key name, one of the keys of KEYMATRIX.
        """
        self.send_command([
                'kbpress %d %d 1' %
                    (ec.KEYMATRIX[keyname][1], ec.KEYMATRIX[keyname][0]),
                'kbpress %d %d 0' %
                    (ec.KEYMATRIX[keyname][1], ec.KEYMATRIX[keyname][0]),
                ])
        # Don't spam the EC console as fast as we can; leave some recovery time
        # in between commands.
        time.sleep(KEYPRESS_RECOVERY_TIME)


    def send_key_string_raw(self, string):
        """Send key strokes consisting of only characters.

        Args:
          string: Raw string.
        """
        for c in string:
            self.key_press(c)


    def send_key_string(self, string):
        """Send key strokes including special keys.

        Args:
          string: Character string including special keys. An example
            is "this is an<tab>example<enter>".
        """
        for m in re.finditer("(<[^>]+>)|([^<>]+)", string):
            sp, raw = m.groups()
            if raw is not None:
                self.send_key_string_raw(raw)
            else:
                self.key_press(sp)


    def reboot(self, flags=''):
        """Reboot EC with given flags.

        Args:
          flags: Optional, a space-separated string of flags passed to the
                 reboot command, including:
                   default: EC soft reboot;
                   'hard': EC hard/cold reboot;
                   'ap-off': Leave AP off after EC reboot (by default, EC turns
                             AP on after reboot if lid is open).

        Raises:
          error.TestError: If the string of flags is invalid.
        """
        for flag in flags.split():
            if flag not in ('hard', 'ap-off'):
                raise error.TestError(
                        'The flag %s of EC reboot command is invalid.' % flag)
        self.send_command("reboot %s" % flags)


    def set_flash_write_protect(self, enable):
        """Set the software write protect of EC flash.

        Args:
          enable: True to enable write protect, False to disable.
        """
        if enable:
            self.send_command("flashwp enable")
        else:
            self.send_command("flashwp disable")


    def set_hostevent(self, codes):
        """Set the EC hostevent codes.

        Args:
          codes: Hostevent codes, HOSTEVENT_*
        """
        self.send_command("hostevent set %#x" % codes)
        # Allow enough time for EC to process input and set flag.
        # See chromium:371631 for details.
        # FIXME: Stop importing time module if this hack becomes obsolete.
        time.sleep(1)


    def enable_console_channel(self, channel):
        """Find console channel mask and enable that channel only

        @param channel: console channel name
        """
        # The 'chan' command returns a list of console channels,
        # their channel masks and channel numbers
        regexp = r'(\d+)\s+([\w]+)\s+\*?\s+{0}'.format(channel)
        l = self.send_command_get_output('chan', [regexp])
        # Use channel mask and append the 0x for proper hex input value
        cmd = 'chan 0x' + l[0][2]
        # Set console to only output the desired channel
        self.send_command(cmd)


    def get_version(self):
        """Get version information from the Chrome EC console.
           Additionally, can be used to verify if EC console is available.
        """
        self.send_command("chan 0")
        # Use "[ \t]" here and not \s because sometimes the version is blank,
        # i.e. 'RO:   \r\n' which matches RO:\s+
        expected_output = [
                "Chip:[ \t]+([^\r\n]*)\r\n", "RO:[ \t]+([^\r\n]*)\r\n",
                "RW_?[AB]?:[ \t]+([^\r\n]*)\r\n", "Build:[ \t]+([^\r\n]*)\r\n"
        ]
        l = self.send_command_get_output("version", expected_output)
        self.send_command("chan 0xffffffff")
        return l

    def check_ro_rw(self, img_exp):
        """Tell if the current EC image matches the given input, 'RW' or 'RO.

        Args:
            img_exp: Expected image type. It should be either 'RW' or 'RO'.
        Return:
            True if the active EC image matches to 'img_exp'.
            False otherwise.
        Raise:
            TestError if img_exp is neither 'RW' nor 'RO'.
        """
        if img_exp not in ['RW', 'RO']:
            raise error.TestError('Arugment img_exp is neither RW nor RO')

        result = self.send_command_get_output('sysinfo', [r'Copy:\s*(RO|RW)'])
        return result[0][1] == img_exp

    def check_feature(self, feature):
        """Return true if EC supports the given feature

        Args:
            feature: feature name as a string as in self.EC_FEATURE.

        Returns:
            True if 'feature' is in EC's feature set.
            False otherwise
        """
        feat_id = self.EC_FEATURE[feature]
        if feat_id < 32:
            feat_start = 0
        else:
            feat_start = 32

        regexp = r'%d-%d:\s*(0x[0-9a-fA-F]{8})' % (feat_start,
                                                   feat_start + 31)

        try:
            result = self.send_command_get_output('feat', [regexp])
        except servo.ResponsiveConsoleError as e:
            logging.warning("feat command is not available: %s", str(e))
            return False

        feat_bitmap = int(result[0][1], 16)

        return ((1 << (feat_id - feat_start)) & feat_bitmap) != 0

    def update_battery_info(self):
        """Get the battery info we care for this test."""
        # The battery parameters we care for this test. The order must match
        # the output of EC battery command.
        battery_params = [
                'V', 'V-desired', 'I', 'I-desired', 'Charging', 'Remaining'
        ]
        regex_str_list = []

        for p in battery_params:
            if p == 'Remaining':
                regex_str_list.append(p + ':\s+(\d+)\s+')
            elif p == 'Charging':
                regex_str_list.append(p + r':\s+(Not Allowed|Allowed)\s+')
            else:
                regex_str_list.append(p +
                                      r':\s+0x[0-9a-f]*\s+=\s+([0-9-]+)\s+')

        # For unknown reasons, servod doesn't always capture the ec
        # command output. It doesn't happen often, but retry if it does.
        retries = 3
        while retries > 0:
            retries -= 1
            try:
                battery_regex_match = self.send_command_get_output(
                        'battery', regex_str_list)
                break
            except (servo.UnresponsiveConsoleError,
                    servo.ResponsiveConsoleError) as e:
                if retries <= 0:
                    raise
                logging.warning('Failed to get battery status. %s', e)
        else:
            battery_regex_match = self.send_command_get_output(
                    'battery', regex_str_list)

        for i in range(len(battery_params)):
            if battery_params[i] == 'Charging':
                self.BATTERY_INFO[
                        battery_params[i]] = battery_regex_match[i][1]
            else:
                self.BATTERY_INFO[battery_params[i]] = int(
                        battery_regex_match[i][1])
        logging.debug('Battery info: %s', self.BATTERY_INFO)

    def get_battery_desired_voltage(self, print_result=True):
        """Get battery desired voltage value."""
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info('Battery desired voltage = %d mV',
                         self.BATTERY_INFO['V-desired'])
        return self.BATTERY_INFO['V-desired']

    def get_battery_desired_current(self, print_result=True):
        """Get battery desired current value."""
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info('Battery desired current = %d mA',
                         self.BATTERY_INFO['I-desired'])
        return self.BATTERY_INFO['I-desired']

    def get_battery_actual_voltage(self, print_result=True):
        """Get the actual voltage from charger to battery."""
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info('Battery actual voltage = %d mV',
                         self.BATTERY_INFO['V'])
        return self.BATTERY_INFO['V']

    def get_battery_actual_current(self, print_result=True):
        """Get the actual current from charger to battery."""
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info('Battery actual current = %d mA',
                         self.BATTERY_INFO['I'])
        return self.BATTERY_INFO['I']

    def get_battery_remaining(self, print_result=True):
        """Get battery charge remaining in mAh."""
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info("Battery charge remaining = %d mAh",
                         self.BATTERY_INFO['Remaining'])
        return self.BATTERY_INFO['Remaining']

    def get_battery_charging_allowed(self, print_result=True):
        """Get the battery charging state.

        Returns True if charging is allowed.
        """
        if not self.BATTERY_INFO:
            self.update_battery_info()
        if print_result:
            logging.info("Battery charging = %s",
                         self.BATTERY_INFO['Charging'])
        if self.BATTERY_INFO['Charging'] == 'Allowed':
            return True
        return False


class ChromeUSBPD(ChromeEC):
    """Manages control of a Chrome USBPD.

    We control the Chrome EC via the UART of a Servo board. Chrome USBPD
    provides many interfaces to set and get its behavior via console commands.
    This class is to abstract these interfaces.
    """

    def __init__(self, servo):
        super(ChromeUSBPD, self).__init__(servo, "usbpd_uart")