| # Copyright 2019 - 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. |
| """OtaTools class.""" |
| |
| import logging |
| import os |
| import tempfile |
| |
| from acloud import errors |
| from acloud.internal.lib import utils |
| |
| logger = logging.getLogger(__name__) |
| |
| _BIN_DIR_NAME = "bin" |
| _LPMAKE = "lpmake" |
| _BUILD_SUPER_IMAGE = "build_super_image" |
| _AVBTOOL = "avbtool" |
| _SGDISK = "sgdisk" |
| _SIMG2IMG = "simg2img" |
| _MK_COMBINED_IMG = "mk_combined_img" |
| _UNPACK_BOOTIMG = "unpack_bootimg" |
| |
| _BUILD_SUPER_IMAGE_TIMEOUT_SECS = 30 |
| _AVBTOOL_TIMEOUT_SECS = 30 |
| _MK_COMBINED_IMG_TIMEOUT_SECS = 180 |
| _UNPACK_BOOTIMG_TIMEOUT_SECS = 30 |
| |
| _MISSING_OTA_TOOLS_MSG = ("%(tool_name)s is not found. Run `make otatools` " |
| "in build environment, or set --local-tool to an " |
| "extracted otatools.zip.") |
| |
| |
| def FindOtaToolsDir(search_paths): |
| """Find OTA tools directory in the search paths. |
| |
| Args: |
| search_paths: List of paths, the directories to search for OTA tools. |
| |
| Returns: |
| The directory containing OTA tools. |
| |
| Raises: |
| errors.CheckPathError if OTA tools are not found. |
| """ |
| for search_path in search_paths: |
| if os.path.isfile(os.path.join(search_path, _BIN_DIR_NAME, |
| _BUILD_SUPER_IMAGE)): |
| return search_path |
| raise errors.CheckPathError(_MISSING_OTA_TOOLS_MSG % |
| {"tool_name": "OTA tool directory"}) |
| |
| |
| def FindOtaTools(search_paths): |
| """Find OTA tools in the search paths. |
| |
| Args: |
| search_paths: List of paths, the directories to search for OTA tools. |
| |
| Returns: |
| An OtaTools object. |
| |
| Raises: |
| errors.CheckPathError if OTA tools are not found. |
| """ |
| return OtaTools(FindOtaToolsDir(search_paths)) |
| |
| |
| def GetImageForPartition(partition_name, image_dir, **image_paths): |
| """Map a partition name to an image path. |
| |
| This function is used with BuildSuperImage or MkCombinedImg to mix |
| image_dir and image_paths into the output file. |
| |
| Args: |
| partition_name: String, e.g., "system", "product", and "vendor". |
| image_dir: String, the directory to search for the images that are not |
| given in image_paths. |
| image_paths: Pairs of partition names and image paths. |
| |
| Returns: |
| The image path if the partition is in image_paths. |
| Otherwise, this function returns the path under image_dir. |
| |
| Raises |
| errors.GetLocalImageError if the image does not exist. |
| """ |
| image_path = (image_paths.get(partition_name) or |
| os.path.join(image_dir, partition_name + ".img")) |
| if not os.path.isfile(image_path): |
| raise errors.GetLocalImageError( |
| "Cannot find image for partition %s" % partition_name) |
| return image_path |
| |
| |
| class OtaTools: |
| """The class that executes OTA tool commands.""" |
| |
| def __init__(self, ota_tools_dir): |
| self._ota_tools_dir = os.path.abspath(ota_tools_dir) |
| |
| def _GetBinary(self, name): |
| """Get an executable file from _ota_tools_dir. |
| |
| Args: |
| name: String, the file name. |
| |
| Returns: |
| String, the absolute path. |
| |
| Raises: |
| errors.NoExecuteCmd if the file does not exist. |
| """ |
| path = os.path.join(self._ota_tools_dir, _BIN_DIR_NAME, name) |
| if not os.path.isfile(path): |
| raise errors.NoExecuteCmd(_MISSING_OTA_TOOLS_MSG % |
| {"tool_name": name}) |
| utils.SetExecutable(path) |
| return path |
| |
| @staticmethod |
| def _RewriteMiscInfo(output_file, input_file, lpmake_path, get_image): |
| """Rewrite lpmake and image paths in misc_info.txt. |
| |
| Misc info consists of multiple lines of <key>=<value>. |
| Sample input_file: |
| lpmake=lpmake |
| dynamic_partition_list= system system_ext product vendor |
| |
| Sample output_file: |
| lpmake=/path/to/lpmake |
| dynamic_partition_list= system system_ext product vendor |
| system_image=/path/to/system.img |
| system_ext_image=/path/to/system_ext.img |
| product_image=/path/to/product.img |
| vendor_image=/path/to/vendor.img |
| |
| This method replaces lpmake with the specified path, and sets |
| *_image for every partition in dynamic_partition_list. |
| |
| Args: |
| output_file: The output file object. |
| input_file: The input file object. |
| lpmake_path: The path to lpmake binary. |
| get_image: A function that takes the partition name as the |
| parameter and returns the image path. |
| """ |
| partition_names = () |
| for line in input_file: |
| split_line = line.strip().split("=", 1) |
| if len(split_line) < 2: |
| split_line = (split_line[0], "") |
| if split_line[0] == "dynamic_partition_list": |
| partition_names = split_line[1].split() |
| elif split_line[0] == "lpmake": |
| output_file.write("lpmake=%s\n" % lpmake_path) |
| continue |
| elif split_line[0].endswith("_image"): |
| continue |
| output_file.write(line) |
| |
| if not partition_names: |
| logger.w("No dynamic partition list in misc info.") |
| |
| for partition_name in partition_names: |
| output_file.write("%s_image=%s\n" % |
| (partition_name, get_image(partition_name))) |
| |
| @utils.TimeExecute(function_description="Build super image") |
| @utils.TimeoutException(_BUILD_SUPER_IMAGE_TIMEOUT_SECS) |
| def BuildSuperImage(self, output_path, misc_info_path, get_image): |
| """Use build_super_image to create a super image. |
| |
| Args: |
| output_path: The path to the output super image. |
| misc_info_path: The path to the misc info that provides parameters |
| to create the super image. |
| get_image: A function that takes the partition name as the |
| parameter and returns the image path. |
| """ |
| build_super_image = self._GetBinary(_BUILD_SUPER_IMAGE) |
| lpmake = self._GetBinary(_LPMAKE) |
| |
| new_misc_info_path = None |
| try: |
| with open(misc_info_path, "r") as misc_info: |
| with tempfile.NamedTemporaryFile( |
| prefix="misc_info_", suffix=".txt", |
| delete=False, mode="w") as new_misc_info: |
| new_misc_info_path = new_misc_info.name |
| self._RewriteMiscInfo(new_misc_info, misc_info, lpmake, |
| get_image) |
| |
| utils.Popen(build_super_image, new_misc_info_path, output_path) |
| finally: |
| if new_misc_info_path: |
| os.remove(new_misc_info_path) |
| |
| @utils.TimeExecute(function_description="Make disabled vbmeta image.") |
| @utils.TimeoutException(_AVBTOOL_TIMEOUT_SECS) |
| def MakeDisabledVbmetaImage(self, output_path): |
| """Use avbtool to create a vbmeta image with verification disabled. |
| |
| Args: |
| output_path: The path to the output vbmeta image. |
| """ |
| avbtool = self._GetBinary(_AVBTOOL) |
| utils.Popen(avbtool, "make_vbmeta_image", |
| "--flag", "2", |
| "--padding_size", "4096", |
| "--output", output_path) |
| |
| @staticmethod |
| def _RewriteSystemQemuConfig(output_file, input_file, get_image): |
| """Rewrite image paths in system-qemu-config.txt. |
| |
| Sample input_file: |
| out/target/product/generic_x86_64/vbmeta.img vbmeta 1 |
| out/target/product/generic_x86_64/super.img super 2 |
| |
| Sample output_file: |
| /path/to/vbmeta.img vbmeta 1 |
| /path/to/super.img super 2 |
| |
| This method replaces the first entry of each line with the path |
| returned by get_image. |
| |
| Args: |
| output_file: The output file object. |
| input_file: The input file object. |
| get_image: A function that takes the partition name as the |
| parameter and returns the image path. |
| """ |
| for line in input_file: |
| split_line = line.split() |
| if len(split_line) == 3: |
| output_file.write("%s %s %s\n" % (get_image(split_line[1]), |
| split_line[1], |
| split_line[2])) |
| else: |
| output_file.write(line) |
| |
| @utils.TimeExecute(function_description="Make combined image") |
| @utils.TimeoutException(_MK_COMBINED_IMG_TIMEOUT_SECS) |
| def MkCombinedImg(self, output_path, system_qemu_config_path, get_image): |
| """Use mk_combined_img to create a disk image. |
| |
| Args: |
| output_path: The path to the output disk image. |
| system_qemu_config: The path to the config that provides the |
| parition information on the disk. |
| get_image: A function that takes the partition name as the |
| parameter and returns the image path. |
| """ |
| mk_combined_img = self._GetBinary(_MK_COMBINED_IMG) |
| sgdisk = self._GetBinary(_SGDISK) |
| simg2img = self._GetBinary(_SIMG2IMG) |
| |
| new_config_path = None |
| try: |
| with open(system_qemu_config_path, "r") as config: |
| with tempfile.NamedTemporaryFile( |
| prefix="system-qemu-config_", suffix=".txt", |
| delete=False, mode="w") as new_config: |
| new_config_path = new_config.name |
| self._RewriteSystemQemuConfig(new_config, config, |
| get_image) |
| |
| mk_combined_img_env = {"SGDISK": sgdisk, "SIMG2IMG": simg2img} |
| utils.Popen(mk_combined_img, |
| "-i", new_config_path, |
| "-o", output_path, |
| env=mk_combined_img_env) |
| finally: |
| if new_config_path: |
| os.remove(new_config_path) |
| |
| @utils.TimeExecute(function_description="Unpack boot image") |
| @utils.TimeoutException(_UNPACK_BOOTIMG_TIMEOUT_SECS) |
| def UnpackBootImg(self, out_dir, boot_img): |
| """Use unpack_bootimg to unpack a boot image to a direcotry. |
| |
| Args: |
| out_dir: The output directory. |
| boot_img: The path to the boot image. |
| """ |
| unpack_bootimg = self._GetBinary(_UNPACK_BOOTIMG) |
| utils.Popen(unpack_bootimg, |
| "--out", out_dir, |
| "--boot_img", boot_img) |
| |
| def MixSuperImage(self, super_image, misc_info, image_dir, |
| system_image=None, system_ext_image=None, |
| product_image=None, system_dlkm_image=None, |
| vendor_image=None, vendor_dlkm_image=None, |
| odm_image=None, odm_dlkm_image=None): |
| """Create mixed super image from device images and given partition |
| images. |
| |
| Args: |
| super_image: Path to the output super image. |
| misc_info: Path to the misc_info.txt. |
| image_dir: Path to image files excluding system image. |
| system_image: Path to the system image. |
| system_ext_image: Path to the system_ext image. |
| product_image: Path to the product image. |
| system_dlkm_image: Path to the system_dlkm image. |
| vendor_image: Path to the vendor image. |
| vendor_dlkm_image: Path to the vendor_dlkm image. |
| odm_image: Path to the odm image. |
| odm_dlkm_image: Path to the odm_dlkm image. |
| """ |
| self.BuildSuperImage( |
| super_image, misc_info, |
| lambda partition: GetImageForPartition( |
| partition, image_dir, |
| system=system_image, |
| system_ext=system_ext_image, |
| product=product_image, |
| system_dlkm=system_dlkm_image, |
| vendor=vendor_image, |
| vendor_dlkm=vendor_dlkm_image, |
| odm=odm_image, |
| odm_dlkm=odm_dlkm_image)) |
