compile_config.py Source File

compile_config.py Source File#

Mobilint SDK qb Compiler: compile_config.py Source File
Mobilint SDK qb Compiler v0.12.0.0
MCS002-EN
from typeguard import typechecked
from . import ConfigABC, is_valid_dir_path, check_save_path
from .resource_management_config import ResourceManagementConfig
from .quantization_config import QuantizationConfig
from .advanced_quantization_config import AdvancedQuantizationConfig
from .llm_config import LlmConfig
from typing import Optional
 
import tempfile
import json
from qubee.calibration.utils_calib import (
    list_np_files_in_json,
    list_np_files_in_txt,
    CalibType,
    check_calib_data,
)
from qubee.version import __version__
from qubee.logging import get_logger
 
logger = get_logger(__name__)
 
_OUTDATED_KWARG_CONFIG_WARNING = ["is_quant_ch"]
 
_OUTDATED_KWARG_CONFIG_ERRORS = {
    "quant_output": "The 'quant_output' argument has been removed. Use just 'quantization_output' instead \n"
    "output (int): Output quantization type index\n"
    "0: Layer - Per-layer quantization\n"
    "1: Ch - Per-channel quantization\n"
    "2: Sigmoid - Sigmoid-based quantization\n By default, this option is set to 0 (Layer).",
    "quantize_percentile": "The 'quantize_percentile' argument has been removed. Use just 'percentile' instead.",
    "is_asym_quant": "'is_asym_quant' is deprecated_option, use 'quantization_method' instead, The quantization_method option only takes integers.\n"
    "quantization_method (int): Calibration method index\n"
    "0: Symmetric per tensor quantization\n"
    "1: Symmetric per channel quantization\n"
    "2: Asymmetric per tensor quantization \n"
    "3: Asymmetric per channel quantization \n By default, this option is set to 1.",
    "is_quant_ch": "'is_quant_ch' is deprecated_option, use 'quantization_method' instead, The quantization_method option only takes integers. \n"
    "quantization_method (int): Calibration method index\n"
    "0: Symmetric per tensor quantization\n"
    "1: Symmetric per channel quantization\n"
    "2: Asymmetric per tensor quantization \n"
    "3: Asymmetric per channel quantization \nIf is_quant_ch is provided, quantization_method is set to 1 when True and 0 when False automatically.",
    "smooth_factor": "'smooth_factor' is deprecated and cannot be used anymore.",
    "quantize_method": "'quantize_method' is deprecated_option, use 'quantization_mode' instead, The quantization_mode option only takes integers. \n"
    "quantization_mode (int): Quantization mode index\n"
    "0: Percentile\n"
    "1: max\n"
    "2: maxPercentile \n"
    "3: fastPercentile \n"
    "4: histogramKL \n"
    "5: histogramMSE \n By default, this option is set to 2 (maxPercentile).",
}
 
 
 
class CompileConfig(ConfigABC):
    optional = False
    py_to_cpp_name = {
        "calib_data_path": "calibData",
        "save_path": "savePath",
        "model_nickname": "modelNickname",
        "save_msgpack_name": "saveMsgpackName",
        "model_path": "modelPath",
        "inference_scheme": "inferenceScheme",
        "save_sample": "saveSample",
        "cpu_offload": "cpuOffload",
        "singlecore_compile": "singlecoreCompile",
        "optimize_option": "optimizeOption",
        "sample_dtype": "sampleDtype",
        "preprocess_dict": "preprocess",
        "start_dram_offset": "startDramOffset",
        "version": "version",
    }
    min_max_check = {"optimize_option": [0, None]}
    inference_scheme_types = ["single", "multi", "global", "global4", "global8"]
    sample_dtypes = ["float", "int8"]
    """
    @brief Unified compilation configuration object for Mobilint MXQ compilation.
 
    @details This config consolidates compilation-related parameters and sub-config class into a single structure.
    """
 
    @typechecked
    def __init__(
        self,
        calib_data_path: str | None,
        use_random_calib: bool,
        save_path: str,
        model_nickname: str,
        save_msgpack_name: str | None,
        model_path: str | None,
        inference_scheme: str,
        save_sample: bool,
        cpu_offload: bool,
        singlecore_compile: bool,
        optimize_option: int,
        sample_dtype: str,
        preprocess_dict: dict | None,
        buffer_mode: int,
        start_dram_offset: int,
        version: str,
        input_shape_dict: dict,
        resource_management: ResourceManagementConfig,
        quantization: QuantizationConfig,
        advanced_quantization: AdvancedQuantizationConfig,
        llm_config: LlmConfig,
        optional: bool = optional,
    ):
        """
        @brief Unified compilation configuration object for Mobilint MXQ compilation.
        @param calib_data_path str | None.
            Path to calibration dataset. Accepts either:<br>
            - a directory containing pre-processed NumPy samples, or<br>
            - a .txt/.json meta file listing NumPy sample file paths.<br>
            If @p use_random_calib is True, this may be None. Otherwise, a valid path is expected for
            quantized compilation workflows.
        @param use_random_calib bool.
            When True, uses randomly generated calibration samples (typically for smoke testing and
            compilability validation). When False, @p calib_data_path is used as the calibration source.
        @param save_path str.
            Output MXQ filename/path to generate.
        @param model_nickname str.
            Model nickname used as a fallback identifier when intermediate results need to be saved.
            It is recommended to set this to the filename component of @p save_path.
        @param save_msgpack_name str | None.
            filename/path to save an intermediate serialized representation (msgpack) of the
            high-level compiled model/graph. Primarily intended for debugging and is now rarely used.
        @param model_path str | None. (deprecated) Outdated argument. Please refer to the model usage in the @p mxq_compile function.
        @param inference_scheme str.
            NPU inference scheme string. Must be one of:<br>
            @c "single", @c "multi", @c "global", @c "global4", @c "global8" (@c "global" and @c "global8" behave identically).
        @param save_sample bool.
            When True, saves representative input/output tensors for debugging.
        @param cpu_offload bool.
            Enables CPU offload for unsupported operator groups/segments during NPU inference.
            This is a beta feature and is recommended only for special cases.
        @param singlecore_compile bool.
            Forces single-core compilation mode.  Commonly used to compile large language models.
        @param optimize_option int. Compiler optimization selector.
        @param sample_dtype string. Data type for saved sample inference outputs: "float" or "int8".
        Primarily intended for debugging and is now rarely used.
        @param preprocess_dict dict | None. Additional preprocessing metadata.
        @param buffer_mode int.
            Buffer serialization mode for saving .mblt file. Common conventions:<br>
            - 0: naïve buffer serialization<br>
            - 1: mmap-backed buffer serialization<br>
        @param start_dram_offset int. Start Dram Address Offset (default = 0).
        @param version str. qubee version.
        @param input_shape_dict dict. Optional HWC input shape dictionary used for multi-shape compilation (for example {"input0": [[224, 224, 3], [256, 256, 3]]}).
        @param resource_management ResourceManagementConfig.
            Resource management configuration class.
        @param quantization QuantizationConfig.
            Quantization configuration class (e.g., calibration strategy, bit-widths, per-tensor/per-channel
            settings, quantization scheme/policies).
        @param advanced_quantization AdvancedQuantizationConfig.
            Advanced quantization configuration block for fine-grained or experimental controls beyond the basic quantization settings.
        @param llm_config LlmConfig. LLM-specific configuration class.
        optional bool. Indicates whether this config is optional. Use the default value defined as a class variable.
 
        """
        super().__init__(optional)
        self.calib_data_path = self._get_calibration_path(
            calib_data_path, use_random_calib
        )
        self.save_path = save_path
        self.model_nickname = model_nickname
        self.save_msgpack_name = save_msgpack_name
        self.model_path = model_path
        self.inference_scheme = inference_scheme
        self.save_sample = save_sample
        self.cpu_offload = cpu_offload
        self.singlecore_compile = singlecore_compile
        self.optimize_option = optimize_option
        self.sample_dtype = sample_dtype
        self.preprocess_dict = preprocess_dict
        self.buffer_mode = buffer_mode
        self.start_dram_offset = start_dram_offset
        self.version = version
        self.input_shape_dict = input_shape_dict
 
        # New unified config structure
        self.resource_management = resource_management
        self.quantization = quantization
        self.advanced_quantization = advanced_quantization
        self.llm_config = llm_config
 
        self.check_valid(
            except_list=[
                "optional",
                "resource_management",
                "quantization",
                "advanced_quantization",
                "llm_config",
                "calib_file",
                "buffer_mode",
                "input_shape_dict",
            ]
        )
 
    @classmethod
    def default_config(cls, **kwargs):
        """
        Create default config with optional parameter overrides.
 
        Args:
            **kwargs: Parameters to override defaults
        """
        # Default values
        defaults = {
            "calib_data_path": None,
            "use_random_calib": False,
            "save_path": "./tmp.mxq",
            "save_msgpack_name": None,
            "model_nickname": "temporary",
            "model_path": None,
            "inference_scheme": "single",
            "save_sample": False,
            "cpu_offload": False,
            "singlecore_compile": False,
            "optimize_option": 0,
            "sample_dtype": "float",
            "preprocess_dict": None,
            "buffer_mode": 1,
            "start_dram_offset": 0,
            "input_shape_dict": dict(),
            "version": __version__,
            "resource_management": ResourceManagementConfig.default_config(),
            "quantization": QuantizationConfig.default_config(),
            "advanced_quantization": AdvancedQuantizationConfig.default_config(),
            "llm_config": LlmConfig.default_config(),
        }
 
        # Override with provided kwargs
        defaults.update({k: v for k, v in kwargs.items() if v is not None})
 
        return CompileConfig(optional=cls.optional, **defaults)
 
    @classmethod
    def from_json(cls, json_path: str, **kwargs):
        """
        Load config from JSON file and merge with additional kwargs.
 
        Args:
            json_path: Path to JSON configuration file
            **kwargs: Additional compile-time parameters (calib_data_path, save_path, etc.)
        """
        with open(json_path, "r") as f:
            config_dict = json.load(f)
 
        # Load sub-configs from JSON
        resource_management = ResourceManagementConfig.from_dict(
            config_dict.get("resourceManagement", {})
        )
        quantization = QuantizationConfig.from_dict(config_dict.get("quantization", {}))
        advanced_quantization = AdvancedQuantizationConfig.from_dict(
            config_dict.get("advancedQuantization", {})
        )
        llm_config = LlmConfig.from_dict(config_dict.get("llmConfig", {}))
        input_shape_dict = config_dict.get("inputShapeDict", {})
 
        # Merge with kwargs (compile-time parameters take precedence)
        return cls(
            resource_management=resource_management,
            quantization=quantization,
            advanced_quantization=advanced_quantization,
            llm_config=llm_config,
            input_shape_dict=kwargs.get("input_shape_dict", input_shape_dict),
            **{k: v for k, v in kwargs.items() if k != "input_shape_dict"},
        )
 
    def check_valid_str_configs(self):
        if self.save_path:
            check_save_path(self.save_path, "mxq")
        if self.save_msgpack_name:
            is_valid_dir_path(self.save_msgpack_name)
        if self.inference_scheme.lower() not in CompileConfig.inference_scheme_types:
            raise ValueError(
                f"Inference_scheme should be one of the {CompileConfig.inference_scheme_types}"
            )
        if self.sample_dtype.lower() not in CompileConfig.sample_dtypes:
            raise ValueError(
                f"Sample data type should be one of the {CompileConfig.sample_dtypes}"
            )
        if self.buffer_mode not in (0, 1):
            raise ValueError(f"buffer mode must be either 0 or 1")
 
    def to_dict(self):
        compile_config_dict = dict()
        for py_name, cpp_name in CompileConfig.py_to_cpp_name.items():
            value = getattr(self, py_name)
            if isinstance(value, str) and py_name in [
                "inference_scheme",
                "sample_dtype",
            ]:
                value = value.lower()
            compile_config_dict[cpp_name] = value
 
        # Add new unified config structure
        compile_config_dict["quant"] = dict()
        compile_config_dict["quant"].update(self.resource_management.to_dict())
        compile_config_dict["quant"].update(self.quantization.to_dict())
        compile_config_dict["quant"].update(self.advanced_quantization.to_dict())
        compile_config_dict["quant"].update(self.llm_config.to_dict())
        compile_config_dict["quant"]["inputShapeDict"] = self.input_shape_dict
 
        return compile_config_dict
 
    def _get_calibration_path(self, calib_data_path: str, use_random_calib: bool):
        if not calib_data_path and not use_random_calib:
            raise ValueError(
                "Please use calib_data_path or enable use_random_calib. Do not leave calib_data_path empty while setting use_random_calib to false."
            )
        if (
            not calib_data_path or use_random_calib
        ):  # we don't have to check the validity of calib_data_path
            return None
        calib_data_case = check_calib_data(calib_data_path)
        if calib_data_case == CalibType.SINGLE_DIR:
            self.calib_file = tempfile.NamedTemporaryFile(suffix=".txt")
            list_np_files_in_txt(calib_data_path, self.calib_file.name)
            calib_data_path = self.calib_file.name
        elif calib_data_case == CalibType.MULTI_DIR:
            self.calib_file = tempfile.NamedTemporaryFile(suffix=".json")
            list_np_files_in_json(calib_data_path, self.calib_file.name)
            calib_data_path = self.calib_file.name
        elif calib_data_case in (CalibType.SINGLE_TXT, CalibType.MULTI_JSON):
            pass
        else:
            raise ValueError(f"Got unexpected calib_data_path={calib_data_path}.")
 
        return calib_data_path
 
 
def get_compile_config(
    calib_data_path=None,
    use_random_calib=False,
    save_path="./tmp.mxq",
    save_msgpack_name=None,
    model_nickname="temporary",
    model_path=None,
    inference_scheme="single",
    save_sample=False,
    cpu_offload=False,
    singlecore_compile=False,
    optimize_option=0,
    sample_dtype="float",
    preprocess_dict=None,
    buffer_mode=1,
    start_dram_offset=0,
    input_shape_dict=dict(),
    resource_management_config: Optional[ResourceManagementConfig] = None,
    quantization_config: Optional[QuantizationConfig] = None,
    llm_config: Optional[LlmConfig] = None,
    advanced_quantization_config: Optional[AdvancedQuantizationConfig] = None,
    version=__version__,
    **kwargs,
):
    """
    @brief Create Compilation Config with partial parameters merged with defaults.
    """
    outdated_keys = _OUTDATED_KWARG_CONFIG_ERRORS.keys() & kwargs.keys()
    if outdated_keys:
        messages = [_OUTDATED_KWARG_CONFIG_ERRORS[k] for k in sorted(outdated_keys)]
        full_message = "\n".join(messages)
        warning_keys = set(_OUTDATED_KWARG_CONFIG_WARNING)
 
        if set(outdated_keys) == warning_keys:
            for old_argument in outdated_keys:
                if old_argument == "is_quant_ch":
                    is_quant = bool(kwargs.get("is_quant_ch", False))
                    kwargs["quantization_method"] = 1 if is_quant else 0
                else:
                    raise NotImplementedError(
                        f"{old_argument} should be properly mapped"
                    )
            logger.warning(full_message)
        else:
            error_keys = set(outdated_keys) - warning_keys
            if not error_keys:
                return
 
            error_messages = [
                _OUTDATED_KWARG_CONFIG_ERRORS[k] for k in sorted(error_keys)
            ]
            error_message = "\n".join(error_messages)
            raise ValueError(error_message)
 
    if resource_management_config is None:
        resource_management_config = ResourceManagementConfig.from_kwargs(**kwargs)
    if quantization_config is None:
        quantization_config = QuantizationConfig.from_kwargs(**kwargs)
    if llm_config is None:
        llm_config = LlmConfig.from_kwargs(**kwargs)
    if advanced_quantization_config is None:
        # AdvancedQuantizationConfig config don't use "from_kwargs" because of shared key names, such as "apply".
        advanced_quantization_config = AdvancedQuantizationConfig.default_config()
 
    return CompileConfig(
        optional=CompileConfig.optional,
        calib_data_path=calib_data_path,
        use_random_calib=use_random_calib,
        save_path=save_path,
        save_msgpack_name=save_msgpack_name,
        model_nickname=model_nickname,
        model_path=model_path,
        inference_scheme=inference_scheme,
        save_sample=save_sample,
        cpu_offload=cpu_offload,
        singlecore_compile=singlecore_compile,
        optimize_option=optimize_option,
        sample_dtype=sample_dtype,
        preprocess_dict=preprocess_dict,
        buffer_mode=buffer_mode,
        start_dram_offset=start_dram_offset,
        version=version,
        input_shape_dict=input_shape_dict,
        resource_management=resource_management_config,
        quantization=quantization_config,
        advanced_quantization=advanced_quantization_config,
        llm_config=llm_config,
    )
 
 