diff options
author | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2022-11-18 16:34:03 +0000 |
---|---|---|
committer | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2022-11-29 14:44:13 +0000 |
commit | 37959522a805a5e23c930ed79aac84920c3cb208 (patch) | |
tree | 484af1240a93c955a72ce2e452432383b6704b56 /src/mlia/tools/metadata | |
parent | 5568f9f000d673ac53e710dcc8991fec6e8a5488 (diff) | |
download | mlia-37959522a805a5e23c930ed79aac84920c3cb208.tar.gz |
Move backends functionality into separate modules
- Move backend management/executor code into module backend_core
- Create separate module for each backend in "backend" module
- Move each backend into corresponding module
- Split Vela wrapper into several submodules
Change-Id: If01b6774aab6501951212541cc5d7f5aa7c97e95
Diffstat (limited to 'src/mlia/tools/metadata')
-rw-r--r-- | src/mlia/tools/metadata/__init__.py | 3 | ||||
-rw-r--r-- | src/mlia/tools/metadata/common.py | 322 | ||||
-rw-r--r-- | src/mlia/tools/metadata/corstone.py | 417 | ||||
-rw-r--r-- | src/mlia/tools/metadata/py_package.py | 84 |
4 files changed, 0 insertions, 826 deletions
diff --git a/src/mlia/tools/metadata/__init__.py b/src/mlia/tools/metadata/__init__.py deleted file mode 100644 index f877e4f..0000000 --- a/src/mlia/tools/metadata/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module for the tools metadata.""" diff --git a/src/mlia/tools/metadata/common.py b/src/mlia/tools/metadata/common.py deleted file mode 100644 index 5019da9..0000000 --- a/src/mlia/tools/metadata/common.py +++ /dev/null @@ -1,322 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module for installation process.""" -from __future__ import annotations - -import logging -from abc import ABC -from abc import abstractmethod -from dataclasses import dataclass -from pathlib import Path -from typing import Callable -from typing import Union - -from mlia.core.errors import ConfigurationError -from mlia.core.errors import InternalError -from mlia.utils.misc import yes - -logger = logging.getLogger(__name__) - - -@dataclass -class InstallFromPath: - """Installation from the local path.""" - - backend_path: Path - - -@dataclass -class DownloadAndInstall: - """Download and install.""" - - eula_agreement: bool = True - - -InstallationType = Union[InstallFromPath, DownloadAndInstall] - - -class Installation(ABC): - """Base class for the installation process of the backends.""" - - @property - @abstractmethod - def name(self) -> str: - """Return name of the backend.""" - - @property - @abstractmethod - def description(self) -> str: - """Return description of the backend.""" - - @property - @abstractmethod - def could_be_installed(self) -> bool: - """Return true if backend could be installed in current environment.""" - - @property - @abstractmethod - def already_installed(self) -> bool: - """Return true if backend is already installed.""" - - @abstractmethod - def supports(self, install_type: InstallationType) -> bool: - """Return true if installation supports requested installation type.""" - - @abstractmethod - def install(self, install_type: InstallationType) -> None: - """Install the backend.""" - - @abstractmethod - def uninstall(self) -> None: - """Uninstall the backend.""" - - -InstallationFilter = Callable[[Installation], bool] - - -class AlreadyInstalledFilter: - """Filter for already installed backends.""" - - def __call__(self, installation: Installation) -> bool: - """Installation filter.""" - return installation.already_installed - - -class ReadyForInstallationFilter: - """Filter for ready to be installed backends.""" - - def __call__(self, installation: Installation) -> bool: - """Installation filter.""" - return installation.could_be_installed and not installation.already_installed - - -class SupportsInstallTypeFilter: - """Filter backends that support certain type of the installation.""" - - def __init__(self, installation_type: InstallationType) -> None: - """Init filter.""" - self.installation_type = installation_type - - def __call__(self, installation: Installation) -> bool: - """Installation filter.""" - return installation.supports(self.installation_type) - - -class SearchByNameFilter: - """Filter installation by name.""" - - def __init__(self, backend_name: str | None) -> None: - """Init filter.""" - self.backend_name = backend_name - - def __call__(self, installation: Installation) -> bool: - """Installation filter.""" - return ( - not self.backend_name - or installation.name.casefold() == self.backend_name.casefold() - ) - - -class InstallationManager(ABC): - """Helper class for managing installations.""" - - @abstractmethod - def install_from(self, backend_path: Path, backend_name: str, force: bool) -> None: - """Install backend from the local directory.""" - - @abstractmethod - def download_and_install( - self, backend_name: str, eula_agreement: bool, force: bool - ) -> None: - """Download and install backends.""" - - @abstractmethod - def show_env_details(self) -> None: - """Show environment details.""" - - @abstractmethod - def backend_installed(self, backend_name: str) -> bool: - """Return true if requested backend installed.""" - - @abstractmethod - def uninstall(self, backend_name: str) -> None: - """Delete the existing installation.""" - - -class InstallationFiltersMixin: - """Mixin for filtering installation based on different conditions.""" - - installations: list[Installation] - - def filter_by(self, *filters: InstallationFilter) -> list[Installation]: - """Filter installations.""" - return [ - installation - for installation in self.installations - if all(filter_(installation) for filter_ in filters) - ] - - def find_by_name(self, backend_name: str) -> list[Installation]: - """Return list of the backends filtered by name.""" - return self.filter_by(SearchByNameFilter(backend_name)) - - def already_installed(self, backend_name: str = None) -> list[Installation]: - """Return list of backends that are already installed.""" - return self.filter_by( - AlreadyInstalledFilter(), - SearchByNameFilter(backend_name), - ) - - def ready_for_installation(self) -> list[Installation]: - """Return list of the backends that could be installed.""" - return self.filter_by(ReadyForInstallationFilter()) - - -class DefaultInstallationManager(InstallationManager, InstallationFiltersMixin): - """Interactive installation manager.""" - - def __init__( - self, installations: list[Installation], noninteractive: bool = False - ) -> None: - """Init the manager.""" - self.installations = installations - self.noninteractive = noninteractive - - def _install( - self, - backend_name: str, - install_type: InstallationType, - prompt: Callable[[Installation], str], - force: bool, - ) -> None: - """Check metadata and install backend.""" - installs = self.find_by_name(backend_name) - - if not installs: - logger.info("Unknown backend '%s'.", backend_name) - logger.info( - "Please run command 'mlia-backend list' to get list of " - "supported backend names." - ) - - return - - if len(installs) > 1: - raise InternalError(f"More than one backend with name {backend_name} found") - - installation = installs[0] - if not installation.supports(install_type): - if isinstance(install_type, InstallFromPath): - logger.info( - "Backend '%s' could not be installed using path '%s'.", - installation.name, - install_type.backend_path, - ) - logger.info( - "Please check that '%s' is a valid path to the installed backend.", - install_type.backend_path, - ) - else: - logger.info( - "Backend '%s' could not be downloaded and installed", - installation.name, - ) - logger.info( - "Please refer to the project's documentation for more details." - ) - - return - - if installation.already_installed and not force: - logger.info("Backend '%s' is already installed.", installation.name) - logger.info("Please, consider using --force option.") - return - - proceed = self.noninteractive or yes(prompt(installation)) - if not proceed: - logger.info("%s installation canceled.", installation.name) - return - - if installation.already_installed and force: - logger.info( - "Force installing %s, so delete the existing " - "installed backend first.", - installation.name, - ) - installation.uninstall() - - installation.install(install_type) - logger.info("%s successfully installed.", installation.name) - - def install_from( - self, backend_path: Path, backend_name: str, force: bool = False - ) -> None: - """Install from the provided directory.""" - - def prompt(install: Installation) -> str: - return ( - f"{install.name} was found in {backend_path}. " - "Would you like to install it?" - ) - - install_type = InstallFromPath(backend_path) - self._install(backend_name, install_type, prompt, force) - - def download_and_install( - self, backend_name: str, eula_agreement: bool = True, force: bool = False - ) -> None: - """Download and install available backends.""" - - def prompt(install: Installation) -> str: - return f"Would you like to download and install {install.name}?" - - install_type = DownloadAndInstall(eula_agreement=eula_agreement) - self._install(backend_name, install_type, prompt, force) - - def show_env_details(self) -> None: - """Print current state of the execution environment.""" - if installed := self.already_installed(): - self._print_installation_list("Installed backends:", installed) - - if could_be_installed := self.ready_for_installation(): - self._print_installation_list( - "Following backends could be installed:", - could_be_installed, - new_section=bool(installed), - ) - - if not installed and not could_be_installed: - logger.info("No backends installed") - - @staticmethod - def _print_installation_list( - header: str, installations: list[Installation], new_section: bool = False - ) -> None: - """Print list of the installations.""" - logger.info("%s%s\n", "\n" if new_section else "", header) - - for installation in installations: - logger.info(" - %s", installation.name) - - def uninstall(self, backend_name: str) -> None: - """Uninstall the backend with name backend_name.""" - installations = self.already_installed(backend_name) - - if not installations: - raise ConfigurationError(f"Backend '{backend_name}' is not installed") - - if len(installations) != 1: - raise InternalError( - f"More than one installed backend with name {backend_name} found" - ) - - installation = installations[0] - installation.uninstall() - - logger.info("%s successfully uninstalled.", installation.name) - - def backend_installed(self, backend_name: str) -> bool: - """Return true if requested backend installed.""" - installations = self.already_installed(backend_name) - - return len(installations) == 1 diff --git a/src/mlia/tools/metadata/corstone.py b/src/mlia/tools/metadata/corstone.py deleted file mode 100644 index df2dcdb..0000000 --- a/src/mlia/tools/metadata/corstone.py +++ /dev/null @@ -1,417 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module for Corstone based FVPs. - -The import of subprocess module raises a B404 bandit error. MLIA usage of -subprocess is needed and can be considered safe hence disabling the security -check. -""" -from __future__ import annotations - -import logging -import platform -import subprocess # nosec -import tarfile -from dataclasses import dataclass -from pathlib import Path -from typing import Callable -from typing import Iterable -from typing import Optional - -import mlia.backend.manager as backend_manager -from mlia.backend.system import remove_system -from mlia.tools.metadata.common import DownloadAndInstall -from mlia.tools.metadata.common import Installation -from mlia.tools.metadata.common import InstallationType -from mlia.tools.metadata.common import InstallFromPath -from mlia.utils.download import DownloadArtifact -from mlia.utils.filesystem import all_files_exist -from mlia.utils.filesystem import all_paths_valid -from mlia.utils.filesystem import copy_all -from mlia.utils.filesystem import get_mlia_resources -from mlia.utils.filesystem import temp_directory -from mlia.utils.filesystem import working_directory - - -logger = logging.getLogger(__name__) - - -@dataclass -class BackendInfo: - """Backend information.""" - - backend_path: Path - copy_source: bool = True - system_config: str | None = None - - -PathChecker = Callable[[Path], Optional[BackendInfo]] -BackendInstaller = Callable[[bool, Path], Path] - - -class BackendMetadata: - """Backend installation metadata.""" - - def __init__( - self, - name: str, - description: str, - system_config: str, - apps_resources: list[str], - fvp_dir_name: str, - download_artifact: DownloadArtifact | None, - supported_platforms: list[str] | None = None, - ) -> None: - """ - Initialize BackendMetadata. - - Members expected_systems and expected_apps are filled automatically. - """ - self.name = name - self.description = description - self.system_config = system_config - self.apps_resources = apps_resources - self.fvp_dir_name = fvp_dir_name - self.download_artifact = download_artifact - self.supported_platforms = supported_platforms - - self.expected_systems = backend_manager.get_all_system_names(name) - self.expected_apps = backend_manager.get_all_application_names(name) - - @property - def expected_resources(self) -> Iterable[Path]: - """Return list of expected resources.""" - resources = [self.system_config, *self.apps_resources] - - return (get_mlia_resources() / resource for resource in resources) - - @property - def supported_platform(self) -> bool: - """Return true if current platform supported.""" - if not self.supported_platforms: - return True - - return platform.system() in self.supported_platforms - - -class BackendInstallation(Installation): - """Backend installation.""" - - def __init__( - self, - backend_runner: backend_manager.BackendRunner, - metadata: BackendMetadata, - path_checker: PathChecker, - backend_installer: BackendInstaller | None, - ) -> None: - """Init the backend installation.""" - self.backend_runner = backend_runner - self.metadata = metadata - self.path_checker = path_checker - self.backend_installer = backend_installer - - @property - def name(self) -> str: - """Return name of the backend.""" - return self.metadata.name - - @property - def description(self) -> str: - """Return description of the backend.""" - return self.metadata.description - - @property - def already_installed(self) -> bool: - """Return true if backend already installed.""" - return self.backend_runner.all_installed( - self.metadata.expected_systems, self.metadata.expected_apps - ) - - @property - def could_be_installed(self) -> bool: - """Return true if backend could be installed.""" - if not self.metadata.supported_platform: - return False - - return all_paths_valid(self.metadata.expected_resources) - - def supports(self, install_type: InstallationType) -> bool: - """Return true if backends supported type of the installation.""" - if isinstance(install_type, DownloadAndInstall): - return self.metadata.download_artifact is not None - - if isinstance(install_type, InstallFromPath): - return self.path_checker(install_type.backend_path) is not None - - return False # type: ignore - - def install(self, install_type: InstallationType) -> None: - """Install the backend.""" - if isinstance(install_type, DownloadAndInstall): - download_artifact = self.metadata.download_artifact - assert download_artifact is not None, "No artifact provided" - - self.download_and_install(download_artifact, install_type.eula_agreement) - elif isinstance(install_type, InstallFromPath): - backend_path = self.path_checker(install_type.backend_path) - assert backend_path is not None, "Unable to resolve backend path" - - self.install_from(backend_path) - else: - raise Exception(f"Unable to install {install_type}") - - def install_from(self, backend_info: BackendInfo) -> None: - """Install backend from the directory.""" - mlia_resources = get_mlia_resources() - - with temp_directory() as tmpdir: - fvp_dist_dir = tmpdir / self.metadata.fvp_dir_name - - system_config = self.metadata.system_config - if backend_info.system_config: - system_config = backend_info.system_config - - resources_to_copy = [mlia_resources / system_config] - if backend_info.copy_source: - resources_to_copy.append(backend_info.backend_path) - - copy_all(*resources_to_copy, dest=fvp_dist_dir) - - self.backend_runner.install_system(fvp_dist_dir) - - for app in self.metadata.apps_resources: - self.backend_runner.install_application(mlia_resources / app) - - def download_and_install( - self, download_artifact: DownloadArtifact, eula_agrement: bool - ) -> None: - """Download and install the backend.""" - with temp_directory() as tmpdir: - try: - downloaded_to = download_artifact.download_to(tmpdir) - except Exception as err: - raise Exception("Unable to download backend artifact") from err - - with working_directory(tmpdir / "dist", create_dir=True) as dist_dir: - with tarfile.open(downloaded_to) as archive: - archive.extractall(dist_dir) - - assert self.backend_installer, ( - f"Backend '{self.metadata.name}' does not support " - "download and installation." - ) - backend_path = self.backend_installer(eula_agrement, dist_dir) - if self.path_checker(backend_path) is None: - raise Exception("Downloaded artifact has invalid structure") - - self.install(InstallFromPath(backend_path)) - - def uninstall(self) -> None: - """Uninstall the backend.""" - remove_system(self.metadata.fvp_dir_name) - - -class PackagePathChecker: - """Package path checker.""" - - def __init__( - self, expected_files: list[str], backend_subfolder: str | None = None - ) -> None: - """Init the path checker.""" - self.expected_files = expected_files - self.backend_subfolder = backend_subfolder - - def __call__(self, backend_path: Path) -> BackendInfo | None: - """Check if directory contains all expected files.""" - resolved_paths = (backend_path / file for file in self.expected_files) - if not all_files_exist(resolved_paths): - return None - - if self.backend_subfolder: - subfolder = backend_path / self.backend_subfolder - - if not subfolder.is_dir(): - return None - - return BackendInfo(subfolder) - - return BackendInfo(backend_path) - - -class StaticPathChecker: - """Static path checker.""" - - def __init__( - self, - static_backend_path: Path, - expected_files: list[str], - copy_source: bool = False, - system_config: str | None = None, - ) -> None: - """Init static path checker.""" - self.static_backend_path = static_backend_path - self.expected_files = expected_files - self.copy_source = copy_source - self.system_config = system_config - - def __call__(self, backend_path: Path) -> BackendInfo | None: - """Check if directory equals static backend path with all expected files.""" - if backend_path != self.static_backend_path: - return None - - resolved_paths = (backend_path / file for file in self.expected_files) - if not all_files_exist(resolved_paths): - return None - - return BackendInfo( - backend_path, - copy_source=self.copy_source, - system_config=self.system_config, - ) - - -class CompoundPathChecker: - """Compound path checker.""" - - def __init__(self, *path_checkers: PathChecker) -> None: - """Init compound path checker.""" - self.path_checkers = path_checkers - - def __call__(self, backend_path: Path) -> BackendInfo | None: - """Iterate over checkers and return first non empty backend info.""" - first_resolved_backend_info = ( - backend_info - for path_checker in self.path_checkers - if (backend_info := path_checker(backend_path)) is not None - ) - - return next(first_resolved_backend_info, None) - - -class Corstone300Installer: - """Helper class that wraps Corstone 300 installation logic.""" - - def __call__(self, eula_agreement: bool, dist_dir: Path) -> Path: - """Install Corstone-300 and return path to the models.""" - with working_directory(dist_dir): - install_dir = "corstone-300" - try: - fvp_install_cmd = [ - "./FVP_Corstone_SSE-300.sh", - "-q", - "-d", - install_dir, - ] - if not eula_agreement: - fvp_install_cmd += [ - "--nointeractive", - "--i-agree-to-the-contained-eula", - ] - - # The following line raises a B603 error for bandit. In this - # specific case, the input is pretty much static and cannot be - # changed byt the user hence disabling the security check for - # this instance - subprocess.check_call(fvp_install_cmd) # nosec - except subprocess.CalledProcessError as err: - raise Exception( - "Error occurred during Corstone-300 installation" - ) from err - - return dist_dir / install_dir - - -def get_corstone_300_installation() -> Installation: - """Get Corstone-300 installation.""" - corstone_300 = BackendInstallation( - backend_runner=backend_manager.BackendRunner(), - # pylint: disable=line-too-long - metadata=BackendMetadata( - name="Corstone-300", - description="Corstone-300 FVP", - system_config="backend_configs/systems/corstone-300/backend-config.json", - apps_resources=[], - fvp_dir_name="corstone_300", - download_artifact=DownloadArtifact( - name="Corstone-300 FVP", - url="https://developer.arm.com/-/media/Arm%20Developer%20Community/Downloads/OSS/FVP/Corstone-300/FVP_Corstone_SSE-300_11.16_26.tgz", - filename="FVP_Corstone_SSE-300_11.16_26.tgz", - version="11.16_26", - sha256_hash="e26139be756b5003a30d978c629de638aed1934d597dc24a17043d4708e934d7", - ), - supported_platforms=["Linux"], - ), - # pylint: enable=line-too-long - path_checker=CompoundPathChecker( - PackagePathChecker( - expected_files=[ - "models/Linux64_GCC-6.4/FVP_Corstone_SSE-300_Ethos-U55", - "models/Linux64_GCC-6.4/FVP_Corstone_SSE-300_Ethos-U65", - ], - backend_subfolder="models/Linux64_GCC-6.4", - ), - StaticPathChecker( - static_backend_path=Path("/opt/VHT"), - expected_files=[ - "VHT_Corstone_SSE-300_Ethos-U55", - "VHT_Corstone_SSE-300_Ethos-U65", - ], - copy_source=False, - system_config=( - "backend_configs/systems/corstone-300-vht/backend-config.json" - ), - ), - ), - backend_installer=Corstone300Installer(), - ) - - return corstone_300 - - -def get_corstone_310_installation() -> Installation: - """Get Corstone-310 installation.""" - corstone_310 = BackendInstallation( - backend_runner=backend_manager.BackendRunner(), - # pylint: disable=line-too-long - metadata=BackendMetadata( - name="Corstone-310", - description="Corstone-310 FVP", - system_config="backend_configs/systems/corstone-310/backend-config.json", - apps_resources=[], - fvp_dir_name="corstone_310", - download_artifact=None, - supported_platforms=["Linux"], - ), - # pylint: enable=line-too-long - path_checker=CompoundPathChecker( - PackagePathChecker( - expected_files=[ - "models/Linux64_GCC-9.3/FVP_Corstone_SSE-310", - "models/Linux64_GCC-9.3/FVP_Corstone_SSE-310_Ethos-U65", - ], - backend_subfolder="models/Linux64_GCC-9.3", - ), - StaticPathChecker( - static_backend_path=Path("/opt/VHT"), - expected_files=[ - "VHT_Corstone_SSE-310", - "VHT_Corstone_SSE-310_Ethos-U65", - ], - copy_source=False, - system_config=( - "backend_configs/systems/corstone-310-vht/backend-config.json" - ), - ), - ), - backend_installer=None, - ) - - return corstone_310 - - -def get_corstone_installations() -> list[Installation]: - """Get Corstone installations.""" - return [ - get_corstone_300_installation(), - get_corstone_310_installation(), - ] diff --git a/src/mlia/tools/metadata/py_package.py b/src/mlia/tools/metadata/py_package.py deleted file mode 100644 index 716b62a..0000000 --- a/src/mlia/tools/metadata/py_package.py +++ /dev/null @@ -1,84 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module for python package based installations.""" -from __future__ import annotations - -from mlia.tools.metadata.common import DownloadAndInstall -from mlia.tools.metadata.common import Installation -from mlia.tools.metadata.common import InstallationType -from mlia.utils.py_manager import get_package_manager - - -class PyPackageBackendInstallation(Installation): - """Backend based on the python package.""" - - def __init__( - self, - name: str, - description: str, - packages_to_install: list[str], - packages_to_uninstall: list[str], - expected_packages: list[str], - ) -> None: - """Init the backend installation.""" - self._name = name - self._description = description - self._packages_to_install = packages_to_install - self._packages_to_uninstall = packages_to_uninstall - self._expected_packages = expected_packages - - self.package_manager = get_package_manager() - - @property - def name(self) -> str: - """Return name of the backend.""" - return self._name - - @property - def description(self) -> str: - """Return description of the backend.""" - return self._description - - @property - def could_be_installed(self) -> bool: - """Check if backend could be installed.""" - return True - - @property - def already_installed(self) -> bool: - """Check if backend already installed.""" - return self.package_manager.packages_installed(self._expected_packages) - - def supports(self, install_type: InstallationType) -> bool: - """Return true if installation supports requested installation type.""" - return isinstance(install_type, DownloadAndInstall) - - def install(self, install_type: InstallationType) -> None: - """Install the backend.""" - if not self.supports(install_type): - raise Exception(f"Unsupported installation type {install_type}") - - self.package_manager.install(self._packages_to_install) - - def uninstall(self) -> None: - """Uninstall the backend.""" - self.package_manager.uninstall(self._packages_to_uninstall) - - -def get_tosa_backend_installation() -> Installation: - """Get TOSA backend installation.""" - return PyPackageBackendInstallation( - name="tosa-checker", - description="Tool to check if a ML model is compatible " - "with the TOSA specification", - packages_to_install=["mlia[tosa]"], - packages_to_uninstall=["tosa-checker"], - expected_packages=["tosa-checker"], - ) - - -def get_pypackage_backend_installations() -> list[Installation]: - """Return list of the backend installations based on python packages.""" - return [ - get_tosa_backend_installation(), - ] |