frontend.py Source File

frontend.py Source File#

Mobilint SDK qb Compiler: frontend.py Source File
Mobilint SDK qb Compiler v1.2
MCS002-EN
 
import os
import pathlib
import warnings
from collections.abc import Callable
from typing import Any, List, Optional, Union
 
import torch
from qbcompiler.compiler.compiler import Compiler
from qbcompiler.compiler.compiler import _quantization_task
from qbcompiler.compiler.utils import validate_target_device
from qbcompiler.configs import (
    CompileConfig,
    CalibrationConfig,
    BitConfig,
    ResourceManagementConfig,
    OptqConfig,
    ModConfig,
    LlmConfig,
    EquivalentTransformationConfig,
    SearchWeightScaleConfig,
    SaveSampleConfig,
    Uint8InputConfig,
    PreprocessingConfig,
)
from qbcompiler.config_resolver import (
    ConfigManager,
    ParserOptions_V2,
    QuantizeRequest_V2,
    ResolvedQuantizeRequest_V2,
    MbltCompileRequest_V2,
    ResolvedMbltCompileRequest_V2,
    UNSET as CONFIG_MANAGER_UNSET,
    apply_compile_config_bindings,
)
from qbcompiler.logging import get_logger
from qbcompiler.progress import emit_progress, progress_context
 
logger = get_logger(__name__)
 
_UNSET = object()
"""Sentinel to distinguish 'not passed' from explicit defaults."""
 
 
def _to_config_manager_unset_V2(value: object) -> object:
    """Translate frontend-local ``_UNSET`` values to ConfigManager ``UNSET``."""
    if value is _UNSET:
        return CONFIG_MANAGER_UNSET
    return value
 
 
def _resolve_deprecated_kwargs(
    *,
    calibration_config,
    optq_config,
    mod_config,
    equivalent_transformation_config,
    search_weight_scale_config,
    uint8_input_config,
    preprocessing_config,
    save_sample_config,
    **kwargs,
):
    """Extract and resolve deprecated parameters from kwargs into modern equivalents.
 
    Mutates kwargs by popping consumed deprecated keys.
    Returns a tuple of resolved sub-config values.
    """
    quantization_config = kwargs.pop("quantization_config", None)
    if quantization_config is not None and calibration_config is None:
        warnings.warn(
            "quantization_config is deprecated. Use calibration_config instead.",
            DeprecationWarning,
            stacklevel=3,
        )
        calibration_config = quantization_config
 
    advanced_quantization_config = kwargs.pop("advanced_quantization_config", None)
    if advanced_quantization_config is not None:
        warnings.warn(
            "advanced_quantization_config is deprecated. Use individual configs "
            "(optq_config, mod_config, equivalent_transformation_config, search_weight_scale_config) instead.",
            DeprecationWarning,
            stacklevel=3,
        )
        if hasattr(advanced_quantization_config, "optq") and optq_config is None:
            optq_config = advanced_quantization_config.optq
        if hasattr(advanced_quantization_config, "mod") and mod_config is None:
            mod_config = advanced_quantization_config.mod
        if (
            hasattr(advanced_quantization_config, "equivalent_transformation")
            and equivalent_transformation_config is None
        ):
            equivalent_transformation_config = (
                advanced_quantization_config.equivalent_transformation
            )
        if (
            hasattr(advanced_quantization_config, "search_weight_scale")
            and search_weight_scale_config is None
        ):
            search_weight_scale_config = (
                advanced_quantization_config.search_weight_scale
            )
 
    input_process_config = kwargs.pop("input_process_config", None)
    if input_process_config is not None:
        warnings.warn(
            "input_process_config is deprecated. Use uint8_input_config and preprocessing_config instead.",
            DeprecationWarning,
            stacklevel=3,
        )
        if hasattr(input_process_config, "uint8_input") and uint8_input_config is None:
            uint8_input_config = input_process_config.uint8_input
        if (
            hasattr(input_process_config, "preprocessing")
            and preprocessing_config is None
        ):
            preprocessing_config = input_process_config.preprocessing
 
    save_sample = kwargs.pop("save_sample", None)
    sample_dtype = kwargs.pop("sample_dtype", None)
    if (
        save_sample is not None or sample_dtype is not None
    ) and save_sample_config is None:
        warnings.warn(
            "save_sample and sample_dtype are deprecated. Use save_sample_config instead.",
            DeprecationWarning,
            stacklevel=3,
        )
        save_sample_kwargs = {}
        if save_sample is not None:
            save_sample_kwargs["apply"] = save_sample
        if sample_dtype is not None:
            save_sample_kwargs["dtype"] = sample_dtype
        save_sample_config = SaveSampleConfig(**save_sample_kwargs)
 
    return (
        calibration_config,
        optq_config,
        mod_config,
        equivalent_transformation_config,
        search_weight_scale_config,
        uint8_input_config,
        preprocessing_config,
        save_sample_config,
    )
 
 
def _normalize_quantize_request_V2(
    *,
    model,
    save_path: Union[str, List[str]] = _UNSET,
    backend: str = "onnx",
    target_device: Optional[str] = None,
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include: bool = False,
    exclude_first_subgraph: bool = False,
    save_subgraph_type: int = 0,
    output_subgraph_path: str = "",
    calib_data_path: Union[str, List[str]] = _UNSET,
    device=_UNSET,
    inference_scheme=_UNSET,
    use_random_calib=_UNSET,
    cpu_offload=_UNSET,
    optimize_option=_UNSET,
    buffer_mode=_UNSET,
    input_shape_dict=_UNSET,
    force_npu_input_reposition=_UNSET,
    force_npu_output_reposition=_UNSET,
    image_channels=_UNSET,
    split_blocks=_UNSET,
    split_parts=_UNSET,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    resource_management_config: Optional[ResourceManagementConfig] = None,
    calibration_config: Optional[CalibrationConfig] = None,
    bit_config: Optional[BitConfig] = None,
    llm_config: Optional[LlmConfig] = None,
    optq_config: Optional[OptqConfig] = None,
    mod_config: Optional[ModConfig] = None,
    equivalent_transformation_config: Optional[EquivalentTransformationConfig] = None,
    search_weight_scale_config: Optional[SearchWeightScaleConfig] = None,
    save_sample_config: Optional[SaveSampleConfig] = None,
    uint8_input_config: Optional[Uint8InputConfig] = None,
    preprocessing_config: Optional[PreprocessingConfig] = None,
    **kwargs,
) -> QuantizeRequest_V2:
    """Build a canonical V2 quantize request without executing side effects."""
    return QuantizeRequest_V2(
        model=model,
        save_path=_to_config_manager_unset_V2(save_path),
        parser_options=ParserOptions_V2(
            backend=backend,
            target_device=target_device,
            feed_dict=feed_dict,
            dynamic_axes=dynamic_axes,
            in_dformats=in_dformats,
            yolo_decode_include=yolo_decode_include,
            exclude_first_subgraph=exclude_first_subgraph,
            extra_kwargs=dict(kwargs),
        ),
        save_subgraph_type=save_subgraph_type,
        output_subgraph_path=output_subgraph_path,
        calib_data_path=_to_config_manager_unset_V2(calib_data_path),
        device=_to_config_manager_unset_V2(device),
        inference_scheme=_to_config_manager_unset_V2(inference_scheme),
        use_random_calib=_to_config_manager_unset_V2(use_random_calib),
        cpu_offload=_to_config_manager_unset_V2(cpu_offload),
        optimize_option=_to_config_manager_unset_V2(optimize_option),
        buffer_mode=_to_config_manager_unset_V2(buffer_mode),
        input_shape_dict=_to_config_manager_unset_V2(input_shape_dict),
        force_npu_input_reposition=_to_config_manager_unset_V2(
            force_npu_input_reposition
        ),
        force_npu_output_reposition=_to_config_manager_unset_V2(
            force_npu_output_reposition
        ),
        image_channels=_to_config_manager_unset_V2(image_channels),
        split_blocks=_to_config_manager_unset_V2(split_blocks),
        split_parts=_to_config_manager_unset_V2(split_parts),
        config_preset=config_preset,
        compile_config=compile_config,
        resource_management_config=resource_management_config,
        calibration_config=calibration_config,
        bit_config=bit_config,
        llm_config=llm_config,
        optq_config=optq_config,
        mod_config=mod_config,
        equivalent_transformation_config=equivalent_transformation_config,
        search_weight_scale_config=search_weight_scale_config,
        save_sample_config=save_sample_config,
        uint8_input_config=uint8_input_config,
        preprocessing_config=preprocessing_config,
    )
 
 
def _normalize_mblt_compile_request_V2(
    *,
    model: str | object,
    mblt_save_path: str,
    backend: str = "onnx",
    target_device: Optional[str] = None,
    device: object = _UNSET,
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include: bool = False,
    cpu_offload: object = _UNSET,
    exclude_first_subgraph: bool = False,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    compilation_mode: Optional[str] = None,
    **kwargs,
) -> MbltCompileRequest_V2:
    """Build a canonical V2 compile-to-mblt request without side effects."""
    return MbltCompileRequest_V2(
        model=model,
        mblt_save_path=mblt_save_path,
        parser_options=ParserOptions_V2(
            backend=backend,
            target_device=target_device,
            feed_dict=feed_dict,
            dynamic_axes=dynamic_axes,
            in_dformats=in_dformats,
            yolo_decode_include=yolo_decode_include,
            exclude_first_subgraph=exclude_first_subgraph,
            extra_kwargs=dict(kwargs),
        ),
        device=_to_config_manager_unset_V2(device),
        cpu_offload=_to_config_manager_unset_V2(cpu_offload),
        config_preset=config_preset,
        compile_config=compile_config,
        compilation_mode=compilation_mode,
    )
 
 
def _is_existing_mblt_input_V2(model: object) -> bool:
    """Return ``True`` when *model* is a readable existing ``.mblt`` file."""
    return isinstance(model, str) and model.endswith(".mblt") and os.path.isfile(model)
 
 
def _validate_mblt_for_mxq_compile_V2(model: str, cpu_offload: object) -> None:
    """Validate an existing ``.mblt`` used as an ``mxq_compile`` input.
 
    Only weight-bearing, properly partitioned ``.mblt`` files produced by
    ``mblt_compile()`` can be quantized. This surfaces actionable guidance for
    the common failure modes up front instead of a cryptic downstream error:
 
    * No weights, or a single subgraph carrying unsupported ops -> error (a
      preview export is not a runnable artifact; regenerate with
      ``mblt_compile()``).
    * ``cpu_offload=False`` with more than one subgraph -> error (the graph
      cannot run without CPU offloading).
    * ``cpu_offload=True`` with a single subgraph -> error (nothing to
      offload; recompile with ``cpu_offload=False``).
    """
 
    def _read_mblt_graph_info(path: str):
        """Return ``(has_weights, subgraphs)`` for the .mblt at *path*.
 
        Both accepted flavors are supported: ``mblt.serialize.SerializeMeta``
        detects the on-disk format from the trailing magic number, then the
        legacy qbcompiler layout is read with qbcompiler's own serializer and
        the mblt-graph layout with the mblt-graph serializer. Only the header
        and graph structure are deserialized — weight data is never read.
        """
        from mblt.serialize import SerializeMeta as MbltSerializeMeta
 
        if MbltSerializeMeta.read_header(path).is_legacy:
            from qbcompiler.model_dict.serialize import SerializeMeta
 
            header = SerializeMeta.get_header(path)
            subgraphs = SerializeMeta.get_model_dict(path, header=header).subgraphs
        else:
            header = MbltSerializeMeta.read_header(path)
            subgraphs = MbltSerializeMeta.get_model_dict(path).subgraphs
        return header.num_buffers > 0, subgraphs
 
    has_weights, subgraphs = _read_mblt_graph_info(model)
    num_subgraphs = len(subgraphs)
    effective_cpu_offload = False if cpu_offload is _UNSET else bool(cpu_offload)
 
    # A file with no weights, or a single subgraph that carries unsupported
    # ops, is a preview export rather than a runnable artifact.
    not_runnable_msg = (
        "The provided .mblt is not a runnable artifact and cannot be quantized "
        "into an MXQ package (it looks like a `save_subgraph_type` preview "
        "export). Recompile a runnable .mblt with `mblt_compile()` and use that "
        "file as the `mxq_compile()` input instead."
    )
 
    if not has_weights:
        raise ValueError(not_runnable_msg)
 
    if num_subgraphs == 1:
        sg = subgraphs[0]
 
        # cpu_offload=True has nothing to offload with a single subgraph.
        if effective_cpu_offload:
            raise ValueError(
                "cpu_offload=True requires the model to contain multiple "
                "subgraphs, but the provided .mblt contains a single subgraph. "
                "Call `mxq_compile()` again with the `cpu_offload` argument set "
                "to False."
            )
 
        # A single subgraph carrying unsupported ops is not NPU-runnable.
        if sg.unsupported or any(op.unsupported for op in sg.operators):
            raise ValueError(not_runnable_msg)
        return
 
    # More than one subgraph: the unsupported subgraphs can only execute when
    # CPU offloading is enabled.
    if not effective_cpu_offload:
        raise ValueError(
            f"cpu_offload=False cannot compile a model with {num_subgraphs} "
            "subgraphs. Call `mxq_compile()` again with the `cpu_offload` "
            "argument set to True"
        )
 
 
def _execute_quantize_request_V2(request: ResolvedQuantizeRequest_V2) -> None:
    """Execute a resolved V2 quantize request via the CompilerV2 pipeline.
 
    Routing:
 
    * ``.mblt`` input on disk → single-stage pipeline of
      :class:`MxqCompileStage` (the dominant ``quantize`` CLI case).
    * Raw model input → two-stage pipeline
      (:class:`MBLTCompileStage` writes an intermediate ``.mblt`` into
      the pipeline's tempdir; :class:`MxqCompileStage` picks it up via
      ``ctx.artifacts["mblt_path"]``).
    """
    logger.debug("MXQ compile: CompilerV2 pipeline path")
    from qbcompiler.compiler.compiler_v2 import (
        CompilerV2,
        MBLTCompileStage,
        MxqCompileStage,
        PipelineStage,
        ResolvedPipeline,
    )
 
    stages: list[PipelineStage] = []
    if not _is_existing_mblt_input_V2(request.model):
        stages.append(
            MBLTCompileStage(
                model=request.model,
                save_path="",  # auto → ctx.tempdir/intermediate.mblt
                parser_options=request.parser_options,
                resolved_device=request.resolved_device,
                resolved_cpu_offload=request.compile_config.cpu_offload,
                target_device=request.parser_options.target_device,
                save_subgraph_type=request.save_subgraph_type,
                output_subgraph_path=request.output_subgraph_path,
            )
        )
        mxq_input = None  # read from ctx.artifacts["mblt_path"]
    else:
        mxq_input = request.model
 
    stages.append(
        MxqCompileStage(
            mblt_path=mxq_input,
            compile_config=request.compile_config,
            target_device=request.parser_options.target_device,
        )
    )
 
    with CompilerV2(ResolvedPipeline(stages=tuple(stages))) as compiler:
        compiler.run()
    torch.cuda.empty_cache()
 
 
def _execute_mblt_compile_request_V2(request: ResolvedMbltCompileRequest_V2) -> None:
    """Execute a resolved V2 compile-to-mblt request via the CompilerV2 pipeline.
 
    Imported lazily so the pipeline module is only loaded on first use.
    """
    from qbcompiler.compiler.compiler_v2 import (
        CompilerV2,
        MBLTCompileStage,
        ResolvedPipeline,
    )
 
    stage = MBLTCompileStage(
        model=request.model,
        save_path=request.mblt_save_path,
        parser_options=request.parser_options,
        resolved_device=request.resolved_device,
        resolved_cpu_offload=request.resolved_cpu_offload,
        target_device=request.resolved_target_device,
        compilation_mode=request.resolved_compilation_mode,
    )
    with CompilerV2(ResolvedPipeline(stages=(stage,))) as compiler:
        # progress_callback is None so ambient progress_context (installed
        # by mblt_compile_with_callback_V2) keeps receiving stage events.
        compiler.run()
 
 
 
 
class Model_Dict(Compiler):
    """
    @brief Wrapper around the Mobilint compiler to support compilation and inference workflows.
 
    @details Capabilities include:
    - Compilation of models into MXQ artifacts runnable on Mobilint NPUs.
    - Inference using the full-precision high-level compiled model on CPU or GPU.
    - Inference with the quantized model on CPU or GPU.
    """
 
    def __init__(
        self,
        model,
        backend="onnx",
        device="cpu",
        feed_dict=None,
        dynamic_axes=None,
        in_dformats=None,
        yolo_decode_include=False,
        exclude_first_subgraph=False,
        **kwargs,
    ):
        """
        @brief Initialize the Mobilint compiler wrapper.
 
        @param model string or model instance. Model path or in-memory model to compile.
        @param backend string. Framework identifier for the model (for example "onnx"). Defaults to "onnx".
        @param device string. Target device for inference ("cpu" or "gpu"). Defaults to "cpu".
        @param feed_dict dict. Example inputs for shape resolution.
        @param dynamic_axes dict. Marks model axes as dynamic.
        @param in_dformats dict. Describes input data formats.
        @param yolo_decode_include bool. Runs YOLO decode on NPU when @c True.
        @param exclude_first_subgraph bool. Applies only when CPU offloading is enabled: exclude the first subgraph from the final graph if it is unsupported.
        @param kwargs dict. Additional compiler arguments.
        """
 
        super().__init__(
            model=model,
            backend=backend,
            device=device,
            feed_dict=feed_dict,
            in_dformats=in_dformats,
            dynamic_axes=dynamic_axes,
            yolo_decode_include=yolo_decode_include,
            exclude_first_subgraph=exclude_first_subgraph,
            **kwargs,
        )
 
    def compile(
        self,
        save_subgraph_type: int = 0,
        output_subgraph_path="",
        compile_config: CompileConfig = None,
        **kwargs,
    ):
        """
        @brief Compile the wrapped model into an MXQ artifact.
 
        @param save_subgraph_type int. Controls optional MBLT exports: 0 disables; 1 saves graph structure; 2 saves
        structure and weights; 3 splits structure into multiple subgraphs; 4 splits structure and weights.
        @param output_subgraph_path string. Destination for exported .mblt when @p save_subgraph_type is 1–4.
        @param compile_config CompileConfig. Compile configuration object.
        @param kwargs dict. Additional arguments forwarded to the compiler.
 
        """
        save_sample = kwargs.pop("save_sample", None)
        sample_dtype = kwargs.pop("sample_dtype", None)
        if save_sample is not None or sample_dtype is not None:
            warnings.warn(
                "save_sample and sample_dtype are deprecated. Use compile_config with saveSample instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            if compile_config is None:
                compile_config = CompileConfig()
            if not compile_config.save_sample.apply:
                save_sample_kwargs = {}
                if save_sample is not None:
                    save_sample_kwargs["apply"] = save_sample
                if sample_dtype is not None:
                    save_sample_kwargs["dtype"] = sample_dtype
                compile_config = compile_config.with_save_sample(**save_sample_kwargs)
 
        super().compile(
            save_subgraph_type=save_subgraph_type,
            output_subgraph_path=output_subgraph_path,
            compile_config=compile_config,
            **kwargs,
        )
 
 
def mxq_compile(
    model,
    target_device: str,
    calib_data_path: Union[str, List[str]] = _UNSET,
    save_subgraph_type: int = 0,
    output_subgraph_path="",
    save_path: Union[str, List[str]] = _UNSET,
    backend="onnx",
    # -- ModelDict params --
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include=False,
    exclude_first_subgraph=False,
    # -- CompileConfig-overridable (_UNSET = defer to config) --
    device=_UNSET,
    inference_scheme=_UNSET,
    use_random_calib=_UNSET,
    cpu_offload=_UNSET,
    optimize_option=_UNSET,
    buffer_mode=_UNSET,
    input_shape_dict=_UNSET,
    force_npu_input_reposition=_UNSET,
    force_npu_output_reposition=_UNSET,
    image_channels=_UNSET,
    split_blocks=_UNSET,
    split_parts=_UNSET,
    # -- Config objects --
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig] = None,
    resource_management_config: Optional[ResourceManagementConfig] = None,
    calibration_config: Optional[CalibrationConfig] = None,
    bit_config: Optional[BitConfig] = None,
    llm_config: Optional[LlmConfig] = None,
    optq_config: Optional[OptqConfig] = None,
    mod_config: Optional[ModConfig] = None,
    equivalent_transformation_config: Optional[EquivalentTransformationConfig] = None,
    search_weight_scale_config: Optional[SearchWeightScaleConfig] = None,
    save_sample_config: Optional[SaveSampleConfig] = None,
    uint8_input_config: Optional[Uint8InputConfig] = None,
    preprocessing_config: Optional[PreprocessingConfig] = None,
    **kwargs,
):
    """
    @brief Compile a model into a Mobilint eXeCUtable (MXQ) package for execution on Mobilint NPUs.
 
    @details When no explicit value is provided for a parameter marked with @c _UNSET, the default
    from CompileConfig is used. This allows a compile_config file or object to supply the value
    without being overridden by function-level defaults.
 
    Configuration is resolved in priority order (highest to lowest):
    1. Explicitly passed function arguments
    2. Individual sub-config objects (calibration_config, llm_config, etc.)
    3. kwargs partial overrides (quantization_method, weight_dtype, etc.)
    4. compile_config (CompileConfig object or JSON/YAML file) or config_preset
    5. CompileConfig field defaults
 
    @param model string or model instance. Model path. When using @c backend="onnx", this should be the path to an ONNX
    model file. When @c backend is "torchscript", provide a TorchScript module; when "torch", provide a standard
    PyTorch model. For @c backend="tf", pass the directory that contains the TensorFlow SavedModel graph and assets. For
    @c backend="tflite", pass the TF Lite model path.
    @param calib_data_path string or list of strings. Path(s) to the calibration dataset. Accepts either a text/json file
    that lists NumPy files or a directory that contains the pre-processed NumPy files.
    @param save_subgraph_type int. Controls optional MBLT subgraph exports: 0 disables exports; 1 saves only the graph
    structure; 2 saves graph structure plus weights; 3 saves the graph structure split into multiple subgraphs; 4 saves
    both structure and weights split into multiple subgraphs. Defaults to 0.
    @param output_subgraph_path string. Destination path for the exported .mblt file when @p save_subgraph_type is 1–4.
    The resulting file can be used for visualization. Defaults to "".
    @param save_path string or list of strings. Output MXQ filename(s). When omitted, defaults to "{model_name}.mxq"
    derived from the model path basename.
    @param backend string. Framework used to generate the Mobilint IR. Must be one of "onnx", "tf", "tflite",
    "torchscript", or "torch". Defaults to "onnx". When @p model is a path to an already-compiled .mblt file (the
    output of @c mblt_compile()), it is compiled directly with MXQ regardless of the @p backend value,
    and the file is validated beforehand; it must be a runnable
    artifact produced by @c mblt_compile() rather than a @c save_subgraph_type preview export.
    @param target_device string. Target NPU device for parser/compiler configuration (for example "aries-rb"). Required by execution paths that parse or quantize a model.
    @param device string. Compilation and inference device: "cpu" or "gpu". When omitted, uses CompileConfig default.
    @param feed_dict dict. Example input tensors for shape inference and inference validation.
    @param dynamic_axes dict. Marks model axes as dynamic. Keys are input names and values map dimension indices to
    aliases (for example {"input": {2: "seq_len"}}).
    @param in_dformats dict. Describes input tensor data formats.
    @param inference_scheme string. NPU inference scheme. One of "single", "multi", "global", "global4", or
    "global8". When omitted, uses CompileConfig default.
    @param yolo_decode_include bool. Determines whether YOLO decode runs on NPU. Defaults to @c False.
    @param use_random_calib bool. Generates random calibration data to validate model compilability.
    When omitted, uses CompileConfig default.
    @param cpu_offload bool. Enables CPU offloading during NPU inference. When omitted, uses CompileConfig default.
    @param optimize_option int. Compiler optimization strategy selector. When omitted, uses CompileConfig default.
    @param buffer_mode int. Buffer serialization mode: 0 uses a naive buffer, 1 uses an mmap-backed buffer.
    When omitted, uses CompileConfig default.
    @param input_shape_dict dict. Multi-shape compilation specification for STT/TTS models only (e.g., Conformer-CTC,
    MeloTTS). The dynamic axis is defined with respect to each input tensor's shape. During compilation, the same axis
    is varied across all model inputs using the provided values. Currently, only a single entry key ("multi_shape0") is
    supported. Example: {"multi_shape0": {"axis": 2, "values": [100, 200, 300]}}
    @param force_npu_input_reposition bool. Force input reposition operations to run on NPU instead of CPU.
    When omitted, uses CompileConfig default.
    @param force_npu_output_reposition bool. Force output reposition operations to run on NPU instead of CPU.
    When omitted, uses CompileConfig default.
    @param image_channels int. Number of image channels (0 for auto-detect).
    When omitted, uses CompileConfig default.
    @param split_blocks list of int. Multi-MXQ split points by transformer block index.
    Only supported for LLM models. When omitted, uses CompileConfig default.
    @param split_parts int. Evenly split transformer blocks into N MXQ parts.
    Only supported for LLM models. When omitted, uses CompileConfig default.
    @param exclude_first_subgraph bool. Applies only when CPU offloading is enabled: exclude the first subgraph from
    the final graph if it is unsupported. Defaults to @c False.
    @param config_preset string. Name of a built-in configuration preset. When provided, loads the preset
    via CompileConfig.from_preset(). Defaults to None (no preset). Available presets:
    - "classification": Image classification models (ResNet, EfficientNet, ViT, etc.)
    - "detection": Object detection models (YOLO, SSD, DETR, etc.)
    - "classification_torchvision": Torchvision classification models with standard preprocessing
    - "yolo_640": YOLO detection models with 640x640 letterbox preprocessing
    - "yolo_1280": YOLO detection models with 1280x1280 letterbox preprocessing
    - "llm": Large Language Models (LLaMA, Qwen, Gemma, etc.)
    - "llm_fast": LLM with faster compilation (less accuracy optimization)
    - "vision_transformer": Vision Transformer models (ViT, DeiT, Swin, etc.)
    - "multimodal": Multimodal models (CLIP, BLIP, LLaVA, etc.)
    @param compile_config CompileConfig or string. CompileConfig object or path to JSON/YAML configuration file
    containing all compilation settings (resourceManagement, calibration, bit, optq, mod, llm, etc.).
    @param resource_management_config ResourceManagementConfig. Resource management configuration object.
    @param calibration_config CalibrationConfig. Calibration configuration object.
    @param bit_config BitConfig. Bit configuration object for quantization precision settings.
    @param llm_config LlmConfig. LLM configuration object.
    @param optq_config OptqConfig. OPTQ (Optimal Brain Quantization) configuration object.
    @param mod_config ModConfig. MOD (Metric-based Optimization and Distillation) configuration object.
    @param equivalent_transformation_config EquivalentTransformationConfig. Configuration for equivalent
    transformations like SmoothQuant.
    @param search_weight_scale_config SearchWeightScaleConfig. Configuration for weight scale search.
    @param save_sample_config SaveSampleConfig. Configuration for sample data generation and saving.
    @param uint8_input_config Uint8InputConfig. Configuration for uint8 input handling.
    @param preprocessing_config PreprocessingConfig. Preprocessing pipeline configuration.
    @param kwargs dict. Additional compiler arguments. Supports partial config overrides such as
    @c quantization_method, @c quantization_mode, @c percentile, @c weight_dtype, @c ram_usage,
    @c max_sequence_length, etc. Also accepts deprecated parameters (@c quantization_config,
    @c advanced_quantization_config, @c input_process_config, @c save_sample, @c sample_dtype)
    which will emit DeprecationWarning.
    @return None.
 
    @par Using configuration
    There are three ways to configure quantization settings:
 
    1. Load all settings from a JSON/YAML config file:
    @code{.py}
    from qbcompiler import mxq_compile
 
    mxq_compile(
        model="path/to/model.onnx",
        target_device="aries-rb",
        calib_data_path="path/to/calib",
        compile_config="path/to/config.json",  # or config.yaml
        device="gpu",
    )
    @endcode
 
    2. Pass individual sub-config objects:
    @code{.py}
    from qbcompiler import mxq_compile
    from qbcompiler.configs import (
        ResourceManagementConfig,
        CalibrationConfig,
        BitConfig,
        OptqConfig,
        ModConfig,
        LlmConfig,
    )
 
    resource_mgmt = ResourceManagementConfig(weight_dtype="float32")
    calib_cfg = CalibrationConfig(method=1, mode=1)
    bit_cfg = BitConfig(...)
    optq_cfg = OptqConfig(apply=True)
    mod_cfg = ModConfig(apply=False)
    llm_cfg = LlmConfig(apply=True)
 
    mxq_compile(
        model="path/to/model.onnx",
        target_device="aries-rb",
        calib_data_path="path/to/calib",
        resource_management_config=resource_mgmt,
        calibration_config=calib_cfg,
        bit_config=bit_cfg,
        optq_config=optq_cfg,
        mod_config=mod_cfg,
        llm_config=llm_cfg,
        device="gpu",
    )
    @endcode
 
    3. Automatically applied partial configuration overrides:
    @code{.py}
    from qbcompiler import mxq_compile
 
    mxq_compile(
        model="path/to/model.onnx",
        target_device="aries-rb",
        calib_data_path="path/to/calib",
        quantization_method=1, # per channel quantization
        quantization_mode=1, # max percentile quantization
        percentile=0.999, # percentile value for max percentile quantization
        quantization_output=0, # per layer quantization for the output layer
        device="gpu",
    )
    @endcode
    Please refer to the mxq_compile function and the quantization configuration section for the meaning of the quantization-related numeric values.
 
 
    @par Compiling models with custom inputs (ONNX/Torch/TensorFlow)
    Provide NumPy inputs whenever the model omits shape information so that qbcompiler can infer unknown dimensions and data
    formats.
 
    @code{.py}
    from qbcompiler import mxq_compile
    import numpy as np
 
    example_input = {
        "input_node_name_1": np.random.randn(1, 3, 224, 224).astype(np.float32),
        "input_node_name_2": np.random.randn(1, 8, 224, 224).astype(np.float32),
        "input_node_name_3": np.random.randn(1, 4, 56, 56).astype(np.float32),
    }
 
    in_dformats = {
        "input_node_name_1": "NCHW",
        "input_node_name_2": "NCHW",
        "input_node_name_3": "NCHW",
    }
 
    onnx_model_path = "path/to/your/model.onnx"
    mxq_compile(
        model=onnx_model_path,
        target_device="aries-rb",
        feed_dict=example_input,
        in_dformats=in_dformats,
        backend="onnx",
        compile_config="path/to/config.json",
    )
    @endcode
    """
    target_device = validate_target_device(target_device)
 
    if _is_existing_mblt_input_V2(model):
        _validate_mblt_for_mxq_compile_V2(model, cpu_offload)
 
    if backend == "onnx" or _is_existing_mblt_input_V2(model):
        mxq_compile_V2(
            model=model,
            calib_data_path=calib_data_path,
            save_subgraph_type=save_subgraph_type,
            output_subgraph_path=output_subgraph_path,
            save_path=save_path,
            backend=backend,
            target_device=target_device,
            feed_dict=feed_dict,
            dynamic_axes=dynamic_axes,
            in_dformats=in_dformats,
            yolo_decode_include=yolo_decode_include,
            exclude_first_subgraph=exclude_first_subgraph,
            device=device,
            inference_scheme=inference_scheme,
            use_random_calib=use_random_calib,
            cpu_offload=cpu_offload,
            optimize_option=optimize_option,
            buffer_mode=buffer_mode,
            input_shape_dict=input_shape_dict,
            force_npu_input_reposition=force_npu_input_reposition,
            force_npu_output_reposition=force_npu_output_reposition,
            image_channels=image_channels,
            split_blocks=split_blocks,
            split_parts=split_parts,
            config_preset=config_preset,
            compile_config=compile_config,
            resource_management_config=resource_management_config,
            calibration_config=calibration_config,
            bit_config=bit_config,
            llm_config=llm_config,
            optq_config=optq_config,
            mod_config=mod_config,
            equivalent_transformation_config=equivalent_transformation_config,
            search_weight_scale_config=search_weight_scale_config,
            save_sample_config=save_sample_config,
            uint8_input_config=uint8_input_config,
            preprocessing_config=preprocessing_config,
            **kwargs,
        )
        return
 
    # --- Handle deprecated kwargs ---
    (
        calibration_config,
        optq_config,
        mod_config,
        equivalent_transformation_config,
        search_weight_scale_config,
        uint8_input_config,
        preprocessing_config,
        save_sample_config,
    ) = _resolve_deprecated_kwargs(
        calibration_config=calibration_config,
        optq_config=optq_config,
        mod_config=mod_config,
        equivalent_transformation_config=equivalent_transformation_config,
        search_weight_scale_config=search_weight_scale_config,
        uint8_input_config=uint8_input_config,
        preprocessing_config=preprocessing_config,
        save_sample_config=save_sample_config,
        **kwargs,
    )
 
    # --- Step 1: Base CompileConfig ---
    if config_preset is not None:
        compile_cfg = CompileConfig.from_preset(config_preset)
    elif compile_config is None:
        compile_cfg = CompileConfig()
    elif isinstance(compile_config, CompileConfig):
        compile_cfg = compile_config
    elif isinstance(compile_config, str):
        compile_cfg = CompileConfig.from_file(compile_config)
    else:
        raise ValueError(
            "compile_config must be a CompileConfig object or a path to JSON/YAML configuration file"
        )
 
    # --- Steps 2-4: Apply sub-config replacements, partial-kwarg
    # overrides, and external named-arg overrides through the unified
    # bindings registry shared with ConfigManager.resolve_mxq_compile.
    # See docs/config-manager-partial-kwargs-bugs.md for the rationale.
    request_values = {
        "model": model,
        "save_path": _to_config_manager_unset_V2(save_path),
        "calib_data_path": _to_config_manager_unset_V2(calib_data_path),
        "use_random_calib": _to_config_manager_unset_V2(use_random_calib),
        "inference_scheme": _to_config_manager_unset_V2(inference_scheme),
        "cpu_offload": _to_config_manager_unset_V2(cpu_offload),
        "optimize_option": _to_config_manager_unset_V2(optimize_option),
        "buffer_mode": _to_config_manager_unset_V2(buffer_mode),
        "input_shape_dict": _to_config_manager_unset_V2(input_shape_dict),
        "device": _to_config_manager_unset_V2(device),
        "force_npu_input_reposition": _to_config_manager_unset_V2(
            force_npu_input_reposition
        ),
        "force_npu_output_reposition": _to_config_manager_unset_V2(
            force_npu_output_reposition
        ),
        "image_channels": _to_config_manager_unset_V2(image_channels),
        "split_blocks": _to_config_manager_unset_V2(split_blocks),
        "split_parts": _to_config_manager_unset_V2(split_parts),
        "resource_management_config": resource_management_config,
        "calibration_config": calibration_config,
        "bit_config": bit_config,
        "llm_config": llm_config,
        "optq_config": optq_config,
        "mod_config": mod_config,
        "equivalent_transformation_config": equivalent_transformation_config,
        "search_weight_scale_config": search_weight_scale_config,
        "save_sample_config": save_sample_config,
        "uint8_input_config": uint8_input_config,
        "preprocessing_config": preprocessing_config,
    }
    compile_cfg, kwargs = apply_compile_config_bindings(
        compile_cfg, kwargs=kwargs, request_values=request_values
    )
 
    # Resolve values from final config for downstream use
    resolved_device = compile_cfg.device
 
    if isinstance(model, str) and model.endswith(".mblt") and os.path.isfile(model):
        _quantization_task(
            model, compile_config=compile_cfg, target_device=target_device
        )
    else:
        model_dict = Model_Dict(
            model=model,
            backend=backend,
            target_device=target_device,
            device=resolved_device,
            feed_dict=feed_dict,
            in_dformats=in_dformats,
            dynamic_axes=dynamic_axes,
            yolo_decode_include=yolo_decode_include,
            exclude_first_subgraph=exclude_first_subgraph,
            **kwargs,
        )
        model_dict.compile(
            save_subgraph_type=save_subgraph_type,
            output_subgraph_path=output_subgraph_path,
            compile_config=compile_cfg,
        )
        torch.cuda.empty_cache()
        del model_dict
 
 
def mblt_compile(
    model: str | Any,
    mblt_save_path: str,
    target_device: str,
    backend="onnx",
    device="cpu",
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include=False,
    cpu_offload=False,
    exclude_first_subgraph=False,
    **kwargs,
):
    """
    @brief Export a model to the Mobilint .mblt format without producing an MXQ package.
 
    @param model string or model instance. Source model or path to compile.
    @param mblt_save_path string. Output path for the .mblt artifact.
    @param backend string. Framework identifier ("onnx", "tf", "tflite", "torchscript", or "torch"). Defaults to
    "onnx".
    @param device string. Compilation device ("cpu" or "gpu"). Defaults to "cpu".
    @param feed_dict dict. Example inputs used for shape inference.
    @param dynamic_axes dict. Declares dynamic axes per input name.
    @param in_dformats dict. Input dataformat metadata.
    @param yolo_decode_include bool. Runs YOLO decode on NPU when True.
    @param cpu_offload bool. Enables CPU offloading for unsupported groups.
    @param exclude_first_subgraph bool. Applies only when CPU offloading is enabled: exclude the first subgraph from the final graph if it is unsupported.
    @param kwargs dict. Additional arguments forwarded to the compiler.
    @return None.
    """
    target_device = validate_target_device(target_device)
    if backend == "onnx":
        mblt_compile_V2(
            model=model,
            mblt_save_path=mblt_save_path,
            backend=backend,
            target_device=target_device,
            device=device,
            feed_dict=feed_dict,
            dynamic_axes=dynamic_axes,
            in_dformats=in_dformats,
            yolo_decode_include=yolo_decode_include,
            cpu_offload=cpu_offload,
            exclude_first_subgraph=exclude_first_subgraph,
            **kwargs,
        )
        return
 
    model_dict = Model_Dict(
        model=model,
        backend=backend,
        target_device=target_device,
        device=device,
        feed_dict=feed_dict,
        dynamic_axes=dynamic_axes,
        in_dformats=in_dformats,
        yolo_decode_include=yolo_decode_include,
        exclude_first_subgraph=exclude_first_subgraph,
        **kwargs,
    )
 
    model_dict.mblt_compile(
        model,
        mblt_save_path=mblt_save_path,
        backend=backend,
        cpu_offload=cpu_offload,
        **kwargs,
    )
 
 
def mxq_compile_V2(
    model,
    target_device: str,
    calib_data_path: Union[str, List[str]] = _UNSET,
    save_subgraph_type: int = 0,
    output_subgraph_path="",
    save_path: Union[str, List[str]] = _UNSET,
    backend="onnx",
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include=False,
    exclude_first_subgraph=False,
    device=_UNSET,
    inference_scheme=_UNSET,
    use_random_calib=_UNSET,
    cpu_offload=_UNSET,
    optimize_option=_UNSET,
    buffer_mode=_UNSET,
    input_shape_dict=_UNSET,
    force_npu_input_reposition=_UNSET,
    force_npu_output_reposition=_UNSET,
    image_channels=_UNSET,
    split_blocks=_UNSET,
    split_parts=_UNSET,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    resource_management_config: Optional[ResourceManagementConfig] = None,
    calibration_config: Optional[CalibrationConfig] = None,
    bit_config: Optional[BitConfig] = None,
    llm_config: Optional[LlmConfig] = None,
    optq_config: Optional[OptqConfig] = None,
    mod_config: Optional[ModConfig] = None,
    equivalent_transformation_config: Optional[EquivalentTransformationConfig] = None,
    search_weight_scale_config: Optional[SearchWeightScaleConfig] = None,
    save_sample_config: Optional[SaveSampleConfig] = None,
    uint8_input_config: Optional[Uint8InputConfig] = None,
    preprocessing_config: Optional[PreprocessingConfig] = None,
    **kwargs,
) -> None:
    """V2 compile/quantize entry point backed only by ConfigManager resolution."""
    request_V2 = _normalize_quantize_request_V2(
        model=model,
        save_path=save_path,
        backend=backend,
        target_device=target_device,
        feed_dict=feed_dict,
        dynamic_axes=dynamic_axes,
        in_dformats=in_dformats,
        yolo_decode_include=yolo_decode_include,
        exclude_first_subgraph=exclude_first_subgraph,
        save_subgraph_type=save_subgraph_type,
        output_subgraph_path=output_subgraph_path,
        calib_data_path=calib_data_path,
        device=device,
        inference_scheme=inference_scheme,
        use_random_calib=use_random_calib,
        cpu_offload=cpu_offload,
        optimize_option=optimize_option,
        buffer_mode=buffer_mode,
        input_shape_dict=input_shape_dict,
        force_npu_input_reposition=force_npu_input_reposition,
        force_npu_output_reposition=force_npu_output_reposition,
        image_channels=image_channels,
        split_blocks=split_blocks,
        split_parts=split_parts,
        config_preset=config_preset,
        compile_config=compile_config,
        resource_management_config=resource_management_config,
        calibration_config=calibration_config,
        bit_config=bit_config,
        llm_config=llm_config,
        optq_config=optq_config,
        mod_config=mod_config,
        equivalent_transformation_config=equivalent_transformation_config,
        search_weight_scale_config=search_weight_scale_config,
        save_sample_config=save_sample_config,
        uint8_input_config=uint8_input_config,
        preprocessing_config=preprocessing_config,
        **kwargs,
    )
    resolved_request_V2 = ConfigManager.resolve_quantize_request_V2(request_V2)
    _execute_quantize_request_V2(resolved_request_V2)
 
 
def mblt_compile_V2(
    model: str | object,
    target_device: str,
    mblt_save_path: str,
    backend="onnx",
    device: object = _UNSET,
    feed_dict=None,
    dynamic_axes=None,
    in_dformats=None,
    yolo_decode_include=False,
    cpu_offload: object = _UNSET,
    exclude_first_subgraph=False,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    compilation_mode: Optional[str] = None,
    **kwargs,
) -> None:
    """V2 compile-to-mblt entry point backed only by ConfigManager resolution.
 
    ``compilation_mode`` is an internal knob: one of
    ``{"release", "dev", "debug"}`` or ``None`` (default) to defer to the
    ``MBLT_APP_ENV`` environment variable. When set it overrides the env
    and drives the parser's ``inference_validation`` / ``device_alloc``
    / ``log_level`` cascade — the same preset as ``MBLT_APP_ENV``. Not
    surfaced in the CLI on purpose; intended for in-process scripts and
    tests that want scoped dev-mode validation without mutating
    process-wide environment.
    """
    request_V2 = _normalize_mblt_compile_request_V2(
        model=model,
        mblt_save_path=mblt_save_path,
        backend=backend,
        target_device=target_device,
        device=device,
        feed_dict=feed_dict,
        dynamic_axes=dynamic_axes,
        in_dformats=in_dformats,
        yolo_decode_include=yolo_decode_include,
        cpu_offload=cpu_offload,
        exclude_first_subgraph=exclude_first_subgraph,
        config_preset=config_preset,
        compile_config=compile_config,
        compilation_mode=compilation_mode,
        **kwargs,
    )
    resolved_request_V2 = ConfigManager.resolve_mblt_compile_request_V2(request_V2)
    _execute_mblt_compile_request_V2(resolved_request_V2)
 
 
def mblt_compile_with_callback_V2(
    model: str,
    target_device: str,
    mblt_save_path: str,
    backend: str = "onnx",
    device: Optional[str] = None,
    cpu_offload: Optional[bool] = None,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    compilation_mode: Optional[str] = None,
    *,
    progress_callback: Callable[[int, str], None],
) -> None:
    """V2 compile-to-mblt entry point with progress callbacks.
 
    Only forwards explicitly-set values so ``mblt_compile_V2``'s ``_UNSET``
    defaults (and the merged ``CompileConfig`` behind them) apply when the
    caller omits a flag.
    """
    call_kwargs: dict[str, Any] = {
        "model": model,
        "mblt_save_path": mblt_save_path,
        "backend": backend,
        "target_device": target_device,
    }
    if device is not None:
        call_kwargs["device"] = device
    if cpu_offload is not None:
        call_kwargs["cpu_offload"] = cpu_offload
    if config_preset is not None:
        call_kwargs["config_preset"] = config_preset
    if compile_config is not None:
        call_kwargs["compile_config"] = compile_config
    if compilation_mode is not None:
        call_kwargs["compilation_mode"] = compilation_mode
 
    with progress_context(progress_callback):
        emit_progress(0, "Initializing compiler")
        mblt_compile_V2(**call_kwargs)
        emit_progress(100, "Complete")
 
 
def mxq_compile_with_callback_V2(
    model: str,
    target_device: str,
    save_path: str,
    backend: str = "onnx",
    device: Optional[str] = None,
    calib_data_path: Optional[Union[str, List[str]]] = None,
    use_random_calib: Optional[bool] = None,
    config_preset: Optional[str] = None,
    compile_config: Optional[CompileConfig | str] = None,
    *,
    progress_callback: Callable[[int, str], None],
) -> None:
    """V2 compile/quantize entry point with progress callbacks."""
    call_kwargs: dict[str, object] = {
        "model": model,
        "save_path": save_path,
        "backend": backend,
    }
    if device is not None:
        call_kwargs["device"] = device
    if target_device is not None:
        call_kwargs["target_device"] = target_device
    if calib_data_path is not None:
        call_kwargs["calib_data_path"] = calib_data_path
    if use_random_calib is not None:
        call_kwargs["use_random_calib"] = use_random_calib
    if config_preset is not None:
        call_kwargs["config_preset"] = config_preset
    if compile_config is not None:
        call_kwargs["compile_config"] = compile_config
 
    with progress_context(progress_callback):
        emit_progress(0, "Initializing compiler")
        mxq_compile_V2(**call_kwargs)
        emit_progress(100, "Complete")
 
 
 
 
def run_mblt_model(
    path_to_mblt: str,
    save_all_outputs: bool = True,
    inference_output_path: Optional[str] = None,
    target_subgraph: Optional[int] = None,
):
    from qbcompiler.model_dict.common.model_dict import ValueDict
    from qbcompiler.model_dict.inference.model import ModelInference
    from qbcompiler.model_dict.serialize import load_mblt_model
 
    assert path_to_mblt.endswith(".mblt")
    if not save_all_outputs:
        raise NotImplementedError(
            f"save_all_outputs={save_all_outputs} is not supported yet."
        )
    if target_subgraph:
        raise NotImplementedError(
            f"target_subgraph={target_subgraph} is not supported yet."
        )
    import numpy
 
    if not inference_output_path:
        inference_output_path = str(pathlib.Path(path_to_mblt).with_suffix(".infer"))
 
    vd = ValueDict()
    md, wd = load_mblt_model(path_to_mblt, load_weight=True)
    for sg in md.subgraphs:
        for iidx in sg.inputs:
            inact = sg.activations[iidx]
            if inact.name in set(md.inputs):
                vd[inact.name] = numpy.random.random(tuple(map(int, inact.src_shape)))
 
    model_inference = ModelInference(
        model_dict=md,
        weight_dict=wd,
        value_dict=vd,
        save_all_outputs=save_all_outputs,
        inference_all_outputs_write_path=inference_output_path,
    )
 
    model_inference.run_inference(use_value_dict_data_for_next_input=False)