From f5b293d0927506c2a979a091bf0d07ecc78fa181 Mon Sep 17 00:00:00 2001 From: Dmitrii Agibov Date: Thu, 8 Sep 2022 14:24:39 +0100 Subject: 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 --- src/mlia/devices/ethosu/advice_generation.py | 10 ++++---- src/mlia/devices/ethosu/advisor.py | 32 ++++++++++++------------- src/mlia/devices/ethosu/config.py | 7 +++--- src/mlia/devices/ethosu/data_analysis.py | 21 ++++++++--------- src/mlia/devices/ethosu/data_collection.py | 18 +++++++------- src/mlia/devices/ethosu/handlers.py | 7 +++--- src/mlia/devices/ethosu/performance.py | 35 ++++++++++++++-------------- src/mlia/devices/ethosu/reporters.py | 19 +++++++-------- 8 files changed, 73 insertions(+), 76 deletions(-) (limited to 'src/mlia/devices/ethosu') 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=[ -- cgit v1.2.1