aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorDmitrii Agibov <dmitrii.agibov@arm.com>2022-11-15 13:19:53 +0000
committerDmitrii Agibov <dmitrii.agibov@arm.com>2022-11-16 09:42:15 +0000
commit302ce432829ae7c25e100a5cca718f0aadbe4fd4 (patch)
tree96e31bc88f795752ecaa6a9672fe512f8b0d9041 /src
parent47fc50576e7040680c19e152592b2c5e5cc297f5 (diff)
downloadmlia-302ce432829ae7c25e100a5cca718f0aadbe4fd4.tar.gz
MLIA-649 Support tosa-checker as a backend
- Add new type of the backend based on python packages - Add installation class for TOSA checker - Update documentation - Extend support of the parameter "force" in the "install" command Change-Id: I95567b75e1cfe85daa1f1c3d359975bb67b2504e
Diffstat (limited to 'src')
-rw-r--r--src/mlia/cli/commands.py23
-rw-r--r--src/mlia/cli/config.py3
-rw-r--r--src/mlia/cli/main.py6
-rw-r--r--src/mlia/cli/options.py2
-rw-r--r--src/mlia/core/errors.py4
-rw-r--r--src/mlia/devices/tosa/operators.py3
-rw-r--r--src/mlia/tools/metadata/common.py173
-rw-r--r--src/mlia/tools/metadata/corstone.py1
-rw-r--r--src/mlia/tools/metadata/py_package.py84
-rw-r--r--src/mlia/utils/py_manager.py62
10 files changed, 256 insertions, 105 deletions
diff --git a/src/mlia/cli/commands.py b/src/mlia/cli/commands.py
index 4be7f3e..09fe9de 100644
--- a/src/mlia/cli/commands.py
+++ b/src/mlia/cli/commands.py
@@ -20,7 +20,6 @@ from __future__ import annotations
import logging
from pathlib import Path
-from typing import cast
from mlia.api import ExecutionContext
from mlia.api import generate_supported_operators_report
@@ -249,29 +248,29 @@ def backend_install(
noninteractive: bool = False,
force: bool = False,
) -> None:
- """Install configuration."""
+ """Install backend."""
logger.info(CONFIG)
manager = get_installation_manager(noninteractive)
- install_from_path = path is not None
-
- if install_from_path:
- manager.install_from(cast(Path, path), name, force)
+ if path is not None:
+ manager.install_from(path, name, force)
else:
eula_agreement = not i_agree_to_the_contained_eula
- manager.download_and_install(name, eula_agreement)
+ manager.download_and_install(name, eula_agreement, force)
-def backend_uninstall(
- name: str,
-) -> None:
- """Uninstall backend(s)."""
+def backend_uninstall(name: str) -> None:
+ """Uninstall backend."""
+ logger.info(CONFIG)
+
manager = get_installation_manager(noninteractive=True)
manager.uninstall(name)
def backend_list() -> None:
- """List backend status."""
+ """List backends status."""
+ logger.info(CONFIG)
+
manager = get_installation_manager(noninteractive=True)
manager.show_env_details()
diff --git a/src/mlia/cli/config.py b/src/mlia/cli/config.py
index 30373e4..6ea9bb4 100644
--- a/src/mlia/cli/config.py
+++ b/src/mlia/cli/config.py
@@ -10,13 +10,14 @@ import mlia.backend.manager as backend_manager
from mlia.tools.metadata.common import DefaultInstallationManager
from mlia.tools.metadata.common import InstallationManager
from mlia.tools.metadata.corstone import get_corstone_installations
+from mlia.tools.metadata.py_package import get_pypackage_backend_installations
logger = logging.getLogger(__name__)
def get_installation_manager(noninteractive: bool = False) -> InstallationManager:
"""Return installation manager."""
- backends = get_corstone_installations()
+ backends = get_corstone_installations() + get_pypackage_backend_installations()
return DefaultInstallationManager(backends, noninteractive=noninteractive)
diff --git a/src/mlia/cli/main.py b/src/mlia/cli/main.py
index 61b8f05..6c74a11 100644
--- a/src/mlia/cli/main.py
+++ b/src/mlia/cli/main.py
@@ -33,6 +33,8 @@ from mlia.cli.options import add_output_options
from mlia.cli.options import add_target_options
from mlia.cli.options import add_tflite_model_options
from mlia.core.context import ExecutionContext
+from mlia.core.errors import ConfigurationError
+from mlia.core.errors import InternalError
logger = logging.getLogger(__name__)
@@ -219,6 +221,10 @@ def run_command(args: argparse.Namespace) -> int:
return 0
except KeyboardInterrupt:
logger.error("Execution has been interrupted")
+ except InternalError as err:
+ logger.error("Internal error: %s", err)
+ except ConfigurationError as err:
+ logger.error(err)
except Exception as err: # pylint: disable=broad-except
logger.error(
"\nExecution finished with error: %s",
diff --git a/src/mlia/cli/options.py b/src/mlia/cli/options.py
index bf2f09b..5eab9aa 100644
--- a/src/mlia/cli/options.py
+++ b/src/mlia/cli/options.py
@@ -154,7 +154,7 @@ def add_backend_install_options(parser: argparse.ArgumentParser) -> None:
"--force",
default=False,
action="store_true",
- help="Force reinstall backend in the specified path",
+ help="Force reinstalling backend in the specified path",
)
parser.add_argument(
"--noninteractive",
diff --git a/src/mlia/core/errors.py b/src/mlia/core/errors.py
index 7d6beb1..d2356c2 100644
--- a/src/mlia/core/errors.py
+++ b/src/mlia/core/errors.py
@@ -7,6 +7,10 @@ class ConfigurationError(Exception):
"""Configuration error."""
+class InternalError(Exception):
+ """Internal error."""
+
+
class FunctionalityNotSupportedError(Exception):
"""Functionality is not supported error."""
diff --git a/src/mlia/devices/tosa/operators.py b/src/mlia/devices/tosa/operators.py
index 03f6fb8..1e4581a 100644
--- a/src/mlia/devices/tosa/operators.py
+++ b/src/mlia/devices/tosa/operators.py
@@ -47,8 +47,7 @@ def get_tosa_compatibility_info(
if checker is None:
raise Exception(
"TOSA checker is not available. "
- "Please make sure that 'tosa_checker' package is installed: "
- "pip install mlia[tosa]"
+ "Please make sure that 'tosa-checker' backend is installed."
)
ops = [
diff --git a/src/mlia/tools/metadata/common.py b/src/mlia/tools/metadata/common.py
index 927be74..5019da9 100644
--- a/src/mlia/tools/metadata/common.py
+++ b/src/mlia/tools/metadata/common.py
@@ -11,9 +11,10 @@ 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__)
@@ -124,7 +125,9 @@ class InstallationManager(ABC):
"""Install backend from the local directory."""
@abstractmethod
- def download_and_install(self, backend_name: str, eula_agreement: bool) -> None:
+ def download_and_install(
+ self, backend_name: str, eula_agreement: bool, force: bool
+ ) -> None:
"""Download and install backends."""
@abstractmethod
@@ -153,29 +156,15 @@ class InstallationFiltersMixin:
if all(filter_(installation) for filter_ in filters)
]
- def could_be_installed_from(
- self, backend_path: Path, backend_name: str
- ) -> list[Installation]:
- """Return installations that could be installed from provided directory."""
- return self.filter_by(
- SupportsInstallTypeFilter(InstallFromPath(backend_path)),
- SearchByNameFilter(backend_name),
- )
-
- def could_be_downloaded_and_installed(
- self, backend_name: str
- ) -> list[Installation]:
- """Return installations that could be downloaded and installed."""
- return self.filter_by(
- SupportsInstallTypeFilter(DownloadAndInstall()),
- SearchByNameFilter(backend_name),
- ReadyForInstallationFilter(),
- )
+ 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)
+ AlreadyInstalledFilter(),
+ SearchByNameFilter(backend_name),
)
def ready_for_installation(self) -> list[Installation]:
@@ -193,83 +182,96 @@ class DefaultInstallationManager(InstallationManager, InstallationFiltersMixin):
self.installations = installations
self.noninteractive = noninteractive
- def choose_installation_for_path(
- self, backend_path: Path, backend_name: str, force: bool
- ) -> Installation | None:
- """Check available installation and select one if possible."""
- installs = self.could_be_installed_from(backend_path, backend_name)
+ 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(
- "Unfortunatelly, it was not possible to automatically "
- "detect type of the installed FVP. "
- "Please, check provided path to the installed FVP."
+ "Please run command 'mlia-backend list' to get list of "
+ "supported backend names."
)
- return None
- if len(installs) != 1:
- names = ",".join(install.name for install in installs)
- logger.info(
- "Unable to correctly detect type of the installed FVP."
- "The following FVPs are detected %s. Installation skipped.",
- names,
- )
- return None
+ return
+
+ if len(installs) > 1:
+ raise InternalError(f"More than one backend with name {backend_name} found")
installation = installs[0]
- if installation.already_installed:
+ 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(
- "%s was found in %s, but it has been already installed "
- "in the ML Inference Advisor.",
+ "Force installing %s, so delete the existing "
+ "installed backend first.",
installation.name,
- backend_path,
)
- return installation if force else None
+ installation.uninstall()
- return installation
+ 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."""
- installation = self.choose_installation_for_path(
- backend_path, backend_name, force
- )
-
- if not installation:
- return
- if force:
- self.uninstall(backend_name)
- logger.info(
- "Force installing %s, so delete the existing installed backend first.",
- installation.name,
+ def prompt(install: Installation) -> str:
+ return (
+ f"{install.name} was found in {backend_path}. "
+ "Would you like to install it?"
)
- prompt = (
- f"{installation.name} was found in {backend_path}. "
- "Would you like to install it?"
- )
- self._install(installation, InstallFromPath(backend_path), prompt)
+ 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
+ self, backend_name: str, eula_agreement: bool = True, force: bool = False
) -> None:
"""Download and install available backends."""
- installations = self.could_be_downloaded_and_installed(backend_name)
- if not installations:
- logger.info("No backends available for the installation.")
- return
+ def prompt(install: Installation) -> str:
+ return f"Would you like to download and install {install.name}?"
- names = ",".join(installation.name for installation in installations)
- logger.info("Following backends are available for downloading: %s", names)
-
- for installation in installations:
- prompt = f"Would you like to download and install {installation.name}?"
- self._install(
- installation, DownloadAndInstall(eula_agreement=eula_agreement), prompt
- )
+ 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."""
@@ -299,24 +301,19 @@ class DefaultInstallationManager(InstallationManager, InstallationFiltersMixin):
def uninstall(self, backend_name: str) -> None:
"""Uninstall the backend with name backend_name."""
installations = self.already_installed(backend_name)
+
if not installations:
- raise Exception("No backend available for uninstall")
- for installation in installations:
- installation.uninstall()
+ raise ConfigurationError(f"Backend '{backend_name}' is not installed")
- def _install(
- self,
- installation: Installation,
- installation_type: InstallationType,
- prompt: str,
- ) -> None:
- proceed = self.noninteractive or yes(prompt)
+ if len(installations) != 1:
+ raise InternalError(
+ f"More than one installed backend with name {backend_name} found"
+ )
- if proceed:
- installation.install(installation_type)
- logger.info("%s successfully installed.", installation.name)
- else:
- logger.info("%s installation canceled.", installation.name)
+ 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."""
diff --git a/src/mlia/tools/metadata/corstone.py b/src/mlia/tools/metadata/corstone.py
index 04b13b5..df2dcdb 100644
--- a/src/mlia/tools/metadata/corstone.py
+++ b/src/mlia/tools/metadata/corstone.py
@@ -209,7 +209,6 @@ class BackendInstallation(Installation):
def uninstall(self) -> None:
"""Uninstall the backend."""
remove_system(self.metadata.fvp_dir_name)
- logger.info("%s successfully uninstalled.", self.name)
class PackagePathChecker:
diff --git a/src/mlia/tools/metadata/py_package.py b/src/mlia/tools/metadata/py_package.py
new file mode 100644
index 0000000..716b62a
--- /dev/null
+++ b/src/mlia/tools/metadata/py_package.py
@@ -0,0 +1,84 @@
+# 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(),
+ ]
diff --git a/src/mlia/utils/py_manager.py b/src/mlia/utils/py_manager.py
new file mode 100644
index 0000000..5f98fcc
--- /dev/null
+++ b/src/mlia/utils/py_manager.py
@@ -0,0 +1,62 @@
+# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
+# SPDX-License-Identifier: Apache-2.0
+"""Util functions for managing python packages."""
+from __future__ import annotations
+
+import sys
+from importlib.metadata import distribution
+from importlib.metadata import PackageNotFoundError
+from subprocess import check_call # nosec
+
+
+class PyPackageManager:
+ """Python package manager."""
+
+ @staticmethod
+ def package_installed(pkg_name: str) -> bool:
+ """Return true if package installed."""
+ try:
+ distribution(pkg_name)
+ except PackageNotFoundError:
+ return False
+
+ return True
+
+ def packages_installed(self, pkg_names: list[str]) -> bool:
+ """Return true if all provided packages installed."""
+ return all(self.package_installed(pkg) for pkg in pkg_names)
+
+ def install(self, pkg_names: list[str]) -> None:
+ """Install provided packages."""
+ if not pkg_names:
+ raise ValueError("No package names provided")
+
+ self._execute_pip_cmd("install", pkg_names)
+
+ def uninstall(self, pkg_names: list[str]) -> None:
+ """Uninstall provided packages."""
+ if not pkg_names:
+ raise ValueError("No package names provided")
+
+ self._execute_pip_cmd("uninstall", ["--yes", *pkg_names])
+
+ @staticmethod
+ def _execute_pip_cmd(subcommand: str, params: list[str]) -> None:
+ """Execute pip command."""
+ assert sys.executable, "Unable to launch pip command"
+
+ check_call(
+ [
+ sys.executable,
+ "-m",
+ "pip",
+ "--disable-pip-version-check",
+ subcommand,
+ *params,
+ ]
+ )
+
+
+def get_package_manager() -> PyPackageManager:
+ """Get python packages manager."""
+ return PyPackageManager()