type.py Source File

type.py Source File#

SDK qb Runtime Library: type.py Source File
SDK qb Runtime Library v0.28
MCS001-EN
 
from typing import List, Optional, Tuple
from enum import Enum
 
import numpy as np
 
import maccel.maccel as _cMaccel
 
 
class Cluster(Enum):
    Cluster0 = _cMaccel.Cluster.Cluster0
    Cluster1 = _cMaccel.Cluster.Cluster1
    Error = _cMaccel.Cluster.Error
 
 
class Core(Enum):
    Core0 = _cMaccel.Core.Core0
    Core1 = _cMaccel.Core.Core1
    Core2 = _cMaccel.Core.Core2
    Core3 = _cMaccel.Core.Core3
    All = _cMaccel.Core.All
    GlobalCore = _cMaccel.Core.GlobalCore
    Error = _cMaccel.Core.Error
 
 
class CoreAllocationPolicy(Enum):
    Auto = _cMaccel.CoreAllocationPolicy.Auto
    Manual = _cMaccel.CoreAllocationPolicy.Manual
 
 
class LatencySetPolicy(Enum):
    Auto = _cMaccel.LatencySetPolicy.Auto
    Manual = _cMaccel.LatencySetPolicy.Manual
 
 
class MaintenancePolicy(Enum):
    Maintain = _cMaccel.MaintenancePolicy.Maintain
    DropExpired = _cMaccel.MaintenancePolicy.DropExpired
    Undefined = _cMaccel.MaintenancePolicy.Undefined
 
 
class SchedulePolicy(Enum):
    FIFO = _cMaccel.SchedulePolicy.FIFO
    LIFO = _cMaccel.SchedulePolicy.LIFO
    ByPriority = _cMaccel.SchedulePolicy.ByPriority
    Undefined = _cMaccel.SchedulePolicy.Undefined
 
 
class Scale:
    def __init__(self, scale: float, is_uniform: bool, scale_list: List[float]):
        self._scale = _cMaccel.Scale()
        self._scale.scale = scale
        self._scale.is_uniform = is_uniform
        self._scale.scale_list = scale_list
 
    @classmethod
    def from_cpp(cls, _scale: _cMaccel.Scale):
        return cls(_scale.scale, _scale.is_uniform, _scale.scale_list)
 
    @property
    def scale_list(self) -> List[float]:
        return self._scale.scale_list
 
    @property
    def scale(self) -> float:
        return self._scale.scale
 
    @property
    def is_uniform(self) -> bool:
        return self._scale.is_uniform
 
    @scale_list.setter
    def scale_list(self, value: List[float]):
        self._scale.scale_list = value
 
    @scale.setter
    def scale(self, value: float):
        self._scale.scale = value
 
    @is_uniform.setter
    def is_uniform(self, value: bool):
        self._scale.is_uniform = value
 
    def __getitem__(self, i: int) -> float:
        return self._scale[i]
 
    def __repr__(self):
        d = {
            "scale": self.scale,
            "is_uniform": self.is_uniform,
            "scale_list": self.scale_list,
        }
        return "{}({})".format(
            self.__class__.__name__,
            ", ".join("{}={}".format(k, v) for k, v in d.items()),
        )
 
 
class CoreId:
    def __init__(self, cluster: Cluster, core: Core):
        self._core_id = _cMaccel.CoreId()
        self._core_id.cluster = cluster.value
        self._core_id.core = core.value
 
    @classmethod
    def from_cpp(cls, _core_id: _cMaccel.CoreId):
        return cls(Cluster(_core_id.cluster), Core(_core_id.core))
 
    @property
    def cluster(self) -> Cluster:
        return Cluster(self._core_id.cluster)
 
    @property
    def core(self) -> Core:
        return Core(self._core_id.core)
 
    @cluster.setter
    def cluster(self, value: Cluster):
        self._core_id.cluster = value.value
 
    @core.setter
    def core(self, value: Core):
        self._core_id.core = value.value
 
    def __eq__(self, other) -> bool:
        return self._core_id == other._core_id
 
    def __lt__(self, other) -> bool:
        return self._core_id < other._core_id
 
    def __repr__(self):
        d = {"cluster": self.cluster, "core": self.core}
        return "{}({})".format(
            self.__class__.__name__,
            ", ".join("{}={}".format(k, v) for k, v in d.items()),
        )
 
 
class Buffer:
    def __init__(self, _buffer: Optional[_cMaccel.Buffer] = None):
        self._buffer = _cMaccel.Buffer() if _buffer is None else _buffer
 
    @property
    def size(self) -> int:
        return self._buffer.size
 
    @size.setter
    def size(self, value: int):
        self._buffer.size = value
 
    def set_buffer(self, arr: np.ndarray):
        self._buffer.set_buffer(np.ascontiguousarray(arr))
 
    def __repr__(self):
        return f"{self.__class__.__name__}(size={self._buffer.size})"
 
 
class CoreMode(Enum):
    Single = _cMaccel.CoreMode.Single
    Multi = _cMaccel.CoreMode.Multi
    Global = _cMaccel.CoreMode.Global
    Global4 = _cMaccel.CoreMode.Global4
    Global8 = _cMaccel.CoreMode.Global8
    Error = _cMaccel.CoreMode.Error
 
 
class BufferInfo:
    def __init__(
        self,
        original_height: int = 0,
        original_width: int = 0,
        original_channel: int = 0,
        reshaped_height: int = 0,
        reshaped_width: int = 0,
        reshaped_channel: int = 0,
        height: int = 0,
        width: int = 0,
        channel: int = 0,
        max_height: int = 0,
        max_width: int = 0,
        max_channel: int = 0,
        max_cache_size: int = 0,
    ):
        self._buffer_info = _cMaccel.BufferInfo()
        self._buffer_info.original_height = original_height
        self._buffer_info.original_width = original_width
        self._buffer_info.original_channel = original_channel
        self._buffer_info.reshaped_height = reshaped_height
        self._buffer_info.reshaped_width = reshaped_width
        self._buffer_info.reshaped_channel = reshaped_channel
        self._buffer_info.height = height
        self._buffer_info.width = width
        self._buffer_info.channel = channel
        self._buffer_info.max_height = max_height
        self._buffer_info.max_width = max_width
        self._buffer_info.max_channel = max_channel
        self._buffer_info.max_cache_size = max_cache_size
 
    @classmethod
    def from_cpp(cls, _buffer_info: _cMaccel.BufferInfo):
        return cls(
            _buffer_info.original_height,
            _buffer_info.original_width,
            _buffer_info.original_channel,
            _buffer_info.reshaped_height,
            _buffer_info.reshaped_width,
            _buffer_info.reshaped_channel,
            _buffer_info.height,
            _buffer_info.width,
            _buffer_info.channel,
            _buffer_info.max_height,
            _buffer_info.max_width,
            _buffer_info.max_channel,
            _buffer_info.max_cache_size,
        )
 
    @property
    def original_height(self) -> int:
        return self._buffer_info.original_height
 
    @property
    def original_width(self) -> int:
        return self._buffer_info.original_width
 
    @property
    def original_channel(self) -> int:
        return self._buffer_info.original_channel
 
    @property
    def reshaped_height(self) -> int:
        return self._buffer_info.reshaped_height
 
    @property
    def reshaped_width(self) -> int:
        return self._buffer_info.reshaped_width
 
    @property
    def reshaped_channel(self) -> int:
        return self._buffer_info.reshaped_channel
 
    @property
    def height(self) -> int:
        return self._buffer_info.height
 
    @property
    def width(self) -> int:
        return self._buffer_info.width
 
    @property
    def channel(self) -> int:
        return self._buffer_info.channel
 
    @property
    def max_height(self) -> int:
        return self._buffer_info.max_height
 
    @property
    def max_width(self) -> int:
        return self._buffer_info.max_width
 
    @property
    def max_channel(self) -> int:
        return self._buffer_info.max_channel
 
    @property
    def max_cache_size(self) -> int:
        return self._buffer_info.max_cache_size
 
    @original_height.setter
    def original_height(self, value: int):
        self._buffer_info.original_height = value
 
    @original_width.setter
    def original_width(self, value: int):
        self._buffer_info.original_width = value
 
    @original_channel.setter
    def original_channel(self, value: int):
        self._buffer_info.original_channel = value
 
    @reshaped_height.setter
    def reshaped_height(self, value: int):
        self._buffer_info.reshaped_height = value
 
    @reshaped_width.setter
    def reshaped_width(self, value: int):
        self._buffer_info.reshaped_width = value
 
    @reshaped_channel.setter
    def reshaped_channel(self, value: int):
        self._buffer_info.reshaped_channel = value
 
    @height.setter
    def height(self, value: int):
        self._buffer_info.height = value
 
    @width.setter
    def width(self, value: int):
        self._buffer_info.width = value
 
    @channel.setter
    def channel(self, value: int):
        self._buffer_info.channel = value
 
    @max_height.setter
    def max_height(self, value: int):
        self._buffer_info.max_height = value
 
    @max_width.setter
    def max_width(self, value: int):
        self._buffer_info.max_width = value
 
    @max_channel.setter
    def max_channel(self, value: int):
        self._buffer_info.max_channel = value
 
    @max_cache_size.setter
    def max_cache_size(self, value: int):
        self._buffer_info.max_cache_size = value
 
    def original_size(self) -> int:
        return self._buffer_info.original_size()
 
    def reshaped_size(self) -> int:
        return self._buffer_info.reshaped_size()
 
    def size(self) -> int:
        return self._buffer_info.size()
 
    def original_shape(self) -> Tuple[int, int, int]:
        return self._buffer_info.original_shape()
 
    def original_shape_chw(self) -> Tuple[int, int, int]:
        return self._buffer_info.original_shape_chw()
 
    def reshaped_shape(self) -> Tuple[int, int, int]:
        return self._buffer_info.reshaped_shape()
 
    def reshaped_shape_chw(self) -> Tuple[int, int, int]:
        return self._buffer_info.reshaped_shape_chw()
 
    def shape(self) -> Tuple[int, int, int]:
        return self._buffer_info.shape()
 
    def shape_chw(self) -> Tuple[int, int, int]:
        return self._buffer_info.shape_chw()
 
    def __repr__(self):
        d = {
            "original_height": self._buffer_info.original_height,
            "original_width": self._buffer_info.original_width,
            "original_channel": self._buffer_info.original_channel,
            "reshaped_height": self._buffer_info.reshaped_height,
            "reshaped_width": self._buffer_info.reshaped_width,
            "reshaped_channel": self._buffer_info.reshaped_channel,
            "height": self._buffer_info.height,
            "width": self._buffer_info.width,
            "channel": self._buffer_info.channel,
            "max_height": self._buffer_info.max_height,
            "max_width": self._buffer_info.max_width,
            "max_channel": self._buffer_info.max_channel,
            "max_cache_size": self._buffer_info.max_cache_size,
        }
        return "{}({})".format(
            self.__class__.__name__,
            ", ".join("{}={}".format(k, v) for k, v in d.items()),
        )
 
 
class ModelConfig:
    def __init__(self, num_cores: Optional[int] = None):
        self._model_config = (
            _cMaccel.ModelConfig()
            if num_cores is None
            else _cMaccel.ModelConfig(num_cores)
        )
 
    def include_all_cores(self) -> bool:
        return self._model_config.include_all_cores()
 
    def exclude_all_cores(self) -> bool:
        return self._model_config.exclude_all_cores()
 
    def include(self, cluster: Cluster, core: Optional[Core] = None) -> bool:
        if core is None:
            return self._model_config.include(cluster.value)
        else:
            return self._model_config.include(cluster.value, core.value)
 
    def exclude(self, cluster: Cluster, core: Optional[Core] = None) -> bool:
        if core is None:
            return self._model_config.exclude(cluster.value)
        else:
            return self._model_config.exclude(cluster.value, core.value)
 
    def set_single_core_mode(
        self, num_cores: Optional[int] = None, core_ids: Optional[List[CoreId]] = None
    ) -> bool:
        if num_cores is not None and core_ids is None:
            return self._model_config.set_single_core_mode(num_cores)
        elif core_ids is not None and num_cores is None:
            return self._model_config.set_single_core_mode(
                [core_id._core_id for core_id in core_ids]
            )
        raise ValueError(
            "`set_single_core_mode` needs either `num_cores` and `core_ids`."
        )
 
    def set_global_core_mode(self, clusters: List[Cluster]) -> bool:
        return self._model_config.set_global_core_mode([c.value for c in clusters])
 
    def set_global4_core_mode(self, clusters: List[Cluster]) -> bool:
        return self._model_config.set_global4_core_mode([c.value for c in clusters])
 
    def set_global8_core_mode(self) -> bool:
        return self._model_config.set_global8_core_mode()
 
    def get_core_mode(self) -> CoreMode:
        return CoreMode(self._model_config.get_core_mode())
 
    def set_multi_core_mode(self, clusters: List[Cluster]) -> bool:
        return self._model_config.set_multi_core_mode([c.value for c in clusters])
 
    def set_auto_mode(self, num_cores: int = 1) -> bool:
        return self._model_config.set_auto_mode(num_cores)
 
    def set_manual_mode(self) -> bool:
        return self._model_config.set_manual_mode()
 
    def get_core_allocation_policy(self) -> CoreAllocationPolicy:
        return CoreAllocationPolicy(self._model_config.get_core_allocation_policy())
 
    def get_num_cores(self) -> int:
        return self._model_config.get_num_cores()
 
    def force_single_npu_bundle(self, npu_bundle_index: int) -> bool:
        return self._model_config.force_single_npu_bundle(npu_bundle_index)
 
    def get_forced_npu_bundle_index(self) -> bool:
        return self._model_config.get_forced_npu_bundle_index()
 
    def set_async_pipeline_enabled(self, enable: bool):
        return self._model_config.set_async_pipeline_enabled(enable)
 
    def get_async_pipeline_enabled(self) -> bool:
        return self._model_config.get_async_pipeline_enabled()
 
    @property
    def schedule_policy(self) -> SchedulePolicy:
        return SchedulePolicy(self._model_config.schedule_policy)
 
    @property
    def latency_set_policy(self) -> LatencySetPolicy:
        return LatencySetPolicy(self._model_config.latency_set_policy)
 
    @property
    def maintenance_policy(self) -> MaintenancePolicy:
        return MaintenancePolicy(self._model_config.maintenance_policy)
 
    @property
    def early_latencies(self) -> List[int]:
        return self._model_config.early_latencies
 
    @property
    def finish_latencies(self) -> List[int]:
        return self._model_config.finish_latencies
 
    @schedule_policy.setter
    def schedule_policy(self, policy: SchedulePolicy):
        self._model_config.schedule_policy = policy.value
 
    @latency_set_policy.setter
    def latency_set_policy(self, policy: LatencySetPolicy):
        self._model_config.latency_set_policy = policy.value
 
    @maintenance_policy.setter
    def maintenance_policy(self, policy: MaintenancePolicy):
        self._model_config.maintenance_policy = policy.value
 
    @early_latencies.setter
    def early_latencies(self, latencies: List[int]):
        self._model_config.early_latencies = latencies
 
    @finish_latencies.setter
    def finish_latencies(self, latencies: List[int]):
        self._model_config.finish_latencies = latencies
 
    def get_core_ids(self) -> List[CoreId]:
        return [
            CoreId(Cluster(core_id.cluster), Core(core_id.core))
            for core_id in self._model_config.core_ids
        ]
 
    def __repr__(self):
        d = {
            "core_mode": self.get_core_mode(),
            "core_allocation_policy": self.get_core_allocation_policy(),
            "core_ids": self.get_core_ids(),
            "num_cores": self.get_num_cores(),
            "forced_npu_bundle_index": self.get_forced_npu_bundle_index(),
        }
        return "{}({})".format(
            self.__class__.__name__,
            ", ".join("{}={}".format(k, v) for k, v in d.items()),
        )
 
 
class LogLevel(Enum):
    DEBUG = _cMaccel.LogLevel.DEBUG
    INFO = _cMaccel.LogLevel.INFO
    WARN = _cMaccel.LogLevel.WARN
    ERR = _cMaccel.LogLevel.ERR
    FATAL = _cMaccel.LogLevel.FATAL
    OFF = _cMaccel.LogLevel.OFF
 
 
def set_log_level(level: LogLevel):
    _cMaccel.set_log_level(level.value)
 
 
def start_tracing_events(path: str) -> bool:
    return _cMaccel.start_tracing_events(path)
 
 
def stop_tracing_events():
    _cMaccel.stop_tracing_events()
 