[Android] Move power implementation from device_utils to battery_utils

build/android/pylib/device/battery_utils.py

 # Copyright 2015 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.
 
 """Provides a variety of device interactions with power.
 """
 # pylint: disable=unused-argument
 
 import collections
 import contextlib
@@ skipping 129 matching lines @@
     returns:
       Dict of UID and power data.
       {
         'uid': uid,
         'data': [1,2,3]
       }
       None if the package is not found in the power data.
     """
     return self.GetPowerData().get(package)
 
-  # TODO(rnephew): Move implementation from device_utils when this is used.
   def GetBatteryInfo(self, timeout=None, retries=None):
     """Gets battery info for the device.
 
     Args:
       timeout: timeout in seconds
       retries: number of retries
     Returns:
       A dict containing various battery information as reported by dumpsys
       battery.
     """
-    return self._device.GetBatteryInfo(timeout=None, retries=None)
+    result = {}
+    # Skip the first line, which is just a header.
+    for line in self._device.RunShellCommand(
+        ['dumpsys', 'battery'], check_return=True)[1:]:
+      # If usb charging has been disabled, an extra line of header exists.
+      if 'UPDATES STOPPED' in line:
+        logging.warning('Dumpsys battery not receiving updates. '
+                        'Run dumpsys battery reset if this is in error.')
+      elif ':' not in line:
+        logging.warning('Unknown line found in dumpsys battery.')

[jbudorick, 2015/04/07 16:56:33]

nit: combine these two logging lines

  logging.warning('Unknown line found in dumpsys battery: "%s"', line)

[rnephew (Reviews Here), 2015/04/07 17:06:14]

Done.

+        logging.warning(line)
+      else:
+        k, v = line.split(': ', 1)

[jbudorick, 2015/04/07 16:56:33]

Why line.split(': ') when you're checking for ':' in line above?

[rnephew (Reviews Here), 2015/04/07 17:06:14]

Done.

+        result[k.strip()] = v.strip()
+    return result
 
-  # TODO(rnephew): Move implementation from device_utils when this is used.
   def GetCharging(self, timeout=None, retries=None):
     """Gets the charging state of the device.
 
     Args:
       timeout: timeout in seconds
       retries: number of retries
     Returns:
       True if the device is charging, false otherwise.
     """
-    return self._device.GetCharging(timeout=None, retries=None)
+    battery_info = self.GetBatteryInfo()
+    for k in ('AC powered', 'USB powered', 'Wireless powered'):
+      if (k in battery_info and
+          battery_info[k].lower() in ('true', '1', 'yes')):
+        return True
+    return False
 
-  # TODO(rnephew): Move implementation from device_utils when this is used.
   def SetCharging(self, enabled, timeout=None, retries=None):
     """Enables or disables charging on the device.
 
     Args:
       enabled: A boolean indicating whether charging should be enabled or
         disabled.
       timeout: timeout in seconds
       retries: number of retries
 
     Raises:
       device_errors.CommandFailedError: If method of disabling charging cannot
         be determined.
     """
-    self._device.SetCharging(enabled, timeout=None, retries=None)
+    if 'charging_config' not in self._cache:
+      for c in _CONTROL_CHARGING_COMMANDS:
+        if self._device.FileExists(c['witness_file']):
+          self._cache['charging_config'] = c
+          break
+      else:
+        raise device_errors.CommandFailedError(
+            'Unable to find charging commands.')
 
-  # TODO(rnephew): Move implementation from device_utils when this is used.
+    if enabled:
+      command = self._cache['charging_config']['enable_command']
+    else:
+      command = self._cache['charging_config']['disable_command']
+
+    def set_and_verify_charging():
+      self._device.RunShellCommand(command, check_return=True)
+      return self.GetCharging() == enabled
+
+    timeout_retry.WaitFor(set_and_verify_charging, wait_period=1)
+
   # TODO(rnephew): Make private when all use cases can use the context manager.
   def DisableBatteryUpdates(self, timeout=None, retries=None):
     """ Resets battery data and makes device appear like it is not
     charging so that it will collect power data since last charge.
 
     Args:
       timeout: timeout in seconds
       retries: number of retries
 
     Raises:
       device_errors.CommandFailedError: When resetting batterystats fails to
         reset power values.
     """
-    self._device.DisableBatteryUpdates(timeout=None, retries=None)
+    def battery_updates_disabled():
+      return self.GetCharging() is False
 
-  # TODO(rnephew): Move implementation from device_utils when this is used.
+    self._device.RunShellCommand(
+        ['dumpsys', 'battery', 'set', 'usb', '1'], check_return=True)
+    self._device.RunShellCommand(
+        ['dumpsys', 'batterystats', '--reset'], check_return=True)
+    battery_data = self._device.RunShellCommand(
+        ['dumpsys', 'batterystats', '--charged', '--checkin'],
+        check_return=True)
+    ROW_TYPE_INDEX = 3
+    PWI_POWER_INDEX = 5
+    for line in battery_data:
+      l = line.split(',')
+      if (len(l) > PWI_POWER_INDEX and l[ROW_TYPE_INDEX] == 'pwi'
+          and l[PWI_POWER_INDEX] != 0):
+        raise device_errors.CommandFailedError(
+            'Non-zero pmi value found after reset.')
+    self._device.RunShellCommand(['dumpsys', 'battery', 'set', 'usb', '0'],
+                                 check_return=True)
+    timeout_retry.WaitFor(battery_updates_disabled, wait_period=1)
+
   # TODO(rnephew): Make private when all use cases can use the context manager.
   def EnableBatteryUpdates(self, timeout=None, retries=None):
     """ Restarts device charging so that dumpsys no longer collects power data.
 
     Args:
       timeout: timeout in seconds
       retries: number of retries
     """
-    self._device.EnableBatteryUpdates(timeout=None, retries=None)
+    def battery_updates_enabled():
+      return self.GetCharging() is True
+
+    self._device.RunShellCommand(['dumpsys', 'battery', 'set', 'usb', '1'],
+                                 check_return=True)
+    self._device.RunShellCommand(['dumpsys', 'battery', 'reset'],
+                                 check_return=True)
+    timeout_retry.WaitFor(battery_updates_enabled, wait_period=1)
 
   @contextlib.contextmanager
   def BatteryMeasurement(self, timeout=None, retries=None):
     """Context manager that enables battery data collection. It makes
     the device appear to stop charging so that dumpsys will start collecting
     power data since last charge. Once the with block is exited, charging is
     resumed and power data since last charge is no longer collected.
 
     Only for devices L and higher.
 
@@ skipping 33 matching lines @@
       battery_level = self.GetBatteryInfo().get('level')
       if battery_level is None:
         logging.warning('Unable to find current battery level.')
         battery_level = 100
       else:
         logging.info('current battery level: %s', battery_level)
         battery_level = int(battery_level)
       return battery_level >= level
 
     timeout_retry.WaitFor(device_charged, wait_period=wait_period)