flash_device.py (6740B)
1 #!/usr/bin/env vpython3 2 # Copyright 2022 The Chromium Authors 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 """Implements commands for flashing a Fuchsia device.""" 6 7 import argparse 8 import logging 9 import os 10 import subprocess 11 import sys 12 13 from typing import Optional, Tuple 14 15 import common 16 from boot_device import BootMode, StateTransitionError, boot_device 17 from common import get_system_info, find_image_in_sdk, \ 18 register_device_args 19 from compatible_utils import get_sdk_hash, running_unattended 20 from lockfile import lock 21 22 # Flash-file lock. Used to restrict number of flash operations per host. 23 # File lock should be marked as stale after 15 mins. 24 _FF_LOCK = os.path.join('/tmp', 'flash.lock') 25 _FF_LOCK_STALE_SECS = 60 * 15 26 _FF_LOCK_ACQ_TIMEOUT = _FF_LOCK_STALE_SECS 27 28 29 def _get_system_info(target: Optional[str], 30 serial_num: Optional[str]) -> Tuple[str, str]: 31 """Retrieves installed OS version from device. 32 33 Args: 34 target: Target to get system info of. 35 serial_num: Serial number of device to get system info of. 36 Returns: 37 Tuple of strings, containing (product, version number). 38 """ 39 40 if running_unattended(): 41 try: 42 boot_device(target, BootMode.REGULAR, serial_num) 43 except (subprocess.CalledProcessError, StateTransitionError): 44 logging.warning('Could not boot device. Assuming in fastboot') 45 return ('', '') 46 47 return get_system_info(target) 48 49 50 def _update_required( 51 os_check, 52 system_image_dir: Optional[str], 53 target: Optional[str], 54 serial_num: Optional[str] = None) -> Tuple[bool, Optional[str]]: 55 """Returns True if a system update is required and path to image dir.""" 56 57 if os_check == 'ignore': 58 return False, system_image_dir 59 if not system_image_dir: 60 raise ValueError('System image directory must be specified.') 61 if not os.path.exists(system_image_dir): 62 logging.warning( 63 'System image directory does not exist. Assuming it\'s ' 64 'a product-bundle name and dynamically searching for ' 65 'image directory') 66 path = find_image_in_sdk(system_image_dir) 67 if not path: 68 raise FileNotFoundError( 69 f'System image directory {system_image_dir} could not' 70 'be found') 71 system_image_dir = path 72 if (os_check == 'check' 73 and get_sdk_hash(system_image_dir) == _get_system_info( 74 target, serial_num)): 75 return False, system_image_dir 76 return True, system_image_dir 77 78 79 def _run_flash_command(system_image_dir: str, target_id: Optional[str]): 80 """Helper function for running `ffx target flash`.""" 81 logging.info('Flashing %s to %s', system_image_dir, target_id) 82 # Flash only with a file lock acquired. 83 # This prevents multiple fastboot binaries from flashing concurrently, 84 # which should increase the odds of flashing success. 85 with lock(_FF_LOCK, timeout=_FF_LOCK_ACQ_TIMEOUT): 86 # The ffx.fastboot.inline_target has negative impact when ffx 87 # discovering devices in fastboot, so it's inlined here to limit its 88 # scope. See the discussion in https://fxbug.dev/issues/317228141. 89 logging.info( 90 'Flash result %s', 91 common.run_ffx_command(cmd=('target', 'flash', '-b', 92 system_image_dir, 93 '--no-bootloader-reboot'), 94 target_id=target_id, 95 configs=['ffx.fastboot.inline_target=true'], 96 capture_output=True).stdout) 97 98 99 def update(system_image_dir: str, 100 os_check: str, 101 target: Optional[str], 102 serial_num: Optional[str] = None) -> None: 103 """Conditionally updates target given. 104 105 Args: 106 system_image_dir: string, path to image directory. 107 os_check: <check|ignore|update>, which decides how to update the device. 108 target: Node-name string indicating device that should be updated. 109 serial_num: String of serial number of device that should be updated. 110 """ 111 needs_update, actual_image_dir = _update_required(os_check, 112 system_image_dir, target, 113 serial_num) 114 logging.info('update_required %s, actual_image_dir %s', needs_update, 115 actual_image_dir) 116 if not needs_update: 117 return 118 if serial_num: 119 boot_device(target, BootMode.BOOTLOADER, serial_num) 120 _run_flash_command(system_image_dir, serial_num) 121 else: 122 _run_flash_command(system_image_dir, target) 123 124 125 def register_update_args(arg_parser: argparse.ArgumentParser, 126 default_os_check: Optional[str] = 'check') -> None: 127 """Register common arguments for device updating.""" 128 serve_args = arg_parser.add_argument_group('update', 129 'device updating arguments') 130 serve_args.add_argument('--system-image-dir', 131 help='Specify the directory that contains the ' 132 'Fuchsia image used to flash the device. Only ' 133 'needs to be specified if "os_check" is not ' 134 '"ignore".') 135 serve_args.add_argument('--serial-num', 136 default=os.environ.get('FUCHSIA_FASTBOOT_SERNUM'), 137 help='Serial number of the device. Should be ' 138 'specified for devices that do not have an image ' 139 'flashed.') 140 serve_args.add_argument('--os-check', 141 choices=['check', 'update', 'ignore'], 142 default=default_os_check, 143 help='Sets the OS version enforcement policy. If ' 144 '"check", then the deployment process will halt ' 145 'if the target\'s version does not match. If ' 146 '"update", then the target device will ' 147 'be reflashed. If "ignore", then the OS version ' 148 'will not be checked.') 149 150 151 def main(): 152 """Stand-alone function for flashing a device.""" 153 parser = argparse.ArgumentParser() 154 register_device_args(parser) 155 register_update_args(parser, default_os_check='update') 156 args = parser.parse_args() 157 update(args.system_image_dir, args.os_check, args.target_id, 158 args.serial_num) 159 160 161 if __name__ == '__main__': 162 sys.exit(main())