unplugged-system/tools/acloud/internal/lib/cvd_utils.py

# Copyright 2022 - 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.

"""Utility functions that process cuttlefish images."""

import collections
import glob
import logging
import os
import posixpath as remote_path
import re
import subprocess
import tempfile

from acloud import errors
from acloud.create import create_common
from acloud.internal import constants
from acloud.internal.lib import ota_tools
from acloud.internal.lib import ssh
from acloud.internal.lib import utils
from acloud.public import report


logger = logging.getLogger(__name__)

# Local build artifacts to be uploaded.
_ARTIFACT_FILES = ["*.img", "bootloader", "kernel"]
# The boot image name pattern corresponds to the use cases:
# - In a cuttlefish build environment, ANDROID_PRODUCT_OUT conatins boot.img
#   and boot-debug.img. The former is the default boot image. The latter is not
#   useful for cuttlefish.
# - In an officially released GKI (Generic Kernel Image) package, the image
#   name is boot-<kernel version>.img.
_BOOT_IMAGE_NAME_PATTERN = r"boot(-[\d.]+)?\.img"
_VENDOR_BOOT_IMAGE_NAME = "vendor_boot.img"
_KERNEL_IMAGE_NAMES = ("kernel", "bzImage", "Image")
_INITRAMFS_IMAGE_NAME = "initramfs.img"
_VENDOR_IMAGE_NAMES = ("vendor.img", "vendor_dlkm.img", "odm.img",
                       "odm_dlkm.img")
VendorImagePaths = collections.namedtuple(
    "VendorImagePaths",
    ["vendor", "vendor_dlkm", "odm", "odm_dlkm"])

# The relative path to the base directory containing cuttelfish images, tools,
# and runtime files. On a GCE instance, the directory is the SSH user's HOME.
GCE_BASE_DIR = "."
_REMOTE_HOST_BASE_DIR_FORMAT = "acloud_cf_%(num)d"
# Relative paths in a base directory.
_REMOTE_IMAGE_DIR = "acloud_image"
_REMOTE_BOOT_IMAGE_PATH = remote_path.join(_REMOTE_IMAGE_DIR, "boot.img")
_REMOTE_VENDOR_BOOT_IMAGE_PATH = remote_path.join(
    _REMOTE_IMAGE_DIR, _VENDOR_BOOT_IMAGE_NAME)
_REMOTE_VBMETA_IMAGE_PATH = remote_path.join(_REMOTE_IMAGE_DIR, "vbmeta.img")
_REMOTE_KERNEL_IMAGE_PATH = remote_path.join(
    _REMOTE_IMAGE_DIR, _KERNEL_IMAGE_NAMES[0])
_REMOTE_INITRAMFS_IMAGE_PATH = remote_path.join(
    _REMOTE_IMAGE_DIR, _INITRAMFS_IMAGE_NAME)
_REMOTE_SUPER_IMAGE_DIR = remote_path.join(_REMOTE_IMAGE_DIR,
                                           "super_image_dir")

# Remote host instance name
_REMOTE_HOST_INSTANCE_NAME_FORMAT = (
    constants.INSTANCE_TYPE_HOST +
    "-%(ip_addr)s-%(num)d-%(build_id)s-%(build_target)s")
_REMOTE_HOST_INSTANCE_NAME_PATTERN = re.compile(
    constants.INSTANCE_TYPE_HOST + r"-(?P<ip_addr>[\d.]+)-(?P<num>\d+)-.+")
# launch_cvd arguments.
_DATA_POLICY_CREATE_IF_MISSING = "create_if_missing"
_DATA_POLICY_ALWAYS_CREATE = "always_create"
_NUM_AVDS_ARG = "-num_instances=%(num_AVD)s"
AGREEMENT_PROMPT_ARG = "-report_anonymous_usage_stats=y"
UNDEFOK_ARG = "-undefok=report_anonymous_usage_stats,config"
# Connect the OpenWrt device via console file.
_ENABLE_CONSOLE_ARG = "-console=true"
# WebRTC args
_WEBRTC_ID = "--webrtc_device_id=%(instance)s"
_WEBRTC_ARGS = ["--start_webrtc", "--vm_manager=crosvm"]
_VNC_ARGS = ["--start_vnc_server=true"]

# Cuttlefish runtime directory is specified by `-instance_dir <runtime_dir>`.
# Cuttlefish tools may create a symbolic link at the specified path.
# The actual location of the runtime directory depends on the version:
#
# In Android 10, the directory is `<runtime_dir>`.
#
# In Android 11 and 12, the directory is `<runtime_dir>.<num>`.
# `<runtime_dir>` is a symbolic link to the first device's directory.
#
# In the latest version, if `--instance-dir <runtime_dir>` is specified, the
# directory is `<runtime_dir>/instances/cvd-<num>`.
# `<runtime_dir>_runtime` and `<runtime_dir>.<num>` are symbolic links.
#
# If `--instance-dir <runtime_dir>` is not specified, the directory is
# `~/cuttlefish/instances/cvd-<num>`.
# `~/cuttlefish_runtime` and `~/cuttelfish_runtime.<num>` are symbolic links.
_LOCAL_LOG_DIR_FORMAT = os.path.join(
    "%(runtime_dir)s", "instances", "cvd-%(num)d", "logs")
# Relative paths in a base directory.
_REMOTE_RUNTIME_DIR_FORMAT = remote_path.join(
    "cuttlefish", "instances", "cvd-%(num)d")
_REMOTE_LEGACY_RUNTIME_DIR_FORMAT = "cuttlefish_runtime.%(num)d"
HOST_KERNEL_LOG = report.LogFile(
    "/var/log/kern.log", constants.LOG_TYPE_KERNEL_LOG, "host_kernel.log")

# Contents of the target_files archive.
_DOWNLOAD_MIX_IMAGE_NAME = "{build_target}-target_files-{build_id}.zip"
_TARGET_FILES_META_DIR_NAME = "META"
_TARGET_FILES_IMAGES_DIR_NAME = "IMAGES"
_MISC_INFO_FILE_NAME = "misc_info.txt"

# ARM flavor build target pattern.
_ARM_TARGET_PATTERN = "arm"


def GetAdbPorts(base_instance_num, num_avds_per_instance):
    """Get ADB ports of cuttlefish.

    Args:
        base_instance_num: An integer or None, the instance number of the first
                           device.
        num_avds_per_instance: An integer or None, the number of devices.

    Returns:
        The port numbers as a list of integers.
    """
    return [constants.CF_ADB_PORT + (base_instance_num or 1) - 1 + index
            for index in range(num_avds_per_instance or 1)]

def GetFastbootPorts(base_instance_num, num_avds_per_instance):
    """Get Fastboot ports of cuttlefish.

    Args:
        base_instance_num: An integer or None, the instance number of the first
                           device.
        num_avds_per_instance: An integer or None, the number of devices.

    Returns:
        The port numbers as a list of integers.
    """
    return [constants.CF_FASTBOOT_PORT + (base_instance_num or 1) - 1 + index
            for index in range(num_avds_per_instance or 1)]

def GetVncPorts(base_instance_num, num_avds_per_instance):
    """Get VNC ports of cuttlefish.

    Args:
        base_instance_num: An integer or None, the instance number of the first
                           device.
        num_avds_per_instance: An integer or None, the number of devices.

    Returns:
        The port numbers as a list of integers.
    """
    return [constants.CF_VNC_PORT + (base_instance_num or 1) - 1 + index
            for index in range(num_avds_per_instance or 1)]


def _UploadImageZip(ssh_obj, remote_dir, image_zip):
    """Upload an image zip to a remote host and a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        image_zip: The path to the image zip.
    """
    remote_cmd = f"/usr/bin/install_zip.sh {remote_dir} < {image_zip}"
    logger.debug("remote_cmd:\n %s", remote_cmd)
    ssh_obj.Run(remote_cmd)


def _UploadImageDir(ssh_obj, remote_dir, image_dir):
    """Upload an image directory to a remote host or a GCE instance.

    The images are compressed for faster upload.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        image_dir: The directory containing the files to be uploaded.
    """
    try:
        images_path = os.path.join(image_dir, "required_images")
        with open(images_path, "r", encoding="utf-8") as images:
            artifact_files = images.read().splitlines()
    except IOError:
        # Older builds may not have a required_images file. In this case
        # we fall back to *.img.
        artifact_files = []
        for file_name in _ARTIFACT_FILES:
            artifact_files.extend(
                os.path.basename(image) for image in glob.glob(
                    os.path.join(image_dir, file_name)))
    # Upload android-info.txt to parse config value.
    artifact_files.append(constants.ANDROID_INFO_FILE)
    cmd = (f"tar -cf - --lzop -S -C {image_dir} {' '.join(artifact_files)} | "
           f"{ssh_obj.GetBaseCmd(constants.SSH_BIN)} -- "
           f"tar -xf - --lzop -S -C {remote_dir}")
    logger.debug("cmd:\n %s", cmd)
    ssh.ShellCmdWithRetry(cmd)


def _UploadCvdHostPackage(ssh_obj, remote_dir, cvd_host_package):
    """Upload a CVD host package to a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        cvd_host_package: The path to the CVD host package.
    """
    if cvd_host_package.endswith(".tar.gz"):
        remote_cmd = f"tar -xzf - -C {remote_dir} < {cvd_host_package}"
        logger.debug("remote_cmd:\n %s", remote_cmd)
        ssh_obj.Run(remote_cmd)
    else:
        cmd = (f"tar -cf - --lzop -S -C {cvd_host_package} . | "
               f"{ssh_obj.GetBaseCmd(constants.SSH_BIN)} -- "
               f"tar -xf - --lzop -S -C {remote_dir}")
        logger.debug("cmd:\n %s", cmd)
        ssh.ShellCmdWithRetry(cmd)


@utils.TimeExecute(function_description="Processing and uploading local images")
def UploadArtifacts(ssh_obj, remote_dir, image_path, cvd_host_package):
    """Upload images and a CVD host package to a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        image_path: A string, the path to the image zip built by `m dist` or
                    the directory containing the images built by `m`.
        cvd_host_package: A string, the path to the CVD host package in gzip.
    """
    if os.path.isdir(image_path):
        _UploadImageDir(ssh_obj, remote_dir, image_path)
    else:
        _UploadImageZip(ssh_obj, remote_dir, image_path)
    _UploadCvdHostPackage(ssh_obj, remote_dir, cvd_host_package)


def FindBootImages(search_path):
    """Find boot and vendor_boot images in a path.

    Args:
        search_path: A path to an image file or an image directory.

    Returns:
        The boot image path and the vendor_boot image path. Each value can be
        None if the path doesn't exist.

    Raises:
        errors.GetLocalImageError if search_path contains more than one boot
        image or the file format is not correct.
    """
    boot_image_path = create_common.FindBootImage(search_path,
                                                  raise_error=False)
    vendor_boot_image_path = os.path.join(search_path, _VENDOR_BOOT_IMAGE_NAME)
    if not os.path.isfile(vendor_boot_image_path):
        vendor_boot_image_path = None

    return boot_image_path, vendor_boot_image_path


def FindKernelImages(search_path):
    """Find kernel and initramfs images in a path.

    Args:
        search_path: A path to an image directory.

    Returns:
        The kernel image path and the initramfs image path. Each value can be
        None if the path doesn't exist.
    """
    paths = [os.path.join(search_path, name) for name in _KERNEL_IMAGE_NAMES]
    kernel_image_path = next((path for path in paths if os.path.isfile(path)),
                             None)

    initramfs_image_path = os.path.join(search_path, _INITRAMFS_IMAGE_NAME)
    if not os.path.isfile(initramfs_image_path):
        initramfs_image_path = None

    return kernel_image_path, initramfs_image_path


@utils.TimeExecute(function_description="Uploading local kernel images.")
def _UploadKernelImages(ssh_obj, remote_dir, search_path):
    """Find and upload kernel or boot images to a remote host or a GCE
    instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        search_path: A path to an image file or an image directory.

    Returns:
        A list of strings, the launch_cvd arguments including the remote paths.

    Raises:
        errors.GetLocalImageError if search_path does not contain kernel
        images.
    """
    # Assume that the caller cleaned up the remote home directory.
    ssh_obj.Run("mkdir -p " + remote_path.join(remote_dir, _REMOTE_IMAGE_DIR))

    kernel_image_path, initramfs_image_path = FindKernelImages(search_path)
    if kernel_image_path and initramfs_image_path:
        remote_kernel_image_path = remote_path.join(
            remote_dir, _REMOTE_KERNEL_IMAGE_PATH)
        remote_initramfs_image_path = remote_path.join(
            remote_dir, _REMOTE_INITRAMFS_IMAGE_PATH)
        ssh_obj.ScpPushFile(kernel_image_path, remote_kernel_image_path)
        ssh_obj.ScpPushFile(initramfs_image_path, remote_initramfs_image_path)
        return ["-kernel_path", remote_kernel_image_path,
                "-initramfs_path", remote_initramfs_image_path]

    boot_image_path, vendor_boot_image_path = FindBootImages(search_path)
    if boot_image_path:
        remote_boot_image_path = remote_path.join(
            remote_dir, _REMOTE_BOOT_IMAGE_PATH)
        ssh_obj.ScpPushFile(boot_image_path, remote_boot_image_path)
        launch_cvd_args = ["-boot_image", remote_boot_image_path]
        if vendor_boot_image_path:
            remote_vendor_boot_image_path = remote_path.join(
                remote_dir, _REMOTE_VENDOR_BOOT_IMAGE_PATH)
            ssh_obj.ScpPushFile(vendor_boot_image_path,
                                remote_vendor_boot_image_path)
            launch_cvd_args.extend(["-vendor_boot_image",
                                    remote_vendor_boot_image_path])
        return launch_cvd_args

    raise errors.GetLocalImageError(
        f"{search_path} is not a boot image or a directory containing images.")


@utils.TimeExecute(function_description="Uploading disabled vbmeta image.")
def _UploadDisabledVbmetaImage(ssh_obj, remote_dir, local_tool_dirs):
    """Upload disabled vbmeta image to a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        local_tool_dirs: A list of local directories containing tools.

    Returns:
        A list of strings, the launch_cvd arguments including the remote paths.

    Raises:
        CheckPathError if local_tool_dirs do not contain OTA tools.
    """
    # Assume that the caller cleaned up the remote home directory.
    ssh_obj.Run("mkdir -p " + remote_path.join(remote_dir, _REMOTE_IMAGE_DIR))

    remote_vbmeta_image_path = remote_path.join(remote_dir,
                                                _REMOTE_VBMETA_IMAGE_PATH)
    with tempfile.NamedTemporaryFile(prefix="vbmeta",
                                     suffix=".img") as temp_file:
        tool_dirs = local_tool_dirs + create_common.GetNonEmptyEnvVars(
                constants.ENV_ANDROID_SOONG_HOST_OUT,
                constants.ENV_ANDROID_HOST_OUT)
        ota = ota_tools.FindOtaTools(tool_dirs)
        ota.MakeDisabledVbmetaImage(temp_file.name)
        ssh_obj.ScpPushFile(temp_file.name, remote_vbmeta_image_path)

    return ["-vbmeta_image", remote_vbmeta_image_path]


def UploadExtraImages(ssh_obj, remote_dir, avd_spec):
    """Find and upload the images specified in avd_spec.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        avd_spec: An AvdSpec object containing extra image paths.

    Returns:
        A list of strings, the launch_cvd arguments including the remote paths.

    Raises:
        errors.GetLocalImageError if any specified image path does not exist.
    """
    extra_img_args = []
    if avd_spec.local_kernel_image:
        extra_img_args += _UploadKernelImages(ssh_obj, remote_dir,
                                              avd_spec.local_kernel_image)
    if avd_spec.local_vendor_image:
        extra_img_args += _UploadDisabledVbmetaImage(ssh_obj, remote_dir,
                                                     avd_spec.local_tool_dirs)
    return extra_img_args


@utils.TimeExecute(function_description="Uploading local super image")
def UploadSuperImage(ssh_obj, remote_dir, super_image_path):
    """Upload a super image to a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        super_image_path: Path to the super image file.

    Returns:
        A list of strings, the launch_cvd arguments including the remote paths.
    """
    # Assume that the caller cleaned up the remote home directory.
    super_image_stem = os.path.basename(super_image_path)
    remote_super_image_dir = remote_path.join(
        remote_dir, _REMOTE_SUPER_IMAGE_DIR)
    remote_super_image_path = remote_path.join(
        remote_super_image_dir, super_image_stem)
    ssh_obj.Run(f"mkdir -p {remote_super_image_dir}")
    cmd = (f"tar -cf - --lzop -S -C {os.path.dirname(super_image_path)} "
           f"{super_image_stem} | "
           f"{ssh_obj.GetBaseCmd(constants.SSH_BIN)} -- "
           f"tar -xf - --lzop -S -C {remote_super_image_dir}")
    ssh.ShellCmdWithRetry(cmd)
    launch_cvd_args = ["-super_image", remote_super_image_path]
    return launch_cvd_args


def CleanUpRemoteCvd(ssh_obj, remote_dir, raise_error):
    """Call stop_cvd and delete the files on a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        raise_error: Whether to raise an error if the remote instance is not
                     running.

    Raises:
        subprocess.CalledProcessError if any command fails.
    """
    home = remote_path.join("$HOME", remote_dir)
    stop_cvd_path = remote_path.join(remote_dir, "bin", "stop_cvd")
    stop_cvd_cmd = f"'HOME={home} {stop_cvd_path}'"
    if raise_error:
        ssh_obj.Run(stop_cvd_cmd)
    else:
        try:
            ssh_obj.Run(stop_cvd_cmd, retry=0)
        except Exception as e:
            logger.debug(
                "Failed to stop_cvd (possibly no running device): %s", e)

    # This command deletes all files except hidden files under HOME.
    # It does not raise an error if no files can be deleted.
    ssh_obj.Run(f"'rm -rf {remote_path.join(remote_dir, '*')}'")


def GetRemoteHostBaseDir(base_instance_num):
    """Get remote base directory by instance number.

    Args:
        base_instance_num: Integer or None, the instance number of the device.

    Returns:
        The remote base directory.
    """
    return _REMOTE_HOST_BASE_DIR_FORMAT % {"num": base_instance_num or 1}


def FormatRemoteHostInstanceName(ip_addr, base_instance_num, build_id,
                                 build_target):
    """Convert an IP address and build info to an instance name.

    Args:
        ip_addr: String, the IP address of the remote host.
        base_instance_num: Integer or None, the instance number of the device.
        build_id: String, the build id.
        build_target: String, the build target, e.g., aosp_cf_x86_64_phone.

    Return:
        String, the instance name.
    """
    return _REMOTE_HOST_INSTANCE_NAME_FORMAT % {
        "ip_addr": ip_addr,
        "num": base_instance_num or 1,
        "build_id": build_id,
        "build_target": build_target}


def ParseRemoteHostAddress(instance_name):
    """Parse IP address from a remote host instance name.

    Args:
        instance_name: String, the instance name.

    Returns:
        The IP address and the base directory as strings.
        None if the name does not represent a remote host instance.
    """
    match = _REMOTE_HOST_INSTANCE_NAME_PATTERN.fullmatch(instance_name)
    if match:
        return (match.group("ip_addr"),
                GetRemoteHostBaseDir(int(match.group("num"))))
    return None


# pylint:disable=too-many-branches
def GetLaunchCvdArgs(avd_spec, config=None):
    """Get launch_cvd arguments for remote instances.

    Args:
        avd_spec: An AVDSpec instance.
        config: A string, the name of the predefined hardware config.
                e.g., "auto", "phone", and "tv".

    Returns:
        A list of strings, arguments of launch_cvd.
    """
    launch_cvd_args = []

    blank_data_disk_size_gb = avd_spec.cfg.extra_data_disk_size_gb
    if blank_data_disk_size_gb and blank_data_disk_size_gb > 0:
        launch_cvd_args.append(
            "-data_policy=" + _DATA_POLICY_CREATE_IF_MISSING)
        launch_cvd_args.append(
            "-blank_data_image_mb=" + str(blank_data_disk_size_gb * 1024))

    if config:
        launch_cvd_args.append("-config=" + config)
    if avd_spec.hw_customize or not config:
        launch_cvd_args.append(
            "-x_res=" + avd_spec.hw_property[constants.HW_X_RES])
        launch_cvd_args.append(
            "-y_res=" + avd_spec.hw_property[constants.HW_Y_RES])
        launch_cvd_args.append(
            "-dpi=" + avd_spec.hw_property[constants.HW_ALIAS_DPI])
        if constants.HW_ALIAS_DISK in avd_spec.hw_property:
            launch_cvd_args.append(
                "-data_policy=" + _DATA_POLICY_ALWAYS_CREATE)
            launch_cvd_args.append(
                "-blank_data_image_mb="
                + avd_spec.hw_property[constants.HW_ALIAS_DISK])
        if constants.HW_ALIAS_CPUS in avd_spec.hw_property:
            launch_cvd_args.append(
                "-cpus=" + str(avd_spec.hw_property[constants.HW_ALIAS_CPUS]))
        if constants.HW_ALIAS_MEMORY in avd_spec.hw_property:
            launch_cvd_args.append(
                "-memory_mb=" +
                str(avd_spec.hw_property[constants.HW_ALIAS_MEMORY]))

    if avd_spec.connect_webrtc:
        launch_cvd_args.extend(_WEBRTC_ARGS)
        if avd_spec.webrtc_device_id:
            launch_cvd_args.append(
                _WEBRTC_ID % {"instance": avd_spec.webrtc_device_id})
    if avd_spec.connect_vnc:
        launch_cvd_args.extend(_VNC_ARGS)
    if avd_spec.openwrt:
        launch_cvd_args.append(_ENABLE_CONSOLE_ARG)
    if avd_spec.num_avds_per_instance > 1:
        launch_cvd_args.append(
            _NUM_AVDS_ARG % {"num_AVD": avd_spec.num_avds_per_instance})
    if avd_spec.base_instance_num:
        launch_cvd_args.append(
            "--base-instance-num=" + str(avd_spec.base_instance_num))
    if avd_spec.launch_args:
        launch_cvd_args.append(avd_spec.launch_args)

    launch_cvd_args.append(UNDEFOK_ARG)
    launch_cvd_args.append(AGREEMENT_PROMPT_ARG)
    return launch_cvd_args


def _GetRemoteRuntimeDirs(ssh_obj, remote_dir, base_instance_num,
                          num_avds_per_instance):
    """Get cuttlefish runtime directories on a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        base_instance_num: An integer, the instance number of the first device.
        num_avds_per_instance: An integer, the number of devices.

    Returns:
        A list of strings, the paths to the runtime directories.
    """
    runtime_dir = remote_path.join(
        remote_dir, _REMOTE_RUNTIME_DIR_FORMAT % {"num": base_instance_num})
    try:
        ssh_obj.Run(f"test -d {runtime_dir}", retry=0)
        return [remote_path.join(remote_dir,
                                 _REMOTE_RUNTIME_DIR_FORMAT %
                                 {"num": base_instance_num + num})
                for num in range(num_avds_per_instance)]
    except subprocess.CalledProcessError:
        logger.debug("%s is not the runtime directory.", runtime_dir)

    legacy_runtime_dirs = [
        remote_path.join(remote_dir, constants.REMOTE_LOG_FOLDER)]
    legacy_runtime_dirs.extend(
        remote_path.join(remote_dir,
                         _REMOTE_LEGACY_RUNTIME_DIR_FORMAT %
                         {"num": base_instance_num + num})
        for num in range(1, num_avds_per_instance))
    return legacy_runtime_dirs


def GetRemoteFetcherConfigJson(remote_dir):
    """Get the config created by fetch_cvd on a remote host or a GCE instance.

    Args:
        remote_dir: The remote base directory.

    Returns:
        An object of report.LogFile.
    """
    return report.LogFile(remote_path.join(remote_dir, "fetcher_config.json"),
                          constants.LOG_TYPE_CUTTLEFISH_LOG)


def _GetRemoteTombstone(runtime_dir, name_suffix):
    """Get log object for tombstones in a remote cuttlefish runtime directory.

    Args:
        runtime_dir: The path to the remote cuttlefish runtime directory.
        name_suffix: The string appended to the log name. It is used to
                     distinguish log files found in different runtime_dirs.

    Returns:
        A report.LogFile object.
    """
    return report.LogFile(remote_path.join(runtime_dir, "tombstones"),
                          constants.LOG_TYPE_DIR,
                          "tombstones-zip" + name_suffix)


def _GetLogType(file_name):
    """Determine log type by file name.

    Args:
        file_name: A file name.

    Returns:
        A string, one of the log types defined in constants.
        None if the file is not a log file.
    """
    if file_name == "kernel.log":
        return constants.LOG_TYPE_KERNEL_LOG
    if file_name == "logcat":
        return constants.LOG_TYPE_LOGCAT
    if file_name.endswith(".log") or file_name == "cuttlefish_config.json":
        return constants.LOG_TYPE_CUTTLEFISH_LOG
    return None


def FindRemoteLogs(ssh_obj, remote_dir, base_instance_num,
                   num_avds_per_instance):
    """Find log objects on a remote host or a GCE instance.

    Args:
        ssh_obj: An Ssh object.
        remote_dir: The remote base directory.
        base_instance_num: An integer or None, the instance number of the first
                           device.
        num_avds_per_instance: An integer or None, the number of devices.

    Returns:
        A list of report.LogFile objects.
    """
    runtime_dirs = _GetRemoteRuntimeDirs(
        ssh_obj, remote_dir,
        (base_instance_num or 1), (num_avds_per_instance or 1))
    logs = []
    for log_path in utils.FindRemoteFiles(ssh_obj, runtime_dirs):
        file_name = remote_path.basename(log_path)
        log_type = _GetLogType(file_name)
        if not log_type:
            continue
        base, ext = remote_path.splitext(file_name)
        # The index of the runtime_dir containing log_path.
        index_str = ""
        for index, runtime_dir in enumerate(runtime_dirs):
            if log_path.startswith(runtime_dir + remote_path.sep):
                index_str = "." + str(index) if index else ""
        log_name = ("full_gce_logcat" + index_str if file_name == "logcat" else
                    base + index_str + ext)

        logs.append(report.LogFile(log_path, log_type, log_name))

    logs.extend(_GetRemoteTombstone(runtime_dir,
                                    ("." + str(index) if index else ""))
                for index, runtime_dir in enumerate(runtime_dirs))
    return logs


def FindLocalLogs(runtime_dir, instance_num):
    """Find log objects in a local runtime directory.

    Args:
        runtime_dir: A string, the runtime directory path.
        instance_num: An integer, the instance number.

    Returns:
        A list of report.LogFile.
    """
    log_dir = _LOCAL_LOG_DIR_FORMAT % {"runtime_dir": runtime_dir,
                                       "num": instance_num}
    if not os.path.isdir(log_dir):
        log_dir = runtime_dir

    logs = []
    for parent_dir, _, file_names in os.walk(log_dir, followlinks=False):
        for file_name in file_names:
            log_path = os.path.join(parent_dir, file_name)
            log_type = _GetLogType(file_name)
            if os.path.islink(log_path) or not log_type:
                continue
            logs.append(report.LogFile(log_path, log_type))
    return logs


def GetRemoteBuildInfoDict(avd_spec):
    """Convert remote build infos to a dictionary for reporting.

    Args:
        avd_spec: An AvdSpec object containing the build infos.

    Returns:
        A dict containing the build infos.
    """
    build_info_dict = {
        key: val for key, val in avd_spec.remote_image.items() if val}

    # kernel_target has a default value. If the user provides kernel_build_id
    # or kernel_branch, then convert kernel build info.
    if (avd_spec.kernel_build_info.get(constants.BUILD_ID) or
            avd_spec.kernel_build_info.get(constants.BUILD_BRANCH)):
        build_info_dict.update(
            {"kernel_" + key: val
             for key, val in avd_spec.kernel_build_info.items() if val}
        )
    build_info_dict.update(
        {"system_" + key: val
         for key, val in avd_spec.system_build_info.items() if val}
    )
    build_info_dict.update(
        {"bootloader_" + key: val
         for key, val in avd_spec.bootloader_build_info.items() if val}
    )
    return build_info_dict


def GetMixBuildTargetFilename(build_target, build_id):
    """Get the mix build target filename.

    Args:
        build_id: String, Build id, e.g. "2263051", "P2804227"
        build_target: String, the build target, e.g. cf_x86_phone-userdebug

    Returns:
        String, a file name, e.g. "cf_x86_phone-target_files-2263051.zip"
    """
    return _DOWNLOAD_MIX_IMAGE_NAME.format(
        build_target=build_target.split('-')[0],
        build_id=build_id)


def FindMiscInfo(image_dir):
    """Find misc info in build output dir or extracted target files.

    Args:
        image_dir: The directory to search for misc info.

    Returns:
        image_dir if the directory structure looks like an output directory
        in build environment.
        image_dir/META if it looks like extracted target files.

    Raises:
        errors.CheckPathError if this function cannot find misc info.
    """
    misc_info_path = os.path.join(image_dir, _MISC_INFO_FILE_NAME)
    if os.path.isfile(misc_info_path):
        return misc_info_path
    misc_info_path = os.path.join(image_dir, _TARGET_FILES_META_DIR_NAME,
                                  _MISC_INFO_FILE_NAME)
    if os.path.isfile(misc_info_path):
        return misc_info_path
    raise errors.CheckPathError(
        f"Cannot find {_MISC_INFO_FILE_NAME} in {image_dir}. The "
        f"directory is expected to be an extracted target files zip or "
        f"{constants.ENV_ANDROID_PRODUCT_OUT}.")


def FindImageDir(image_dir):
    """Find images in build output dir or extracted target files.

    Args:
        image_dir: The directory to search for images.

    Returns:
        image_dir if the directory structure looks like an output directory
        in build environment.
        image_dir/IMAGES if it looks like extracted target files.

    Raises:
        errors.GetLocalImageError if this function cannot find any image.
    """
    if glob.glob(os.path.join(image_dir, "*.img")):
        return image_dir
    subdir = os.path.join(image_dir, _TARGET_FILES_IMAGES_DIR_NAME)
    if glob.glob(os.path.join(subdir, "*.img")):
        return subdir
    raise errors.GetLocalImageError(
        "Cannot find images in %s." % image_dir)


def IsArmImage(image):
    """Check if the image is built for ARM.

    Args:
        image: Image meta info.

    Returns:
        A boolean, whether the image is for ARM.
    """
    return _ARM_TARGET_PATTERN in image.get("build_target", "")


def FindVendorImages(image_dir):
    """Find vendor, vendor_dlkm, odm, and odm_dlkm image in build output dir.

    Args:
        image_dir: The directory to search for images.

    Returns:
        An object of VendorImagePaths.

    Raises:
        errors.GetLocalImageError if this function cannot find images.
    """

    image_paths = []
    for image_name in _VENDOR_IMAGE_NAMES:
        image_path = os.path.join(image_dir, image_name)
        if not os.path.isfile(image_path):
            raise errors.GetLocalImageError(
                f"Cannot find {image_path} in {image_dir}.")
        image_paths.append(image_path)

    return VendorImagePaths(*image_paths)