catapult/telemetry/telemetry/internal/platform/power_monitor/android_power_monitor_controller.py

# Copyright 2014 The Chromium 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 atexit
import logging

from telemetry.internal.platform.power_monitor import android_power_monitor_base

def _ReenableChargingIfNeeded(battery):
  if not battery.GetCharging():
    battery.SetCharging(True)

class AndroidPowerMonitorController(
    android_power_monitor_base.AndroidPowerMonitorBase):
  """
  PowerMonitor that acts as facade for a list of PowerMonitor objects and uses
  the first available one.
  """
  def __init__(self, power_monitors, battery):
    super(AndroidPowerMonitorController, self).__init__()
    self._candidate_power_monitors = power_monitors
    self._active_monitors = []
    self._battery = battery
    atexit.register(_ReenableChargingIfNeeded, self._battery)

  def CanMonitorPower(self):
    return any(m.CanMonitorPower() for m in self._candidate_power_monitors)

  def StartMonitoringPower(self, browser):
    # TODO(rnephew): re-add assert when crbug.com/553601 is solved and
    # StopMonitoringPower is called in the correct place.
    if self._active_monitors:
      logging.warning('StopMonitoringPower() not called when expected. Last '
                      'results are likely not reported.')
      self.StopMonitoringPower()
    self._CheckStart()
    self._ChargingOff(self._battery)
    self._active_monitors = (
        [m for m in self._candidate_power_monitors if m.CanMonitorPower()])
    assert self._active_monitors, 'No available monitor.'
    for monitor in self._active_monitors:
      monitor.StartMonitoringPower(browser)

  @staticmethod
  def _MergePowerResults(combined_results, monitor_results):
    """
    Merges monitor_results into combined_results and leaves monitor_results
    values if there are merge conflicts.
    """
    def _CheckDuplicateKeys(dict_one, dict_two, ignore_list=None):
      for key in dict_one:
        if key in dict_two and key not in ignore_list:
          logging.warning('Found multiple instances of %s in power monitor '
                          'entries. Using newest one.', key)
    # Sub level power entries.
    for part in ['platform_info', 'component_utilization']:
      if part in monitor_results:
        _CheckDuplicateKeys(combined_results[part], monitor_results[part])
        combined_results[part].update(monitor_results[part])

    # Top level power entries.
    platform_info = combined_results['platform_info'].copy()
    comp_utilization = combined_results['component_utilization'].copy()
    _CheckDuplicateKeys(
        combined_results, monitor_results,
        ['identifier', 'platform_info', 'component_utilization'])
    combined_results.update(monitor_results)
    combined_results['platform_info'] = platform_info
    combined_results['component_utilization'] = comp_utilization

  def StopMonitoringPower(self):
    self._CheckStop()
    self._ChargingOn(self._battery)
    try:
      results = {'platform_info': {}, 'component_utilization': {}}
      for monitor in self._active_monitors:
        self._MergePowerResults(results, monitor.StopMonitoringPower())
      return results
    finally:
      self._active_monitors = []

  def _ChargingOff(self, battery):
    battery.SetCharging(False)

  def _ChargingOn(self, battery):
    if battery.GetCharging():
      logging.warning('Charging re-enabled during test.'
                      'Results may be inaccurate.')
    battery.SetCharging(True)