blob: 403780d6920891e90a27c3df9155d3a0fb32c98c [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2016 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.
"""A script to keep track of devices across builds and report state."""
import argparse
import json
import logging
import os
import re
import sys
if __name__ == '__main__':
sys.path.append(
os.path.abspath(
os.path.join(os.path.dirname(__file__), '..', '..', '..')))
from devil.android import battery_utils
from devil.android import device_denylist
from devil.android import device_errors
from devil.android import device_list
from devil.android import device_utils
from devil.android.sdk import adb_wrapper
from devil.android.tools import script_common
from devil.constants import exit_codes
from devil.utils import logging_common
from devil.utils import lsusb
logger = logging.getLogger(__name__)
_RE_DEVICE_ID = re.compile(r'Device ID = (\d+)')
def IsDenylisted(serial, denylist):
return denylist and serial in denylist.Read()
def _BatteryStatus(device, denylist):
battery_info = {}
try:
battery = battery_utils.BatteryUtils(device)
battery_info = battery.GetBatteryInfo(timeout=5)
battery_level = int(battery_info.get('level', 100))
if battery_level < 15:
logger.error('Critically low battery level (%d)', battery_level)
battery = battery_utils.BatteryUtils(device)
if not battery.GetCharging():
battery.SetCharging(True)
if denylist:
denylist.Extend([device.adb.GetDeviceSerial()], reason='low_battery')
except (device_errors.CommandFailedError,
device_errors.DeviceUnreachableError):
logger.exception('Failed to get battery information for %s', str(device))
return battery_info
def DeviceStatus(devices, denylist):
"""Generates status information for the given devices.
Args:
devices: The devices to generate status for.
denylist: The current device denylist.
Returns:
A dict of the following form:
{
'<serial>': {
'serial': '<serial>',
'adb_status': str,
'usb_status': bool,
'denylisted': bool,
# only if the device is connected and not denylisted
'type': ro.build.product,
'build': ro.build.id,
'build_detail': ro.build.fingerprint,
'battery': {
...
},
'imei_slice': str,
'wifi_ip': str,
},
...
}
"""
adb_devices = {
a[0].GetDeviceSerial(): a
for a in adb_wrapper.AdbWrapper.Devices(
desired_state=None, long_list=True)
}
usb_devices = set(lsusb.get_android_devices())
def denylisting_device_status(device):
serial = device.adb.GetDeviceSerial()
adb_status = (adb_devices[serial][1]
if serial in adb_devices else 'missing')
usb_status = bool(serial in usb_devices)
device_status = {
'serial': serial,
'adb_status': adb_status,
'usb_status': usb_status,
}
if not IsDenylisted(serial, denylist):
if adb_status == 'device':
try:
build_product = device.build_product
build_id = device.build_id
build_fingerprint = device.build_fingerprint
build_description = device.build_description
wifi_ip = device.GetProp('dhcp.wlan0.ipaddress')
battery_info = _BatteryStatus(device, denylist)
try:
imei_slice = device.GetIMEI()
except device_errors.CommandFailedError:
logging.exception('Unable to fetch IMEI for %s.', str(device))
imei_slice = 'unknown'
if (device.product_name == 'mantaray'
and battery_info.get('AC powered', None) != 'true'):
logger.error('Mantaray device not connected to AC power.')
device_status.update({
'ro.build.product': build_product,
'ro.build.id': build_id,
'ro.build.fingerprint': build_fingerprint,
'ro.build.description': build_description,
'battery': battery_info,
'imei_slice': imei_slice,
'wifi_ip': wifi_ip,
})
except (device_errors.CommandFailedError,
device_errors.DeviceUnreachableError):
logger.exception('Failure while getting device status for %s.',
str(device))
if denylist:
denylist.Extend([serial], reason='status_check_failure')
except device_errors.CommandTimeoutError:
logger.exception('Timeout while getting device status for %s.',
str(device))
if denylist:
denylist.Extend([serial], reason='status_check_timeout')
elif denylist:
denylist.Extend([serial],
reason=adb_status if usb_status else 'offline')
device_status['denylisted'] = IsDenylisted(serial, denylist)
return device_status
parallel_devices = device_utils.DeviceUtils.parallel(devices)
statuses = parallel_devices.pMap(denylisting_device_status).pGet(None)
return statuses
def _LogStatuses(statuses):
# Log the state of all devices.
for status in statuses:
logger.info(status['serial'])
adb_status = status.get('adb_status')
denylisted = status.get('denylisted')
logger.info(' USB status: %s',
'online' if status.get('usb_status') else 'offline')
logger.info(' ADB status: %s', adb_status)
logger.info(' Denylisted: %s', str(denylisted))
if adb_status == 'device' and not denylisted:
logger.info(' Device type: %s', status.get('ro.build.product'))
logger.info(' OS build: %s', status.get('ro.build.id'))
logger.info(' OS build fingerprint: %s',
status.get('ro.build.fingerprint'))
logger.info(' Battery state:')
for k, v in status.get('battery', {}).items():
logger.info(' %s: %s', k, v)
logger.info(' IMEI slice: %s', status.get('imei_slice'))
logger.info(' WiFi IP: %s', status.get('wifi_ip'))
def _WriteBuildbotFile(file_path, statuses):
buildbot_path, _ = os.path.split(file_path)
if os.path.exists(buildbot_path):
with open(file_path, 'w') as f:
for status in statuses:
try:
if status['adb_status'] == 'device':
f.write(
'{serial} {adb_status} {build_product} {build_id} '
'{temperature:.1f}C {level}%\n'.format(
serial=status['serial'],
adb_status=status['adb_status'],
build_product=status['type'],
build_id=status['build'],
temperature=float(status['battery']['temperature']) / 10,
level=status['battery']['level']))
elif status.get('usb_status', False):
f.write('{serial} {adb_status}\n'.format(
serial=status['serial'], adb_status=status['adb_status']))
else:
f.write('{serial} offline\n'.format(serial=status['serial']))
except Exception: # pylint: disable=broad-except
pass
def GetExpectedDevices(known_devices_files):
expected_devices = set()
try:
for path in known_devices_files:
if os.path.exists(path):
expected_devices.update(device_list.GetPersistentDeviceList(path))
else:
logger.warning('Could not find known devices file: %s', path)
except IOError:
logger.warning('Problem reading %s, skipping.', path)
logger.info('Expected devices:')
for device in expected_devices:
logger.info(' %s', device)
return expected_devices
def AddArguments(parser):
parser.add_argument(
'--json-output', help='Output JSON information into a specified file.')
parser.add_argument('--denylist-file', help='Device denylist JSON file.')
parser.add_argument(
'--known-devices-file',
action='append',
default=[],
dest='known_devices_files',
help='Path to known device lists.')
parser.add_argument(
'--buildbot-path',
'-b',
default='/home/chrome-bot/.adb_device_info',
help='Absolute path to buildbot file location')
parser.add_argument(
'-w',
'--overwrite-known-devices-files',
action='store_true',
help='If set, overwrites known devices files wiht new '
'values.')
def main():
parser = argparse.ArgumentParser()
logging_common.AddLoggingArguments(parser)
script_common.AddEnvironmentArguments(parser)
AddArguments(parser)
args = parser.parse_args()
logging_common.InitializeLogging(args)
script_common.InitializeEnvironment(args)
denylist = (device_denylist.Denylist(args.denylist_file)
if args.denylist_file else None)
expected_devices = GetExpectedDevices(args.known_devices_files)
usb_devices = set(lsusb.get_android_devices())
devices = [
device_utils.DeviceUtils(s) for s in expected_devices.union(usb_devices)
]
statuses = DeviceStatus(devices, denylist)
# Log the state of all devices.
_LogStatuses(statuses)
# Update the last devices file(s).
if args.overwrite_known_devices_files:
for path in args.known_devices_files:
device_list.WritePersistentDeviceList(
path, [status['serial'] for status in statuses])
# Write device info to file for buildbot info display.
_WriteBuildbotFile(args.buildbot_path, statuses)
# Dump the device statuses to JSON.
if args.json_output:
with open(args.json_output, 'wb') as f:
f.write(
json.dumps(
statuses, indent=4, sort_keys=True, separators=(',', ': ')))
live_devices = [
status['serial'] for status in statuses
if (status['adb_status'] == 'device'
and not IsDenylisted(status['serial'], denylist))
]
# If all devices failed, or if there are no devices, it's an infra error.
if not live_devices:
logger.error('No available devices.')
return 0 if live_devices else exit_codes.INFRA
if __name__ == '__main__':
sys.exit(main())