aboutsummaryrefslogtreecommitdiff
path: root/src/mlia/devices/ethosu
diff options
context:
space:
mode:
authorDmitrii Agibov <dmitrii.agibov@arm.com>2022-09-08 14:24:39 +0100
committerDmitrii Agibov <dmitrii.agibov@arm.com>2022-09-09 17:21:48 +0100
commitf5b293d0927506c2a979a091bf0d07ecc78fa181 (patch)
tree4de585b7cb6ed34da8237063752270189a730a41 /src/mlia/devices/ethosu
parentcde0c6ee140bd108849bff40467d8f18ffc332ef (diff)
downloadmlia-f5b293d0927506c2a979a091bf0d07ecc78fa181.tar.gz
MLIA-386 Simplify typing in the source code
- Enable deferred annotations evaluation - Use builtin types for type hints whenever possible - Use | syntax for union types - Rename mlia.core._typing into mlia.core.typing Change-Id: I3f6ffc02fa069c589bdd9e8bddbccd504285427a
Diffstat (limited to 'src/mlia/devices/ethosu')
-rw-r--r--src/mlia/devices/ethosu/advice_generation.py10
-rw-r--r--src/mlia/devices/ethosu/advisor.py32
-rw-r--r--src/mlia/devices/ethosu/config.py7
-rw-r--r--src/mlia/devices/ethosu/data_analysis.py21
-rw-r--r--src/mlia/devices/ethosu/data_collection.py18
-rw-r--r--src/mlia/devices/ethosu/handlers.py7
-rw-r--r--src/mlia/devices/ethosu/performance.py35
-rw-r--r--src/mlia/devices/ethosu/reporters.py19
8 files changed, 73 insertions, 76 deletions
diff --git a/src/mlia/devices/ethosu/advice_generation.py b/src/mlia/devices/ethosu/advice_generation.py
index 0b1352b..dee1650 100644
--- a/src/mlia/devices/ethosu/advice_generation.py
+++ b/src/mlia/devices/ethosu/advice_generation.py
@@ -1,9 +1,9 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Ethos-U advice generation."""
+from __future__ import annotations
+
from functools import singledispatchmethod
-from typing import List
-from typing import Union
from mlia.core.advice_generation import Advice
from mlia.core.advice_generation import advice_category
@@ -146,8 +146,8 @@ class EthosUAdviceProducer(FactBasedAdviceProducer):
@staticmethod
def get_next_optimization_targets(
- opt_type: List[OptimizationSettings],
- ) -> List[OptimizationSettings]:
+ opt_type: list[OptimizationSettings],
+ ) -> list[OptimizationSettings]:
"""Get next optimization targets."""
next_targets = (item.next_target() for item in opt_type)
@@ -173,7 +173,7 @@ class EthosUStaticAdviceProducer(ContextAwareAdviceProducer):
def produce_advice(self, data_item: DataItem) -> None:
"""Do not process passed data items."""
- def get_advice(self) -> Union[Advice, List[Advice]]:
+ def get_advice(self) -> Advice | list[Advice]:
"""Return predefined advice based on category."""
advice_per_category = {
AdviceCategory.PERFORMANCE: [
diff --git a/src/mlia/devices/ethosu/advisor.py b/src/mlia/devices/ethosu/advisor.py
index b7b8305..be58de7 100644
--- a/src/mlia/devices/ethosu/advisor.py
+++ b/src/mlia/devices/ethosu/advisor.py
@@ -1,14 +1,11 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Ethos-U MLIA module."""
+from __future__ import annotations
+
from pathlib import Path
from typing import Any
-from typing import Dict
-from typing import List
-from typing import Optional
-from typing import Union
-from mlia.core._typing import PathOrFileLike
from mlia.core.advice_generation import AdviceProducer
from mlia.core.advisor import DefaultInferenceAdvisor
from mlia.core.advisor import InferenceAdvisor
@@ -18,6 +15,7 @@ from mlia.core.context import ExecutionContext
from mlia.core.data_analysis import DataAnalyzer
from mlia.core.data_collection import DataCollector
from mlia.core.events import Event
+from mlia.core.typing import PathOrFileLike
from mlia.devices.ethosu.advice_generation import EthosUAdviceProducer
from mlia.devices.ethosu.advice_generation import EthosUStaticAdviceProducer
from mlia.devices.ethosu.config import EthosUConfiguration
@@ -40,13 +38,13 @@ class EthosUInferenceAdvisor(DefaultInferenceAdvisor):
"""Return name of the advisor."""
return "ethos_u_inference_advisor"
- def get_collectors(self, context: Context) -> List[DataCollector]:
+ def get_collectors(self, context: Context) -> list[DataCollector]:
"""Return list of the data collectors."""
model = self.get_model(context)
device = self._get_device(context)
backends = self._get_backends(context)
- collectors: List[DataCollector] = []
+ collectors: list[DataCollector] = []
if AdviceCategory.OPERATORS in context.advice_category:
collectors.append(EthosUOperatorCompatibility(model, device))
@@ -75,20 +73,20 @@ class EthosUInferenceAdvisor(DefaultInferenceAdvisor):
return collectors
- def get_analyzers(self, context: Context) -> List[DataAnalyzer]:
+ def get_analyzers(self, context: Context) -> list[DataAnalyzer]:
"""Return list of the data analyzers."""
return [
EthosUDataAnalyzer(),
]
- def get_producers(self, context: Context) -> List[AdviceProducer]:
+ def get_producers(self, context: Context) -> list[AdviceProducer]:
"""Return list of the advice producers."""
return [
EthosUAdviceProducer(),
EthosUStaticAdviceProducer(),
]
- def get_events(self, context: Context) -> List[Event]:
+ def get_events(self, context: Context) -> list[Event]:
"""Return list of the startup events."""
model = self.get_model(context)
device = self._get_device(context)
@@ -103,7 +101,7 @@ class EthosUInferenceAdvisor(DefaultInferenceAdvisor):
return get_target(target_profile)
- def _get_optimization_settings(self, context: Context) -> List[List[dict]]:
+ def _get_optimization_settings(self, context: Context) -> list[list[dict]]:
"""Get optimization settings."""
return self.get_parameter( # type: ignore
EthosUOptimizationPerformance.name(),
@@ -113,7 +111,7 @@ class EthosUInferenceAdvisor(DefaultInferenceAdvisor):
context=context,
)
- def _get_backends(self, context: Context) -> Optional[List[str]]:
+ def _get_backends(self, context: Context) -> list[str] | None:
"""Get list of backends."""
return self.get_parameter( # type: ignore
self.name(),
@@ -127,8 +125,8 @@ class EthosUInferenceAdvisor(DefaultInferenceAdvisor):
def configure_and_get_ethosu_advisor(
context: ExecutionContext,
target_profile: str,
- model: Union[Path, str],
- output: Optional[PathOrFileLike] = None,
+ model: str | Path,
+ output: PathOrFileLike | None = None,
**extra_args: Any,
) -> InferenceAdvisor:
"""Create and configure Ethos-U advisor."""
@@ -158,12 +156,12 @@ _DEFAULT_OPTIMIZATION_TARGETS = [
def _get_config_parameters(
- model: Union[Path, str],
+ model: str | Path,
target_profile: str,
**extra_args: Any,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
"""Get configuration parameters for the advisor."""
- advisor_parameters: Dict[str, Any] = {
+ advisor_parameters: dict[str, Any] = {
"ethos_u_inference_advisor": {
"model": model,
"target_profile": target_profile,
diff --git a/src/mlia/devices/ethosu/config.py b/src/mlia/devices/ethosu/config.py
index cecbb27..e44dcdc 100644
--- a/src/mlia/devices/ethosu/config.py
+++ b/src/mlia/devices/ethosu/config.py
@@ -1,9 +1,10 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Ethos-U configuration."""
+from __future__ import annotations
+
import logging
from typing import Any
-from typing import Dict
from mlia.devices.config import IPConfiguration
from mlia.tools.vela_wrapper import resolve_compiler_config
@@ -38,7 +39,7 @@ class EthosUConfiguration(IPConfiguration):
)
@property
- def resolved_compiler_config(self) -> Dict[str, Any]:
+ def resolved_compiler_config(self) -> dict[str, Any]:
"""Resolve compiler configuration."""
return resolve_compiler_config(self.compiler_options)
@@ -63,7 +64,7 @@ def get_target(target_profile: str) -> EthosUConfiguration:
return EthosUConfiguration(target_profile)
-def _check_target_data_complete(target_data: Dict[str, Any]) -> None:
+def _check_target_data_complete(target_data: dict[str, Any]) -> None:
"""Check if profile contains all needed data."""
mandatory_keys = {"target", "mac", "system_config", "memory_mode"}
missing_keys = sorted(mandatory_keys - target_data.keys())
diff --git a/src/mlia/devices/ethosu/data_analysis.py b/src/mlia/devices/ethosu/data_analysis.py
index 9ed32ff..8d88cf7 100644
--- a/src/mlia/devices/ethosu/data_analysis.py
+++ b/src/mlia/devices/ethosu/data_analysis.py
@@ -1,11 +1,10 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Ethos-U data analysis module."""
+from __future__ import annotations
+
from dataclasses import dataclass
from functools import singledispatchmethod
-from typing import Dict
-from typing import List
-from typing import Union
from mlia.core.common import DataItem
from mlia.core.data_analysis import Fact
@@ -19,7 +18,7 @@ from mlia.tools.vela_wrapper import Operators
class HasCPUOnlyOperators(Fact):
"""Model has CPU only operators."""
- cpu_only_ops: List[str]
+ cpu_only_ops: list[str]
@dataclass
@@ -38,8 +37,8 @@ class AllOperatorsSupportedOnNPU(Fact):
class PerfMetricDiff:
"""Performance metric difference."""
- original_value: Union[int, float]
- optimized_value: Union[int, float]
+ original_value: int | float
+ optimized_value: int | float
@property
def diff(self) -> float:
@@ -69,15 +68,15 @@ class PerfMetricDiff:
class OptimizationDiff:
"""Optimization performance impact."""
- opt_type: List[OptimizationSettings]
- opt_diffs: Dict[str, PerfMetricDiff]
+ opt_type: list[OptimizationSettings]
+ opt_diffs: dict[str, PerfMetricDiff]
@dataclass
class OptimizationResults(Fact):
"""Optimization results."""
- diffs: List[OptimizationDiff]
+ diffs: list[OptimizationDiff]
class EthosUDataAnalyzer(FactExtractor):
@@ -113,13 +112,13 @@ class EthosUDataAnalyzer(FactExtractor):
orig_memory = orig.memory_usage
orig_cycles = orig.npu_cycles
- diffs: List[OptimizationDiff] = []
+ diffs: list[OptimizationDiff] = []
for opt_type, opt_perf_metrics in optimizations:
opt = opt_perf_metrics.in_kilobytes()
opt_memory = opt.memory_usage
opt_cycles = opt.npu_cycles
- opt_diffs: Dict[str, PerfMetricDiff] = {}
+ opt_diffs: dict[str, PerfMetricDiff] = {}
if orig_memory and opt_memory:
opt_diffs.update(
diff --git a/src/mlia/devices/ethosu/data_collection.py b/src/mlia/devices/ethosu/data_collection.py
index 291f1b8..6ddebac 100644
--- a/src/mlia/devices/ethosu/data_collection.py
+++ b/src/mlia/devices/ethosu/data_collection.py
@@ -1,10 +1,10 @@
# 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 typing import List
-from typing import Optional
from mlia.core.context import Context
from mlia.core.data_collection import ContextAwareDataCollector
@@ -59,7 +59,7 @@ class EthosUPerformance(ContextAwareDataCollector):
self,
model: Path,
device: EthosUConfiguration,
- backends: Optional[List[str]] = None,
+ backends: list[str] | None = None,
) -> None:
"""Init performance data collector."""
self.model = model
@@ -87,7 +87,7 @@ class OptimizeModel:
"""Helper class for model optimization."""
def __init__(
- self, context: Context, opt_settings: List[OptimizationSettings]
+ self, context: Context, opt_settings: list[OptimizationSettings]
) -> None:
"""Init helper."""
self.context = context
@@ -115,8 +115,8 @@ class EthosUOptimizationPerformance(ContextAwareDataCollector):
self,
model: Path,
device: EthosUConfiguration,
- optimizations: List[List[dict]],
- backends: Optional[List[str]] = None,
+ optimizations: list[list[dict]],
+ backends: list[str] | None = None,
) -> None:
"""Init performance optimizations data collector."""
self.model = model
@@ -124,7 +124,7 @@ class EthosUOptimizationPerformance(ContextAwareDataCollector):
self.optimizations = optimizations
self.backends = backends
- def collect_data(self) -> Optional[OptimizationPerformanceMetrics]:
+ def collect_data(self) -> OptimizationPerformanceMetrics | None:
"""Collect performance metrics for the optimizations."""
logger.info("Estimate performance ...")
@@ -164,8 +164,8 @@ class EthosUOptimizationPerformance(ContextAwareDataCollector):
@staticmethod
def _parse_optimization_params(
- optimizations: List[List[dict]],
- ) -> List[List[OptimizationSettings]]:
+ 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")
diff --git a/src/mlia/devices/ethosu/handlers.py b/src/mlia/devices/ethosu/handlers.py
index ee0b809..48f9a2e 100644
--- a/src/mlia/devices/ethosu/handlers.py
+++ b/src/mlia/devices/ethosu/handlers.py
@@ -1,12 +1,13 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Event handler."""
+from __future__ import annotations
+
import logging
-from typing import Optional
-from mlia.core._typing import PathOrFileLike
from mlia.core.events import CollectedDataEvent
from mlia.core.handlers import WorkflowEventsHandler
+from mlia.core.typing import PathOrFileLike
from mlia.devices.ethosu.events import EthosUAdvisorEventHandler
from mlia.devices.ethosu.events import EthosUAdvisorStartedEvent
from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics
@@ -20,7 +21,7 @@ logger = logging.getLogger(__name__)
class EthosUEventHandler(WorkflowEventsHandler, EthosUAdvisorEventHandler):
"""CLI event handler."""
- def __init__(self, output: Optional[PathOrFileLike] = None) -> None:
+ def __init__(self, output: PathOrFileLike | None = None) -> None:
"""Init event handler."""
super().__init__(ethos_u_formatters, output)
diff --git a/src/mlia/devices/ethosu/performance.py b/src/mlia/devices/ethosu/performance.py
index a73045a..e89a65a 100644
--- a/src/mlia/devices/ethosu/performance.py
+++ b/src/mlia/devices/ethosu/performance.py
@@ -1,13 +1,12 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Performance estimation."""
+from __future__ import annotations
+
import logging
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
-from typing import List
-from typing import Optional
-from typing import Tuple
from typing import Union
import mlia.backend.manager as backend_manager
@@ -49,11 +48,11 @@ class MemorySizeType(Enum):
class MemoryUsage:
"""Memory usage metrics."""
- sram_memory_area_size: Union[int, float]
- dram_memory_area_size: Union[int, float]
- unknown_memory_area_size: Union[int, float]
- on_chip_flash_memory_area_size: Union[int, float]
- off_chip_flash_memory_area_size: Union[int, float]
+ sram_memory_area_size: int | float
+ dram_memory_area_size: int | float
+ unknown_memory_area_size: int | float
+ on_chip_flash_memory_area_size: int | float
+ off_chip_flash_memory_area_size: int | float
memory_size_type: MemorySizeType = MemorySizeType.BYTES
_default_columns = [
@@ -64,7 +63,7 @@ class MemoryUsage:
"Off chip flash used",
]
- def in_kilobytes(self) -> "MemoryUsage":
+ def in_kilobytes(self) -> MemoryUsage:
"""Return memory usage with values in kilobytes."""
if self.memory_size_type == MemorySizeType.KILOBYTES:
return self
@@ -91,10 +90,10 @@ class PerformanceMetrics:
"""Performance metrics."""
device: EthosUConfiguration
- npu_cycles: Optional[NPUCycles]
- memory_usage: Optional[MemoryUsage]
+ npu_cycles: NPUCycles | None
+ memory_usage: MemoryUsage | None
- def in_kilobytes(self) -> "PerformanceMetrics":
+ def in_kilobytes(self) -> PerformanceMetrics:
"""Return metrics with memory usage in KiB."""
if self.memory_usage is None:
return PerformanceMetrics(self.device, self.npu_cycles, self.memory_usage)
@@ -109,8 +108,8 @@ class OptimizationPerformanceMetrics:
"""Optimization performance metrics."""
original_perf_metrics: PerformanceMetrics
- optimizations_perf_metrics: List[
- Tuple[List[OptimizationSettings], PerformanceMetrics]
+ optimizations_perf_metrics: list[
+ tuple[list[OptimizationSettings], PerformanceMetrics]
]
@@ -124,7 +123,7 @@ class VelaPerformanceEstimator(
self.context = context
self.device = device
- def estimate(self, model: Union[Path, ModelConfiguration]) -> MemoryUsage:
+ def estimate(self, model: Path | ModelConfiguration) -> MemoryUsage:
"""Estimate performance."""
logger.info("Getting the memory usage metrics ...")
@@ -160,7 +159,7 @@ class CorstonePerformanceEstimator(
self.device = device
self.backend = backend
- def estimate(self, model: Union[Path, ModelConfiguration]) -> NPUCycles:
+ def estimate(self, model: Path | ModelConfiguration) -> NPUCycles:
"""Estimate performance."""
logger.info("Getting the performance metrics for '%s' ...", self.backend)
logger.info(
@@ -212,7 +211,7 @@ class EthosUPerformanceEstimator(
self,
context: Context,
device: EthosUConfiguration,
- backends: Optional[List[str]] = None,
+ backends: list[str] | None = None,
) -> None:
"""Init performance estimator."""
self.context = context
@@ -228,7 +227,7 @@ class EthosUPerformanceEstimator(
)
self.backends = set(backends)
- def estimate(self, model: Union[Path, ModelConfiguration]) -> PerformanceMetrics:
+ def estimate(self, model: Path | ModelConfiguration) -> PerformanceMetrics:
"""Estimate performance."""
model_path = (
Path(model.model_path) if isinstance(model, ModelConfiguration) else model
diff --git a/src/mlia/devices/ethosu/reporters.py b/src/mlia/devices/ethosu/reporters.py
index b3aea24..f11430c 100644
--- a/src/mlia/devices/ethosu/reporters.py
+++ b/src/mlia/devices/ethosu/reporters.py
@@ -1,12 +1,11 @@
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Reports module."""
+from __future__ import annotations
+
from collections import defaultdict
from typing import Any
from typing import Callable
-from typing import List
-from typing import Tuple
-from typing import Union
from mlia.core.advice_generation import Advice
from mlia.core.reporting import BytesCell
@@ -52,7 +51,7 @@ def report_operators_stat(operators: Operators) -> Report:
)
-def report_operators(ops: List[Operator]) -> Report:
+def report_operators(ops: list[Operator]) -> Report:
"""Return table representation for the list of operators."""
columns = [
Column("#", only_for=["plain_text"]),
@@ -235,11 +234,11 @@ def report_device_details(device: EthosUConfiguration) -> Report:
)
-def metrics_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
+def metrics_as_records(perf_metrics: list[PerformanceMetrics]) -> list[tuple]:
"""Convert perf metrics object into list of records."""
perf_metrics = [item.in_kilobytes() for item in perf_metrics]
- def _cycles_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
+ def _cycles_as_records(perf_metrics: list[PerformanceMetrics]) -> list[tuple]:
metric_map = defaultdict(list)
for metrics in perf_metrics:
if not metrics.npu_cycles:
@@ -253,7 +252,7 @@ def metrics_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
for name, values in metric_map.items()
]
- def _memory_usage_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
+ def _memory_usage_as_records(perf_metrics: list[PerformanceMetrics]) -> list[tuple]:
metric_map = defaultdict(list)
for metrics in perf_metrics:
if not metrics.memory_usage:
@@ -276,7 +275,7 @@ def metrics_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
if all(val > 0 for val in values)
]
- def _data_beats_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
+ def _data_beats_as_records(perf_metrics: list[PerformanceMetrics]) -> list[tuple]:
metric_map = defaultdict(list)
for metrics in perf_metrics:
if not metrics.npu_cycles:
@@ -308,7 +307,7 @@ def metrics_as_records(perf_metrics: List[PerformanceMetrics]) -> List[Tuple]:
def report_perf_metrics(
- perf_metrics: Union[PerformanceMetrics, List[PerformanceMetrics]]
+ perf_metrics: PerformanceMetrics | list[PerformanceMetrics],
) -> Report:
"""Return comparison table for the performance metrics."""
if isinstance(perf_metrics, PerformanceMetrics):
@@ -361,7 +360,7 @@ def report_perf_metrics(
)
-def report_advice(advice: List[Advice]) -> Report:
+def report_advice(advice: list[Advice]) -> Report:
"""Generate report for the advice."""
return Table(
columns=[