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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Ethos-U MLIA module."""
from pathlib import Path
from typing import List
from typing import Optional
from mlia.core.advice_generation import AdviceProducer
from mlia.core.advisor import InferenceAdvisor
from mlia.core.common import AdviceCategory
from mlia.core.context import Context
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.devices.ethosu.advice_generation import EthosUAdviceProducer
from mlia.devices.ethosu.advice_generation import EthosUStaticAdviceProducer
from mlia.devices.ethosu.config import EthosUConfiguration
from mlia.devices.ethosu.config import get_target
from mlia.devices.ethosu.data_analysis import EthosUDataAnalyzer
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
class EthosUInferenceAdvisor(InferenceAdvisor, ParameterResolverMixin):
"""Ethos-U Inference Advisor."""
@classmethod
def name(cls) -> str:
"""Return name of the advisor."""
return "ethos_u_inference_advisor"
def configure(self, context: Context) -> WorkflowExecutor:
"""Configure advisor execution."""
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 context.any_category_enabled(
AdviceCategory.OPERATORS,
AdviceCategory.ALL,
):
collectors.append(EthosUOperatorCompatibility(model, device))
if context.category_enabled(AdviceCategory.PERFORMANCE):
collectors.append(EthosUPerformance(model, device, backends))
if context.any_category_enabled(
AdviceCategory.OPTIMIZATION,
AdviceCategory.ALL,
):
optimization_settings = self._get_optimization_settings(context)
collectors.append(
EthosUOptimizationPerformance(
model, device, optimization_settings, backends
)
)
return collectors
@staticmethod
def _get_analyzers() -> List[DataAnalyzer]:
"""Return data analyzers."""
return [
EthosUDataAnalyzer(),
]
@staticmethod
def _get_advice_producers() -> List[AdviceProducer]:
"""Return advice producers."""
return [
EthosUAdviceProducer(),
EthosUStaticAdviceProducer(),
]
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
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
EthosUOptimizationPerformance.name(),
"optimizations",
expected_type=list,
expected=False,
context=context,
)
def _get_backends(self, context: Context) -> Optional[List[str]]:
"""Get list of backends."""
return self.get_parameter( # type: ignore
self.name(),
"backends",
expected_type=list,
expected=False,
context=context,
)
|
