diff options
author | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2022-07-21 14:06:03 +0100 |
---|---|---|
committer | Benjamin Klimczak <benjamin.klimczak@arm.com> | 2022-08-19 10:23:23 +0100 |
commit | a8ee1aee3e674c78a77801d1bf2256881ab6b4b9 (patch) | |
tree | 8463b24ba0446a49b3e012477b0834c3b5415b86 /src/mlia/core/handlers.py | |
parent | 76ec769ad8f8ed53ec3ff829fdd34d53db8229fd (diff) | |
download | mlia-a8ee1aee3e674c78a77801d1bf2256881ab6b4b9.tar.gz |
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
Diffstat (limited to 'src/mlia/core/handlers.py')
-rw-r--r-- | src/mlia/core/handlers.py | 166 |
1 files changed, 166 insertions, 0 deletions
diff --git a/src/mlia/core/handlers.py b/src/mlia/core/handlers.py new file mode 100644 index 0000000..e576f74 --- /dev/null +++ b/src/mlia/core/handlers.py @@ -0,0 +1,166 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Event handlers module.""" +import logging +from typing import Any +from typing import Callable +from typing import List +from typing import Optional + +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 ActionFinishedEvent +from mlia.core.events import ActionStartedEvent +from mlia.core.events import AdviceStageFinishedEvent +from mlia.core.events import AdviceStageStartedEvent +from mlia.core.events import AnalyzedDataEvent +from mlia.core.events import CollectedDataEvent +from mlia.core.events import DataAnalysisStageFinishedEvent +from mlia.core.events import DataAnalysisStageStartedEvent +from mlia.core.events import DataCollectionStageFinishedEvent +from mlia.core.events import DataCollectionStageStartedEvent +from mlia.core.events import DataCollectorSkippedEvent +from mlia.core.events import EventDispatcher +from mlia.core.events import ExecutionFailedEvent +from mlia.core.events import ExecutionFinishedEvent +from mlia.core.events import ExecutionStartedEvent +from mlia.core.reporting import Report +from mlia.core.reporting import Reporter +from mlia.core.reporting import resolve_output_format +from mlia.utils.console import create_section_header + + +logger = logging.getLogger(__name__) + + +class SystemEventsHandler(EventDispatcher): + """System events handler.""" + + def on_execution_started(self, event: ExecutionStartedEvent) -> None: + """Handle ExecutionStarted event.""" + + def on_execution_finished(self, event: ExecutionFinishedEvent) -> None: + """Handle ExecutionFinished event.""" + + def on_execution_failed(self, event: ExecutionFailedEvent) -> None: + """Handle ExecutionFailed event.""" + + def on_data_collection_stage_started( + self, event: DataCollectionStageStartedEvent + ) -> None: + """Handle DataCollectionStageStarted event.""" + + def on_data_collection_stage_finished( + self, event: DataCollectionStageFinishedEvent + ) -> None: + """Handle DataCollectionStageFinished event.""" + + def on_data_collector_skipped(self, event: DataCollectorSkippedEvent) -> None: + """Handle DataCollectorSkipped event.""" + + def on_data_analysis_stage_started( + self, event: DataAnalysisStageStartedEvent + ) -> None: + """Handle DataAnalysisStageStartedEvent event.""" + + def on_data_analysis_stage_finished( + self, event: DataAnalysisStageFinishedEvent + ) -> None: + """Handle DataAnalysisStageFinishedEvent event.""" + + def on_advice_stage_started(self, event: AdviceStageStartedEvent) -> None: + """Handle AdviceStageStarted event.""" + + def on_advice_stage_finished(self, event: AdviceStageFinishedEvent) -> None: + """Handle AdviceStageFinished event.""" + + def on_collected_data(self, event: CollectedDataEvent) -> None: + """Handle CollectedData event.""" + + def on_analyzed_data(self, event: AnalyzedDataEvent) -> None: + """Handle AnalyzedData event.""" + + def on_action_started(self, event: ActionStartedEvent) -> None: + """Handle ActionStarted event.""" + + def on_action_finished(self, event: ActionFinishedEvent) -> None: + """Handle ActionFinished event.""" + + +_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 __init__( + self, + formatter_resolver: Callable[[Any], Callable[[Any], Report]], + output: Optional[PathOrFileLike] = None, + ) -> None: + """Init event handler.""" + output_format = resolve_output_format(output) + self.reporter = Reporter(formatter_resolver, output_format) + self.output = output + + self.advice: List[Advice] = [] + + 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) + + @staticmethod + def report_generated(output: PathOrFileLike) -> None: + """Log report generation.""" + logger.info(_REPORT_GENERATION_MSG) + logger.info("Report(s) and advice list saved to: %s", output) + + def on_data_analysis_stage_finished( + self, event: DataAnalysisStageFinishedEvent + ) -> None: + """Handle DataAnalysisStageFinished event.""" + logger.info(_MODEL_ANALYSIS_RESULTS_MSG) + + self.reporter.print_delayed() + + def on_advice_event(self, event: AdviceEvent) -> None: + """Handle Advice event.""" + self.advice.append(event.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: + self.report_generated(self.output) |