type.py Source File

type.py Source File#

SDK qb Runtime Library: type.py Source File
SDK qb Runtime Library v1.4
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 activation buffer slots will be reset after `set_auto_core_mode` is called.
534
535 @return True if the mode was successfully set, False otherwise.
536 """
538
539 @overload
540 def set_single_core_mode(self, num_cores: int) -> bool: ...
541
542 @overload
543 def set_single_core_mode(self, core_ids: List[CoreId]) -> bool: ...
544
545 def set_single_core_mode(self, num_cores=None, core_ids=None) -> bool:
546 """
547 @brief Sets the model to use single-core mode for inference with a specified number
548 of local cores.
549
550 In single-core mode, each local core executes model inference independently.
551 The number of cores used is specified by the `num_cores` parameter, and the core
552 allocation policy is set to `CoreAllocationPolicy.Auto`, meaning the model will be
553 automatically allocated to available local cores when the model is launched to the
554 NPU, specifically when the `Model.launch()` function is called. Or The user can
555 specify a list of CoreIds to determine which cores to use for inference.
556
557 @param[in] num_cores The number of local cores to use for inference.
558 @param[in] core_ids A list of CoreIds to be used for model inference.
559
560 @return True if the mode was successfully set, False otherwise.
561
562 @note Supported call forms (provide exactly one of `num_cores` / `core_ids`):
563 - `set_single_core_mode(num_cores)`:
564 auto-allocate `num_cores` local cores. (deprecated in latest models)
565 - `set_single_core_mode(core_ids)`:
566 use exactly the cores listed in `core_ids`.
567 """
568 # set_single_core_mode(core_ids) constructor
569 if isinstance(num_cores, (list, tuple)) and core_ids is None:
570 num_cores, core_ids = None, num_cores
571 if num_cores is not None and core_ids is None:
572 return self._model_config.set_single_core_mode(num_cores)
573 elif core_ids is not None and num_cores is None:
574 return self._model_config.set_single_core_mode(
575 [core_id._core_id for core_id in core_ids]
576 )
577 raise ValueError(
578 "`set_single_core_mode` needs either `num_cores` or `core_ids`."
579 )
580
581 def set_global_core_mode(self, clusters: List[Cluster]) -> bool:
582 """@deprecated"""
583 return self._model_config.set_global_core_mode([c.value for c in clusters])
584
586 self, clusters: List[Cluster] = [Cluster.Cluster0, Cluster.Cluster1]
587 ) -> bool:
588 """
589 @brief Sets the model to use global4-core mode for inference with a specified set
590 of NPU clusters.
591
592 For Aries NPU, there are two clusters, each consisting of four local cores. In
593 global4-core mode, four local cores within the same cluster work together to
594 execute the model inference.
595
596 @param[in] clusters A list of clusters to be used for model inference.
597
598 @return True if the mode was successfully set, False otherwise.
599 """
600 return self._model_config.set_global4_core_mode([c.value for c in clusters])
601
602 def set_global8_core_mode(self) -> bool:
603 """
604 @brief Sets the model to use global8-core mode for inference.
605
606 For Aries NPU, there are two clusters, each consisting of four local cores. In
607 global8-core mode, all eight local cores across the two clusters work together to
608 execute the model inference.
609
610 @return True if the mode was successfully set, False otherwise.
611 """
613
614 def get_core_mode(self) -> CoreMode:
615 """
616 @brief Gets the core mode to be applied to the model.
617
618 This reflects the core mode that will be used when the model is created.
619
620 @return The `CoreMode` to be applied to the model.
621 """
623
625 self, clusters: List[Cluster] = [Cluster.Cluster0, Cluster.Cluster1]
626 ) -> bool:
627 """
628 @brief Sets the model to use multi-core mode for batch inference.
629
630 In multi-core mode, on Aries NPU, the four local cores within a cluster work
631 together to process batch inference tasks efficiently. This mode is optimized for
632 batch processing.
633
634 @param[in] clusters A list of clusters to be used for multi-core batch inference.
635
636 @return True if the mode was successfully set, False otherwise.
637 """
638 return self._model_config.set_multi_core_mode([c.value for c in clusters])
639
640 def get_core_allocation_policy(self) -> CoreAllocationPolicy:
641 """
642 @brief Gets the core allocation policy to be applied to the model.
643
644 This reflects the core allocation policy that will be used when the model is
645 created.
646
647 @return The `CoreAllocationPolicy` to be applied to the model.
648 """
650
651 def get_num_cores(self) -> int:
652 """
653 @brief Gets the number of cores to be allocated for the model.
654
655 This represents the number of cores that will be allocated for inference
656 when the model is launched to the NPU.
657
658 @return The number of cores to be allocated for the model.
659 """
660 return self._model_config.get_num_cores()
661
662 def force_single_npu_bundle(self, npu_bundle_index: int) -> bool:
663 """
664 @brief Forces the use of a specific NPU bundle.
665
666 This function forces the selection of a specific NPU bundle. If a non-negative
667 index is provided, the corresponding NPU bundle is selected and runs without CPU
668 offloading. If -1 is provided, all NPU bundles are used with CPU offloading
669 enabled.
670
671 @param[in] npu_bundle_index The index of the NPU bundle to force. A non-negative
672 integer selects a specific NPU bundle (runs without CPU
673 offloading), or -1 to enable all NPU bundles with CPU
674 offloading.
675
676 @return True if the index is valid and the NPU bundle is successfully set,
677 False if the index is invalid (less than -1).
678 """
679 return self._model_config.force_single_npu_bundle(npu_bundle_index)
680
682 """
683 @brief Retrieves the index of the forced NPU bundle.
684
685 This function returns the index of the NPU bundle that has been forced using the
686 `force_single_npu_bundle` function. If no NPU bundle is forced, the returned value
687 will be -1.
688
689 @return The index of the forced NPU bundle, or -1 if no bundle is forced.
690 """
692
693 def set_async_pipeline_enabled(self, enable: bool) -> None:
694 """
695 @brief Enables or disables the asynchronous pipeline required for asynchronous
696 inference.
697
698 Call this function with `enable` set to `True` if you intend to use
699 `Model.infer_async()`, as the asynchronous pipeline is necessary for their operation.
700
701 If you are only using synchronous inference, such as `Model.infer()` or
702 `Model.infer_to_float()`, it is recommended to keep the asynchronous pipeline disabled
703 to avoid unnecessary overhead.
704
705 @param[in] enable Set to `True` to enable the asynchronous pipeline; set to `False`
706 to disable it.
707 """
708 return self._model_config.set_async_pipeline_enabled(enable)
709
710 def get_async_pipeline_enabled(self) -> bool:
711 """
712 @brief Returns whether the asynchronous pipeline is enabled in this configuration.
713
714 @return `True` if the asynchronous pipeline is enabled; `False` otherwise.
715 """
717
718 def set_activation_slots(self, num: int) -> None:
719 """
720 @brief Sets activation buffer slots for multi-activation supported model.
721
722 all this function if you want to set the number of activation buffer slots manually.
723
724 If you do not call this function, the default number of activation buffer slots
725 is set differently depending on the CoreMode.
726
727 - `CoreMode.Single` : 2 * (the number of target core ids)
728 - `CoreMode.Multi` : 2 * (the number of target clusters)
729 - `CoreMode.Global4` : 2 * (the number of target clusters)
730 - `CoreMode.Global8` : 2
731
732 @note This function has no effect on MXQ file in version earlier than MXQv7.
733
734 @note Currently, LLM model's activation slot is fixed to 1 and ignoring `count`.
735
736 @param[in] count Multi activation counts. Must be >= 1.
737 """
738 return self._model_config.set_activation_slots(num)
739
740 def get_activation_slots(self) -> int:
741 """
742 @brief Returns activation buffer slot count.
743
744 @note This function has no meaning on MXQ file in version earlier than MXQv7.
745
746 @return Activation buffer slot count.
747 """
749
750 def get_core_ids(self) -> List[CoreId]:
751 """
752 @brief Returns the list of NPU CoreIds to be used for model inference.
753
754 The returned list reflects only what was set on this ModelConfig.
755 What it contains depends on the configured CoreMode:
756 - Single mode via `set_single_core_mode(core_ids)`: the specified core_ids.
757 - Single mode via `set_single_core_mode(num_cores)`: empty. The actual cores
758 are resolved from the target NPU when the model is created.
759 - Multi, Global4, Global8, and Auto modes: empty. Use `get_clusters()` for the
760 cluster-based modes.
761
762 @return A list of NPU CoreIds.
763 """
764 return [
765 CoreId(Cluster(core_id.cluster), Core(core_id.core))
766 for core_id in self._model_config.core_ids
767 ]
768
769 def get_clusters(self) -> List[Cluster]:
770 """
771 @brief Returns the list of clusters to be used for model inference.
772
773 The returned list reflects only what was set on this ModelConfig.
774 What it contains depends on the configured CoreMode:
775 - Multi mode via `set_multi_core_mode()` and Global4 mode via
776 `set_global4_core_mode()`: the specified clusters (defaults to all clusters).
777 - Single, Global8, and Auto modes: empty.
778
779 @return A list of NPU Clusters.
780 """
781 return [Cluster(cluster) for cluster in self._model_config.clusters]
782
783 def __repr__(self):
784 d = {
785 "core_mode": self.get_core_mode(),
786 "core_allocation_policy": self.get_core_allocation_policy(),
787 "core_ids": self.get_core_ids(),
788 "clusters": self.get_clusters(),
789 "num_cores": self.get_num_cores(),
790 "forced_npu_bundle_index": self.get_forced_npu_bundle_index(),
791 }
792 return "{}({})".format(
793 self.__class__.__name__,
794 ", ".join("{}={}".format(k, v) for k, v in d.items()),
795 )
796
797
798class LogLevel(Enum):
799 """@brief LogLevel"""
800
801 Debug = _cQbRuntime.LogLevel.Debug
802 Info = _cQbRuntime.LogLevel.Info
803 Warn = _cQbRuntime.LogLevel.Warn
804 Err = _cQbRuntime.LogLevel.Err
805 Fatal = _cQbRuntime.LogLevel.Fatal
806 Off = _cQbRuntime.LogLevel.Off
807
808
809def set_log_level(level: LogLevel):
810 _cQbRuntime.set_log_level(level.value)
811
812
813class CacheType(Enum):
814 """@brief CacheType"""
815
816 Default = _cQbRuntime.CacheType.Default
817 Batch = _cQbRuntime.CacheType.Batch
818 Error = _cQbRuntime.CacheType.Error
819
820
822 """@brief Struct representing KV-cache information."""
823
824 def __init__(
825 self,
826 cache_type: CacheType = CacheType.Error,
827 name: str = "",
828 layer_hash: str = "",
829 size: int = 0,
830 num_batches: int = 0,
831 ):
832 self._cache_info = _cQbRuntime.CacheInfo()
833 self._cache_info.cache_type = cache_type.value
834 self._cache_info.name = name
835 self._cache_info.layer_hash = layer_hash
836 self._cache_info.size = size
837 self._cache_info.num_batches = num_batches
838
839 @classmethod
840 def from_cpp(cls, _cache_info: _cQbRuntime.CacheInfo):
841 return cls(
842 CacheType(_cache_info.cache_type),
843 _cache_info.name,
844 _cache_info.layer_hash,
845 _cache_info.size,
846 _cache_info.num_batches,
847 )
848
849 @property
850 def cache_type(self) -> CacheType:
851 return CacheType(self._cache_info.cache_type)
852
853 @property
854 def name(self) -> str:
855 return self._cache_info.name
856
857 @property
858 def layer_hash(self) -> str:
859 return self._cache_info.layer_hash
860
861 @property
862 def size(self) -> int:
863 return self._cache_info.size
864
865 @property
866 def num_batches(self) -> int:
867 return self._cache_info.num_batches
868
869 @cache_type.setter
870 def cache_type(self, value: CacheType):
871 self._cache_info.cache_type = value.value
872
873 @name.setter
874 def name(self, value: str):
875 self._cache_info.name = value
876
877 @layer_hash.setter
878 def layer_hash(self, value: str):
879 self._cache_info.layer_hash = value
880
881 @size.setter
882 def size(self, value: int):
883 self._cache_info.size = value
884
885 @num_batches.setter
886 def num_batches(self, value: int):
887 self._cache_info.num_batches = value
888
889
890class DataType(Enum):
891 """@brief DataType"""
892
893 Float32 = _cQbRuntime.DataType.Float32
894 Float16 = _cQbRuntime.DataType.Float16
895 Int8 = _cQbRuntime.DataType.Int8
896 Uint8 = _cQbRuntime.DataType.Uint8
897 BInt16 = _cQbRuntime.DataType.BInt16
898 Error = _cQbRuntime.DataType.Error
899
900
902 """@brief Struct containing BatchLLM parameters."""
903
904 def __init__(
905 self,
906 sequence_length: int = 0,
907 cache_size: int = 0,
908 cache_id: int = 0,
909 ):
910 self._batch_param = _cQbRuntime.BatchParam()
911 self._batch_param.sequence_length = sequence_length
912 self._batch_param.cache_size = cache_size
913 self._batch_param.cache_id = cache_id
914
915 def __repr__(self):
916 d = {
917 "sequence_length": self.sequence_length,
918 "cache_size": self.cache_size,
919 "cache_id": self.cache_id,
920 }
921 return "{}({})".format(
922 self.__class__.__name__,
923 ", ".join("{}={}".format(k, v) for k, v in d.items()),
924 )
925
926 @property
927 def sequence_length(self) -> int:
928 return self._batch_param.sequence_length
929
930 @property
931 def cache_size(self) -> int:
932 return self._batch_param.cache_size
933
934 @property
935 def cache_id(self) -> int:
936 return self._batch_param.cache_id
937
938 @sequence_length.setter
939 def sequence_length(self, value: int):
940 self._batch_param.sequence_length = value
941
942 @cache_size.setter
943 def cache_size(self, value: int):
944 self._batch_param.cache_size = value
945
946 @cache_id.setter
947 def cache_id(self, value: int):
948 self._batch_param.cache_id = value
949
950
951def start_tracing_events(path: str) -> bool:
952 """
953 @brief Starts event tracing and prepares to save the trace log to a specified file.
954
955 The trace log is recorded in "Chrome Tracing JSON format," which can be
956 viewed at https://ui.perfetto.dev/.
957
958 The trace log is not written immediately; it is saved only when
959 stop_tracing_events() is called.
960
961 @param[in] path The file path where the trace log should be stored.
962 @return True if tracing starts successfully, False otherwise.
963 """
964 return _cQbRuntime.start_tracing_events(path)
965
966
968 """
969 @brief Stops event tracing and writes the recorded trace log.
970
971 This function finalizes tracing and saves the collected trace data
972 to the file specified when start_tracing_events() was called.
973 """
974 _cQbRuntime.stop_tracing_events()
975
976
977def get_model_summary(mxq_path: str) -> str:
978 """
979 @brief Generates a structured summary of the specified MXQ model.
980
981 Returns an overview of the model contained in the MXQ file, including:
982 - Target NPU device
983 - Supported core modes and their associated cores
984 - The total number of model variants
985 - For each variant:
986 - Input and output tensor shapes
987 - A list of layers with their types, output shapes, and input layer indices
988
989 The summary is returned as a human-readable string in a table and is useful for
990 inspecting model compatibility, structure, and input/output shapes.
991
992 @param[in] mxq_path Path to the MXQ model file.
993 @return A formatted string containing the model summary.
994 """
995 return _cQbRuntime.get_model_summary(mxq_path)
996
997
998def get_available_devices() -> List[str]:
999 """
1000 @brief Get the device names of the NPU devices detected on the system.
1001
1002 @return The detected device names (e.g. "aries-rb", "regulus-ra").
1003 Empty if no NPU device is detected.
1004 """
1005 return _cQbRuntime.get_available_devices()
1006
1007
1008def get_available_device_numbers(device_name: str = "auto") -> List[int]:
1009 """
1010 @brief Get the available device numbers of the given device.
1011
1012 @param device_name The device name ("auto" or a name such as "aries", "aries-rb",
1013 "regulus", "regulus-ra"; case-insensitive).
1014 @return The available device numbers. Empty if `device_name` cannot be resolved.
1015 """
1016 return _cQbRuntime.get_available_device_numbers(device_name)
1017
1018
1019
Struct containing BatchLLM parameters.
Definition type.py:901
int sequence_length(self)
Definition type.py:927
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:821
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:750
bool get_forced_npu_bundle_index(self)
Retrieves the index of the forced NPU bundle.
Definition type.py:681
bool set_global8_core_mode(self)
Sets the model to use global8-core mode for inference.
Definition type.py:602
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:626
bool set_global_core_mode(self, List[Cluster] clusters)
Definition type.py:581
bool force_single_npu_bundle(self, int npu_bundle_index)
Forces the use of a specific NPU bundle.
Definition type.py:662
__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:769
CoreMode get_core_mode(self)
Gets the core mode to be applied to the model.
Definition type.py:614
None set_async_pipeline_enabled(self, bool enable)
Enables or disables the asynchronous pipeline required for asynchronous inference.
Definition type.py:693
bool get_async_pipeline_enabled(self)
Returns whether the asynchronous pipeline is enabled in this configuration.
Definition type.py:710
int get_activation_slots(self)
Returns activation buffer slot count.
Definition type.py:740
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:718
int get_num_cores(self)
Gets the number of cores to be allocated for the model.
Definition type.py:651
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:587
CoreAllocationPolicy get_core_allocation_policy(self)
Gets the core allocation policy to be applied to the model.
Definition type.py:640
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:998
str get_model_summary(str mxq_path)
Generates a structured summary of the specified MXQ model.
Definition type.py:977
List[int] get_available_device_numbers(str device_name="auto")
Get the available device numbers of the given device.
Definition type.py:1008
bool start_tracing_events(str path)
Starts event tracing and prepares to save the trace log to a specified file.
Definition type.py:951
stop_tracing_events()
Stops event tracing and writes the recorded trace log.
Definition type.py:967