aboutsummaryrefslogtreecommitdiff
path: root/src/mlia/devices/ethosu/data_collection.py
diff options
context:
space:
mode:
authorDmitrii Agibov <dmitrii.agibov@arm.com>2022-11-18 17:21:09 +0000
committerDmitrii Agibov <dmitrii.agibov@arm.com>2022-11-29 14:44:13 +0000
commit6a88ee5315b4ce5b023370c1e55e48bf9f2b6f67 (patch)
tree88edabf90228724f4fe2944b0ab23859d824a880 /src/mlia/devices/ethosu/data_collection.py
parenta34163c9d9a5cc0416bcaea2ebf8383bda9d505c (diff)
downloadmlia-6a88ee5315b4ce5b023370c1e55e48bf9f2b6f67.tar.gz
Rename modules
- Rename module "mlia.devices" into "mlia.target" - Rename module "mlia.target.ethosu" into "mlia.target.ethos_u" - Rename module "mlia.target.cortexa" into "mlia.target.cortex_a" - Rename and update tests Change-Id: I6dca7c8646d881f739fb6b5914d1cc7e45e63dc2
Diffstat (limited to 'src/mlia/devices/ethosu/data_collection.py')
-rw-r--r--src/mlia/devices/ethosu/data_collection.py187
1 files changed, 0 insertions, 187 deletions
diff --git a/src/mlia/devices/ethosu/data_collection.py b/src/mlia/devices/ethosu/data_collection.py
deleted file mode 100644
index d68eadb..0000000
--- a/src/mlia/devices/ethosu/data_collection.py
+++ /dev/null
@@ -1,187 +0,0 @@
-# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
-# SPDX-License-Identifier: Apache-2.0
-"""Data collection module for Ethos-U."""
-from __future__ import annotations
-
-import logging
-from pathlib import Path
-
-from mlia.backend.vela.compat import Operators
-from mlia.backend.vela.compat import supported_operators
-from mlia.core.context import Context
-from mlia.core.data_collection import ContextAwareDataCollector
-from mlia.core.errors import FunctionalityNotSupportedError
-from mlia.core.performance import estimate_performance
-from mlia.devices.ethosu.config import EthosUConfiguration
-from mlia.devices.ethosu.performance import EthosUPerformanceEstimator
-from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics
-from mlia.devices.ethosu.performance import PerformanceMetrics
-from mlia.nn.tensorflow.config import get_keras_model
-from mlia.nn.tensorflow.config import get_tflite_model
-from mlia.nn.tensorflow.config import KerasModel
-from mlia.nn.tensorflow.optimizations.select import get_optimizer
-from mlia.nn.tensorflow.optimizations.select import OptimizationSettings
-from mlia.nn.tensorflow.utils import save_keras_model
-from mlia.utils.logging import log_action
-from mlia.utils.types import is_list_of
-
-logger = logging.getLogger(__name__)
-
-
-class EthosUOperatorCompatibility(ContextAwareDataCollector):
- """Collect operator compatibility information."""
-
- def __init__(self, model: Path, device: EthosUConfiguration) -> None:
- """Init operator compatibility data collector."""
- self.model = model
- self.device = device
-
- def collect_data(self) -> Operators:
- """Collect operator compatibility information."""
- tflite_model = get_tflite_model(self.model, self.context)
-
- with log_action("Checking operator compatibility ..."):
- return supported_operators(
- Path(tflite_model.model_path), self.device.compiler_options
- )
-
- @classmethod
- def name(cls) -> str:
- """Return name of the collector."""
- return "ethos_u_operator_compatibility"
-
-
-class EthosUPerformance(ContextAwareDataCollector):
- """Collect performance metrics."""
-
- def __init__(
- self,
- model: Path,
- device: EthosUConfiguration,
- backends: list[str] | None = None,
- ) -> None:
- """Init performance data collector."""
- self.model = model
- self.device = device
- self.backends = backends
-
- def collect_data(self) -> PerformanceMetrics:
- """Collect model performance metrics."""
- tflite_model = get_tflite_model(self.model, self.context)
- estimator = EthosUPerformanceEstimator(
- self.context,
- self.device,
- self.backends,
- )
-
- return estimator.estimate(tflite_model)
-
- @classmethod
- def name(cls) -> str:
- """Return name of the collector."""
- return "ethos_u_performance"
-
-
-class OptimizeModel:
- """Helper class for model optimization."""
-
- def __init__(
- self, context: Context, opt_settings: list[OptimizationSettings]
- ) -> None:
- """Init helper."""
- self.context = context
- self.opt_settings = opt_settings
-
- def __call__(self, keras_model: KerasModel) -> KerasModel:
- """Run optimization."""
- optimizer = get_optimizer(keras_model, self.opt_settings)
-
- opts_as_str = ", ".join(str(opt) for opt in self.opt_settings)
- logger.info("Applying model optimizations - [%s]", opts_as_str)
- optimizer.apply_optimization()
-
- model = optimizer.get_model()
- model_path = self.context.get_model_path("optimized_model.h5")
- save_keras_model(model, model_path)
-
- return KerasModel(model_path)
-
-
-class EthosUOptimizationPerformance(ContextAwareDataCollector):
- """Collect performance metrics for the optimizations."""
-
- def __init__(
- self,
- model: Path,
- device: EthosUConfiguration,
- optimizations: list[list[dict]],
- backends: list[str] | None = None,
- ) -> None:
- """Init performance optimizations data collector."""
- self.model = model
- self.device = device
- self.optimizations = optimizations
- self.backends = backends
-
- def collect_data(self) -> OptimizationPerformanceMetrics | None:
- """Collect performance metrics for the optimizations."""
- logger.info("Estimate performance ...")
-
- if not self.optimizations:
- raise FunctionalityNotSupportedError(
- reason="Unable to estimate model optimizations impact",
- description="No optimization targets provided",
- )
-
- opt_settings = self._parse_optimization_params(self.optimizations)
-
- try:
- keras_model = get_keras_model(self.model, self.context)
- except NotImplementedError as err:
- raise FunctionalityNotSupportedError(
- reason="Unable to run model optimizations",
- description=f"{self.model} is not a Keras model and "
- "could not be converted to a Keras model",
- ) from err
-
- optimizers = [OptimizeModel(self.context, opts) for opts in opt_settings]
-
- estimator = EthosUPerformanceEstimator(
- self.context,
- self.device,
- self.backends,
- )
- original_metrics, *optimized_metrics = estimate_performance(
- keras_model, estimator, optimizers # type: ignore
- )
-
- result = OptimizationPerformanceMetrics(
- original_perf_metrics=original_metrics,
- optimizations_perf_metrics=list(zip(opt_settings, optimized_metrics)),
- )
- return result
-
- @staticmethod
- def _parse_optimization_params(
- optimizations: list[list[dict]],
- ) -> list[list[OptimizationSettings]]:
- """Parse optimization parameters."""
- if not is_list_of(optimizations, list):
- raise Exception("Optimization parameters expected to be a list")
-
- return [
- [
- OptimizationSettings(
- item.get("optimization_type"), # type: ignore
- item.get("optimization_target"), # type: ignore
- item.get("layers_to_optimized"),
- )
- for item in opt_configuration
- ]
- for opt_configuration in optimizations
- ]
-
- @classmethod
- def name(cls) -> str:
- """Return name of the collector."""
- return "ethos_u_model_optimizations"