type.py Source File

type.py Source File#

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