Blame for gguf.py

"""Minimal GGUF reader and llama.cpp iMatrix loader.

The V2 quantizer works on float tensors, so GGUF ingestion is deliberately a
thin, read-only layer:

* parse the GGUF header, metadata and tensor directory;
* dequantize the common source types (F32, F16, BF16, Q8_0, Q4_0 and Q2_K);
* read llama.cpp ``.gguf``/legacy ``.dat`` importance matrices and normalize
  each vector to mean 1.0, matching ``imatrix_reader.h``.

No GGUF writer is provided here; the quantizer emits Q2_K arrays that can be
packed by an external GGUF writer.
"""

from __future__ import annotations

import re
import struct
import time
from dataclasses import dataclass
from pathlib import Path

import numpy as np

from .q2k import QK_K, blocks_to_bytes, dequantize_params_batch, unpack_codes
from .q3_k import Q3_K_BLOCK_BYTES, QK3_K, dequantize_q3_k_bytes, quantize_tensor_q3_k
from .q4_0 import Q4_0_BLOCK_BYTES, QK4_0, quantize_tensor_q4_0
from .q1_0 import QK1_0, Q1_0_BLOCK_BYTES, quantize_tensor_q1_0

GGUF_MAGIC = 0x46554747
DEFAULT_ALIGNMENT = 32

GGML_TYPE_F32 = 0
GGML_TYPE_F16 = 1
GGML_TYPE_Q4_0 = 2
GGML_TYPE_Q4_1 = 3
GGML_TYPE_Q5_0 = 6
GGML_TYPE_Q5_1 = 7
GGML_TYPE_Q8_0 = 8
GGML_TYPE_Q2_K = 10
GGML_TYPE_Q3_K = 11
GGML_TYPE_BF16 = 30
GGML_TYPE_Q1_0 = 41

TYPE_NAME = {
    GGML_TYPE_F32: "F32",
    GGML_TYPE_F16: "F16",
    GGML_TYPE_Q4_0: "Q4_0",
    GGML_TYPE_Q4_1: "Q4_1",
    GGML_TYPE_Q5_0: "Q5_0",
    GGML_TYPE_Q5_1: "Q5_1",
    GGML_TYPE_Q8_0: "Q8_0",
    GGML_TYPE_Q2_K: "Q2_K",
    GGML_TYPE_Q3_K: "Q3_K",
    GGML_TYPE_BF16: "BF16",
    GGML_TYPE_Q1_0: "Q1_0",
}

BLOCK_SIZE = {
    GGML_TYPE_F32: 1,
    GGML_TYPE_F16: 1,
    GGML_TYPE_BF16: 1,
    GGML_TYPE_Q4_0: 32,
    GGML_TYPE_Q4_1: 32,
    GGML_TYPE_Q5_0: 32,
    GGML_TYPE_Q5_1: 32,
    GGML_TYPE_Q8_0: 32,
    GGML_TYPE_Q2_K: QK_K,
    GGML_TYPE_Q3_K: QK3_K,
    GGML_TYPE_Q1_0: QK1_0,
}
BLOCK_BYTES = {
    GGML_TYPE_F32: 4,
    GGML_TYPE_F16: 2,
    GGML_TYPE_BF16: 2,
    GGML_TYPE_Q4_0: 18,
    GGML_TYPE_Q4_1: 20,
    GGML_TYPE_Q5_0: 22,
    GGML_TYPE_Q5_1: 24,
    GGML_TYPE_Q8_0: 34,
    GGML_TYPE_Q2_K: 84,
    GGML_TYPE_Q3_K: Q3_K_BLOCK_BYTES,
    GGML_TYPE_Q1_0: Q1_0_BLOCK_BYTES,
}
TYPE_BPW = {
    GGML_TYPE_F32: 32.0,
    GGML_TYPE_F16: 16.0,
    GGML_TYPE_BF16: 16.0,
    GGML_TYPE_Q8_0: 8.5,
    GGML_TYPE_Q5_1: 6.0,
    GGML_TYPE_Q5_0: 5.5,
    GGML_TYPE_Q4_1: 5.0,
    GGML_TYPE_Q4_0: 4.5,
    GGML_TYPE_Q2_K: 2.625,
    GGML_TYPE_Q3_K: 3.4375,
    GGML_TYPE_Q1_0: 1.125,
    12: 4.5,     # Q4_K
    13: 5.5,     # Q5_K
    14: 6.5625,  # Q6_K
    15: 8.5,     # Q8_K
    16: 2.0625,  # IQ2_XXS
    17: 2.3125,  # IQ2_XS
    18: 3.0625,  # IQ3_XXS
    19: 1.5625,  # IQ1_S
    20: 4.5,     # IQ4_NL
    21: 3.4375,  # IQ3_S
    22: 2.5625,  # IQ2_S
    23: 4.25,    # IQ4_XS
}


def source_type_bpw(ttype: int) -> float:
    if ttype in TYPE_BPW:
        return TYPE_BPW[ttype]
    if ttype in BLOCK_SIZE and ttype in BLOCK_BYTES:
        return (BLOCK_BYTES[ttype] * 8.0) / BLOCK_SIZE[ttype]
    return 16.0
SUPPORTED_READ_TYPES = {
    GGML_TYPE_F32,
    GGML_TYPE_F16,
    GGML_TYPE_BF16,
    GGML_TYPE_Q4_0,
    GGML_TYPE_Q8_0,
    GGML_TYPE_Q2_K,
    GGML_TYPE_Q3_K,
    GGML_TYPE_Q1_0,
}


@dataclass(slots=True)
class TensorInfo:
    name: str
    dims: tuple[int, ...]
    type: int
    offset: int

    @property
    def n_elements(self) -> int:
        n = 1
        for d in self.dims:
            n *= int(d)
        return n

    @property
    def row_width(self) -> int:
        return int(self.dims[0]) if self.dims else 1

    @property
    def type_name(self) -> str:
        return TYPE_NAME.get(self.type, f"type_{self.type}")

    @property
    def n_bytes(self) -> int:
        if self.type not in BLOCK_SIZE:
            raise ValueError(f"unsupported GGUF type {self.type} ({self.type_name})")
        block = BLOCK_SIZE[self.type]
        if self.n_elements % block != 0:
            raise ValueError(f"{self.name}: element count is not block-aligned")
        return (self.n_elements // block) * BLOCK_BYTES[self.type]


def _align(value: int, alignment: int = DEFAULT_ALIGNMENT) -> int:
    return (value + alignment - 1) & ~(alignment - 1)


def _read_string(f) -> str:
    n = struct.unpack("<Q", f.read(8))[0]
    return f.read(n).decode("utf-8", errors="replace")


def _read_value(f, vtype: int):
    if vtype == 0:
        return struct.unpack("<B", f.read(1))[0]
    if vtype == 1:
        return struct.unpack("<b", f.read(1))[0]
    if vtype == 2:
        return struct.unpack("<H", f.read(2))[0]
    if vtype == 3:
        return struct.unpack("<h", f.read(2))[0]
    if vtype == 4:
        return struct.unpack("<I", f.read(4))[0]
    if vtype == 5:
        return struct.unpack("<i", f.read(4))[0]
    if vtype == 6:
        return struct.unpack("<f", f.read(4))[0]
    if vtype == 7:
        return bool(struct.unpack("<B", f.read(1))[0])
    if vtype == 8:
        return _read_string(f)
    if vtype == 9:
        subtype = struct.unpack("<I", f.read(4))[0]
        length = struct.unpack("<Q", f.read(8))[0]
        if subtype == 8:
            for _ in range(length):
                _read_string(f)
        else:
            sizes = {0: 1, 1: 1, 2: 2, 3: 2, 4: 4, 5: 4, 6: 4, 7: 1, 10: 8, 11: 8, 12: 8}
            f.read(sizes.get(subtype, 4) * length)
        return None
    if vtype == 10:
        return struct.unpack("<Q", f.read(8))[0]
    if vtype == 11:
        return struct.unpack("<q", f.read(8))[0]
    if vtype == 12:
        return struct.unpack("<d", f.read(8))[0]
    raise ValueError(f"unknown GGUF value type {vtype}")


class GGUFReader:
    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.metadata: dict[str, object] = {}
        self.kv_pairs: list[tuple[str, int, bytes]] = []
        self.tensors: list[TensorInfo] = []
        with self.path.open("rb") as f:
            magic, version = struct.unpack("<II", f.read(8))
            if magic != GGUF_MAGIC:
                raise ValueError(f"{self.path} is not a GGUF file")
            self.version = version
            self.kv_start = f.tell()
            n_tensors, n_kv = struct.unpack("<QQ", f.read(16))
            for _ in range(n_kv):
                key = _read_string(f)
                vtype = struct.unpack("<I", f.read(4))[0]
                value_start = f.tell()
                self.metadata[key] = _read_value(f, vtype)
                value_end = f.tell()
                f.seek(value_start)
                raw = f.read(value_end - value_start)
                f.seek(value_end)
                self.kv_pairs.append((key, vtype, raw))
            self.kv_end = f.tell()
            for _ in range(n_tensors):
                name = _read_string(f)
                n_dims = struct.unpack("<I", f.read(4))[0]
                dims = tuple(struct.unpack("<Q", f.read(8))[0] for _ in range(n_dims))
                ttype = struct.unpack("<I", f.read(4))[0]
                offset = struct.unpack("<Q", f.read(8))[0]
                self.tensors.append(TensorInfo(name, dims, ttype, offset))
            self.info_end = f.tell()
            self.alignment = int(self.metadata.get("general.alignment", DEFAULT_ALIGNMENT))
            self.data_start = _align(self.info_end, self.alignment)
        self._by_name = {ti.name: ti for ti in self.tensors}

    def info(self, name: str) -> TensorInfo:
        try:
            return self._by_name[name]
        except KeyError:
            raise KeyError(f"tensor '{name}' not found in {self.path}") from None

    def list_tensors(self) -> list[TensorInfo]:
        return list(self.tensors)

    def read_tensor_raw(self, name: str) -> bytes:
        ti = self.info(name)
        with self.path.open("rb") as f:
            f.seek(self.data_start + ti.offset)
            raw = f.read(ti.n_bytes)
        if len(raw) != ti.n_bytes:
            raise ValueError(f"{name}: truncated tensor data")
        return raw

    def copy_tensor_raw_to(self, name: str, out, chunk: int = 1 << 20) -> None:
        ti = self.info(name)
        remaining = ti.n_bytes
        with self.path.open("rb") as f:
            f.seek(self.data_start + ti.offset)
            while remaining > 0:
                data = f.read(min(chunk, remaining))
                if not data:
                    raise ValueError(f"{name}: truncated tensor data")
                out.write(data)
                remaining -= len(data)

    def read_tensor(self, name: str) -> tuple[np.ndarray, int]:
        raw = self.read_tensor_raw(name)
        return _dequantize(raw, self.info(name)), self.info(name).row_width

    def iter_tensor_row_chunks(self, name: str, max_elements: int = 8_388_608):
        """Stream a tensor in row-aligned float32 chunks to bound peak RAM."""
        ti = self.info(name)
        row = ti.row_width
        blk = BLOCK_SIZE[ti.type]
        if row <= 0 or row % blk != 0 or ti.n_elements % row != 0:
            x, r = self.read_tensor(name)
            yield x, r, 0, ti.n_elements // max(r, 1)
            return
        n_rows = ti.n_elements // row
        rows_per_chunk = max(1, max_elements // row)
        bytes_per_row = (row // blk) * BLOCK_BYTES[ti.type]
        with self.path.open("rb") as f:
            for r0 in range(0, n_rows, rows_per_chunk):
                r1 = min(r0 + rows_per_chunk, n_rows)
                chunk_elems = (r1 - r0) * row
                chunk_bytes = (r1 - r0) * bytes_per_row
                f.seek(self.data_start + ti.offset + r0 * bytes_per_row)
                raw = f.read(chunk_bytes)
                if len(raw) != chunk_bytes:
                    raise ValueError(f"{name}: truncated tensor data")
                sub_ti = TensorInfo(ti.name, (chunk_elems,), ti.type, 0)
                yield _dequantize(raw, sub_ti), row, r0, r1


def _dequantize(raw: bytes, ti: TensorInfo) -> np.ndarray:
    t = ti.type
    if t == GGML_TYPE_F32:
        return np.frombuffer(raw, dtype="<f4").astype(np.float32)
    if t == GGML_TYPE_F16:
        return np.frombuffer(raw, dtype="<f2").astype(np.float32)
    if t == GGML_TYPE_BF16:
        bits = np.frombuffer(raw, dtype="<u2").astype(np.uint32) << 16
        return bits.view(np.float32)
    if t == GGML_TYPE_Q8_0:
        blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 34)
        d = blocks[:, :2].copy().view("<f2").reshape(-1).astype(np.float32)
        q = blocks[:, 2:].view(np.int8).astype(np.float32)
        return (d[:, None] * q).reshape(-1)
    if t == GGML_TYPE_Q4_0:
        blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 18)
        d = blocks[:, :2].copy().view("<f2").reshape(-1).astype(np.float32)
        qs = blocks[:, 2:]
        low = (qs & 0x0F).astype(np.float32) - 8.0
        high = (qs >> 4).astype(np.float32) - 8.0
        q = np.concatenate([low, high], axis=1)
        return (d[:, None] * q).reshape(-1)
    if t == GGML_TYPE_Q1_0:
        blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, Q1_0_BLOCK_BYTES)
        d = blocks[:, :2].copy().view("<f2").reshape(-1, 1).astype(np.float32)
        bits = np.unpackbits(blocks[:, 2:], axis=1, bitorder="little")[:, :QK1_0].astype(np.float32)
        return (d * (bits * 2.0 - 1.0)).reshape(-1)
    if t == GGML_TYPE_Q3_K:
        n_blocks = len(raw) // Q3_K_BLOCK_BYTES
        return dequantize_q3_k_bytes(raw, n_blocks)
    if t == GGML_TYPE_Q2_K:
        blocks = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 84)
        n = blocks.shape[0]
        scales = blocks[:, :16]
        qs = blocks[:, 16:80]
        d = blocks[:, 80:82].copy().view("<f2").reshape(-1).astype(np.float32)
        dmin = blocks[:, 82:84].copy().view("<f2").reshape(-1).astype(np.float32)
        codes = np.empty((n, QK_K), dtype=np.uint8)
        for base in (0, 128):
            q = qs[:, base // 4 : base // 4 + 32]
            for s_lvl in range(4):
                lo = base + 32 * s_lvl
                codes[:, lo : lo + 32] = (q >> np.uint8(2 * s_lvl)) & np.uint8(3)
        return dequantize_params_batch(d, dmin, scales & 0x0F, scales >> 4, codes).reshape(-1)
    raise ValueError(f"unsupported source GGUF type {ti.type_name} for {ti.name}")


def read_imatrix(path: str | Path) -> dict[str, np.ndarray]:
    """Read a llama.cpp imatrix (GGUF) or legacy .dat file.

    Returns ``name -> float32 vector`` normalized to mean 1.0, exactly the
    per-input-column ``E[a^2]`` weight used by the C engine.
    """
    path = Path(path)
    imat: dict[str, np.ndarray] = {}
    with path.open("rb") as f:
        magic = struct.unpack("<I", f.read(4))[0]
        if magic == GGUF_MAGIC:
            f.seek(0)
            reader = GGUFReader(path)
            sums: dict[str, np.ndarray] = {}
            counts: dict[str, np.ndarray] = {}
            with reader.path.open("rb") as fh:
                for ti in reader.tensors:
                    if not (ti.name.endswith(".in_sum2") or ti.name.endswith(".counts")):
                        continue
                    fh.seek(reader.data_start + ti.offset)
                    raw = fh.read(ti.n_bytes)
                    data = np.frombuffer(raw, dtype="<f4").astype(np.float32)
                    if ti.name.endswith(".in_sum2"):
                        sums[ti.name[: -len(".in_sum2")]] = data
                    else:
                        counts[ti.name[: -len(".counts")]] = data
            for base, sum2 in sums.items():
                c = counts.get(base)
                if c is not None and c.size and float(c.flat[0]) > 0.0:
                    values = sum2 / float(c.flat[0])
                else:
                    values = np.ones_like(sum2)
                mean = float(values.mean())
                imat[base] = values / mean if mean > 1.0e-30 else np.ones_like(values)
        else:
            f.seek(0)
            n_entries = struct.unpack("<i", f.read(4))[0]
            for _ in range(n_entries):
                name_len = struct.unpack("<i", f.read(4))[0]
                name = f.read(name_len).decode("utf-8", errors="replace")
                n_values = struct.unpack("<i", f.read(4))[0]
                struct.unpack("<i", f.read(4))[0]
                values = np.frombuffer(f.read(n_values * 4), dtype="<f4").astype(np.float32)
                mean = float(values.mean())
                imat[name] = values / mean if mean > 1.0e-30 else np.ones_like(values)
    return imat


def lookup_imatrix(imat: dict[str, np.ndarray], name: str) -> np.ndarray | None:
    if name in imat:
        return imat[name]
    for suffix in (".weight", ".in_sum2", ".counts"):
        alt = name[: -len(suffix)] if name.endswith(suffix) else name + suffix
        if alt in imat:
            return imat[alt]
    return None


def expand_importance(
    values: np.ndarray | None, n_elements: int, row_width: int
) -> np.ndarray | None:
    if values is None:
        return None
    v = np.asarray(values, dtype=np.float32).ravel()
    if v.size == n_elements:
        return v
    if row_width > 0 and v.size == row_width and n_elements % row_width == 0:
        return np.tile(v, n_elements // row_width)
    raise ValueError(
        f"importance length {v.size} does not match {n_elements} elements "
        f"(row width {row_width})"
    )


_BLOCK_PREFIX_RE = re.compile(r"^blk\.\d+\.")


def tensor_type_key(name: str) -> str:
    """Return the layer-independent tensor type (e.g. ``blk.4.ffn_up.weight`` -> ``ffn_up``)."""
    key = _BLOCK_PREFIX_RE.sub("", name)
    for suffix in (".weight", ".bias"):
        if key.endswith(suffix):
            return key[: -len(suffix)]
    return key


def _is_skipped_metadata_tensor(name: str, dims: tuple[int, ...]) -> bool:
    if len(dims) < 2:
        return True
    if "norm" in name or ".bias" in name:
        return True
    if "ffn_gate_inp" in name:
        return True
    if name.startswith("mm.") or name.startswith("a.") or name.startswith("v."):
        return True
    if "altup" in name or "laurel" in name or "layer_output_scale" in name:
        return True
    if "ssm_a" in name.split(".") or "A_log" in name:
        return True
    if "ssm_dt" in name or "dt_bias" in name:
        return True
    if "ssm_conv1d" in name or "conv1d.weight" in name:
        return True
    return False


def _is_embd_or_head(name: str) -> bool:
    return "token_embd" in name or name == "output.weight" or name.endswith(".output.weight")


def _is_attention_projection(name: str) -> bool:
    """True for attention and recurrent/SSM projections (Q/K/V/O, QKV, gates, SSM state)."""
    if "norm" in name or ".bias" in name:
        return False
    return any(
        tok in name
        for tok in (
            ".attn_",
            ".ssm_",
            "self_attn.",
            "linear_attn.",
            "attention.w",
            "attn.q",
            "attn.k",
            "attn.v",
            "attn.o",
            "attn.wq",
            "attn.wk",
            "attn.wv",
            "attn.wo",
        )
    )


def _is_tiny_ssm_gate(name: str, dims: tuple[int, ...]) -> bool:
    """True for low-rank recurrent step-size/decay gates (e.g. ssm_alpha, ssm_beta, ssm_f_a/b)."""
    if len(dims) < 2 or max(int(d) for d in dims) < 1024:
        return False
    if any(
        k in name
        for k in ("ssm_alpha", "ssm_beta", "ssm_f_a", "ssm_f_b", "ssm_g_a", "ssm_g_b")
    ):
        return True
    if ".ssm_" in name and min(int(dims[0]), int(dims[1])) <= 128:
        return True
    return False


def is_q2k_eligible(name: str, dims: tuple[int, ...]) -> bool:
    if _is_skipped_metadata_tensor(name, dims):
        return False
    if _is_embd_or_head(name):
        return False
    if dims[0] % QK_K != 0:
        return False
    n_elements = 1
    for d in dims:
        n_elements *= int(d)
    return n_elements >= QK_K and n_elements % QK_K == 0


def is_q3_k_eligible(name: str, dims: tuple[int, ...]) -> bool:
    """True for 256-aligned 2D/3D weight tensors eligible for Q3_K."""
    if _is_skipped_metadata_tensor(name, dims):
        return False
    if _is_embd_or_head(name):
        return False
    if dims[0] % QK3_K != 0:
        return False
    n_elements = 1
    for d in dims:
        n_elements *= int(d)
    return n_elements >= QK3_K and n_elements % QK3_K == 0


def is_q4_0_eligible(
    name: str,
    dims: tuple[int, ...],
    include_embd: bool = True,
    promote_attn: bool = False,
    allow_q2k_tensors: bool = False,
) -> bool:
    """True for 32-aligned 2D/3D weight tensors eligible for Q4_0."""
    if _is_skipped_metadata_tensor(name, dims):
        return False
    is_embd = _is_embd_or_head(name)
    if is_embd and not include_embd:
        return False
    if not allow_q2k_tensors and not is_embd and not (promote_attn and _is_attention_projection(name)):
        if is_q2k_eligible(name, dims):
            return False
    if dims[0] % QK4_0 != 0:
        return False
    n_elements = 1
    for d in dims:
        n_elements *= int(d)
    return n_elements >= QK4_0 and n_elements % QK4_0 == 0


def is_q1_0_eligible(
    name: str,
    dims: tuple[int, ...],
    include_embd: bool = True,
) -> bool:
    """True for 128-aligned 2D/3D weight tensors eligible for Q1_0."""
    if _is_skipped_metadata_tensor(name, dims):
        return False
    if _is_tiny_ssm_gate(name, dims):
        return False
    if _is_embd_or_head(name) and not include_embd:
        return False
    if len(dims) < 2 or dims[0] % QK1_0 != 0:
        return False
    n_elements = 1
    for d in dims:
        n_elements *= int(d)
    return n_elements >= QK1_0 and n_elements % QK1_0 == 0


def target_quant_type(
    name: str,
    dims: tuple[int, ...],
    enable_q4_fallback: bool = True,
    quantize_embd_q4: bool = True,
    promote_attn_q4: bool = False,
    quant_format: str = "mixed",
    source_type: int | None = None,
    custom_quant_map: dict[str, str] | None = None,
) -> str | None:
    """Determine target quantization format ('Q1_0', 'Q2_K', 'Q4_0', or None for Copy).

    Modes:
    * 'mixed' (default): all Q2_K-eligible tensors -> Q2_K; non-Q2_K 32-aligned
      tensors (and embeddings/head if quantize_embd_q4) -> Q4_0.
    * 'mixed_q1': FFN/expert 128-aligned tensors -> Q1_0; attention/SSM projections,
      embeddings/head, and non-128-aligned 32-aligned tensors -> Q4_0.
    * 'custom': per-tensor-type overrides from ``custom_quant_map``; unlisted
      tensor types fall back to the 'mixed' plan.
    * 'q4_0': all 32-aligned eligible weight tensors -> Q4_0.
    * 'q2_k': only Q2_K-eligible tensors -> Q2_K; everything else copied.
    * 'q3_k': only Q3_K-eligible tensors -> Q3_K; everything else copied.
    * 'q1_0': all 128-aligned eligible weight tensors -> Q1_0.

    ``custom_quant_map`` maps a layer-independent tensor type (see
    :func:`tensor_type_key`) to 'Q1_0', 'Q2_K', 'Q3_K', 'Q4_0', 'Copy', or 'Auto'.
    Non-Auto entries override the selected mode for that tensor type.  A target
    is applied only when the tensor is actually eligible for it; explicit
    'Copy' (and ineligible targets) leave the tensor at its source precision.

    Never promotes a lower-bit source tensor to a higher-bit target format
    (e.g., a Q2_K source tensor is never re-quantized to Q4_0).
    """
    fmt = "mixed" if quant_format == "custom" else quant_format
    tgt: str | None = None
    custom_choice = None
    if custom_quant_map:
        custom_choice = custom_quant_map.get(tensor_type_key(name))
    if custom_choice is not None and str(custom_choice).strip().lower() not in ("", "auto"):
        choice = str(custom_choice).strip().upper()
        if choice == "Q1_0":
            if is_q1_0_eligible(name, dims, include_embd=True):
                tgt = "Q1_0"
        elif choice == "Q2_K":
            if is_q2k_eligible(name, dims):
                tgt = "Q2_K"
        elif choice == "Q3_K":
            if is_q3_k_eligible(name, dims):
                tgt = "Q3_K"
        elif choice == "Q4_0":
            if is_q4_0_eligible(name, dims, include_embd=True, allow_q2k_tensors=True):
                tgt = "Q4_0"
        # 'COPY', unknown values, and ineligible targets leave tgt as None.
    elif fmt == "q1_0":
        if promote_attn_q4 and _is_attention_projection(name) and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
        elif _is_embd_or_head(name) and quantize_embd_q4 and enable_q4_fallback and is_q4_0_eligible(
            name, dims, include_embd=True, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
        elif is_q1_0_eligible(name, dims, include_embd=not quantize_embd_q4):
            tgt = "Q1_0"
        elif enable_q4_fallback and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
    elif fmt == "mixed_q1":
        if (_is_attention_projection(name) or _is_tiny_ssm_gate(name, dims)) and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
        elif _is_embd_or_head(name):
            if quantize_embd_q4 and is_q4_0_eligible(name, dims, include_embd=True):
                tgt = "Q4_0"
        elif is_q1_0_eligible(name, dims, include_embd=False):
            tgt = "Q1_0"
        elif enable_q4_fallback and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=True, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
    elif fmt == "q4_0":
        if is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, allow_q2k_tensors=True
        ):
            tgt = "Q4_0"
    elif fmt == "q2_k":
        if is_q2k_eligible(name, dims):
            tgt = "Q2_K"
    elif fmt == "q3_k":
        if is_q3_k_eligible(name, dims):
            tgt = "Q3_K"
    else:
        if promote_attn_q4 and _is_attention_projection(name) and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=True
        ):
            tgt = "Q4_0"
        elif is_q2k_eligible(name, dims):
            tgt = "Q2_K"
        elif enable_q4_fallback and is_q4_0_eligible(
            name, dims, include_embd=quantize_embd_q4, promote_attn=False
        ):
            tgt = "Q4_0"

    if tgt is not None and source_type is not None:
        src_bpw = source_type_bpw(int(source_type))
        tgt_bpw = {"Q2_K": 2.625, "Q3_K": 3.4375, "Q4_0": 4.5, "Q1_0": 1.125}[tgt]
        if src_bpw <= tgt_bpw + 1.0e-6:
            return None
    return tgt


def _write_string(f, value: str) -> None:
    data = value.encode("utf-8")
    f.write(struct.pack("<Q", len(data)))
    f.write(data)


def write_quantized_gguf(
    input_path: str | Path,
    output_path: str | Path,
    quantizer,
    imat: dict[str, np.ndarray] | None = None,
    quantize_all: bool = True,
    tensor_names: set[str] | None = None,
    limit: int = 0,
    verbose: bool = True,
    progress_callback=None,
    stage_callback=None,
    stop_event=None,
) -> dict[str, float]:
    """Quantize a source GGUF with the V2 engine and emit a valid GGUF.

    Eligible tensors become Q2_K; everything else is copied byte-for-byte,
    including the metadata/tokenizer section.  Returns aggregate statistics.
    """
    reader = GGUFReader(input_path)
    alignment = reader.alignment
    names = set(tensor_names) if tensor_names else None
    if names is not None:
        missing = names - set(reader._by_name)
        if missing:
            raise KeyError(f"tensors not found: {sorted(missing)}")

    cfg = getattr(quantizer, "config", None)
    enable_q4 = bool(getattr(cfg, "enable_q4_fallback", True))
    embd_q4 = bool(getattr(cfg, "quantize_embd_q4", True))
    attn_q4 = bool(getattr(cfg, "promote_attn_q4", False))
    q_format = str(getattr(cfg, "quant_format", "mixed"))
    custom_map = getattr(cfg, "custom_quant_map", None) or None

    plan: list[str | None] = []
    quantized_count = 0
    for ti in reader.tensors:
        target_fmt: str | None = None
        if names is not None:
            if ti.name in names:
                target_fmt = target_quant_type(
                    ti.name,
                    ti.dims,
                    enable_q4_fallback=enable_q4,
                    quantize_embd_q4=embd_q4,
                    promote_attn_q4=attn_q4,
                    quant_format=q_format,
                    source_type=ti.type,
                    custom_quant_map=custom_map,
                )
                custom_choice = None
                if custom_map:
                    custom_choice = custom_map.get(tensor_type_key(ti.name))
                explicit_custom = (
                    custom_choice is not None
                    and str(custom_choice).strip().lower() not in ("", "auto")
                )
                if target_fmt is None and not explicit_custom and q_format != "q3_k":
                    src_bpw = source_type_bpw(ti.type)
                    if q_format == "q1_0" and len(ti.dims) >= 2 and ti.dims[0] % QK1_0 == 0 and src_bpw > 1.125:
                        target_fmt = "Q1_0"
                    elif q_format == "q4_0" and len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and src_bpw > 4.5:
                        target_fmt = "Q4_0"
                    elif len(ti.dims) >= 2 and ti.dims[0] % QK_K == 0 and ti.n_elements % QK_K == 0 and src_bpw > 2.625:
                        target_fmt = "Q2_K"
                    elif len(ti.dims) >= 2 and ti.dims[0] % QK4_0 == 0 and ti.n_elements % QK4_0 == 0 and src_bpw > 4.5:
                        target_fmt = "Q4_0"
                    elif verbose:
                        print(f"  copy (already <= target precision or unaligned): {ti.name}")
        elif quantize_all:
            target_fmt = target_quant_type(
                ti.name,
                ti.dims,
                enable_q4_fallback=enable_q4,
                quantize_embd_q4=embd_q4,
                promote_attn_q4=attn_q4,
                quant_format=q_format,
                source_type=ti.type,
                custom_quant_map=custom_map,
            )
        if target_fmt and ti.type not in SUPPORTED_READ_TYPES:
            if verbose:
                print(f"  copy (unsupported source {ti.type_name}): {ti.name}")
            target_fmt = None
        if target_fmt and limit and quantized_count >= limit:
            target_fmt = None
        if target_fmt:
            quantized_count += 1
        plan.append(target_fmt)

    out_infos = []
    data_offset = 0
    total_planned_blocks = 0
    for ti, fmt in zip(reader.tensors, plan):
        if fmt == "Q2_K":
            n_blocks = ti.n_elements // QK_K
            out_type = GGML_TYPE_Q2_K
            out_size = n_blocks * 84
            total_planned_blocks += n_blocks
        elif fmt == "Q3_K":
            n_blocks = ti.n_elements // QK3_K
            out_type = GGML_TYPE_Q3_K
            out_size = n_blocks * Q3_K_BLOCK_BYTES
            total_planned_blocks += max(1, n_blocks // 16)
        elif fmt == "Q4_0":
            n_blocks = ti.n_elements // QK4_0
            out_type = GGML_TYPE_Q4_0
            out_size = n_blocks * Q4_0_BLOCK_BYTES
            total_planned_blocks += max(1, n_blocks // 8)
        elif fmt == "Q1_0":
            n_blocks = ti.n_elements // QK1_0
            out_type = GGML_TYPE_Q1_0
            out_size = n_blocks * Q1_0_BLOCK_BYTES
            total_planned_blocks += max(1, n_blocks // 32)
        else:
            out_type = ti.type
            out_size = ti.n_bytes
        out_infos.append((ti, out_type, out_size, data_offset))
        data_offset = _align(data_offset + out_size, alignment)

    has_q2k = any(fmt == "Q2_K" for fmt in plan)
    has_q3k = any(fmt == "Q3_K" for fmt in plan)
    has_q1_0 = any(fmt == "Q1_0" for fmt in plan)
    # 11 = MOSTLY_Q3_K, 10 = MOSTLY_Q2_K, 2 = MOSTLY_Q4_0, 40 = MOSTLY_Q1_0
    ftype_code = 11 if has_q3k else (10 if has_q2k else (40 if has_q1_0 else 2))
    kv_pairs = []
    for key, vtype, raw in reader.kv_pairs:
        if quantized_count and key == "general.file_type" and vtype == 4:
            raw = struct.pack("<I", ftype_code)
        elif quantized_count and key == "general.quantization_version" and vtype == 4:
            raw = struct.pack("<I", 2)
        kv_pairs.append((key, vtype, raw))

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    total_rmse = 0.0
    total_quantized = 0
    total_q2k = 0
    total_q3k = 0
    total_q4_0 = 0
    total_q1_0 = 0
    total_elements = 0
    completed_planned_blocks = 0
    model_t0 = time.time()
    with output_path.open("wb") as fout:
        fout.write(struct.pack("<IIQQ", GGUF_MAGIC, reader.version, len(out_infos), len(kv_pairs)))
        for key, vtype, raw in kv_pairs:
            _write_string(fout, key)
            fout.write(struct.pack("<I", vtype))
            fout.write(raw)
        for ti, out_type, out_size, offset in out_infos:
            _write_string(fout, ti.name)
            fout.write(struct.pack("<I", len(ti.dims)))
            for d in ti.dims:
                fout.write(struct.pack("<Q", int(d)))
            fout.write(struct.pack("<I", out_type))
            fout.write(struct.pack("<Q", offset))

        data_start = _align(fout.tell(), alignment)
        fout.write(b"\x00" * (data_start - fout.tell()))

        for idx, (ti, out_type, out_size, offset) in enumerate(out_infos):
            if stop_event is not None and stop_event.is_set():
                raise InterruptedError("Quantization cancelled by user")
            target = data_start + offset
            if target > fout.tell():
                fout.write(b"\x00" * (target - fout.tell()))
            fmt = plan[idx]
            if fmt == "Q2_K":
                t_blocks = ti.n_elements // QK_K
                has_imat = imat is not None and lookup_imatrix(imat, ti.name) is not None

                def _emit_stage(ev: dict, _i=idx + 1, _ti=ti, _base=completed_planned_blocks, _tb=t_blocks):
                    if stage_callback is None:
                        return
                    phase_name = ev.get("phase", "")
                    done_b = int(ev.get("done", 0))
                    tot_b = max(int(ev.get("total", _tb)), 1)
                    if phase_name.startswith("Phase 2b") or "Phase 1–2b" in phase_name:
                        frac = 0.80 * (done_b / tot_b)
                    elif phase_name.startswith("Phase 3"):
                        frac = 0.82
                    elif "Phase 4" in phase_name and not phase_name.startswith("Phase 4.8"):
                        frac = 0.82 + 0.17 * (done_b / tot_b)
                    elif phase_name.startswith("Phase 4.8") or phase_name.startswith("Phase 5"):
                        frac = 1.0
                    else:
                        frac = 0.0
                    eff_blocks = _base + frac * _tb
                    elapsed_model = max(time.time() - model_t0, 1.0e-6)
                    model_rate = eff_blocks / elapsed_model
                    rem_blocks = max(total_planned_blocks - eff_blocks, 0.0)
                    model_eta = rem_blocks / model_rate if model_rate > 1.0e-6 else 0.0
                    model_pct = 100.0 * eff_blocks / max(total_planned_blocks, 1)
                    stage_callback(
                        _i,
                        len(out_infos),
                        _ti,
                        {
                            **ev,
                            "model_pct": model_pct,
                            "model_eta": model_eta,
                            "model_done_blocks": int(eff_blocks),
                            "model_total_blocks": total_planned_blocks,
                        },
                    )

                _emit_stage(
                    {
                        "phase": f"Reading & Dequantizing Source ({ti.type_name})",
                        "done": 0,
                        "total": t_blocks,
                        "detail": f"dims={ti.dims} elements={ti.n_elements:,} ({ti.n_bytes / 1048576:.2f} MB) imatrix={'yes' if has_imat else 'no'}",
                    }
                )
                x, row = reader.read_tensor(ti.name)
                imp = None
                if imat is not None:
                    raw_imp = lookup_imatrix(imat, ti.name)
                    if raw_imp is not None:
                        imp = expand_importance(raw_imp, x.size, row)
                res = quantizer.quantize(
                    x,
                    importance=imp,
                    row_width=row,
                    status_callback=_emit_stage if stage_callback is not None else None,
                    stop_event=stop_event,
                )
                completed_planned_blocks += t_blocks
                payload = blocks_to_bytes(res.scales, res.qs, res.dm)
                if len(payload) != out_size:
                    raise RuntimeError(f"{ti.name}: Q2_K payload size mismatch")
                fout.write(payload)
                total_rmse += res.stats.rmse * res.stats.rmse * x.size
                total_elements += x.size
                total_quantized += 1
                total_q2k += 1
                if verbose:
                    print(
                        f"  Q2_K {ti.name:48s} row={row:6d} "
                        f"rmse={res.stats.rmse:.6e} imatrix={'yes' if imp is not None else 'no'}"
                    )
                if progress_callback is not None:
                    progress_callback(
                        idx + 1,
                        len(out_infos),
                        ti.name,
                        "Q2_K",
                        float(res.stats.rmse),
                        imp is not None,
                    )
            elif fmt == "Q3_K":
                t_blocks_q3 = ti.n_elements // QK3_K
                eff_tb = max(1, t_blocks_q3 // 16)
                raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None
                has_imat = raw_imp is not None and raw_imp.size == ti.row_width
                written_bytes = 0
                sse_q3 = 0.0
                done_q3_blks = 0
                t_q3_start = time.time()

                for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608):
                    if stop_event is not None and stop_event.is_set():
                        raise InterruptedError("Quantization cancelled by user")
                    imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None
                    q3_bytes, nb_chunk, chunk_rmse = quantize_tensor_q3_k(
                        x_chunk, importance=imp_chunk, cfg=cfg, chunk_blocks=8192
                    )
                    fout.write(q3_bytes)
                    written_bytes += len(q3_bytes)
                    sse_q3 += (chunk_rmse * chunk_rmse) * x_chunk.size
                    done_q3_blks += nb_chunk
                    if stage_callback is not None:
                        frac = done_q3_blks / max(t_blocks_q3, 1)
                        eff_blocks = completed_planned_blocks + frac * eff_tb
                        elapsed_model = max(time.time() - model_t0, 1.0e-6)
                        model_rate = eff_blocks / elapsed_model
                        model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6)
                        el_q3 = max(time.time() - t_q3_start, 1.0e-6)
                        r_q3 = done_q3_blks / el_q3
                        eta_q3 = (t_blocks_q3 - done_q3_blks) / max(r_q3, 1.0e-6)
                        stage_callback(
                            idx + 1,
                            len(out_infos),
                            ti,
                            {
                                "phase": "Q3_K: Analog + WLS + Sieve/Viterbi + FP16-ULP Quantization",
                                "done": done_q3_blks,
                                "total": t_blocks_q3,
                                "detail": f"blk {done_q3_blks:,}/{t_blocks_q3:,} ({r_q3:.0f} blk/s) rmse={chunk_rmse:.6e}",
                                "rate": r_q3,
                                "eta": eta_q3,
                                "rmse": chunk_rmse,
                                "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))),
                                "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1),
                                "model_eta": model_eta,
                                "model_done_blocks": int(eff_blocks),
                                "model_total_blocks": total_planned_blocks,
                            },
                        )

                if written_bytes != out_size:
                    raise RuntimeError(f"{ti.name}: Q3_K payload size mismatch")
                completed_planned_blocks += eff_tb
                q3_rmse = float(np.sqrt(sse_q3 / ti.n_elements))
                total_rmse += sse_q3
                total_elements += ti.n_elements
                total_quantized += 1
                total_q3k += 1
                if verbose:
                    print(
                        f"  Q3_K {ti.name:48s} row={ti.row_width:6d} "
                        f"rmse={q3_rmse:.6e} imatrix={'yes' if has_imat else 'no'}"
                    )
                if progress_callback is not None:
                    progress_callback(
                        idx + 1,
                        len(out_infos),
                        ti.name,
                        "Q3_K",
                        q3_rmse,
                        has_imat,
                    )
            elif fmt == "Q4_0":
                t_blocks_q4 = ti.n_elements // QK4_0
                eff_tb = max(1, t_blocks_q4 // 8)
                raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None
                has_imat = raw_imp is not None and raw_imp.size == ti.row_width
                written_bytes = 0
                sse_q4 = 0.0
                done_q4_blks = 0
                t_q4_start = time.time()

                for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608):
                    if stop_event is not None and stop_event.is_set():
                        raise InterruptedError("Quantization cancelled by user")
                    imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None
                    q4_bytes, nb_chunk, chunk_rmse = quantize_tensor_q4_0(
                        x_chunk, importance=imp_chunk, cfg=cfg, chunk_blocks=16384
                    )
                    fout.write(q4_bytes)
                    written_bytes += len(q4_bytes)
                    sse_q4 += (chunk_rmse * chunk_rmse) * x_chunk.size
                    done_q4_blks += nb_chunk
                    if stage_callback is not None:
                        frac = done_q4_blks / max(t_blocks_q4, 1)
                        eff_blocks = completed_planned_blocks + frac * eff_tb
                        elapsed_model = max(time.time() - model_t0, 1.0e-6)
                        model_rate = eff_blocks / elapsed_model
                        model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6)
                        el_q4 = max(time.time() - t_q4_start, 1.0e-6)
                        r_q4 = done_q4_blks / el_q4
                        eta_q4 = (t_blocks_q4 - done_q4_blks) / max(r_q4, 1.0e-6)
                        stage_callback(
                            idx + 1,
                            len(out_infos),
                            ti,
                            {
                                "phase": "Q4_0: Analog + WLS + FP16-ULP Quantization",
                                "done": done_q4_blks,
                                "total": t_blocks_q4,
                                "detail": f"blk {done_q4_blks:,}/{t_blocks_q4:,} ({r_q4:.0f} blk/s) rmse={chunk_rmse:.6e}",
                                "rate": r_q4,
                                "eta": eta_q4,
                                "rmse": chunk_rmse,
                                "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))),
                                "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1),
                                "model_eta": model_eta,
                                "model_done_blocks": int(eff_blocks),
                                "model_total_blocks": total_planned_blocks,
                            },
                        )

                if written_bytes != out_size:
                    raise RuntimeError(f"{ti.name}: Q4_0 payload size mismatch")
                completed_planned_blocks += eff_tb
                q4_rmse = float(np.sqrt(sse_q4 / ti.n_elements))
                total_rmse += sse_q4
                total_elements += ti.n_elements
                total_quantized += 1
                total_q4_0 += 1
                if verbose:
                    print(
                        f"  Q4_0 {ti.name:48s} row={ti.row_width:6d} "
                        f"rmse={q4_rmse:.6e} imatrix={'yes' if has_imat else 'no'}"
                    )
                if progress_callback is not None:
                    progress_callback(
                        idx + 1,
                        len(out_infos),
                        ti.name,
                        "Q4_0",
                        q4_rmse,
                        has_imat,
                    )
            elif fmt == "Q1_0":
                t_blocks_q1 = ti.n_elements // QK1_0
                eff_tb = max(1, t_blocks_q1 // 32)
                raw_imp = lookup_imatrix(imat, ti.name) if imat is not None else None
                has_imat = raw_imp is not None and raw_imp.size == ti.row_width
                written_bytes = 0
                sse_q1 = 0.0
                done_q1_blks = 0
                t_q1_start = time.time()

                for x_chunk, row, _r0, _r1 in reader.iter_tensor_row_chunks(ti.name, max_elements=8_388_608):
                    if stop_event is not None and stop_event.is_set():
                        raise InterruptedError("Quantization cancelled by user")
                    imp_chunk = expand_importance(raw_imp, x_chunk.size, row) if has_imat else None
                    q1_bytes, nb_chunk, chunk_rmse = quantize_tensor_q1_0(
                        x_chunk, importance=imp_chunk, cfg=cfg
                    )
                    fout.write(q1_bytes)
                    written_bytes += len(q1_bytes)
                    sse_q1 += (chunk_rmse * chunk_rmse) * x_chunk.size
                    done_q1_blks += nb_chunk
                    if stage_callback is not None:
                        frac = done_q1_blks / max(t_blocks_q1, 1)
                        eff_blocks = completed_planned_blocks + frac * eff_tb
                        elapsed_model = max(time.time() - model_t0, 1.0e-6)
                        model_rate = eff_blocks / elapsed_model
                        model_eta = max(total_planned_blocks - eff_blocks, 0.0) / max(model_rate, 1.0e-6)
                        el_q1 = max(time.time() - t_q1_start, 1.0e-6)
                        r_q1 = done_q1_blks / el_q1
                        eta_q1 = (t_blocks_q1 - done_q1_blks) / max(r_q1, 1.0e-6)
                        stage_callback(
                            idx + 1,
                            len(out_infos),
                            ti,
                            {
                                "phase": "Q1_0: Analog + WLS + Sieve/Viterbi + FP16-ULP Quantization",
                                "done": done_q1_blks,
                                "total": t_blocks_q1,
                                "detail": f"blk {done_q1_blks:,}/{t_blocks_q1:,} ({r_q1:.0f} blk/s) rmse={chunk_rmse:.6e}",
                                "rate": r_q1,
                                "eta": eta_q1,
                                "rmse": chunk_rmse,
                                "sigma": float(np.sqrt(np.mean(x_chunk.astype(np.float64) ** 2))),
                                "model_pct": 100.0 * eff_blocks / max(total_planned_blocks, 1),
                                "model_eta": model_eta,
                                "model_done_blocks": int(eff_blocks),
                                "model_total_blocks": total_planned_blocks,
                            },
                        )

                if written_bytes != out_size:
                    raise RuntimeError(f"{ti.name}: Q1_0 payload size mismatch")
                completed_planned_blocks += eff_tb
                q1_rmse = float(np.sqrt(sse_q1 / ti.n_elements))
                total_rmse += sse_q1
                total_elements += ti.n_elements
                total_quantized += 1
                total_q1_0 += 1
                if verbose:
                    print(
                        f"  Q1_0 {ti.name:48s} row={ti.row_width:6d} "
                        f"rmse={q1_rmse:.6e} imatrix={'yes' if has_imat else 'no'}"
                    )
                if progress_callback is not None:
                    progress_callback(
                        idx + 1,
                        len(out_infos),
                        ti.name,
                        "Q1_0",
                        q1_rmse,
                        has_imat,
                    )
            else:
                if stage_callback is not None:
                    stage_callback(
                        idx + 1,
                        len(out_infos),
                        ti,
                        {
                            "phase": f"Passthrough Copy ({ti.type_name})",
                            "done": 1,
                            "total": 1,
                            "detail": f"dims={ti.dims} ({ti.n_bytes / 1048576:.2f} MB)",
                        },
                    )
                reader.copy_tensor_raw_to(ti.name, fout)
                if verbose and idx % 25 == 0:
                    print(f"  copy {ti.name}")
                if progress_callback is not None:
                    progress_callback(idx + 1, len(out_infos), ti.name, "copy", None, False)

    rmse = float(np.sqrt(total_rmse / total_elements)) if total_elements else 0.0
    return {
        "tensors": float(len(out_infos)),
        "quantized": float(total_quantized),
        "q2_k": float(total_q2k),
        "q3_k": float(total_q3k),
        "q4_0": float(total_q4_0),
        "q1_0": float(total_q1_0),
        "elements": float(total_elements),
        "rmse": rmse,
    }