From a8ee1aee3e674c78a77801d1bf2256881ab6b4b9 Mon Sep 17 00:00:00 2001 From: Dmitrii Agibov Date: Thu, 21 Jul 2022 14:06:03 +0100 Subject: MLIA-549 Refactor API module to support several target profiles - Move target specific details out of API module - Move common logic for workflow event handler into a separate class Change-Id: Ic4a22657b722af1c1fead1d478f606ac57325788 --- src/mlia/devices/ethosu/advisor.py | 159 ++++++++++++++++++++++------------- src/mlia/devices/ethosu/handlers.py | 100 +--------------------- src/mlia/devices/ethosu/reporters.py | 4 +- 3 files changed, 105 insertions(+), 158 deletions(-) (limited to 'src/mlia/devices/ethosu') diff --git a/src/mlia/devices/ethosu/advisor.py b/src/mlia/devices/ethosu/advisor.py index e93858f..b7b8305 100644 --- a/src/mlia/devices/ethosu/advisor.py +++ b/src/mlia/devices/ethosu/advisor.py @@ -2,18 +2,22 @@ # SPDX-License-Identifier: Apache-2.0 """Ethos-U MLIA module.""" 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 from mlia.core.common import AdviceCategory from mlia.core.context import Context +from mlia.core.context import ExecutionContext from mlia.core.data_analysis import DataAnalyzer from mlia.core.data_collection import DataCollector -from mlia.core.mixins import ParameterResolverMixin -from mlia.core.workflow import DefaultWorkflowExecutor -from mlia.core.workflow import WorkflowExecutor +from mlia.core.events import Event from mlia.devices.ethosu.advice_generation import EthosUAdviceProducer from mlia.devices.ethosu.advice_generation import EthosUStaticAdviceProducer from mlia.devices.ethosu.config import EthosUConfiguration @@ -23,10 +27,12 @@ from mlia.devices.ethosu.data_collection import EthosUOperatorCompatibility from mlia.devices.ethosu.data_collection import EthosUOptimizationPerformance from mlia.devices.ethosu.data_collection import EthosUPerformance from mlia.devices.ethosu.events import EthosUAdvisorStartedEvent +from mlia.devices.ethosu.handlers import EthosUEventHandler from mlia.nn.tensorflow.utils import is_tflite_model +from mlia.utils.types import is_list_of -class EthosUInferenceAdvisor(InferenceAdvisor, ParameterResolverMixin): +class EthosUInferenceAdvisor(DefaultInferenceAdvisor): """Ethos-U Inference Advisor.""" @classmethod @@ -34,34 +40,12 @@ class EthosUInferenceAdvisor(InferenceAdvisor, ParameterResolverMixin): """Return name of the advisor.""" return "ethos_u_inference_advisor" - def configure(self, context: Context) -> WorkflowExecutor: - """Configure advisor execution.""" - model = self._get_model(context) + 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 = self._get_collectors(context, model, device, backends) - analyzers = self._get_analyzers() - producers = self._get_advice_producers() - - return DefaultWorkflowExecutor( - context, - collectors, - analyzers, - producers, - before_start_events=[ - EthosUAdvisorStartedEvent(device=device, model=model), - ], - ) - - def _get_collectors( - self, - context: Context, - model: Path, - device: EthosUConfiguration, - backends: Optional[List[str]], - ) -> List[DataCollector]: - """Get collectors.""" collectors: List[DataCollector] = [] if AdviceCategory.OPERATORS in context.advice_category: @@ -91,51 +75,34 @@ class EthosUInferenceAdvisor(InferenceAdvisor, ParameterResolverMixin): return collectors - @staticmethod - def _get_analyzers() -> List[DataAnalyzer]: - """Return data analyzers.""" + def get_analyzers(self, context: Context) -> List[DataAnalyzer]: + """Return list of the data analyzers.""" return [ EthosUDataAnalyzer(), ] - @staticmethod - def _get_advice_producers() -> List[AdviceProducer]: - """Return advice producers.""" + 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]: + """Return list of the startup events.""" + model = self.get_model(context) + device = self._get_device(context) + + return [ + EthosUAdvisorStartedEvent(device=device, model=model), + ] + def _get_device(self, context: Context) -> EthosUConfiguration: """Get device.""" - device_params = self.get_parameter( - self.name(), - "device", - expected_type=dict, - context=context, - ) - - try: - target_profile = device_params["target_profile"] - except KeyError as err: - raise Exception("Unable to get device details") from err + target_profile = self.get_target_profile(context) return get_target(target_profile) - def _get_model(self, context: Context) -> Path: - """Get path to the model.""" - model_param = self.get_parameter( - self.name(), - "model", - expected_type=str, - context=context, - ) - - if not (model := Path(model_param)).exists(): - raise Exception(f"Path {model} does not exist") - - return model - def _get_optimization_settings(self, context: Context) -> List[List[dict]]: """Get optimization settings.""" return self.get_parameter( # type: ignore @@ -155,3 +122,75 @@ class EthosUInferenceAdvisor(InferenceAdvisor, ParameterResolverMixin): expected=False, context=context, ) + + +def configure_and_get_ethosu_advisor( + context: ExecutionContext, + target_profile: str, + model: Union[Path, str], + output: Optional[PathOrFileLike] = None, + **extra_args: Any, +) -> InferenceAdvisor: + """Create and configure Ethos-U advisor.""" + if context.event_handlers is None: + context.event_handlers = [EthosUEventHandler(output)] + + if context.config_parameters is None: + context.config_parameters = _get_config_parameters( + model, target_profile, **extra_args + ) + + return EthosUInferenceAdvisor() + + +_DEFAULT_OPTIMIZATION_TARGETS = [ + { + "optimization_type": "pruning", + "optimization_target": 0.5, + "layers_to_optimize": None, + }, + { + "optimization_type": "clustering", + "optimization_target": 32, + "layers_to_optimize": None, + }, +] + + +def _get_config_parameters( + model: Union[Path, str], + target_profile: str, + **extra_args: Any, +) -> Dict[str, Any]: + """Get configuration parameters for the advisor.""" + advisor_parameters: Dict[str, Any] = { + "ethos_u_inference_advisor": { + "model": model, + "target_profile": target_profile, + }, + } + + # Specifying backends is optional (default is used) + backends = extra_args.get("backends") + if backends is not None: + if not is_list_of(backends, str): + raise Exception("Backends value has wrong format") + + advisor_parameters["ethos_u_inference_advisor"]["backends"] = backends + + optimization_targets = extra_args.get("optimization_targets") + if not optimization_targets: + optimization_targets = _DEFAULT_OPTIMIZATION_TARGETS + + if not is_list_of(optimization_targets, dict): + raise Exception("Optimization targets value has wrong format") + + advisor_parameters.update( + { + "ethos_u_model_optimizations": { + "optimizations": [optimization_targets], + }, + } + ) + + return advisor_parameters diff --git a/src/mlia/devices/ethosu/handlers.py b/src/mlia/devices/ethosu/handlers.py index 7a0c31c..ee0b809 100644 --- a/src/mlia/devices/ethosu/handlers.py +++ b/src/mlia/devices/ethosu/handlers.py @@ -2,101 +2,27 @@ # SPDX-License-Identifier: Apache-2.0 """Event handler.""" import logging -from pathlib import Path -from typing import Dict -from typing import List from typing import Optional -from mlia.core._typing import OutputFormat from mlia.core._typing import PathOrFileLike -from mlia.core.advice_generation import Advice -from mlia.core.advice_generation import AdviceEvent -from mlia.core.events import AdviceStageFinishedEvent -from mlia.core.events import AdviceStageStartedEvent from mlia.core.events import CollectedDataEvent -from mlia.core.events import DataAnalysisStageFinishedEvent -from mlia.core.events import DataCollectionStageStartedEvent -from mlia.core.events import DataCollectorSkippedEvent -from mlia.core.events import ExecutionFailedEvent -from mlia.core.events import ExecutionStartedEvent -from mlia.core.events import SystemEventsHandler -from mlia.core.reporting import Reporter +from mlia.core.handlers import WorkflowEventsHandler from mlia.devices.ethosu.events import EthosUAdvisorEventHandler from mlia.devices.ethosu.events import EthosUAdvisorStartedEvent from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics from mlia.devices.ethosu.performance import PerformanceMetrics -from mlia.devices.ethosu.reporters import find_appropriate_formatter +from mlia.devices.ethosu.reporters import ethos_u_formatters from mlia.tools.vela_wrapper import Operators -from mlia.utils.console import create_section_header logger = logging.getLogger(__name__) -ADV_EXECUTION_STARTED = create_section_header("ML Inference Advisor started") -MODEL_ANALYSIS_MSG = create_section_header("Model Analysis") -MODEL_ANALYSIS_RESULTS_MSG = create_section_header("Model Analysis Results") -ADV_GENERATION_MSG = create_section_header("Advice Generation") -REPORT_GENERATION_MSG = create_section_header("Report Generation") - - -class WorkflowEventsHandler(SystemEventsHandler): - """Event handler for the system events.""" - - def on_execution_started(self, event: ExecutionStartedEvent) -> None: - """Handle ExecutionStarted event.""" - logger.info(ADV_EXECUTION_STARTED) - - def on_execution_failed(self, event: ExecutionFailedEvent) -> None: - """Handle ExecutionFailed event.""" - raise event.err - - def on_data_collection_stage_started( - self, event: DataCollectionStageStartedEvent - ) -> None: - """Handle DataCollectionStageStarted event.""" - logger.info(MODEL_ANALYSIS_MSG) - - def on_advice_stage_started(self, event: AdviceStageStartedEvent) -> None: - """Handle AdviceStageStarted event.""" - logger.info(ADV_GENERATION_MSG) - - def on_data_collector_skipped(self, event: DataCollectorSkippedEvent) -> None: - """Handle DataCollectorSkipped event.""" - logger.info("Skipped: %s", event.reason) - class EthosUEventHandler(WorkflowEventsHandler, EthosUAdvisorEventHandler): """CLI event handler.""" def __init__(self, output: Optional[PathOrFileLike] = None) -> None: """Init event handler.""" - output_format = self.resolve_output_format(output) - - self.reporter = Reporter(find_appropriate_formatter, output_format) - self.output = output - self.advice: List[Advice] = [] - - def on_advice_stage_finished(self, event: AdviceStageFinishedEvent) -> None: - """Handle AdviceStageFinishedEvent event.""" - self.reporter.submit( - self.advice, - show_title=False, - show_headers=False, - space="between", - table_style="no_borders", - ) - - self.reporter.generate_report(self.output) - - if self.output is not None: - logger.info(REPORT_GENERATION_MSG) - logger.info("Report(s) and advice list saved to: %s", self.output) - - def on_data_analysis_stage_finished( - self, event: DataAnalysisStageFinishedEvent - ) -> None: - """Handle DataAnalysisStageFinished event.""" - logger.info(MODEL_ANALYSIS_RESULTS_MSG) - self.reporter.print_delayed() + super().__init__(ethos_u_formatters, output) def on_collected_data(self, event: CollectedDataEvent) -> None: """Handle CollectedDataEvent event.""" @@ -106,7 +32,7 @@ class EthosUEventHandler(WorkflowEventsHandler, EthosUAdvisorEventHandler): self.reporter.submit([data_item.ops, data_item], delay_print=True) if isinstance(data_item, PerformanceMetrics): - self.reporter.submit(data_item, delay_print=True) + self.reporter.submit(data_item, delay_print=True, space=True) if isinstance(data_item, OptimizationPerformanceMetrics): original_metrics = data_item.original_perf_metrics @@ -123,24 +49,6 @@ class EthosUEventHandler(WorkflowEventsHandler, EthosUAdvisorEventHandler): space=True, ) - def on_advice_event(self, event: AdviceEvent) -> None: - """Handle Advice event.""" - self.advice.append(event.advice) - def on_ethos_u_advisor_started(self, event: EthosUAdvisorStartedEvent) -> None: """Handle EthosUAdvisorStarted event.""" self.reporter.submit(event.device) - - @staticmethod - def resolve_output_format(output: Optional[PathOrFileLike]) -> OutputFormat: - """Resolve output format based on the output name.""" - output_format: OutputFormat = "plain_text" - - if isinstance(output, str): - output_path = Path(output) - output_formats: Dict[str, OutputFormat] = {".csv": "csv", ".json": "json"} - - if (suffix := output_path.suffix) in output_formats: - return output_formats[suffix] - - return output_format diff --git a/src/mlia/devices/ethosu/reporters.py b/src/mlia/devices/ethosu/reporters.py index d28c68f..b3aea24 100644 --- a/src/mlia/devices/ethosu/reporters.py +++ b/src/mlia/devices/ethosu/reporters.py @@ -374,7 +374,7 @@ def report_advice(advice: List[Advice]) -> Report: ) -def find_appropriate_formatter(data: Any) -> Callable[[Any], Report]: +def ethos_u_formatters(data: Any) -> Callable[[Any], Report]: """Find appropriate formatter for the provided data.""" if isinstance(data, PerformanceMetrics) or is_list_of(data, PerformanceMetrics, 2): return report_perf_metrics @@ -392,7 +392,7 @@ def find_appropriate_formatter(data: Any) -> Callable[[Any], Report]: return report_device_details if isinstance(data, (list, tuple)): - formatters = [find_appropriate_formatter(item) for item in data] + formatters = [ethos_u_formatters(item) for item in data] return CompoundFormatter(formatters) raise Exception(f"Unable to find appropriate formatter for {data}") -- cgit v1.2.1