aboutsummaryrefslogtreecommitdiff
path: root/src/mlia/devices/cortexa/handlers.py
blob: 7ed2b75b2994f6389b6668c041a2a095aa96fa6a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 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 mlia.core.events import CollectedDataEvent
from mlia.core.handlers import WorkflowEventsHandler
from mlia.core.typing import PathOrFileLike
from mlia.devices.cortexa.events import CortexAAdvisorEventHandler
from mlia.devices.cortexa.events import CortexAAdvisorStartedEvent
from mlia.devices.cortexa.operators import CortexACompatibilityInfo
from mlia.devices.cortexa.reporters import cortex_a_formatters
from mlia.nn.tensorflow.tflite_compat import TFLiteCompatibilityInfo

logger = logging.getLogger(__name__)


class CortexAEventHandler(WorkflowEventsHandler, CortexAAdvisorEventHandler):
    """CLI event handler."""

    def __init__(self, output: PathOrFileLike | None = None) -> None:
        """Init event handler."""
        super().__init__(cortex_a_formatters, output)

    def on_collected_data(self, event: CollectedDataEvent) -> None:
        """Handle CollectedDataEvent event."""
        data_item = event.data_item

        if isinstance(data_item, CortexACompatibilityInfo):
            self.reporter.submit(data_item.operators, delay_print=True)

        if isinstance(data_item, TFLiteCompatibilityInfo) and not data_item.compatible:
            self.reporter.submit(data_item, delay_print=True)

    def on_cortex_a_advisor_started(self, event: CortexAAdvisorStartedEvent) -> None:
        """Handle CortexAAdvisorStarted event."""
        self.reporter.submit(event.device)