o3de/cmake/Tools/Platform/Android/android_deployment.py

#
# All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
# its licensors.
#
# For complete copyright and license terms please see the LICENSE at the root of this
# distribution (the "License"). All use of this software is governed by the License,
# or, if provided, by the license below or the license accompanying this file. Do not
# remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#

import datetime
import logging
import os
import json
import platform
import subprocess
import sys
import time
import pathlib

from distutils.version import LooseVersion

# Resolve the common python module
ROOT_DEV_PATH = os.path.realpath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
if ROOT_DEV_PATH not in sys.path:
    sys.path.append(ROOT_DEV_PATH)

from cmake.Tools import common
from cmake.Tools.Platform.Android import android_support

# The following is the list of known android external storage paths that we will attempt to verify on a device and
# return the first one that is detected
KNOWN_ANDROID_EXTERNAL_STORAGE_PATHS = [
    '/sdcard/',
    '/storage/emulated/0/',
    '/storage/emulated/legacy/',
    '/storage/sdcard0/',
    '/storage/self/primary/',
]

ANDROID_TARGET_TIMESTAMP_FILENAME = 'deploy.timestamp'


class AndroidDeployment(object):
    """
    Class to manage the deployment of game assets to an android device (Separately from the APK)
    """

    DEPLOY_APK_ONLY = 'APK'
    DEPLOY_ASSETS_ONLY = 'ASSETS'
    DEPLOY_BOTH = 'BOTH'

    def __init__(self, dev_root, build_dir, configuration, android_device_filter, clean_deploy, android_sdk_path, deployment_type, game_name=None, asset_mode=None, asset_type=None, embedded_assets=True, is_unit_test=False):
        """
        Initialize the Android Deployment Worker

        :param dev_root:                The dev-root of the engine
        :param android_device_filter:   An optional list of devices to filter on the connected devices to deploy to. If not supplied, deploy to all devices
        :param clean_deploy:            Option to clean the target device's assets before deploying the game's assets from the host
        :param android_sdk_path:        Path to the android SDK (to use the adb tool)
        :param deployment_type:         The type of deployment (DEPLOY_APK_ONLY, DEPLOY_ASSETS_ONLY, or DEPLOY_BOTH)
        :param game_name:               The name of the game whose assets are being deployed. None if is_test_project is True
        :param asset_mode:              The asset mode of deployment (LOOSE, PAK, VFS). None if is_test_project is True
        :param asset_type:              The asset type. None if is_test_project is True
        :param embedded_assets:         Boolean to indicate if the assets are embedded in the APK or not
        :param is_unit_test:            Boolean to indicate if this is a unit test deployment
        """

        self.dev_root = pathlib.Path(dev_root)

        self.build_dir = self.dev_root / build_dir

        self.configuration = configuration

        self.game_name = game_name

        self.asset_mode = asset_mode

        self.asset_type = asset_type

        self.clean_deploy = clean_deploy

        self.embedded_assets = embedded_assets

        self.deployment_type = deployment_type

        self.is_test_project = is_unit_test

        if not self.is_test_project:

            if embedded_assets:
                # If the assets are embedded, then warn that both APK and ASSETS will be deployed even if 'BOTH' is not specified
                if deployment_type in (AndroidDeployment.DEPLOY_APK_ONLY, AndroidDeployment.DEPLOY_ASSETS_ONLY):
                    logging.warning(f"Deployment type of {deployment_type} set but the assets are embedded in the APK. Both the APK and the Assets will be deployed.")

            if asset_mode == 'PAK':
                self.local_asset_path = self.dev_root / 'Pak' / f'{game_name.lower()}_{asset_type}_paks'
            else:
                self.local_asset_path = self.dev_root / game_name / 'Cache' / asset_type

            assert game_name is not None, f"'game_name' is required"
            self.game_name = game_name

            assert asset_mode is not None, f"'asset_mode' is required"
            self.asset_mode = asset_mode

            assert asset_type is not None, f"'asset_type' is required"
            self.asset_type = asset_type

            self.files_in_asset_path = list(self.local_asset_path.glob('**/*'))

            self.android_settings = AndroidDeployment.read_android_settings(self.dev_root, game_name)
        else:
            self.local_asset_path = None

            if asset_mode:
                logging.warning(f"'asset_mode' argument '{asset_mode}' ignored for unit test deployment.")
            if asset_type:
                logging.warning(f"'asset_type' argument '{asset_type}' ignored for unit test deployment.")

            self.files_in_asset_path = []

        self.apk_path = self.build_dir / 'app' / 'build' / 'outputs' / 'apk' / configuration / f'app-{configuration}.apk'

        self.android_device_filter = [android_device.strip() for android_device in android_device_filter.split(',')] if android_device_filter else []

        self.adb_path = AndroidDeployment.resolve_adb_tool(pathlib.Path(android_sdk_path))

        self.adb_started = False

    @staticmethod
    def read_android_settings(dev_root, game_name):
        """
        Read and parse the project.json file into a dictionary to process the specific attributes needed for the manifest template

        :param dev_root:    The dev root we are working from
        :param game_name:   Name of the game under the dev root
        :return: The android settings for the game project if any
        """
        game_folder = dev_root / game_name
        game_folder_project_properties_path = game_folder / 'project.json'
        game_project_properties_content = game_folder_project_properties_path.resolve(strict=True)\
                                                                             .read_text(encoding=common.DEFAULT_TEXT_READ_ENCODING,
                                                                                        errors=common.ENCODING_ERROR_HANDLINGS)
        # Extract the key attributes we need to process and build up our environment table
        game_project_json = json.loads(game_project_properties_content)
        android_settings = game_project_json.get('android_settings', {})

        return android_settings

    @staticmethod
    def resolve_adb_tool(android_sdk_path):
        """
        Resolve the location of the adb tool based on the input Android SDK Path
        :param android_sdk_path: The android SDK path to search for the adb tool
        :return: The absolute path to the adb tool
        """
        adb_target = 'adb.exe' if platform.system() == 'Windows' else 'adb'
        check_adb_target = android_sdk_path / 'platform-tools' / adb_target
        if not check_adb_target.exists():
            raise common.LmbrCmdError(f"Invalid Android SDK path '{str(android_sdk_path)}': Unable to locate '{adb_target}'.")
        return check_adb_target

    def get_android_project_settings(self, key_name, default_value):
        return self.android_settings.get(key_name, default_value)

    def adb_call(self, arg_list, device_id=None):
        """
        Wrapper to execute the adb command-line tool

        :param arg_list:    Argument list to send to the tool
        :param device_id:   Optional device id (from the 'get_target_android_devices' call) to invoke the call to.
        :return: The stdout result of the call
        """

        if isinstance(arg_list, str):
            arg_list = [arg_list]

        call_arguments = [str(self.adb_path.resolve())]

        if device_id:
            call_arguments.extend(['-s', device_id])

        call_arguments.extend(arg_list)

        try:
            output = subprocess.check_output(call_arguments,
                                             shell=True,
                                             stderr=subprocess.PIPE).decode(common.DEFAULT_TEXT_READ_ENCODING,
                                                                               common.ENCODING_ERROR_HANDLINGS)
            return output
        except subprocess.CalledProcessError as err:
            raise common.LmbrCmdError(err.stderr.decode(common.DEFAULT_TEXT_READ_ENCODING,
                                                        common.ENCODING_ERROR_HANDLINGS))

    def adb_shell(self, command, device_id):
        """
        Special wrapper around calling "adb shell" which will invoke a shell command on the android device

        :param command:         The shell command to invoke on the android device
        :param device_id:       The device id (from the 'get_target_android_devices' call) to invoke the shell call on
        :return: The stdout result of the call
        """
        shell_command = ['shell', command]

        return self.adb_call(shell_command, device_id=device_id)

    def adb_ls(self, path, device_id, args=None):
        """
        Request an 'ls' call on the android device

        :param path:        The path to perform the 'ls' call on
        :param device_id:   device id (from the 'get_target_android_devices' call) to invoke the shell call on
        :param args:        Additional args to pass into the l'ls' call
        :return: Tuple of Boolean result of the call and the output of the call
        """
        error_messages = [
            'No such file or directory',
            'Permission denied'
        ]

        shell_command = ['ls']

        if args:
            shell_command.extend(args)

        shell_command.append(path)

        logging.debug(f"Testing {device_id}: ls {' '.join(shell_command)}")
        raw_output = self.adb_shell(command=' '.join(shell_command),
                                    device_id=device_id)

        if not raw_output:
            logging.debug('adb_ls: No output given')
            return False, None

        if raw_output is None or any([error for error in error_messages if error in raw_output]):
            logging.debug('adb_ls: Error message found')
            status = False
        else:
            logging.debug('adb_ls: Command was successful')
            status = True

        return status, raw_output

    def get_target_android_devices(self):
        """
        Gets all of the connected android devices with adb, filtered by the set optional device filter
        :return: list of serial numbers of optionally filtered connected devices.
        """

        connected_devices = []

        # Call adb to get the device list and process the raw response
        raw_devices_output = self.adb_call("devices")
        if not raw_devices_output:
            raise common.LmbrCmdError("Error getting connected devices through adb")

        device_output_list = raw_devices_output.split(os.linesep)
        for device_output in device_output_list:
            if any(x in device_output for x in ['List', '*']):
                logging.debug(f"Skipping the following line as it has 'List', '*' or 'emulator' in it: {device_output}")
                continue

            device_serial = device_output.split()
            if device_serial:
                if 'unauthorized' in device_output.lower():
                    logging.warning(f"Device {device_serial[0]} is not authorized for development access. Please reconnect the device and check for a confirmation dialog.")
                elif device_serial[0] in self.android_device_filter or not self.android_device_filter:
                    connected_devices.append(device_serial[0])
                else:
                    logging.debug(f"Skipping filtered out Device {device_serial[0]} .")

        if not connected_devices:
            raise common.LmbrCmdError("No connected android devices found")

        return connected_devices

    def check_known_android_paths(self, device_id):
        """
        Look for a known android path that is writeable and return the first one that is found

        :param device_id:   The device id (from the 'get_target_android_devices' call) to invoke the shell call on
        :return: The first available android path if found, None if not
        """
        for path in KNOWN_ANDROID_EXTERNAL_STORAGE_PATHS:
            logging.debug(f"Checking known path '{path}' on device '{device_id}'")

            # Test the path by performing an 'ls' call on it and checking if an error is returned from the result
            result, output = self.adb_ls(path=path,
                                         args=None,
                                         device_id=device_id)
            if result:
                return path[:-1]

        return None

    def detect_device_storage_path(self, device_id):
        """
        Uses the device's environment variable "EXTERNAL_STORAGE" to determine the correct
        path to public storage that has write permissions.  If at any point does the env var
        validation fail, fallback to checking known possible paths to external storage.
        :param device_id:
        :return: The first available storage device
        """

        external_storage = self.adb_shell(command="set | grep EXTERNAL_STORAGE",
                                          device_id=device_id)
        if not external_storage:
            logging.debug(f"Unable to get 'EXTERNAL_STORAGE' environment from device '{device_id}'. Falling back to known android paths.")
            return self.check_known_android_paths(device_id)

        # Given the 'EXTERNAL_STORAGE' environment, parse out the value and validate it
        storage_path_key_value = external_storage.split('=')
        if len(storage_path_key_value) != 2:
            logging.debug(f"The value for 'EXTERNAL_STORAGE' environment from device '{device_id}' does not represent a valid key-value pair: {storage_path_key_value}. Falling back to known android paths")
            return self.check_known_android_paths(device_id)
        # Check the existence and permissions issue of the storage path
        storage_path = storage_path_key_value[1].strip()
        is_external_valid, _ = self.adb_ls(path=storage_path,
                                           device_id=device_id)
        if is_external_valid:
            return storage_path

        # The set external path has an issue, attempt to determine its real path through an adb shell call
        logging.debug(f"The path specified in EXTERNAL_STORAGE seems to have permission issues, attempting to resolve with realpath for device {device_id}.")
        real_path = self.adb_shell(command=f'realpath {storage_path}',
                                   device_id=device_id)
        if not real_path:
            logging.debug(f"Unable to determine the real path '{storage_path}' (from EXTERNAL_STORAGE) for {self.game_name} on device {device_id}. Falling back to known android paths")
            return self.check_known_android_paths(device_id)
        real_path = real_path.strip()
        is_external_valid, _ = self.adb_ls(path=real_path,
                                           device_id=device_id)
        if is_external_valid:
            return real_path

        logging.debug(f'Unable to validate the resolved EXTERNAL_STORAGE environment variable path for device {device_id}.')
        return self.check_known_android_paths(device_id)

    def get_device_file_timestamp(self, remote_file_path, device_id):
        """
        Get the integer timestamp value of a file from a given device.

        :param remote_file_path:    The path to the timestamp file on the android device
        :param device_id:           The device id (from the 'get_target_android_devices' call) to invoke the shell call on
        :return:    The time value if found,  None if not
        """
        try:
            timestamp_string = self.adb_shell(command=f'cat {remote_file_path}',
                                              device_id=device_id).strip()
        except (subprocess.CalledProcessError, AttributeError):
            return None

        if not timestamp_string:
            return None

        for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M:%S.%f'):
            try:
                target_time = time.mktime(time.strptime(timestamp_string, fmt))
                break
            except ValueError:
                target_time = None

        return target_time

    def update_device_file_timestamp(self, relative_assets_path, device_id):
        """
        Update the device timestamp file on an android device to track files that need updating on pushes

        :param relative_assets_path:    The relative path to the assets on the android device
        :param device_id:               The device id (from the 'get_target_android_devices' call) to invoke the shell call on
        """
        timestamp_str = str(datetime.datetime.now())
        logging.debug(f"Updating timestamp on device {device_id} to {timestamp_str}")

        local_timestamp_file_path = self.local_asset_path / ANDROID_TARGET_TIMESTAMP_FILENAME
        local_timestamp_file_path.write_text(timestamp_str)
        target_timestamp_file_path = f'{relative_assets_path}/{ANDROID_TARGET_TIMESTAMP_FILENAME}'

        self.adb_call(arg_list=['push', str(local_timestamp_file_path), target_timestamp_file_path],
                      device_id=device_id)

    @staticmethod
    def should_copy_file(check_path, check_time):
        """
        Check if a source file should be copied, by checking if its timestamp is newer than the 'check_time'
        :param check_path:  The path to the source file whose timestamp will be evaluated
        :param check_time:  The baseline 'check_time' value to compare the source file timestamp against
        :return: True if the source file is newer than the baseline 'check_time', False if not
        """

        if not check_path.is_file():
            return False

        stat_src = check_path.stat()
        should_copy = stat_src.st_mtime >= check_time
        return should_copy

    def check_package_installed(self, package_name, target_device):
        """
        Checks if the package for the game is currently installed or not
        @param package_name:    The name of the package to search for
        @param target_device:   The target device to search for the package on
        @return: True if there an existing package on the device with the same package name, false if not
        """
        output_result = self.adb_call(['shell', 'cmd', 'package', 'list', 'packages', package_name],
                                      target_device)
        return output_result != ''

    def install_apk_to_device(self, target_device):
        """
        Install the APK to a target device
        @param target_device: The device id of the connected device to install to
        """
        if self.is_test_project:
            android_package_name = android_support.TEST_RUNNER_PACKAGE_NAME
        else:
            android_package_name = self.get_android_project_settings(key_name='package_name',
                                                                     default_value='com.lumberyard.sdk')

        if self.clean_deploy and self.check_package_installed(android_package_name, target_device):
            logging.info(f"Device '{target_device}': Uninstalling pre-existing APK for {self.game_name} ...")
            self.adb_call(arg_list=['uninstall', android_package_name],
                          device_id=target_device)

        logging.info(f"Device '{target_device}': Installing APK for {self.game_name} ...")
        self.adb_call(arg_list=['install', '-t', '-r', str(self.apk_path.resolve())],
                      device_id=target_device)

    def install_assets_to_device(self, detected_storage, target_device):
        """
        Install the assets for the game to a target device

        @param detected_storage:    The detected storage path on the target device
        @param target_device:       The ID of the target device
        """
        assert not self.is_test_project

        android_package_name = self.get_android_project_settings(key_name='package_name',
                                                                 default_value='com.lumberyard.sdk')
        relative_assets_path = f'Android/data/{android_package_name}/files'
        output_target = f'{detected_storage}/{relative_assets_path}'
        device_timestamp_file = f'{output_target}/{ANDROID_TARGET_TIMESTAMP_FILENAME}'

        # Track the current timestamp if possible to see if we can incrementally push files rather
        # than always pushing all files
        target_timestamp = self.get_device_file_timestamp(remote_file_path=device_timestamp_file,
                                                          device_id=target_device)

        if self.clean_deploy:
            logging.info(f"Device '{target_device}': Cleaning target assets before deployment...")
            self.adb_shell(command=f'rm -rf {output_target}',
                           device_id=target_device)
            logging.info(f"Device '{target_device}': Target cleaned.")

        settings_registry_src = self.build_dir / 'app/src/main/assets/Registry'
        settings_registry_dst = f'{output_target}/Registry'

        if self.clean_deploy or not target_timestamp:
            logging.info(f"Device '{target_device}': Pushing {len(self.files_in_asset_path)} files from {str(self.local_asset_path.resolve())} to device ...")
            paths_to_deploy = [(str(self.local_asset_path.resolve()), output_target),
                               (str(settings_registry_src), settings_registry_dst)]
            for path_to_deploy, target_path in paths_to_deploy:
                try:
                    self.adb_call(arg_list=['push', str(path_to_deploy), target_path],
                                  device_id=target_device)
                except subprocess.CalledProcessError as err:
                    # Something went wrong, clean up before leaving
                    self.adb_shell(command=f'rm -rf {output_target}',
                                   device_id=target_device)
                    raise err

        else:
            # If no clean was specified, individually inspect all files to see if it needs to be updated
            files_to_copy = []
            for asset_file in self.files_in_asset_path:
                # TODO: Check if the target exists in the destination as well?
                if AndroidDeployment.should_copy_file(asset_file, target_timestamp):
                    files_to_copy.append(asset_file)

            if len(files_to_copy) > 0:
                logging.info(f"Copying {len(files_to_copy)} assets to device  {target_device}")

            for src_path in files_to_copy:
                relative_path = os.path.relpath(str(src_path), str(self.local_asset_path)).replace('\\', '/')
                target_path = f"{output_target}/{relative_path}"
                self.adb_call(arg_list=['push', str(src_path), target_path],
                              device_id=target_device)

            # Always update the settings registry
            self.adb_call(arg_list=['push', str(settings_registry_src), settings_registry_dst],
                          device_id=target_device)

        self.update_device_file_timestamp(relative_assets_path=output_target,
                                          device_id=target_device)

    def execute(self):
        """
        Execute the asset deployment
        """

        if self.is_test_project:
            if not self.apk_path.is_file():
                raise common.LmbrCmdError(f"Missing apk for {android_support.TEST_RUNNER_PROJECT} ({str(self.apk_path)}). Make sure it is built and is set as a signed APK.")
        else:
            if self.embedded_assets or self.deployment_type in (AndroidDeployment.DEPLOY_APK_ONLY, AndroidDeployment.DEPLOY_BOTH):
                if not self.apk_path.is_file():
                    raise common.LmbrCmdError(f"Missing apk for game {self.game_name} ({str(self.apk_path)}). Make sure it is built and is set as a signed APK.")

            if not self.embedded_assets or self.deployment_type in (AndroidDeployment.DEPLOY_ASSETS_ONLY, AndroidDeployment.DEPLOY_BOTH):
                if not self.local_asset_path.is_dir():
                    raise common.LmbrCmdError(f"Missing {self.asset_type} assets for game {self.game_name} .")

        try:
            logging.debug("Starting ADB Server")

            self.adb_call('start-server')
            self.adb_started = True

            # Get the list of target devices to deploy to
            target_devices = self.get_target_android_devices()
            if not target_devices:
                raise common.LmbrCmdError("No connected and eligible android devices found")

            for target_device in target_devices:
                detected_storage = self.detect_device_storage_path(target_device)
                if not detected_storage:
                    logging.warning(f"Unable to resolve storage path for device '{target_device}'. Skipping.")
                    continue

                if self.is_test_project:
                    # If this is the unit test runner, then only install the APK, assets are not applicable
                    self.install_apk_to_device(target_device=target_device)
                else:
                    # Otherwise install the apk and assets based on the deployment type
                    if self.embedded_assets or self.deployment_type in (AndroidDeployment.DEPLOY_APK_ONLY, AndroidDeployment.DEPLOY_BOTH):
                        self.install_apk_to_device(target_device=target_device)

                    if not self.embedded_assets and self.deployment_type in (AndroidDeployment.DEPLOY_ASSETS_ONLY, AndroidDeployment.DEPLOY_BOTH):
                        if self.deployment_type == AndroidDeployment.DEPLOY_ASSETS_ONLY:
                            # If we are deploying assets only without an APK, make sure the APK is even installed first
                            android_package_name = self.get_android_project_settings(key_name='package_name',
                                                                                     default_value='com.lumberyard.sdk')
                            if not self.check_package_installed(package_name=android_package_name,
                                                                target_device=target_device):
                                raise common.LmbrCmdError(f"Unable to locate APK for {self.game_name} on device '{target_device}'. Make sure it is installed "
                                                          f"first before installing the assets.")

                        self.install_assets_to_device(detected_storage=detected_storage,
                                                      target_device=target_device)

                logging.info(f"{self.game_name} deployed to device {target_device}")
        finally:
            if self.adb_started:
                self.adb_call('kill-server')