app.py

import os
import struct
from collections import Counter

import gradio as gr
import requests
from huggingface_hub import HfApi, hf_hub_url


DEFAULT_REPO = "AdvancedDataIntelligence/adi-qwen3.5-4b-glm5.2-general-GGUF"
DEFAULT_FILE = "adi-qwen3.5-4b-glm5.2-general-q4_k_m.gguf"
MAX_RANGE_BYTES = 256 * 1024 * 1024

GGUF_TYPES = {
    0: ("UINT8", "B", 1),
    1: ("INT8", "b", 1),
    2: ("UINT16", "H", 2),
    3: ("INT16", "h", 2),
    4: ("UINT32", "I", 4),
    5: ("INT32", "i", 4),
    6: ("FLOAT32", "f", 4),
    7: ("BOOL", "?", 1),
    8: ("STRING", None, None),
    9: ("ARRAY", None, None),
    10: ("UINT64", "Q", 8),
    11: ("INT64", "q", 8),
    12: ("FLOAT64", "d", 8),
}

TENSOR_TYPES = {
    0: "F32",
    1: "F16",
    2: "Q4_0",
    3: "Q4_1",
    6: "Q5_0",
    7: "Q5_1",
    8: "Q8_0",
    9: "Q8_1",
    10: "Q2_K",
    11: "Q3_K",
    12: "Q4_K",
    13: "Q5_K",
    14: "Q6_K",
    15: "Q8_K",
    16: "IQ2_XXS",
    17: "IQ2_XS",
    18: "IQ3_XXS",
    19: "IQ1_S",
    20: "IQ4_NL",
    21: "IQ3_S",
    22: "IQ2_S",
    23: "IQ4_XS",
    24: "I8",
    25: "I16",
    26: "I32",
    27: "I64",
    28: "F64",
    29: "IQ1_M",
    30: "BF16",
    31: "TQ1_0",
    32: "TQ2_0",
}


class NeedMoreData(Exception):
    pass


class Reader:
    def __init__(self, data):
        self.data = data
        self.pos = 0

    def require(self, size):
        if self.pos + size > len(self.data):
            raise NeedMoreData

    def read(self, size):
        self.require(size)
        chunk = self.data[self.pos : self.pos + size]
        self.pos += size
        return chunk

    def unpack(self, fmt):
        size = struct.calcsize("<" + fmt)
        return struct.unpack("<" + fmt, self.read(size))[0]

    def string(self):
        length = self.unpack("Q")
        return self.read(length).decode("utf-8", errors="replace")


def fetch_prefix(repo_id, filename, revision, size):
    token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN")
    headers = {"Range": f"bytes=0-{size - 1}"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    url = hf_hub_url(repo_id=repo_id, filename=filename, revision=revision)
    response = requests.get(url, headers=headers, allow_redirects=True, timeout=60)
    response.raise_for_status()
    return response.content


def file_size(repo_id, filename, revision):
    info = HfApi().model_info(repo_id=repo_id, revision=revision, files_metadata=True)
    for sibling in info.siblings:
        if sibling.rfilename == filename:
            return sibling.size
    return None


def read_scalar(reader, value_type):
    type_name, fmt, size = GGUF_TYPES.get(value_type, (f"UNKNOWN_{value_type}", None, None))
    if value_type == 8:
        return reader.string()
    if value_type == 9:
        item_type = reader.unpack("I")
        item_count = reader.unpack("Q")
        values = []
        for _ in range(min(item_count, 64)):
            values.append(read_scalar(reader, item_type))
        if item_count > 64:
            skip_value(reader, item_type, item_count - 64)
            values.append(f"... {item_count - 64} more")
        return values
    if fmt is None:
        raise ValueError(f"Unsupported GGUF metadata type {type_name}")
    return reader.unpack(fmt)


def skip_value(reader, value_type, count=1):
    for _ in range(count):
        if value_type == 8:
            length = reader.unpack("Q")
            reader.read(length)
        elif value_type == 9:
            item_type = reader.unpack("I")
            item_count = reader.unpack("Q")
            skip_value(reader, item_type, item_count)
        else:
            _name, _fmt, size = GGUF_TYPES.get(value_type, (None, None, None))
            if size is None:
                raise ValueError(f"Unsupported GGUF metadata type {value_type}")
            reader.read(size)


def parse_gguf(data):
    reader = Reader(data)
    if reader.read(4) != b"GGUF":
        raise ValueError("File does not start with GGUF magic bytes.")

    version = reader.unpack("I")
    tensor_count = reader.unpack("Q")
    metadata_count = reader.unpack("Q")

    metadata = {}
    metadata_types = {}
    for _ in range(metadata_count):
        key = reader.string()
        value_type = reader.unpack("I")
        metadata[key] = read_scalar(reader, value_type)
        metadata_types[key] = GGUF_TYPES.get(value_type, (str(value_type), None, None))[0]

    tensor_types = Counter()
    tensor_names = []
    tensor_shapes = []
    for _ in range(tensor_count):
        name = reader.string()
        dims_count = reader.unpack("I")
        dims = [reader.unpack("Q") for _ in range(dims_count)]
        tensor_type_id = reader.unpack("I")
        offset = reader.unpack("Q")
        tensor_type = TENSOR_TYPES.get(tensor_type_id, f"TYPE_{tensor_type_id}")
        tensor_types[tensor_type] += 1
        if len(tensor_names) < 80:
            tensor_names.append(name)
            tensor_shapes.append(
                {
                    "name": name,
                    "shape": " x ".join(str(d) for d in dims),
                    "type": tensor_type,
                    "offset": offset,
                }
            )

    return {
        "version": version,
        "tensor_count": tensor_count,
        "metadata_count": metadata_count,
        "metadata": metadata,
        "metadata_types": metadata_types,
        "tensor_types": dict(tensor_types),
        "tensor_names": tensor_names,
        "tensor_shapes": tensor_shapes,
        "header_bytes_read": reader.pos,
    }


def inspect_gguf(repo_id, filename, revision):
    repo_id = repo_id.strip()
    filename = filename.strip()
    revision = (revision or "main").strip()
    if not repo_id or not filename:
        raise gr.Error("Repo ID and GGUF filename are required.")

    last_error = None
    data = b""
    for size in (2, 4, 8, 16, 32, 64, 128, 256):
        byte_count = size * 1024 * 1024
        try:
            data = fetch_prefix(repo_id, filename, revision, byte_count)
            parsed = parse_gguf(data)
            break
        except NeedMoreData as exc:
            last_error = exc
            if byte_count >= MAX_RANGE_BYTES:
                raise gr.Error("GGUF header is larger than 256 MB; cannot inspect safely.")
        except requests.HTTPError as exc:
            raise gr.Error(f"Could not fetch GGUF prefix: HTTP {exc.response.status_code}")
    else:
        raise gr.Error(f"Could not parse GGUF metadata: {last_error}")

    size_bytes = file_size(repo_id, filename, revision)
    summary = summarize(repo_id, filename, revision, size_bytes, parsed)
    metadata_rows = metadata_table(parsed["metadata"], parsed["metadata_types"])
    tensor_rows = tensor_table(parsed["tensor_shapes"])
    return summary, metadata_rows, tensor_rows, parsed


def pick(metadata, suffixes):
    for suffix in suffixes:
        for key, value in metadata.items():
            if key.endswith(suffix):
                return value
    return None


def summarize(repo_id, filename, revision, size_bytes, parsed):
    metadata = parsed["metadata"]
    arch = metadata.get("general.architecture", "unknown")
    model_name = metadata.get("general.name", filename)
    quant = metadata.get("general.file_type")
    quant_name = TENSOR_TYPES.get(quant, quant) if isinstance(quant, int) else quant
    ctx_train = pick(metadata, [".context_length"])
    block_count = pick(metadata, [".block_count"])
    embedding = pick(metadata, [".embedding_length"])
    head_count = pick(metadata, [".attention.head_count"])
    tokenizer = metadata.get("tokenizer.ggml.model", "unknown")
    tensor_mix = ", ".join(
        f"{name}: {count}" for name, count in sorted(parsed["tensor_types"].items())
    )

    warnings = compatibility_notes(arch, metadata, parsed["tensor_names"])
    size_text = format_bytes(size_bytes) if size_bytes else "unknown"

    return f"""# {model_name}

**Repo:** `{repo_id}`  
**File:** `{filename}`  
**Revision:** `{revision}`  
**Size:** `{size_text}`  
**GGUF version:** `{parsed["version"]}`  
**Architecture:** `{arch}`  
**File type:** `{quant_name}`  
**Tensor count:** `{parsed["tensor_count"]}`  
**Metadata entries:** `{parsed["metadata_count"]}`  
**Header bytes inspected:** `{format_bytes(parsed["header_bytes_read"])}`  

## Model Shape

- Training/context length: `{ctx_train}`
- Blocks/layers: `{block_count}`
- Embedding length: `{embedding}`
- Attention heads: `{head_count}`
- Tokenizer: `{tokenizer}`

## Tensor Types

`{tensor_mix or "none"}`

## Compatibility Notes

{warnings}
"""


def compatibility_notes(arch, metadata, tensor_names):
    lower_arch = str(arch).lower()
    keys = " ".join(metadata.keys()).lower()
    names = " ".join(tensor_names[:80]).lower()
    haystack = f"{lower_arch} {keys} {names}"
    notes = []

    if "qwen3" in haystack and any(term in haystack for term in ("ssm", "mamba", "gated_delta", "gated-delta")):
        notes.append(
            "- Qwen3/Qwen3.5 hybrid SSM or gated-delta signals detected. Use a very recent llama.cpp build."
        )
    elif "qwen3" in haystack:
        notes.append("- Qwen3-family metadata detected. Prefer recent llama.cpp/llama-cpp-python builds.")

    if any(term in haystack for term in ("ssm", "mamba", "gated_delta", "gated-delta")):
        notes.append("- SSM/Mamba-style metadata detected; older llama.cpp builds may fail at model load.")

    if "tokenizer.ggml.tokens" not in metadata:
        notes.append("- Token list is not present in the inspected metadata prefix.")

    if not notes:
        notes.append("- No obvious metadata-level compatibility red flags found.")

    return "\n".join(notes)


def metadata_table(metadata, metadata_types):
    rows = []
    for key in sorted(metadata):
        value = metadata[key]
        if isinstance(value, list):
            value_text = f"[{len(value)} items] " + repr(value[:8])
        else:
            value_text = repr(value)
        if len(value_text) > 500:
            value_text = value_text[:497] + "..."
        rows.append([key, metadata_types.get(key, ""), value_text])
    return rows


def tensor_table(tensor_shapes):
    return [[item["name"], item["shape"], item["type"], item["offset"]] for item in tensor_shapes]


def format_bytes(value):
    if value is None:
        return "unknown"
    value = float(value)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if value < 1024 or unit == "TB":
            return f"{value:.2f} {unit}" if unit != "B" else f"{int(value)} B"
        value /= 1024


with gr.Blocks(title="ADI GGUF Inspector", fill_width=True) as demo:
    gr.Markdown("# ADI GGUF Inspector")
    with gr.Row():
        repo = gr.Textbox(label="Model repo", value=DEFAULT_REPO)
        filename = gr.Textbox(label="GGUF filename", value=DEFAULT_FILE)
        revision = gr.Textbox(label="Revision", value="main")
    inspect_btn = gr.Button("Inspect", variant="primary")
    summary = gr.Markdown()
    with gr.Tabs():
        with gr.Tab("Metadata"):
            metadata = gr.Dataframe(
                headers=["Key", "Type", "Value"],
                datatype=["str", "str", "str"],
                wrap=True,
                interactive=False,
            )
        with gr.Tab("Tensors"):
            tensors = gr.Dataframe(
                headers=["Name", "Shape", "Type", "Offset"],
                datatype=["str", "str", "str", "number"],
                wrap=True,
                interactive=False,
            )
        with gr.Tab("Raw"):
            raw = gr.JSON()

    inspect_btn.click(
        inspect_gguf,
        inputs=[repo, filename, revision],
        outputs=[summary, metadata, tensors, raw],
    )


if __name__ == "__main__":
    demo.launch()