path: root/src/mlia/devices/ethosu/reporters.py

# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Reports module."""
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
from mlia.core.reporting import Cell
from mlia.core.reporting import ClockCell
from mlia.core.reporting import Column
from mlia.core.reporting import CompoundFormatter
from mlia.core.reporting import CyclesCell
from mlia.core.reporting import Format
from mlia.core.reporting import NestedReport
from mlia.core.reporting import Report
from mlia.core.reporting import ReportItem
from mlia.core.reporting import SingleRow
from mlia.core.reporting import Table
from mlia.devices.ethosu.config import EthosUConfiguration
from mlia.devices.ethosu.performance import PerformanceMetrics
from mlia.tools.vela_wrapper import Operator
from mlia.tools.vela_wrapper import Operators
from mlia.utils.console import style_improvement
from mlia.utils.types import is_list_of


def report_operators_stat(operators: Operators) -> Report:
    """Return table representation for the ops stats."""
    columns = [
        Column("Number of operators", alias="num_of_operators"),
        Column("Number of NPU supported operators", "num_of_npu_supported_operators"),
        Column("Unsupported ops ratio", "npu_unsupported_ratio"),
    ]
    rows = [
        (
            operators.total_number,
            operators.npu_supported_number,
            Cell(
                operators.npu_unsupported_ratio * 100,
                fmt=Format(str_fmt="{0:.0f}%".format),
            ),
        )
    ]

    return SingleRow(
        columns, rows, name="Operators statistics", alias="operators_stats"
    )


def report_operators(ops: List[Operator]) -> Report:
    """Return table representation for the list of operators."""
    columns = [
        Column("#", only_for=["plain_text"]),
        Column(
            "Operator name",
            alias="operator_name",
            fmt=Format(wrap_width=30),
        ),
        Column(
            "Operator type",
            alias="operator_type",
            fmt=Format(wrap_width=25),
        ),
        Column(
            "Placement",
            alias="placement",
            fmt=Format(wrap_width=20),
        ),
        Column(
            "Notes",
            alias="notes",
            fmt=Format(wrap_width=35),
        ),
    ]

    rows = [
        (
            i + 1,
            op.name,
            op.op_type,
            Cell(
                "NPU" if (npu := op.run_on_npu.supported) else "CPU",
                Format(style=style_improvement(npu)),
            ),
            Table(
                columns=[
                    Column(
                        "Note",
                        alias="note",
                        fmt=Format(wrap_width=35),
                    )
                ],
                rows=[
                    (Cell(item, Format(str_fmt=lambda x: f"* {x}")),)
                    for reason in op.run_on_npu.reasons
                    for item in reason
                    if item
                ],
                name="Notes",
            ),
        )
        for i, op in enumerate(ops)
    ]

    return Table(columns, rows, name="Operators", alias="operators")


def report_device_details(device: EthosUConfiguration) -> Report:
    """Return table representation for the device."""
    compiler_config = device.resolved_compiler_config

    memory_settings = [
        ReportItem(
            "Const mem area",
            "const_mem_area",
            compiler_config["const_mem_area"],
        ),
        ReportItem(
            "Arena mem area",
            "arena_mem_area",
            compiler_config["arena_mem_area"],
        ),
        ReportItem(
            "Cache mem area",
            "cache_mem_area",
            compiler_config["cache_mem_area"],
        ),
        ReportItem(
            "Arena cache size",
            "arena_cache_size",
            BytesCell(compiler_config["arena_cache_size"]),
        ),
    ]

    mem_areas_settings = [
        ReportItem(
            f"{mem_area_name}",
            mem_area_name,
            None,
            nested_items=[
                ReportItem(
                    "Clock scales",
                    "clock_scales",
                    mem_area_settings["clock_scales"],
                ),
                ReportItem(
                    "Burst length",
                    "burst_length",
                    BytesCell(mem_area_settings["burst_length"]),
                ),
                ReportItem(
                    "Read latency",
                    "read_latency",
                    CyclesCell(mem_area_settings["read_latency"]),
                ),
                ReportItem(
                    "Write latency",
                    "write_latency",
                    CyclesCell(mem_area_settings["write_latency"]),
                ),
            ],
        )
        for mem_area_name, mem_area_settings in compiler_config["memory_area"].items()
    ]

    system_settings = [
        ReportItem(
            "Accelerator clock",
            "accelerator_clock",
            ClockCell(compiler_config["core_clock"]),
        ),
        ReportItem(
            "AXI0 port",
            "axi0_port",
            compiler_config["axi0_port"],
        ),
        ReportItem(
            "AXI1 port",
            "axi1_port",
            compiler_config["axi1_port"],
        ),
        ReportItem(
            "Memory area settings", "memory_area", None, nested_items=mem_areas_settings
        ),
    ]

    arch_settings = [
        ReportItem(
            "Permanent storage mem area",
            "permanent_storage_mem_area",
            compiler_config["permanent_storage_mem_area"],
        ),
        ReportItem(
            "Feature map storage mem area",
            "feature_map_storage_mem_area",
            compiler_config["feature_map_storage_mem_area"],
        ),
        ReportItem(
            "Fast storage mem area",
            "fast_storage_mem_area",
            compiler_config["fast_storage_mem_area"],
        ),
    ]

    return NestedReport(
        "Device information",
        "device",
        [
            ReportItem("Target", alias="target", value=device.target),
            ReportItem("MAC", alias="mac", value=device.mac),
            ReportItem(
                "Memory mode",
                alias="memory_mode",
                value=compiler_config["memory_mode"],
                nested_items=memory_settings,
            ),
            ReportItem(
                "System config",
                alias="system_config",
                value=compiler_config["system_config"],
                nested_items=system_settings,
            ),
            ReportItem(
                "Architecture settings",
                "arch_settings",
                None,
                nested_items=arch_settings,
            ),
        ],
    )


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]:
        metric_map = defaultdict(list)
        for metrics in perf_metrics:
            if not metrics.npu_cycles:
                return []
            metric_map["NPU active cycles"].append(metrics.npu_cycles.npu_active_cycles)
            metric_map["NPU idle cycles"].append(metrics.npu_cycles.npu_idle_cycles)
            metric_map["NPU total cycles"].append(metrics.npu_cycles.npu_total_cycles)

        return [
            (name, *(Cell(value, Format(str_fmt="12,d")) for value in values), "cycles")
            for name, values in metric_map.items()
        ]

    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:
                return []
            metric_map["SRAM used"].append(metrics.memory_usage.sram_memory_area_size)
            metric_map["DRAM used"].append(metrics.memory_usage.dram_memory_area_size)
            metric_map["Unknown memory area used"].append(
                metrics.memory_usage.unknown_memory_area_size
            )
            metric_map["On-chip flash used"].append(
                metrics.memory_usage.on_chip_flash_memory_area_size
            )
            metric_map["Off-chip flash used"].append(
                metrics.memory_usage.off_chip_flash_memory_area_size
            )

        return [
            (name, *(Cell(value, Format(str_fmt="12.2f")) for value in values), "KiB")
            for name, values in metric_map.items()
            if all(val > 0 for val in values)
        ]

    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:
                return []
            metric_map["NPU AXI0 RD data beat received"].append(
                metrics.npu_cycles.npu_axi0_rd_data_beat_received
            )
            metric_map["NPU AXI0 WR data beat written"].append(
                metrics.npu_cycles.npu_axi0_wr_data_beat_written
            )
            metric_map["NPU AXI1 RD data beat received"].append(
                metrics.npu_cycles.npu_axi1_rd_data_beat_received
            )

        return [
            (name, *(Cell(value, Format(str_fmt="12,d")) for value in values), "beats")
            for name, values in metric_map.items()
        ]

    return [
        metrics
        for metrics_func in (
            _memory_usage_as_records,
            _cycles_as_records,
            _data_beats_as_records,
        )
        for metrics in metrics_func(perf_metrics)
    ]


def report_perf_metrics(
    perf_metrics: Union[PerformanceMetrics, List[PerformanceMetrics]]
) -> Report:
    """Return comparison table for the performance metrics."""
    if isinstance(perf_metrics, PerformanceMetrics):
        perf_metrics = [perf_metrics]

    rows = metrics_as_records(perf_metrics)

    if len(perf_metrics) == 2:
        return Table(
            columns=[
                Column("Metric", alias="metric", fmt=Format(wrap_width=30)),
                Column("Original", alias="original", fmt=Format(wrap_width=15)),
                Column("Optimized", alias="optimized", fmt=Format(wrap_width=15)),
                Column("Unit", alias="unit", fmt=Format(wrap_width=15)),
                Column("Improvement (%)", alias="improvement"),
            ],
            rows=[
                (
                    metric,
                    original_value,
                    optimized_value,
                    unit,
                    Cell(
                        (
                            diff := 100
                            - (optimized_value.value / original_value.value * 100)
                        ),
                        Format(str_fmt="15.2f", style=style_improvement(diff > 0)),
                    )
                    if original_value.value != 0
                    else None,
                )
                for metric, original_value, optimized_value, unit in rows
            ],
            name="Performance metrics",
            alias="performance_metrics",
            notes="IMPORTANT: The performance figures above refer to NPU only",
        )

    return Table(
        columns=[
            Column("Metric", alias="metric", fmt=Format(wrap_width=30)),
            Column("Value", alias="value", fmt=Format(wrap_width=15)),
            Column("Unit", alias="unit", fmt=Format(wrap_width=15)),
        ],
        rows=rows,
        name="Performance metrics",
        alias="performance_metrics",
        notes="IMPORTANT: The performance figures above refer to NPU only",
    )


def report_advice(advice: List[Advice]) -> Report:
    """Generate report for the advice."""
    return Table(
        columns=[
            Column("#", only_for=["plain_text"]),
            Column("Advice", alias="advice_message"),
        ],
        rows=[(i + 1, a.messages) for i, a in enumerate(advice)],
        name="Advice",
        alias="advice",
    )


def find_appropriate_formatter(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

    if is_list_of(data, Advice):
        return report_advice

    if is_list_of(data, Operator):
        return report_operators

    if isinstance(data, Operators):
        return report_operators_stat

    if isinstance(data, EthosUConfiguration):
        return report_device_details

    if isinstance(data, (list, tuple)):
        formatters = [find_appropriate_formatter(item) for item in data]
        return CompoundFormatter(formatters)

    raise Exception(f"Unable to find appropriate formatter for {data}")