1 files changed, 154 insertions, 0 deletions
diff --git a/src/mlia/devices/ethosu/data_analysis.py b/src/mlia/devices/ethosu/data_analysis.py
new file mode 100644
index 0000000..9ed32ff
--- /dev/null
+++ b/src/mlia/devices/ethosu/data_analysis.py
@@ -0,0 +1,154 @@
+# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
+# SPDX-License-Identifier: Apache-2.0
+"""Ethos-U data analysis module."""
+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
+from mlia.core.data_analysis import FactExtractor
+from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics
+from mlia.nn.tensorflow.optimizations.select import OptimizationSettings
+from mlia.tools.vela_wrapper import Operators
+
+
+@dataclass
+class HasCPUOnlyOperators(Fact):
+    """Model has CPU only operators."""
+
+    cpu_only_ops: List[str]
+
+
+@dataclass
+class HasUnsupportedOnNPUOperators(Fact):
+    """Model has unsupported on NPU operators."""
+
+    npu_unsupported_ratio: float
+
+
+@dataclass
+class AllOperatorsSupportedOnNPU(Fact):
+    """All model's operators supported on NPU."""
+
+
+@dataclass
+class PerfMetricDiff:
+    """Performance metric difference."""
+
+    original_value: Union[int, float]
+    optimized_value: Union[int, float]
+
+    @property
+    def diff(self) -> float:
+        """Difference between metrics."""
+        if self.original_value == 0:
+            return 0
+
+        return 100 - ((self.optimized_value / self.original_value) * 100)
+
+    @property
+    def improved(self) -> bool:
+        """Return true if metric improved."""
+        return self.diff > 0
+
+    @property
+    def degraded(self) -> bool:
+        """Return true if metric degraded."""
+        return self.diff < 0
+
+    @property
+    def same(self) -> bool:
+        """Return true if metric stays the same."""
+        return self.diff == 0
+
+
+@dataclass
+class OptimizationDiff:
+    """Optimization performance impact."""
+
+    opt_type: List[OptimizationSettings]
+    opt_diffs: Dict[str, PerfMetricDiff]
+
+
+@dataclass
+class OptimizationResults(Fact):
+    """Optimization results."""
+
+    diffs: List[OptimizationDiff]
+
+
+class EthosUDataAnalyzer(FactExtractor):
+    """Ethos-U data analyzer."""
+
+    @singledispatchmethod
+    def analyze_data(self, data_item: DataItem) -> None:
+        """Analyse the data."""
+
+    @analyze_data.register
+    def analyze_operator_compatibility(self, operators: Operators) -> None:
+        """Analyse operator compatibility information."""
+        cpu_only = [op.op_type for op in operators.ops if op.cpu_only]
+        if cpu_only:
+            self.add_fact(HasCPUOnlyOperators(cpu_only))
+
+        if operators.npu_unsupported_ratio != 0:
+            self.add_fact(HasUnsupportedOnNPUOperators(operators.npu_unsupported_ratio))
+
+        if operators.npu_unsupported_ratio == 0:
+            self.add_fact(AllOperatorsSupportedOnNPU())
+
+    @analyze_data.register
+    def analyze_optimization_results(
+        self, optimization_results: OptimizationPerformanceMetrics
+    ) -> None:
+        """Analyse optimization performance metrics."""
+        optimizations = optimization_results.optimizations_perf_metrics
+        if not optimizations:
+            return
+
+        orig = optimization_results.original_perf_metrics.in_kilobytes()
+        orig_memory = orig.memory_usage
+        orig_cycles = orig.npu_cycles
+
+        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] = {}
+
+            if orig_memory and opt_memory:
+                opt_diffs.update(
+                    {
+                        "sram": PerfMetricDiff(
+                            orig_memory.sram_memory_area_size,
+                            opt_memory.sram_memory_area_size,
+                        ),
+                        "dram": PerfMetricDiff(
+                            orig_memory.dram_memory_area_size,
+                            opt_memory.dram_memory_area_size,
+                        ),
+                        "on_chip_flash": PerfMetricDiff(
+                            orig_memory.on_chip_flash_memory_area_size,
+                            opt_memory.on_chip_flash_memory_area_size,
+                        ),
+                        "off_chip_flash": PerfMetricDiff(
+                            orig_memory.off_chip_flash_memory_area_size,
+                            opt_memory.off_chip_flash_memory_area_size,
+                        ),
+                    }
+                )
+            if orig_cycles and opt_cycles:
+                opt_diffs["npu_total_cycles"] = PerfMetricDiff(
+                    orig_cycles.npu_total_cycles,
+                    opt_cycles.npu_total_cycles,
+                )
+
+            diff = OptimizationDiff(opt_type=opt_type, opt_diffs=opt_diffs)
+            diffs.append(diff)
+
+        self.add_fact(OptimizationResults(diffs))