# !/usr/bin/python # Copyright 2017 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Fastboot Interface Implementation using subprocess library.""" import os import subprocess import sys import threading import fastboot_exceptions CREATE_NO_WINDOW = 0x08000000 def _GetCurrentPath(): if getattr(sys, 'frozen', False): # we are running in a bundle path = sys._MEIPASS # pylint: disable=protected-access else: # we are running in a normal Python environment path = os.path.dirname(os.path.abspath(__file__)) return path class FastbootDevice(object): """An abstracted fastboot device object. Attributes: serial_number: The serial number of the fastboot device. """ current_path = _GetCurrentPath() fastboot_command = os.path.join(current_path, 'fastboot.exe') HOST_OS = 'Windows' @staticmethod def ListDevices(): """List all fastboot devices. Returns: A list of serial numbers for all the fastboot devices. """ try: out = subprocess.check_output( [FastbootDevice.fastboot_command, 'devices'], creationflags=CREATE_NO_WINDOW) device_serial_numbers = (out.replace('\tfastboot', '') .rstrip().splitlines()) # filter out empty string return filter(None, device_serial_numbers) except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) def __init__(self, serial_number): """Initiate the fastboot device object. Args: serial_number: The serial number of the fastboot device. """ self.serial_number = serial_number # Lock to make sure only one fastboot command can be issued to one device # at one time. self._lock = threading.Lock() def Reboot(self): """Reboot the device into fastboot mode. Returns: The command output. """ try: self._lock.acquire() out = subprocess.check_output( [FastbootDevice.fastboot_command, '-s', self.serial_number, 'reboot-bootloader'], creationflags=CREATE_NO_WINDOW) return out except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() def Oem(self, oem_command, err_to_out): """"Run an OEM command. Args: oem_command: The OEM command to run. err_to_out: Whether to redirect stderr to stdout. Returns: The result message for the OEM command. Raises: FastbootFailure: If failure happens during the command. """ try: self._lock.acquire() # We need to redirect the output no matter err_to_out is set # So that FastbootFailure can catch the right error. return subprocess.check_output( [ FastbootDevice.fastboot_command, '-s', self.serial_number, 'oem', oem_command ], stderr=subprocess.STDOUT, creationflags=CREATE_NO_WINDOW) except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() def Flash(self, partition, file_path): """Flash a file to a partition. Args: file_path: The partition file to be flashed. partition: The partition to be flashed. Returns: The output for the fastboot command required. Raises: FastbootFailure: If failure happens during the command. """ try: self._lock.acquire() return subprocess.check_output( [ FastbootDevice.fastboot_command, '-s', self.serial_number, 'flash', partition, file_path ], creationflags=CREATE_NO_WINDOW) except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() def Upload(self, file_path): """Pulls a file from the fastboot device to the local file system. Args: file_path: The file path of the file system that the remote file would be pulled to. Returns: The output for the fastboot command required. Raises: FastbootFailure: If failure happens during the command. """ try: self._lock.acquire() return subprocess.check_output( [ FastbootDevice.fastboot_command, '-s', self.serial_number, 'get_staged', file_path ], creationflags=CREATE_NO_WINDOW) except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() def Download(self, file_path): """Push a file from the file system to the fastboot device. Args: file_path: The file path of the file on the local file system that would be pushed to fastboot device. Returns: The output for the fastboot command required. Raises: FastbootFailure: If failure happens during the command. """ try: self._lock.acquire() return subprocess.check_output( [ FastbootDevice.fastboot_command, '-s', self.serial_number, 'stage', file_path ], creationflags=CREATE_NO_WINDOW) except subprocess.CalledProcessError as e: raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() def GetVar(self, var): """Get a variable from the device. Note that the return value is in stderr instead of stdout. Args: var: The name of the variable. Returns: The value for the variable. Raises: FastbootFailure: If failure happens during the command. """ try: # Fastboot getvar command's output would be in stderr instead of stdout. # Need to redirect stderr to stdout. self._lock.acquire() # Need the shell=True flag for windows, otherwise it hangs. out = subprocess.check_output( [ FastbootDevice.fastboot_command, '-s', self.serial_number, 'getvar', var ], stderr=subprocess.STDOUT, shell=True, creationflags=CREATE_NO_WINDOW) except subprocess.CalledProcessError as e: # Since we redirected stderr, we should print stdout here. raise fastboot_exceptions.FastbootFailure(e.output) finally: self._lock.release() if var == 'at-vboot-state': # For the result of vboot-state, it does not follow the standard. return out lines = out.splitlines() for line in lines: if line.startswith(var + ': '): value = line.replace(var + ': ', '').replace('\r', '') return value @staticmethod def GetHostOs(): return FastbootDevice.HOST_OS def Disconnect(self): """Disconnect from the fastboot device.""" pass def __del__(self): self.Disconnect()