#!/usr/bin/env vpython3
# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Implements commands for flashing a Fuchsia device."""
import argparse
import logging
import os
import subprocess
import sys
from typing import Optional, Tuple
import common
from boot_device import BootMode, StateTransitionError, boot_device
from common import get_system_info, find_image_in_sdk, \
register_device_args
from compatible_utils import get_sdk_hash, running_unattended
from lockfile import lock
# Flash-file lock. Used to restrict number of flash operations per host.
# File lock should be marked as stale after 15 mins.
_FF_LOCK = os.path.join('/tmp', 'flash.lock')
_FF_LOCK_STALE_SECS = 60 * 15
_FF_LOCK_ACQ_TIMEOUT = _FF_LOCK_STALE_SECS
def _get_system_info(target: Optional[str],
serial_num: Optional[str]) -> Tuple[str, str]:
"""Retrieves installed OS version from device.
Args:
target: Target to get system info of.
serial_num: Serial number of device to get system info of.
Returns:
Tuple of strings, containing (product, version number).
"""
if running_unattended():
try:
boot_device(target, BootMode.REGULAR, serial_num)
except (subprocess.CalledProcessError, StateTransitionError):
logging.warning('Could not boot device. Assuming in fastboot')
return ('', '')
return get_system_info(target)
def _update_required(
os_check,
system_image_dir: Optional[str],
target: Optional[str],
serial_num: Optional[str] = None) -> Tuple[bool, Optional[str]]:
"""Returns True if a system update is required and path to image dir."""
if os_check == 'ignore':
return False, system_image_dir
if not system_image_dir:
raise ValueError('System image directory must be specified.')
if not os.path.exists(system_image_dir):
logging.warning(
'System image directory does not exist. Assuming it\'s '
'a product-bundle name and dynamically searching for '
'image directory')
path = find_image_in_sdk(system_image_dir)
if not path:
raise FileNotFoundError(
f'System image directory {system_image_dir} could not'
'be found')
system_image_dir = path
if (os_check == 'check'
and get_sdk_hash(system_image_dir) == _get_system_info(
target, serial_num)):
return False, system_image_dir
return True, system_image_dir
def _run_flash_command(system_image_dir: str, target_id: Optional[str]):
"""Helper function for running `ffx target flash`."""
logging.info('Flashing %s to %s', system_image_dir, target_id)
# Flash only with a file lock acquired.
# This prevents multiple fastboot binaries from flashing concurrently,
# which should increase the odds of flashing success.
with lock(_FF_LOCK, timeout=_FF_LOCK_ACQ_TIMEOUT):
# The ffx.fastboot.inline_target has negative impact when ffx
# discovering devices in fastboot, so it's inlined here to limit its
# scope. See the discussion in https://fxbug.dev/issues/317228141.
logging.info(
'Flash result %s',
common.run_ffx_command(cmd=('target', 'flash', '-b',
system_image_dir,
'--no-bootloader-reboot'),
target_id=target_id,
configs=['ffx.fastboot.inline_target=true'],
capture_output=True).stdout)
def update(system_image_dir: str,
os_check: str,
target: Optional[str],
serial_num: Optional[str] = None) -> None:
"""Conditionally updates target given.
Args:
system_image_dir: string, path to image directory.
os_check: <check|ignore|update>, which decides how to update the device.
target: Node-name string indicating device that should be updated.
serial_num: String of serial number of device that should be updated.
"""
needs_update, actual_image_dir = _update_required(os_check,
system_image_dir, target,
serial_num)
logging.info('update_required %s, actual_image_dir %s', needs_update,
actual_image_dir)
if not needs_update:
return
if serial_num:
boot_device(target, BootMode.BOOTLOADER, serial_num)
_run_flash_command(system_image_dir, serial_num)
else:
_run_flash_command(system_image_dir, target)
def register_update_args(arg_parser: argparse.ArgumentParser,
default_os_check: Optional[str] = 'check') -> None:
"""Register common arguments for device updating."""
serve_args = arg_parser.add_argument_group('update',
'device updating arguments')
serve_args.add_argument('--system-image-dir',
help='Specify the directory that contains the '
'Fuchsia image used to flash the device. Only '
'needs to be specified if "os_check" is not '
'"ignore".')
serve_args.add_argument('--serial-num',
default=os.environ.get('FUCHSIA_FASTBOOT_SERNUM'),
help='Serial number of the device. Should be '
'specified for devices that do not have an image '
'flashed.')
serve_args.add_argument('--os-check',
choices=['check', 'update', 'ignore'],
default=default_os_check,
help='Sets the OS version enforcement policy. If '
'"check", then the deployment process will halt '
'if the target\'s version does not match. If '
'"update", then the target device will '
'be reflashed. If "ignore", then the OS version '
'will not be checked.')
def main():
"""Stand-alone function for flashing a device."""
parser = argparse.ArgumentParser()
register_device_args(parser)
register_update_args(parser, default_os_check='update')
args = parser.parse_args()
update(args.system_image_dir, args.os_check, args.target_id,
args.serial_num)
if __name__ == '__main__':
sys.exit(main())
Codebase Browser
chromium