type.py Source File

type.py Source File#

SDK qb Runtime Library: type.py Source File
SDK qb Runtime Library v1.5
MCS001-KR
type.py
Go to the documentation of this file.
1
4
5from typing import List, Optional, Tuple, overload
6from enum import Enum
7
8import numpy as np
9
10import qbruntime.qbruntime as _cQbRuntime
11
12
15
16
17class Cluster(Enum):
18 """
19 @brief Enumerates clusters in the ARIES NPU.
20
21 @note The ARIES NPU consists of two clusters, each containing one global core and
22 four local cores, totaling eight local cores. REGULUS has only a single cluster
23 (Cluster0) with one local core (Core0).
24 """
25
26 Cluster0 = _cQbRuntime.Cluster.Cluster0
27 Cluster1 = _cQbRuntime.Cluster.Cluster1
28 Error = _cQbRuntime.Cluster.Error
29
30
31class Core(Enum):
32 """
33 @brief Enumerates cores within a cluster in the ARIES NPU.
34
35 @note The ARIES NPU consists of two clusters, each containing one global core and
36 four local cores, totaling eight local cores. REGULUS has only a single cluster
37 (Cluster0) with one local core (Core0).
38 """
39
40 Core0 = _cQbRuntime.Core.Core0
41 Core1 = _cQbRuntime.Core.Core1
42 Core2 = _cQbRuntime.Core.Core2
43 Core3 = _cQbRuntime.Core.Core3
44 All = _cQbRuntime.Core.All
45 GlobalCore = _cQbRuntime.Core.GlobalCore
46 Error = _cQbRuntime.Core.Error
47
48
50 """@brief Core allocation policy"""
51
52 Auto = _cQbRuntime.CoreAllocationPolicy.Auto
53 Manual = _cQbRuntime.CoreAllocationPolicy.Manual
54
55
56class Scale:
57 """@brief Struct for scale values."""
58
59 def __init__(
60 self,
61 scale: float,
62 is_uniform: bool,
63 scale_list: List[float],
64 zero_point: int = 0,
65 is_asymmetric: bool = False,
66 zero_points: Optional[List[int]] = None,
67 ):
68 self._scale = _cQbRuntime.Scale()
69 self._scale.scale = scale
70 self._scale.is_uniform = is_uniform
71 self._scale.scale_list = scale_list
72 self._scale.zero_point = zero_point
73 self._scale.is_asymmetric = is_asymmetric
74 self._scale.zero_points = zero_points if zero_points is not None else []
75
76 @classmethod
77 def from_cpp(cls, _scale: _cQbRuntime.Scale):
78 return cls(
79 _scale.scale,
80 _scale.is_uniform,
81 _scale.scale_list,
82 _scale.zero_point,
83 _scale.is_asymmetric,
84 _scale.zero_points,
85 )
86
87 @property
88 def scale_list(self) -> List[float]:
89 return self._scale.scale_list
90
91 @property
92 def scale(self) -> float:
93 return self._scale.scale
94
95 @property
96 def is_uniform(self) -> bool:
97 return self._scale.is_uniform
98
99 @property
100 def zero_points(self) -> List[int]:
101 """Per-channel zero points for asymmetric quantization."""
102 return self._scale.zero_points
103
104 @property
105 def zero_point(self) -> int:
106 """Uniform zero point for asymmetric quantization."""
107 return self._scale.zero_point
108
109 @property
110 def is_asymmetric(self) -> bool:
111 """Indicates whether asymmetric quantization is used."""
112 return self._scale.is_asymmetric
113
114 @scale_list.setter
115 def scale_list(self, value: List[float]):
116 self._scale.scale_list = value
117
118 @scale.setter
119 def scale(self, value: float):
120 self._scale.scale = value
121
122 @is_uniform.setter
123 def is_uniform(self, value: bool):
124 self._scale.is_uniform = value
125
126 @zero_points.setter
127 def zero_points(self, value: List[int]):
128 self._scale.zero_points = value
129
130 @zero_point.setter
131 def zero_point(self, value: int):
132 self._scale.zero_point = value
133
134 @is_asymmetric.setter
135 def is_asymmetric(self, value: bool):
136 self._scale.is_asymmetric = value
137
138 def __getitem__(self, i: int) -> float:
139 """
140 @brief Returns the scale value at the specified index.
141
142 @param[in] i Index.
143 @return Scale value.
144 """
145 return self._scale[i]
146
147 def __repr__(self):
148 d = {
149 "scale": self.scale,
150 "is_uniform": self.is_uniform,
151 "scale_list": self.scale_list,
152 "zero_point": self.zero_point,
153 "is_asymmetric": self.is_asymmetric,
154 "zero_points": self.zero_points,
155 }
156 return "{}({})".format(
157 self.__class__.__name__,
158 ", ".join("{}={}".format(k, v) for k, v in d.items()),
159 )
160
161
162class CoreId:
163 """
164 @brief Represents a unique identifier for an NPU core.
165
166 A CoreId consists of a Cluster and a Core, identifying a specific core
167 within an NPU.
168 """
169
170 def __init__(self, cluster: Cluster, core: Core):
171 self._core_id = _cQbRuntime.CoreId()
172 self._core_id.cluster = cluster.value
173 self._core_id.core = core.value
174
175 @classmethod
176 def from_cpp(cls, _core_id: _cQbRuntime.CoreId):
177 return cls(Cluster(_core_id.cluster), Core(_core_id.core))
178
179 @property
180 def cluster(self) -> Cluster:
181 return Cluster(self._core_id.cluster)
182
183 @property
184 def core(self) -> Core:
185 return Core(self._core_id.core)
186
187 @cluster.setter
188 def cluster(self, value: Cluster):
189 self._core_id.cluster = value.value
190
191 @core.setter
192 def core(self, value: Core):
193 self._core_id.core = value.value
194
195 def __eq__(self, other) -> bool:
196 """
197 @brief Checks if two CoreId objects are equal.
198
199 @return True if both CoreId objects are identical, False otherwise.
200 """
201 return self._core_id == other._core_id
202
203 def __lt__(self, other) -> bool:
204 """
205 @brief Compares two CoreId objects for ordering.
206
207 @return True if this CoreId is less than the given CoreId, False otherwise.
208 """
209 return self._core_id < other._core_id
210
211 def __repr__(self):
212 d = {"cluster": self.cluster, "core": self.core}
213 return "{}({})".format(
214 self.__class__.__name__,
215 ", ".join("{}={}".format(k, v) for k, v in d.items()),
216 )
217
218
219class Buffer:
220 """
221 @brief A simple byte-sized buffer.
222
223 This struct represents a contiguous block of memory for storing byte-sized data.
224 """
225
226 def __init__(self, _buffer: Optional[_cQbRuntime.Buffer] = None):
227 self._buffer = _cQbRuntime.Buffer() if _buffer is None else _buffer
228
229 @property
230 def size(self) -> int:
231 return self._buffer.size
232
233 @size.setter
234 def size(self, value: int):
235 self._buffer.size = value
236
237 def set_buffer(self, arr: np.ndarray):
238 self._buffer.set_buffer(np.ascontiguousarray(arr))
239
240 def __repr__(self):
241 return f"{self.__class__.__name__}(size={self._buffer.size})"
242
243
244class CoreMode(Enum):
245 """
246 @brief Defines the core mode for NPU execution.
247
248 Supported core modes include single-core, multi-core, global4-core, and global8-core.
249 For detailed explanations of each mode, refer to the following functions:
250
251 - `ModelConfig.set_auto_core_mode()`
252 - `ModelConfig.set_single_core_mode()`
253 - `ModelConfig.set_multi_core_mode()`
254 - `ModelConfig.set_global4_core_mode()`
255 - `ModelConfig.set_global8_core_mode()`
256 """
257
258 Single = _cQbRuntime.CoreMode.Single
259 Multi = _cQbRuntime.CoreMode.Multi
260 Global = _cQbRuntime.CoreMode.Global
261 Global4 = _cQbRuntime.CoreMode.Global4
262 Global8 = _cQbRuntime.CoreMode.Global8
263 Auto = _cQbRuntime.CoreMode.Auto
264 Error = _cQbRuntime.CoreMode.Error
265
266
268 """@brief Struct representing input/output buffer information."""
269
270 def __init__(
271 self,
272 original_height: int = 0,
273 original_width: int = 0,
274 original_channel: int = 0,
275 reshaped_height: int = 0,
276 reshaped_width: int = 0,
277 reshaped_channel: int = 0,
278 height: int = 0,
279 width: int = 0,
280 channel: int = 0,
281 max_height: int = 0,
282 max_width: int = 0,
283 max_channel: int = 0,
284 max_cache_size: int = 0,
285 ):
286 self._buffer_info = _cQbRuntime.BufferInfo()
287 self._buffer_info.original_height = original_height
288 self._buffer_info.original_width = original_width
289 self._buffer_info.original_channel = original_channel
290 self._buffer_info.reshaped_height = reshaped_height
291 self._buffer_info.reshaped_width = reshaped_width
292 self._buffer_info.reshaped_channel = reshaped_channel
293 self._buffer_info.height = height
294 self._buffer_info.width = width
295 self._buffer_info.channel = channel
296 self._buffer_info.max_height = max_height
297 self._buffer_info.max_width = max_width
298 self._buffer_info.max_channel = max_channel
299 self._buffer_info.max_cache_size = max_cache_size
300
301 @classmethod
302 def from_cpp(cls, _buffer_info: _cQbRuntime.BufferInfo):
303 return cls(
304 _buffer_info.original_height,
305 _buffer_info.original_width,
306 _buffer_info.original_channel,
307 _buffer_info.reshaped_height,
308 _buffer_info.reshaped_width,
309 _buffer_info.reshaped_channel,
310 _buffer_info.height,
311 _buffer_info.width,
312 _buffer_info.channel,
313 _buffer_info.max_height,
314 _buffer_info.max_width,
315 _buffer_info.max_channel,
316 _buffer_info.max_cache_size,
317 )
318
319 @property
320 def original_height(self) -> int:
321 """Height of original input/output"""
322 return self._buffer_info.original_height
323
324 @property
325 def original_width(self) -> int:
326 """Width of original input/output"""
327 return self._buffer_info.original_width
328
329 @property
330 def original_channel(self) -> int:
331 """Channel of original input/output"""
332 return self._buffer_info.original_channel
333
334 @property
335 def reshaped_height(self) -> int:
336 """Height of reshaped input/output"""
337 return self._buffer_info.reshaped_height
338
339 @property
340 def reshaped_width(self) -> int:
341 """Width of reshaped input/output"""
342 return self._buffer_info.reshaped_width
343
344 @property
345 def reshaped_channel(self) -> int:
346 """Channel of reshaped input/output"""
347 return self._buffer_info.reshaped_channel
348
349 @property
350 def height(self) -> int:
351 """Height of NPU input/output"""
352 return self._buffer_info.height
353
354 @property
355 def width(self) -> int:
356 """Width of NPU input/output"""
357 return self._buffer_info.width
358
359 @property
360 def channel(self) -> int:
361 """Channel of NPU input/output"""
362 return self._buffer_info.channel
363
364 @property
365 def max_height(self) -> int:
366 """Maximum height of original input/output if data is sequential."""
367 return self._buffer_info.max_height
368
369 @property
370 def max_width(self) -> int:
371 """Maximum width of original input/output if data is sequential."""
372 return self._buffer_info.max_width
373
374 @property
375 def max_channel(self) -> int:
376 """Maximum channel of original input/output if data is sequential."""
377 return self._buffer_info.max_channel
378
379 @property
380 def max_cache_size(self) -> int:
381 """Maximum KV-cache size, relevant for LLM models using KV cache."""
382 return self._buffer_info.max_cache_size
383
384 @original_height.setter
385 def original_height(self, value: int):
386 self._buffer_info.original_height = value
387
388 @original_width.setter
389 def original_width(self, value: int):
390 self._buffer_info.original_width = value
391
392 @original_channel.setter
393 def original_channel(self, value: int):
394 self._buffer_info.original_channel = value
395
396 @reshaped_height.setter
397 def reshaped_height(self, value: int):
398 self._buffer_info.reshaped_height = value
399
400 @reshaped_width.setter
401 def reshaped_width(self, value: int):
402 self._buffer_info.reshaped_width = value
403
404 @reshaped_channel.setter
405 def reshaped_channel(self, value: int):
406 self._buffer_info.reshaped_channel = value
407
408 @height.setter
409 def height(self, value: int):
410 self._buffer_info.height = value
411
412 @width.setter
413 def width(self, value: int):
414 self._buffer_info.width = value
415
416 @channel.setter
417 def channel(self, value: int):
418 self._buffer_info.channel = value
419
420 @max_height.setter
421 def max_height(self, value: int):
422 self._buffer_info.max_height = value
423
424 @max_width.setter
425 def max_width(self, value: int):
426 self._buffer_info.max_width = value
427
428 @max_channel.setter
429 def max_channel(self, value: int):
430 self._buffer_info.max_channel = value
431
432 @max_cache_size.setter
433 def max_cache_size(self, value: int):
434 self._buffer_info.max_cache_size = value
435
436 def original_size(self) -> int:
437 """
438 @brief Returns the total size of the original input/output.
439
440 @return The data size.
441 """
442 return self._buffer_info.original_size()
443
444 def reshaped_size(self) -> int:
445 """
446 @brief Returns the total size of the reshaped input/output.
447
448 @return The data size.
449 """
450 return self._buffer_info.reshaped_size()
451
452 def size(self) -> int:
453 """
454 @brief Returns the total size of the NPU input/output.
455
456 @return The data size.
457 """
458 return self._buffer_info.size()
459
460 def original_shape(self) -> Tuple[int, int, int]:
461 return self._buffer_info.original_shape()
462
463 def original_shape_chw(self) -> Tuple[int, int, int]:
464 return self._buffer_info.original_shape_chw()
465
466 def reshaped_shape(self) -> Tuple[int, int, int]:
467 return self._buffer_info.reshaped_shape()
468
469 def reshaped_shape_chw(self) -> Tuple[int, int, int]:
470 return self._buffer_info.reshaped_shape_chw()
471
472 def shape(self) -> Tuple[int, int, int]:
473 return self._buffer_info.shape()
474
475 def shape_chw(self) -> Tuple[int, int, int]:
476 return self._buffer_info.shape_chw()
477
478 def __repr__(self):
479 d = {
480 "original_height": self._buffer_info.original_height,
481 "original_width": self._buffer_info.original_width,
482 "original_channel": self._buffer_info.original_channel,
483 "reshaped_height": self._buffer_info.reshaped_height,
484 "reshaped_width": self._buffer_info.reshaped_width,
485 "reshaped_channel": self._buffer_info.reshaped_channel,
486 "height": self._buffer_info.height,
487 "width": self._buffer_info.width,
488 "channel": self._buffer_info.channel,
489 "max_height": self._buffer_info.max_height,
490 "max_width": self._buffer_info.max_width,
491 "max_channel": self._buffer_info.max_channel,
492 "max_cache_size": self._buffer_info.max_cache_size,
493 }
494 return "{}({})".format(
495 self.__class__.__name__,
496 ", ".join("{}={}".format(k, v) for k, v in d.items()),
497 )
498
499
501 """
502 @brief Configures a core mode and core allocation of a model for NPU inference.
503 The `ModelConfig` class provides methods for setting a core mode and allocating
504 cores for NPU inference. Supported core modes are single-core, multi-core,
505 global4-core, and global8-core. Users can also specify which cores to allocate for
506 the model. Additionally, the configuration offers an option to enforce the use of a
507 specific NPU bundle.
508
509 @note Deprecated functions are included for backward compatibility, but it is
510 recommended to use the newer core mode configuration methods.
511 """
512
513 def __init__(self, num_cores: Optional[int] = None):
514 """
515 @brief Default constructor. This default-constructed object is initially set to
516 auto-core mode.
517 """
518 self._model_config = (
519 _cQbRuntime.ModelConfig()
520 if num_cores is None
521 else _cQbRuntime.ModelConfig(num_cores)
522 )
523
524 def set_auto_core_mode(self) -> bool:
525 """
526 @brief Sets the model to detect CoreMode automatically.
527
528 In auto-core mode, the model automatically detects a supported CoreMode
529 while using all available NPU cores.
530
531 @note If the model has more than one CoreMode, `CoreMode.Auto` is not supported.
532
533 @note The activation buffer slot count is reset to -1 when `set_auto_core_mode`
534 is called. A value of -1 means "not set": when the model is created, the
535 slot count is decided from the detected CoreMode of each bundle, following
536 the same per-CoreMode defaults as `set_activation_slots` describes. Call
537 `set_activation_slots` after `set_auto_core_mode` to override it.
538
539 @return True if the mode was successfully set, False otherwise.
540 """
542
543 @overload
544 def set_single_core_mode(self, num_cores: int) -> bool: ...
545
546 @overload
547 def set_single_core_mode(self, core_ids: List[CoreId]) -> bool: ...
548
549 def set_single_core_mode(self, num_cores=None, core_ids=None) -> bool:
550 """
551 @brief Sets the model to use single-core mode for inference with a specified number
552 of local cores.
553
554 In single-core mode, each local core executes model inference independently.
555 The number of cores used is specified by the `num_cores` parameter, and the core
556 allocation policy is set to `CoreAllocationPolicy.Auto`, meaning the model will be
557 automatically allocated to available local cores when the model is launched to the
558 NPU, specifically when the `Model.launch()` function is called. Or The user can
559 specify a list of CoreIds to determine which cores to use for inference.
560
561 @param[in] num_cores The number of local cores to use for inference.
562 @param[in] core_ids A list of CoreIds to be used for model inference.
563
564 @return True if the mode was successfully set, False otherwise.
565
566 @note Supported call forms (provide exactly one of `num_cores` / `core_ids`):
567 - `set_single_core_mode(num_cores)`:
568 auto-allocate `num_cores` local cores. (deprecated in latest models)
569 - `set_single_core_mode(core_ids)`:
570 use exactly the cores listed in `core_ids`.
571 """
572 # set_single_core_mode(core_ids) constructor
573 if isinstance(num_cores, (list, tuple)) and core_ids is None:
574 num_cores, core_ids = None, num_cores
575 if num_cores is not None and core_ids is None:
576 return self._model_config.set_single_core_mode(num_cores)
577 elif core_ids is not None and num_cores is None:
578 return self._model_config.set_single_core_mode(
579 [core_id._core_id for core_id in core_ids]
580 )
581 raise ValueError(
582 "`set_single_core_mode` needs either `num_cores` or `core_ids`."
583 )
584
585 def set_global_core_mode(self, clusters: List[Cluster]) -> bool:
586 """@deprecated"""
587 return self._model_config.set_global_core_mode([c.value for c in clusters])
588
590 self, clusters: List[Cluster] = [Cluster.Cluster0, Cluster.Cluster1]
591 ) -> bool:
592 """
593 @brief Sets the model to use global4-core mode for inference with a specified set
594 of NPU clusters.
595
596 For Aries NPU, there are two clusters, each consisting of four local cores. In
597 global4-core mode, four local cores within the same cluster work together to
598 execute the model inference.
599
600 @param[in] clusters A list of clusters to be used for model inference.
601
602 @return True if the mode was successfully set, False otherwise.
603 """
604 return self._model_config.set_global4_core_mode([c.value for c in clusters])
605
606 def set_global8_core_mode(self) -> bool:
607 """
608 @brief Sets the model to use global8-core mode for inference.
609
610 For Aries NPU, there are two clusters, each consisting of four local cores. In
611 global8-core mode, all eight local cores across the two clusters work together to
612 execute the model inference.
613
614 @return True if the mode was successfully set, False otherwise.
615 """
617
618 def get_core_mode(self) -> CoreMode:
619 """
620 @brief Gets the core mode to be applied to the model.
621
622 This reflects the core mode that will be used when the model is created.
623
624 @return The `CoreMode` to be applied to the model.
625 """
627
629 self, clusters: List[Cluster] = [Cluster.Cluster0, Cluster.Cluster1]
630 ) -> bool:
631 """
632 @brief Sets the model to use multi-core mode for batch inference.
633
634 In multi-core mode, on Aries NPU, the four local cores within a cluster work
635 together to process batch inference tasks efficiently. This mode is optimized for
636 batch processing.
637
638 @param[in] clusters A list of clusters to be used for multi-core batch inference.
639
640 @return True if the mode was successfully set, False otherwise.
641 """
642 return self._model_config.set_multi_core_mode([c.value for c in clusters])
643
644 def get_core_allocation_policy(self) -> CoreAllocationPolicy:
645 """
646 @brief Gets the core allocation policy to be applied to the model.
647
648 This reflects the core allocation policy that will be used when the model is
649 created.
650
651 @return The `CoreAllocationPolicy` to be applied to the model.
652 """
654
655 def get_num_cores(self) -> int:
656 """
657 @brief Gets the number of cores to be allocated for the model.
658
659 This represents the number of cores that will be allocated for inference
660 when the model is launched to the NPU.
661
662 @return The number of cores to be allocated for the model.
663 """
664 return self._model_config.get_num_cores()
665
666 def force_single_npu_bundle(self, npu_bundle_index: int) -> bool:
667 """
668 @brief Forces the use of a specific NPU bundle.
669
670 This function forces the selection of a specific NPU bundle. If a non-negative
671 index is provided, the corresponding NPU bundle is selected and runs without CPU
672 offloading. If -1 is provided, all NPU bundles are used with CPU offloading
673 enabled.
674
675 @param[in] npu_bundle_index The index of the NPU bundle to force. A non-negative
676 integer selects a specific NPU bundle (runs without CPU
677 offloading), or -1 to enable all NPU bundles with CPU
678 offloading.
679
680 @return True if the index is valid and the NPU bundle is successfully set,
681 False if the index is invalid (less than -1).
682 """
683 return self._model_config.force_single_npu_bundle(npu_bundle_index)
684
686 """
687 @brief Retrieves the index of the forced NPU bundle.
688
689 This function returns the index of the NPU bundle that has been forced using the
690 `force_single_npu_bundle` function. If no NPU bundle is forced, the returned value
691 will be -1.
692
693 @return The index of the forced NPU bundle, or -1 if no bundle is forced.
694 """
696
697 def set_async_pipeline_enabled(self, enable: bool) -> None:
698 """
699 @brief Enables or disables the asynchronous pipeline required for asynchronous
700 inference.
701
702 Call this function with `enable` set to `True` if you intend to use
703 `Model.infer_async()`, as the asynchronous pipeline is necessary for their operation.
704
705 If you are only using synchronous inference, such as `Model.infer()` or
706 `Model.infer_to_float()`, it is recommended to keep the asynchronous pipeline disabled
707 to avoid unnecessary overhead.
708
709 @param[in] enable Set to `True` to enable the asynchronous pipeline; set to `False`
710 to disable it.
711 """
712 return self._model_config.set_async_pipeline_enabled(enable)
713
714 def get_async_pipeline_enabled(self) -> bool:
715 """
716 @brief Returns whether the asynchronous pipeline is enabled in this configuration.
717
718 @return `True` if the asynchronous pipeline is enabled; `False` otherwise.
719 """
721
722 def set_activation_slots(self, num: int) -> None:
723 """
724 @brief Sets activation buffer slots for multi-activation supported model.
725
726 all this function if you want to set the number of activation buffer slots manually.
727
728 If you do not call this function, the default number of activation buffer slots
729 is set differently depending on the CoreMode.
730
731 - `CoreMode.Single` : 2 * (the number of target core ids)
732 - `CoreMode.Multi` : 2 * (the number of target clusters)
733 - `CoreMode.Global4` : 2 * (the number of target clusters)
734 - `CoreMode.Global8` : 2
735
736 @note This function has no effect on MXQ file in version earlier than MXQv7.
737
738 @note Currently, LLM model's activation slot is fixed to 1 and ignoring `count`.
739
740 @param[in] count Multi activation counts. Must be >= 1.
741 """
742 return self._model_config.set_activation_slots(num)
743
744 def get_activation_slots(self) -> int:
745 """
746 @brief Returns activation buffer slot count.
747
748 @note This function has no meaning on MXQ file in version earlier than MXQv7.
749
750 @return Activation buffer slot count.
751 """
753
754 def get_core_ids(self) -> List[CoreId]:
755 """
756 @brief Returns the list of NPU CoreIds to be used for model inference.
757
758 The returned list reflects only what was set on this ModelConfig.
759 What it contains depends on the configured CoreMode:
760 - Single mode via `set_single_core_mode(core_ids)`: the specified core_ids.
761 - Single mode via `set_single_core_mode(num_cores)`: empty. The actual cores
762 are resolved from the target NPU when the model is created.
763 - Multi, Global4, Global8, and Auto modes: empty. Use `get_clusters()` for the
764 cluster-based modes.
765
766 @return A list of NPU CoreIds.
767 """
768 return [
769 CoreId(Cluster(core_id.cluster), Core(core_id.core))
770 for core_id in self._model_config.core_ids
771 ]
772
773 def get_clusters(self) -> List[Cluster]:
774 """
775 @brief Returns the list of clusters to be used for model inference.
776
777 The returned list reflects only what was set on this ModelConfig.
778 What it contains depends on the configured CoreMode:
779 - Multi mode via `set_multi_core_mode()` and Global4 mode via
780 `set_global4_core_mode()`: the specified clusters (defaults to all clusters).
781 - Single, Global8, and Auto modes: empty.
782
783 @return A list of NPU Clusters.
784 """
785 return [Cluster(cluster) for cluster in self._model_config.clusters]
786
787 def __repr__(self):
788 d = {
789 "core_mode": self.get_core_mode(),
790 "core_allocation_policy": self.get_core_allocation_policy(),
791 "core_ids": self.get_core_ids(),
792 "clusters": self.get_clusters(),
793 "num_cores": self.get_num_cores(),
794 "forced_npu_bundle_index": self.get_forced_npu_bundle_index(),
795 }
796 return "{}({})".format(
797 self.__class__.__name__,
798 ", ".join("{}={}".format(k, v) for k, v in d.items()),
799 )
800
801
802class LogLevel(Enum):
803 """@brief LogLevel"""
804
805 Debug = _cQbRuntime.LogLevel.Debug
806 Info = _cQbRuntime.LogLevel.Info
807 Warn = _cQbRuntime.LogLevel.Warn
808 Err = _cQbRuntime.LogLevel.Err
809 Fatal = _cQbRuntime.LogLevel.Fatal
810 Off = _cQbRuntime.LogLevel.Off
811
812
813def set_log_level(level: LogLevel):
814 _cQbRuntime.set_log_level(level.value)
815
816
817class CacheType(Enum):
818 """@brief CacheType"""
819
820 Default = _cQbRuntime.CacheType.Default
821 Batch = _cQbRuntime.CacheType.Batch
822 FixedTailCache = _cQbRuntime.CacheType.FixedTailCache
823 Error = _cQbRuntime.CacheType.Error
824
825
827 """@brief Struct representing KV-cache information."""
828
829 def __init__(
830 self,
831 cache_type: CacheType = CacheType.Error,
832 name: str = "",
833 layer_hash: str = "",
834 size: int = 0,
835 num_batches: int = 0,
836 ):
837 self._cache_info = _cQbRuntime.CacheInfo()
838 self._cache_info.cache_type = cache_type.value
839 self._cache_info.name = name
840 self._cache_info.layer_hash = layer_hash
841 self._cache_info.size = size
842 self._cache_info.num_batches = num_batches
843
844 @classmethod
845 def from_cpp(cls, _cache_info: _cQbRuntime.CacheInfo):
846 return cls(
847 CacheType(_cache_info.cache_type),
848 _cache_info.name,
849 _cache_info.layer_hash,
850 _cache_info.size,
851 _cache_info.num_batches,
852 )
853
854 @property
855 def cache_type(self) -> CacheType:
856 return CacheType(self._cache_info.cache_type)
857
858 @property
859 def name(self) -> str:
860 return self._cache_info.name
861
862 @property
863 def layer_hash(self) -> str:
864 return self._cache_info.layer_hash
865
866 @property
867 def size(self) -> int:
868 return self._cache_info.size
869
870 @property
871 def num_batches(self) -> int:
872 return self._cache_info.num_batches
873
874 @cache_type.setter
875 def cache_type(self, value: CacheType):
876 self._cache_info.cache_type = value.value
877
878 @name.setter
879 def name(self, value: str):
880 self._cache_info.name = value
881
882 @layer_hash.setter
883 def layer_hash(self, value: str):
884 self._cache_info.layer_hash = value
885
886 @size.setter
887 def size(self, value: int):
888 self._cache_info.size = value
889
890 @num_batches.setter
891 def num_batches(self, value: int):
892 self._cache_info.num_batches = value
893
894
895class DataType(Enum):
896 """@brief DataType"""
897
898 Float32 = _cQbRuntime.DataType.Float32
899 Float16 = _cQbRuntime.DataType.Float16
900 Int8 = _cQbRuntime.DataType.Int8
901 Uint8 = _cQbRuntime.DataType.Uint8
902 BInt16 = _cQbRuntime.DataType.BInt16
903 Error = _cQbRuntime.DataType.Error
904
905
907 """@brief Struct containing BatchLLM parameters."""
908
909 def __init__(
910 self,
911 sequence_length: int = 0,
912 cache_size: int = 0,
913 cache_id: int = 0,
914 ):
915 self._batch_param = _cQbRuntime.BatchParam()
916 self._batch_param.sequence_length = sequence_length
917 self._batch_param.cache_size = cache_size
918 self._batch_param.cache_id = cache_id
919
920 def __repr__(self):
921 d = {
922 "sequence_length": self.sequence_length,
923 "cache_size": self.cache_size,
924 "cache_id": self.cache_id,
925 }
926 return "{}({})".format(
927 self.__class__.__name__,
928 ", ".join("{}={}".format(k, v) for k, v in d.items()),
929 )
930
931 @property
932 def sequence_length(self) -> int:
933 return self._batch_param.sequence_length
934
935 @property
936 def cache_size(self) -> int:
937 return self._batch_param.cache_size
938
939 @property
940 def cache_id(self) -> int:
941 return self._batch_param.cache_id
942
943 @sequence_length.setter
944 def sequence_length(self, value: int):
945 self._batch_param.sequence_length = value
946
947 @cache_size.setter
948 def cache_size(self, value: int):
949 self._batch_param.cache_size = value
950
951 @cache_id.setter
952 def cache_id(self, value: int):
953 self._batch_param.cache_id = value
954
955
956def start_tracing_events(path: str) -> bool:
957 """
958 @brief Starts event tracing and prepares to save the trace log to a specified file.
959
960 The trace log is recorded in "Chrome Tracing JSON format," which can be
961 viewed at https://ui.perfetto.dev/.
962
963 The trace log is not written immediately; it is saved only when
964 stop_tracing_events() is called.
965
966 @param[in] path The file path where the trace log should be stored.
967 @return True if tracing starts successfully, False otherwise.
968 """
969 return _cQbRuntime.start_tracing_events(path)
970
971
973 """
974 @brief Stops event tracing and writes the recorded trace log.
975
976 This function finalizes tracing and saves the collected trace data
977 to the file specified when start_tracing_events() was called.
978 """
979 _cQbRuntime.stop_tracing_events()
980
981
982def get_model_summary(mxq_path: str) -> str:
983 """
984 @brief Generates a structured summary of the specified MXQ model.
985
986 Returns an overview of the model contained in the MXQ file, including:
987 - Target NPU device
988 - Supported core modes and their associated cores
989 - The total number of model variants
990 - For each variant:
991 - Input and output tensor shapes
992 - A list of layers with their types, output shapes, and input layer indices
993
994 The summary is returned as a human-readable string in a table and is useful for
995 inspecting model compatibility, structure, and input/output shapes.
996
997 @param[in] mxq_path Path to the MXQ model file.
998 @return A formatted string containing the model summary.
999 """
1000 return _cQbRuntime.get_model_summary(mxq_path)
1001
1002
1003def get_available_devices() -> List[str]:
1004 """
1005 @brief Get the device names of the NPU devices detected on the system.
1006
1007 @return The detected device names (e.g. "aries-rb", "regulus-ra").
1008 Empty if no NPU device is detected.
1009 """
1010 return _cQbRuntime.get_available_devices()
1011
1012
1013def get_available_device_numbers(device_name: str = "auto") -> List[int]:
1014 """
1015 @brief Get the available device numbers of the given device.
1016
1017 @param device_name The device name ("auto" or a name such as "aries", "aries-rb",
1018 "regulus", "regulus-ra"; case-insensitive).
1019 @return The available device numbers. Empty if `device_name` cannot be resolved.
1020 """
1021 return _cQbRuntime.get_available_device_numbers(device_name)
1022
1023
1024
Struct containing BatchLLM parameters.
Definition type.py:906
int sequence_length(self)
Definition type.py:932
Struct representing input/output buffer information.
Definition type.py:267
int reshaped_width(self)
Width of reshaped input/output.
Definition type.py:340
int max_channel(self)
Maximum channel of original input/output if data is sequential.
Definition type.py:375
int original_height(self)
Height of original input/output.
Definition type.py:320
int width(self)
Width of NPU input/output.
Definition type.py:355
int max_cache_size(self)
Maximum KV-cache size, relevant for LLM models using KV cache.
Definition type.py:380
int max_height(self)
Maximum height of original input/output if data is sequential.
Definition type.py:365
int original_size(self)
Returns the total size of the original input/output.
Definition type.py:436
int height(self)
Height of NPU input/output.
Definition type.py:350
int original_width(self)
Width of original input/output.
Definition type.py:325
int channel(self)
Channel of NPU input/output.
Definition type.py:360
int size(self)
Returns the total size of the NPU input/output.
Definition type.py:452
int max_width(self)
Maximum width of original input/output if data is sequential.
Definition type.py:370
int reshaped_channel(self)
Channel of reshaped input/output.
Definition type.py:345
int original_channel(self)
Channel of original input/output.
Definition type.py:330
int reshaped_size(self)
Returns the total size of the reshaped input/output.
Definition type.py:444
int reshaped_height(self)
Height of reshaped input/output.
Definition type.py:335
A simple byte-sized buffer.
Definition type.py:219
Struct representing KV-cache information.
Definition type.py:826
Enumerates clusters in the ARIES NPU.
Definition type.py:17
Core allocation policy.
Definition type.py:49
Represents a unique identifier for an NPU core.
Definition type.py:162
bool __eq__(self, other)
Checks if two CoreId objects are equal.
Definition type.py:195
Cluster cluster(self)
Definition type.py:180
Core core(self)
Definition type.py:184
bool __lt__(self, other)
Compares two CoreId objects for ordering.
Definition type.py:203
Defines the core mode for NPU execution.
Definition type.py:244
Enumerates cores within a cluster in the ARIES NPU.
Definition type.py:31
Configures a core mode and core allocation of a model for NPU inference.
Definition type.py:500
List[CoreId] get_core_ids(self)
Returns the list of NPU CoreIds to be used for model inference.
Definition type.py:754
bool get_forced_npu_bundle_index(self)
Retrieves the index of the forced NPU bundle.
Definition type.py:685
bool set_global8_core_mode(self)
Sets the model to use global8-core mode for inference.
Definition type.py:606
bool set_multi_core_mode(self, List[Cluster] clusters=[Cluster.Cluster0, Cluster.Cluster1])
Sets the model to use multi-core mode for batch inference.
Definition type.py:630
bool set_global_core_mode(self, List[Cluster] clusters)
Definition type.py:585
bool force_single_npu_bundle(self, int npu_bundle_index)
Forces the use of a specific NPU bundle.
Definition type.py:666
__init__(self, Optional[int] num_cores=None)
Default constructor.
Definition type.py:513
List[Cluster] get_clusters(self)
Returns the list of clusters to be used for model inference.
Definition type.py:773
CoreMode get_core_mode(self)
Gets the core mode to be applied to the model.
Definition type.py:618
None set_async_pipeline_enabled(self, bool enable)
Enables or disables the asynchronous pipeline required for asynchronous inference.
Definition type.py:697
bool get_async_pipeline_enabled(self)
Returns whether the asynchronous pipeline is enabled in this configuration.
Definition type.py:714
int get_activation_slots(self)
Returns activation buffer slot count.
Definition type.py:744
bool set_auto_core_mode(self)
Sets the model to detect CoreMode automatically.
Definition type.py:524
None set_activation_slots(self, int num)
Sets activation buffer slots for multi-activation supported model.
Definition type.py:722
int get_num_cores(self)
Gets the number of cores to be allocated for the model.
Definition type.py:655
bool set_global4_core_mode(self, List[Cluster] clusters=[Cluster.Cluster0, Cluster.Cluster1])
Sets the model to use global4-core mode for inference with a specified set of NPU clusters.
Definition type.py:591
CoreAllocationPolicy get_core_allocation_policy(self)
Gets the core allocation policy to be applied to the model.
Definition type.py:644
Struct for scale values.
Definition type.py:56
List[int] zero_points(self)
Per-channel zero points for asymmetric quantization.
Definition type.py:100
int zero_point(self)
Uniform zero point for asymmetric quantization.
Definition type.py:105
bool is_asymmetric(self)
Indicates whether asymmetric quantization is used.
Definition type.py:110
bool is_uniform(self)
Definition type.py:96
float scale(self)
Definition type.py:92
float __getitem__(self, int i)
Returns the scale value at the specified index.
Definition type.py:138
List[float] scale_list(self)
Definition type.py:88
List[str] get_available_devices()
Get the device names of the NPU devices detected on the system.
Definition type.py:1003
str get_model_summary(str mxq_path)
Generates a structured summary of the specified MXQ model.
Definition type.py:982
List[int] get_available_device_numbers(str device_name="auto")
Get the available device numbers of the given device.
Definition type.py:1013
bool start_tracing_events(str path)
Starts event tracing and prepares to save the trace log to a specified file.
Definition type.py:956
stop_tracing_events()
Stops event tracing and writes the recorded trace log.
Definition type.py:972