cros/power/power_telemetry_logger.py

# Lint as: python2, python3
# 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 power measurement with telemetry devices."""

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

import collections
import csv
import datetime
from distutils import sysconfig
import json
import logging
import numpy
import os
import re
import shutil
import six
import string
import subprocess
import threading
import time

import powerlog

from servo import measure_power

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.power import power_status
from autotest_lib.client.cros.power import power_telemetry_utils as utils
from autotest_lib.server.cros.power import power_dashboard


# If a sample has over 10% NaN values, the data might be very unreliable if
# interpolation is applied.
ACCEPTABLE_NAN_RATIO = 0.1

# If a sample has more than these NaN values in sequence, the data is also not
# reliable.
MAX_CONSECUTIVE_NAN_READINGS = 5

# If for over a second no values can be read, the data is also not reliable.
MAX_NAN_GAP_S = 1

# Dictionary to make passing the default arguments for loggers to the NaN
# interpolation utility easy.
INTERPOLATION_ARGS = {'max_nan_ratio': ACCEPTABLE_NAN_RATIO,
                      'max_sample_gap': MAX_CONSECUTIVE_NAN_READINGS,
                      'max_sample_time_gap': MAX_NAN_GAP_S}

def ts_processing(ts_str):
    """Parse autotest log timestamp into local time seconds since epoch.

    @param ts_str: a timestamp string from client.DEBUG file in local time zone.
    @return seconds since epoch, inserting the current year because ts_str does
            not include year. This introduces error if client side test is
            running across the turn of the year.
    """
    ts = datetime.datetime.strptime(ts_str, '%m/%d %H:%M:%S.%f ')
    # TODO(mqg): fix the wrong year at turn of the year.
    ts = ts.replace(year=datetime.datetime.today().year)
    return time.mktime(ts.timetuple()) + ts.microsecond / 1e6


class PowerTelemetryLogger(object):
    """A helper class for power autotests requiring telemetry devices.

    Telemetry: external pieces of hardware which help with measuring power
    data on the Chrome device. This is not to be confused with library
    telemetry.core, which is a required library / dependency for autotests
    involving Chrome and / or ARC. Examples of power telemetry devices include
    Servo and Sweetberry.

    This logger class detects telemetry devices connected to the DUT. It will
    then start and stop the measurement, trim the excessive power telemetry
    device data and report the data back to the workstation and the dashboard
    """

    DASHBOARD_UPLOAD_URL = 'http://chrome-power.appspot.com'
    DEFAULT_START = r'starting test\(run_once\(\)\), test details follow'
    DEFAULT_END = r'The test has completed successfully'

    def __init__(self, config, resultsdir, host):
        """Init PowerTelemetryLogger.

        @param config: the args argument from test_that in a dict. Settings for
                       power telemetry devices.
                       required data: {'test': 'test_TestName.tag'}
        @param resultsdir: path to directory where current autotest results are
                           stored, e.g. /tmp/test_that_results/
                           results-1-test_TestName.tag/test_TestName.tag/
                           results/
        @param host: CrosHost object representing the DUT.
        """
        logging.debug('%s initialize.', self.__class__.__name__)
        self._resultsdir = resultsdir
        self._host = host
        self._tagged_testname = config['test']
        self._pdash_note = config.get('pdash_note', '')

    def start_measurement(self):
        """Start power telemetry devices."""
        self._start_measurement()
        logging.info('%s starts.', self.__class__.__name__)
        self._start_ts = time.time()

    def _start_measurement(self):
        """Start power telemetry devices."""
        raise NotImplementedError('Subclasses must implement '
                '_start_measurement.')

    def end_measurement(self, client_test_dir):
        """End power telemetry devices.

        End power measurement with telemetry devices, get the power telemetry
        device data, trim the data recorded outside of actual testing, and
        upload statistics to dashboard.

        @param client_test_dir: directory of the client side test.
        """
        self._end_measurement()
        logging.info('%s finishes.', self.__class__.__name__)
        checkpoint_logger = self._get_client_test_checkpoint_logger(
                client_test_dir)
        start_ts, end_ts = self._get_client_test_ts(client_test_dir)
        loggers = self._load_and_trim_data(start_ts, end_ts)
        # Call export after trimming to only export trimmed data.
        self._export_data_locally(client_test_dir,
                                  checkpoint_logger.checkpoint_data)
        self._upload_data(loggers, checkpoint_logger)

    def _end_measurement(self):
        """End power telemetry devices."""
        raise NotImplementedError('Subclasses must implement _end_measurement.')

    def _export_data_locally(self, client_test_dir, checkpoint_data=None):
        """Slot for the logger to export measurements locally."""
        raise NotImplementedError('Subclasses must implement '
                                  '_export_data_locally.')

    def _get_client_test_ts(self, client_test_dir):
        """Determine the start and end timestamp for the telemetry data.

        Power telemetry devices will run through the entire autotest, including
        the overhead time, but we only need the measurements of actual testing,
        so parse logs from client side test to determine the start and end
        timestamp for the telemetry data.

        @param client_test_dir: directory of the client side test.
        @return (start_ts, end_ts)
                start_ts: the start timestamp of the client side test in seconds
                          since epoch or None.
                end_ts: the end timestamp of the client side test in seconds
                        since epoch or None.
        """
        if not os.path.isdir(client_test_dir):
            logging.error('Cannot find client side test dir %s, no need to '
                          'trim power telemetry measurements.', client_test_dir)
            return (None, None)

        # Use timestamp in client side test power_log.json as start & end
        # timestamp.
        power_log_path = os.path.join(client_test_dir, 'results',
                                      'power_log.json')
        start_ts, end_ts = self._get_power_log_ts(power_log_path)
        if start_ts and end_ts:
            self._start_ts = start_ts
            return (start_ts, end_ts)

        # Parse timestamp in client side test debug log and use as start & end
        # timestamp.
        client_test_name = os.path.basename(client_test_dir)
        debug_file_path = os.path.join(client_test_dir, 'debug',
                                       '%s.DEBUG' % client_test_name)
        start_ts, end_ts = self._get_debug_log_ts(debug_file_path)
        if start_ts:
            self._start_ts = start_ts
        return (start_ts, end_ts)

    def _get_debug_log_ts(self, debug_file_path):
        """Parse client side test start and end timestamp from debug log.

        @param debug_file_path: path to client side test debug log.
        @return (start_ts, end_ts)
                start_ts: the start timestamp of the client side test in seconds
                          since epoch or None.
                end_ts: the end timestamp of the client side test in seconds
                        since epoch or None.
        """
        default_test_events = collections.defaultdict(dict)
        custom_test_events = collections.defaultdict(dict)
        default_test_events['start']['str'] = self.DEFAULT_START
        default_test_events['end']['str'] = self.DEFAULT_END
        custom_test_events['start']['str'] = utils.CUSTOM_START
        custom_test_events['end']['str'] = utils.CUSTOM_END
        for event in default_test_events:
            default_test_events[event]['re'] = re.compile(r'([\d\s\./:]+).+' +
                    default_test_events[event]['str'])
            default_test_events[event]['match'] = False
        for event in custom_test_events:
            custom_test_events[event]['re'] = re.compile(r'.*' +
                    custom_test_events[event]['str'] + r'\s+([\d\.]+)')
        events_ts = {
            'start': None,
            'end': None,
        }

        try:
            with open(debug_file_path, 'r') as debug_log:

                for line in debug_log:
                    for event in default_test_events:
                        match = default_test_events[event]['re'].match(line)
                        if match:
                            default_test_events[event]['ts'] = \
                                    ts_processing(match.group(1))
                            default_test_events[event]['match'] = True
                    for event in custom_test_events:
                        match = custom_test_events[event]['re'].match(line)
                        if match:
                            custom_test_events[event]['ts'] = \
                                    float(match.group(1))

            for event in default_test_events:
                if not default_test_events[event]['match']:
                    raise error.TestWarn('Cannot find %s timestamp in client '
                                         'side test debug log.')

            for event in events_ts:
                events_ts[event] = default_test_events[event].get(
                        'ts', events_ts[event])
                events_ts[event] = custom_test_events[event].get(
                        'ts', events_ts[event])

            return (events_ts['start'], events_ts['end'])

        except Exception as exc:
            logging.warning('Client side test debug log %s does not contain '
                            'valid start and end timestamp, see exception: %s',
                            debug_file_path, exc)
            return (None, None)

    def _get_power_log_ts(self, power_log_path):
        """Parse client side test start and end timestamp from power_log.json.

        @param power_log_path: path to client side test power_log.json.
        @return (start_ts, end_ts)
                start_ts: the start timestamp of the client side test in seconds
                          since epoch or None.
                end_ts: the end timestamp of the client side test in seconds
                        since epoch or None.
        """
        try:
            with open(power_log_path, 'r') as power_log:
                power_log_str = power_log.read()
                json_decoder = json.JSONDecoder()
                power_log_obj = []

                idx = 0
                start_ts = list()
                end_ts = list()
                while idx < len(power_log_str):
                    log_entry, idx = json_decoder.raw_decode(power_log_str, idx)
                    start_ts.append(log_entry['timestamp'])
                    end_ts.append(log_entry['timestamp'] +
                                  log_entry['power']['sample_duration'] *
                                  log_entry['power']['sample_count'])

                return (min(start_ts), max(end_ts))
        except Exception as exc:
            logging.warning('Client side test power_log %s does not contain '
                            'valid start and end timestamp, see exception: %s',
                            power_log_path, exc)
            return (None, None)

    def _load_and_trim_data(self, start_ts, end_ts):
        """Load data and trim data.

        Load and format data recorded by power telemetry devices. Trim data if
        necessary.

        @param start_ts: start timestamp in seconds since epoch, None if no
                         need to trim data.
        @param end_ts: end timestamp in seconds since epoch, None if no need to
                       trim data.
        @return a list of loggers, where each logger contains raw power data and
                statistics.

        logger format:
        {
            'sample_count' : 60,
            'sample_duration' : 60,
            'data' : {
                'domain_1' : [ 111.11, 123.45 , ... , 99.99 ],
                ...
                'domain_n' : [ 3999.99, 4242.42, ... , 4567.89 ]
            },
            'average' : {
                'domain_1' : 100.00,
                ...
                'domain_n' : 4300.00
            },
            'unit' : {
                'domain_1' : 'milliwatt',
                ...
                'domain_n' : 'milliwatt'
            },
            'type' : {
                'domain_1' : 'servod',
                ...
                'domain_n' : 'servod'
            },
        }
        """
        raise NotImplementedError('Subclasses must implement '
                '_load_and_trim_data and return a list of loggers.')

    def _get_client_test_checkpoint_logger(self, client_test_dir):
        """Load the client-side test checkpoints.

        The key data we need is the checkpoint_logger.checkpoint_data object.
        This is a dictionary that contains for each key a list of [start, end]
        timestamps (seconds since epoch) for a checkpoint.
        Note: should there be issues loading the data, the checkpoint logger
        will still be returned, but it will be empty. Code that relies on the
        returned object here and wants to make sure its valid, needs to check
        against the |checkpoint_logger.checkpoint_data| being empty, as it
        will never be None

        Returns: CheckpointLogger object with client endpoints, or empty data
        """
        client_test_resultsdir = os.path.join(client_test_dir, 'results')
        checkpoint_logger = power_status.get_checkpoint_logger_from_file(
                resultsdir=client_test_resultsdir)
        return checkpoint_logger

    def _upload_data(self, loggers, checkpoint_logger):
        """Upload the data to dashboard.

        @param loggers: a list of loggers, where each logger contains raw power
                        data and statistics.
        """

        for logger in loggers:
            pdash = power_dashboard.PowerTelemetryLoggerDashboard(
                    logger=logger, testname=self._tagged_testname,
                    host=self._host, start_ts=self._start_ts,
                    checkpoint_logger=checkpoint_logger,
                    resultsdir=self._resultsdir,
                    uploadurl=self.DASHBOARD_UPLOAD_URL, note=self._pdash_note)
            pdash.upload()


class PacTelemetryLogger(PowerTelemetryLogger):
    """This logger class measures power via pacman debugger."""

    def __init__(self, config, resultsdir, host):
        """Init PacTelemetryLogger.

        @param config: the args argument from test_that in a dict. Settings for
                       power telemetry devices.
                       required data:
                       {'test': 'test_TestName.tag',
                        'config': PAC address and sense resistor .py file location,
                        'mapping: DUT power rail mapping csv file,
                        'gpio': gpio}
        @param resultsdir: path to directory where current autotest results are
                           stored, e.g. /tmp/test_that_results/
                           results-1-test_TestName.tag/test_TestName.tag/
                           results/
        @param host: CrosHost object representing the DUT.

        @raises error.TestError if problem running pacman.py
        """
        super(PacTelemetryLogger, self).__init__(config, resultsdir, host)
        required_args = ['config', 'mapping', 'gpio']
        for arg in required_args:
            if arg not in config:
                msg = 'Missing required arguments for PacTelemetryLogger: %s' % arg
                raise error.TestError(msg)
        self._pac_config_file = config['config']
        self._pac_mapping_file = config['mapping']
        self._pac_gpio_file = config['gpio']
        self._resultsdir = resultsdir
        self.pac_path = self._get_pacman_install_path()
        self.pac_data_path = os.path.join(resultsdir, 'pac')

        os.makedirs(self.pac_data_path, exist_ok=True)

        # Check if pacman is able to run
        try:
            subprocess.check_output('pacman.py', timeout=5, cwd=self.pac_path)
        except subprocess.CalledProcessError as e:
            msg = 'Error running pacman.py '\
                  'Check dependencies have been installed'
            logging.error(msg)
            logging.error(e.output)
            raise error.TestError(e)

    def _start_measurement(self):
        """Start a pacman thread with the given config, mapping, and gpio files."""

        self._log = open(os.path.join(self.pac_data_path, "pac.log"), "a")

        self._pacman_args = [
                '--config', self._pac_config_file, '--mapping',
                self._pac_mapping_file, '--gpio', self._pac_gpio_file,
                '--output', self.pac_data_path
        ]

        logging.debug('Starting pacman process')
        cmds = ['pacman.py'] + self._pacman_args
        logging.debug(cmds)

        self._pacman_process = subprocess.Popen(cmds,
                                                cwd=self.pac_path,
                                                stdout=self._log,
                                                stderr=self._log)

    def _end_measurement(self):
        """Stop pacman thread. This will dump and process the accumulators."""
        self._pacman_process.send_signal(2)
        self._pacman_process.wait(timeout=10)
        self._load_and_trim_data(None, None)
        self._export_data_locally(self._resultsdir)

        self._log.close()

    def _get_pacman_install_path(self):
        """Return the absolute path of pacman on the host.

        @raises error.TestError if pacman is not in PATH
        """
        pac_path = shutil.which('pacman.py')
        if pac_path == None:
            msg = 'Unable to locate pacman.py \n'\
                  'Check pacman.py is in PATH'
            logging.error(msg)
            raise error.TestNAError(msg)
        return os.path.dirname(pac_path)

    def _load_and_trim_data(self, start_ts, end_ts):
        """Load data and trim data.

        Load and format data recorded by power telemetry devices. Trim data if
        necessary.

        @param start_ts: start timestamp in seconds since epoch, None if no
                         need to trim data.
        @param end_ts: end timestamp in seconds since epoch, None if no need to
                       trim data.
        @return a list of loggers, where each logger contains raw power data and
                statistics.

        @raises TestError when unable to locate or open pacman accumulator results

        logger format:
        {
            'sample_count' : 60,
            'sample_duration' : 60,
            'data' : {
                'domain_1' : [ 111.11, 123.45 , ... , 99.99 ],
                ...
                'domain_n' : [ 3999.99, 4242.42, ... , 4567.89 ]
            },
            'average' : {
                'domain_1' : 100.00,
                ...
                'domain_n' : 4300.00
            },
            'unit' : {
                'domain_1' : 'milliwatt',
                ...
                'domain_n' : 'milliwatt'
            },
            'type' : {
                'domain_1' : 'servod',
                ...
                'domain_n' : 'servod'
            },
        }
        """
        loggers = list()
        accumulator_path = os.path.join(self.pac_data_path, 'accumulatorData.csv')
        if not os.path.exists(accumulator_path):
            raise error.TestError('Unable to locate pacman results!')
        # Load resulting pacman csv file
        try:
            with open(accumulator_path, 'r') as csvfile:
                reader = csv.reader(csvfile, delimiter=',')
                # Capture the first line
                schema = next(reader)
                # First column is an index
                schema[0] = 'index'
                # Place data into a dictionary
                self._accumulator_data = list()
                for row in reader:
                    measurement = dict(zip(schema, row))
                    self._accumulator_data.append(measurement)
        except OSError:
            raise error.TestError('Unable to open pacman accumulator results!')

        # Match required logger format
        log = {
                'sample_count': 1,
                'sample_duration': float(self._accumulator_data[0]['tAccum']),
                'data': {
                        x['Rail']: [float(x['Average Power (w)'])]
                        for x in self._accumulator_data
                },
                'average': {
                        x['Rail']: float(x['Average Power (w)'])
                        for x in self._accumulator_data
                },
                'unit': {x['Rail']: 'watts'
                         for x in self._accumulator_data},
                'type': {x['Rail']: 'pacman'
                         for x in self._accumulator_data},
        }
        loggers.append(log)
        return loggers

    def output_pacman_aggregates(self, test):
        """This outputs all the processed aggregate values to the results-chart.json

        @param test: the test.test object to use when outputting the
                    performance values to results-chart.json
        """
        for rail in self._accumulator_data:
            test.output_perf_value(rail['Rail'],
                                   float(rail['Average Power (w)']),
                                   units='watts',
                                   replace_existing_values=True)

    def _export_data_locally(self, client_test_dir, checkpoint_data=None):
        """Slot for the logger to export measurements locally."""
        self._local_pac_data_path = os.path.join(client_test_dir,
                                                 'pacman_data')
        shutil.copytree(self.pac_data_path, self._local_pac_data_path)

    def _upload_data(self, loggers, checkpoint_logger):
        """
        _upload_data is defined as a pass as a hot-fix to external partners' lack
        of access to the power_dashboard URL
        """
        pass


class ServodTelemetryLogger(PowerTelemetryLogger):
    """This logger class measures power by querying a servod instance."""

    DEFAULT_INA_RATE = 20.0
    DEFAULT_VBAT_RATE = 60.0

    def __init__(self, config, resultsdir, host):
        """Init ServodTelemetryLogger.

        @param config: the args argument from test_that in a dict. Settings for
                       power telemetry devices.
                       required data:
                       {'test': 'test_TestName.tag',
                        'servo_host': host of servod instance,
                        'servo_port: port that the servod instance is on}
        @param resultsdir: path to directory where current autotest results are
                           stored, e.g. /tmp/test_that_results/
                           results-1-test_TestName.tag/test_TestName.tag/
                           results/
        @param host: CrosHost object representing the DUT.
        """
        super(ServodTelemetryLogger, self).__init__(config, resultsdir, host)

        self._servo_host = host.servo._servo_host.hostname
        self._servo_port = host.servo._servo_host.servo_port
        self._ina_rate = float(config.get('ina_rate', self.DEFAULT_INA_RATE))
        self._vbat_rate = float(config.get('vbat_rate', self.DEFAULT_VBAT_RATE))
        self._pm = measure_power.PowerMeasurement(host=self._servo_host,
                                                  port=self._servo_port,
                                                  adc_rate=self._ina_rate,
                                                  vbat_rate=self._vbat_rate)

    def _start_measurement(self):
        """Start power measurement by querying servod."""
        setup_done = self._pm.MeasurePower()
        # Block the main thread until setup is done and measurement has started.
        setup_done.wait()

    def _end_measurement(self):
        """End querying servod."""
        self._pm.FinishMeasurement()

    def _export_data_locally(self, client_test_dir, checkpoint_data=None):
        """Output formatted text summaries to test results directory.

        @param client_test_dir: path to the client test output
        @param checkpoint_data: dict, checkpoint data. data is list of tuples
                                of (start,end) format for the timesteps
        """
        # At this point the PowerMeasurement unit has been processed. Dump its
        # formatted summaries into the results directory.
        power_summaries_dir = os.path.join(self._resultsdir, 'power_summaries')
        self._pm.SaveSummary(outdir=power_summaries_dir)
        # After the main summaries are exported, we also want to export one
        # for each checkpoint. As each checkpoint might contain multiple
        # entries, the checkpoint name is expanded by a digit.
        def export_checkpoint(name, start, end):
            """Helper to avoid code duplication for 0th and next cases."""
            self._pm.SaveTrimmedSummary(tag=name,
                                        tstart=start,
                                        tend=end,
                                        outdir=power_summaries_dir)

        if checkpoint_data:
            for checkpoint_name, checkpoint_list in checkpoint_data.items():
                # Export the first entry without any sort of name change.
                tstart, tend = checkpoint_list[0]
                export_checkpoint(checkpoint_name, tstart, tend)
                for suffix, checkpoint_element in enumerate(
                        checkpoint_list[1:], start=1):
                    # Export subsequent entries with a suffix
                    tstart, tend = checkpoint_element
                    export_checkpoint('%s%d' % (checkpoint_name, suffix),
                                      tstart, tend)

    def _load_and_trim_data(self, start_ts, end_ts):
        """Load data and trim data.

        Load and format data recorded by servod. Trim data if necessary.
        """
        self._pm.ProcessMeasurement(start_ts, end_ts)

        summary = self._pm.GetSummary()
        raw_data = self._pm.GetRawData()

        loggers = list()

        # Domains in summary/raw_data that carry no power-data.
        metadata_domains = ['Sample_msecs', 'time', 'timeline']

        for source in summary:
            tl = raw_data[source]['timeline']
            samples = len(tl)
            data = {
                k[:-3] if k.endswith('_mw') else k: v
                for k, v in six.iteritems(raw_data[source])
                if k not in metadata_domains
            }

            # Add the timeline of this measurement to the interpolation
            # arguments. This is to detect and reject large measurement gaps.
            # See above for details or in power_telemetry_utils.
            INTERPOLATION_ARGS['timeline'] = tl

            try:
                # Smoothen out data to remove any NaN values by interpolating
                # the missing values. If too many values are NaN, or too many
                # values are NaN consecutively, fail the test.
                # Here r stands for rail and d stands for data.
                data = {r: utils.interpolate_missing_data(d,
                                                          **INTERPOLATION_ARGS)
                        for r, d in six.iteritems(data)}
            except utils.TelemetryUtilsError as e:
                raise error.TestFail('Issue at source %s: %s' % (source,
                                                                 str(e)))

            ave = {
                k[:-3] if k.endswith('_mw') else k: v['mean']
                for k, v in six.iteritems(summary[source])
                if k not in metadata_domains
            }
            if samples > 1:
                # Having more than one sample allows the code to properly set a
                # sample duration.
                sample_duration = (tl[-1] - tl[0]) / (samples - 1)
            else:
                # In thise case, it seems that there is only one sample as the
                # difference between start and end is 0. Use the entire duration
                # of the test as the sample start/end
                sample_duration = end_ts - start_ts

            logger = {
                # All data domains should have same sample count.
                'sample_count': summary[source]['time']['count'],
                'sample_duration': sample_duration,
                'data': data,
                'average': ave,
                # TODO(mqg): hard code the units for now because we are only
                # dealing with power so far. When we start to work with voltage
                # or current, read the units from the .json files.
                'unit': {k: 'milliwatt' for k in data},
                'type': {k: 'servod' for k in data},
            }

            loggers.append(logger)

        return loggers


class PowerlogTelemetryLogger(PowerTelemetryLogger):
    """This logger class measures power with Sweetberry via powerlog tool.
    """

    DEFAULT_SWEETBERRY_INTERVAL = 20.0
    SWEETBERRY_CONFIG_DIR = os.path.join(
            sysconfig.get_python_lib(standard_lib=False), 'servo', 'data')

    def __init__(self, config, resultsdir, host):
        """Init PowerlogTelemetryLogger.

        @param config: the args argument from test_that in a dict. Settings for
                       power telemetry devices.
                       required data: {'test': 'test_TestName.tag'}
        @param resultsdir: path to directory where current autotest results are
                           stored, e.g. /tmp/test_that_results/
                           results-1-test_TestName.tag/test_TestName.tag/
                           results/
        @param host: CrosHost object representing the DUT.
        """
        super(PowerlogTelemetryLogger, self).__init__(config, resultsdir, host)

        self._interval = float(config.get('sweetberry_interval',
                                          self.DEFAULT_SWEETBERRY_INTERVAL))
        self._logdir = os.path.join(resultsdir, 'sweetberry_log')
        self._end_flag = threading.Event()
        self._sweetberry_serial = config.get('sweetberry_serial', None)
        if 'sweetberry_config' in config:
            self._sweetberry_config = config['sweetberry_config']
        else:
            board = self._host.get_board().replace('board:', '')
            hardware_rev = self._host.get_hardware_revision()
            self._sweetberry_config = board + '_' + hardware_rev
        board_path, scenario_path = \
                self._get_sweetberry_config_path(self._sweetberry_config)
        self._sweetberry_thread = SweetberryThread(
                board=board_path,
                scenario=scenario_path,
                interval=self._interval,
                stats_json_dir=self._logdir,
                end_flag=self._end_flag,
                serial=self._sweetberry_serial)
        self._sweetberry_thread.setDaemon(True)

    def _start_measurement(self):
        """Start power measurement with Sweetberry via powerlog tool."""
        self._sweetberry_thread.start()

    def _export_data_locally(self, client_test_dir, checkpoint_data=None):
        """Output formatted text summaries locally."""
        #TODO(crbug.com/978665): implement this.
        pass

    def _end_measurement(self):
        """End querying Sweetberry."""
        self._end_flag.set()
        # Sweetberry thread should theoretically finish within 1 self._interval
        # but giving 2 here to be more lenient.
        self._sweetberry_thread.join(self._interval * 2)
        if self._sweetberry_thread.is_alive():
            logging.warning('%s %s thread did not finish. There might be extra '
                            'data at the end.', self.__class__.__name__,
                            self._sweetberry_thread.name)

    def _load_and_trim_data(self, start_ts, end_ts):
        """Load data and trim data.

        Load and format data recorded by powerlog tool. Trim data if necessary.
        """
        if not os.path.exists(self._logdir):
            logging.error('Cannot find %s, no Sweetberry measurements exist, '
                          'not uploading to dashboard.', self._logdir)
            return

        trimmed_log_dirs = list()
        # Adding a padding to both start and end timestamp because the timestamp
        # of each data point is taken at the end of its corresponding interval.
        start_ts = start_ts + self._interval / 2 if start_ts else 0
        end_ts = end_ts + self._interval / 2 if end_ts else time.time()
        for dirname in os.listdir(self._logdir):
            if dirname.startswith('sweetberry'):
                sweetberry_ts = float(string.lstrip(dirname, 'sweetberry'))
                if start_ts <= sweetberry_ts <= end_ts:
                    trimmed_log_dirs.append(dirname)

        data = collections.defaultdict(list)
        for sweetberry_file in sorted(trimmed_log_dirs):
            fname = os.path.join(self._logdir, sweetberry_file, 'summary.json')
            with open(fname, 'r') as f:
                d = json.load(f)
                for k, v in six.iteritems(d):
                    data[k].append(v['mean'])

        logger = {
            # All data domains should have same sample count.
            'sample_count': len(next(six.itervalues(data))),
            'sample_duration': self._interval,
            'data': data,
            'average': {k: numpy.average(v) for k, v in six.iteritems(data)},
            # TODO(mqg): hard code the units for now because we are only dealing
            # with power so far. When we start to work with voltage or current,
            # read the units from the .json files.
            'unit': {k: 'milliwatt' for k in data},
            'type': {k: 'sweetberry' for k in data},
        }

        return [logger]

    def _get_sweetberry_config_path(self, filename):
        """Get the absolute path for Sweetberry board and scenario file.

        @param filename: string of Sweetberry config filename.
        @return a tuple of the path to Sweetberry board file and the path to
                Sweetberry scenario file.
        @raises error.TestError if board file or scenario file does not exist in
                file system.
        """
        board_path = os.path.join(self.SWEETBERRY_CONFIG_DIR,
                                  '%s.board' % filename)
        if not os.path.isfile(board_path):
            msg = 'Sweetberry board file %s does not exist.' % board_path
            raise error.TestError(msg)

        scenario_path = os.path.join(self.SWEETBERRY_CONFIG_DIR,
                                     '%s.scenario' % filename)
        if not os.path.isfile(scenario_path):
            msg = 'Sweetberry scenario file %s does not exist.' % scenario_path
            raise error.TestError(msg)
        return (board_path, scenario_path)


class SweetberryThread(threading.Thread):
    """A thread that starts and ends Sweetberry measurement."""

    def __init__(self, board, scenario, interval, stats_json_dir, end_flag,
                 serial=None):
        """Initialize the Sweetberry thread.

        Once started, this thread will invoke Sweetberry powerlog tool every
        [interval] seconds, which will sample each rail in [scenario] file
        multiple times and write the average of those samples in json format to
        [stats_json_dir]. The resistor size of each power rail is specified in
        [board] file.

        See go/sweetberry and go/sweetberry-readme for more details.

        @param board: file name for Sweetberry board file.
        @param scenario: file name for Sweetberry scenario file.
        @param interval: time of each Sweetberry run cycle; print Sweetberry
                         data every <interval> seconds.
        @param stats_json_dir: directory to store Sweetberry stats in json.
        @param end_flag: event object, stop Sweetberry measurement when this is
                         set.
        @param serial: serial number of sweetberry.
        """
        threading.Thread.__init__(self, name='Sweetberry')
        self._end_flag = end_flag
        self._interval = interval
        self._argv = ['--board', board,
                      '--config', scenario,
                      '--save_stats_json', stats_json_dir,
                      '--no_print_raw_data',
                      '--mW']
        if serial:
            self._argv.extend(['--serial', serial])

    def run(self):
        """Start Sweetberry measurement until end_flag is set."""
        logging.debug('Sweetberry starts.')
        loop = 0
        start_timestamp = time.time()
        while not self._end_flag.is_set():
            # TODO(mqg): in the future use more of powerlog components
            # explicitly, make a long call and harvest data from Sweetberry,
            # instead of using it like a command line tool now.
            loop += 1
            next_loop_start_timestamp = start_timestamp + loop * self._interval
            current_timestamp = time.time()
            this_loop_duration = next_loop_start_timestamp - current_timestamp
            powerlog.main(self._argv + ['--seconds', str(this_loop_duration)])
        logging.debug('Sweetberry stops.')