feat(civitai): name gguf downloads by their header quant

A GGUF's variant is its quant, never metadata.fp, which is the dtype it
was quantized from (a Q4_K_M upload tagged fp8 was saved as -fp8). The
name route now reads the GGUF header the way it reads safetensors
headers: general.file_type gives the exact quant (Q4_K_M) and files
without it, such as Ideogram 4 GGUFs with an empty KV block, fall back
to the dominant tensor type (Q4_0). When the header cannot be read the
metadata or url quantType still applies; fp never does.

- peek_civitai gains a v2/v3 GGUF reader over a 1 MB ranged read, with
  the same id-keyed cache and the quant token in the "quant" field
- peek_headers dispatches by extension, so .gguf siblings are probed
  alongside .safetensors ones
This commit is contained in:
CalamitousFelicitousness
2026-09-30 03:22:02 +01:00
parent 87e62e6545
commit f49595da0a
4 changed files with 400 additions and 74 deletions
+19 -7
View File
@@ -52,11 +52,19 @@ def url_quant_type(url: str) -> str | None:
return values[0] if values else None
def is_gguf(f: CivitFile) -> bool:
return f.metadata.format == 'GGUF' or f.name.lower().endswith('.gguf')
def file_variant(f: CivitFile) -> str | None:
"""Precision for safetensors is metadata.fp; GGUF quants carry metadata.quantType
on the model endpoint only, so fall back to the quantType query param of the
download URL (CivitAI's own variant discriminator) when metadata lacks it."""
return f.metadata.fp or f.metadata.quant_type or url_quant_type(f.download_url)
"""Precision for safetensors is metadata.fp. A GGUF's variant is its quant:
metadata.quantType (model endpoint only) or the quantType query param of
the download URL; its metadata.fp is the dtype it was quantized from, so it
never names the file."""
quant = f.metadata.quant_type or url_quant_type(f.download_url)
if is_gguf(f):
return quant
return f.metadata.fp or quant
def split_words(text: str) -> str:
@@ -102,7 +110,7 @@ def role_from_metadata(metadata: dict | None, context: NameContext) -> str | Non
def peek_targets(files: list[CivitFile]) -> list[CivitFile]:
return [f for f in files if f.name.lower().endswith('.safetensors')]
return [f for f in files if f.name.lower().endswith(('.safetensors', '.gguf'))]
def precision_from_dtype(dtype: str | None) -> str | None:
@@ -116,11 +124,15 @@ def precision_claim_satisfied(claimed: str, actual: str) -> bool:
def apply_peeks(context: NameContext, files: list[CivitFile], peeks: dict[int, dict]) -> NameContext:
"""Fold header probes into the context: roles from __metadata__, and a
generic metadata variant upgraded to the dtype-exact token."""
"""Fold header probes into the context: roles from __metadata__, a GGUF's
quant from its header, and a generic metadata variant upgraded to the
dtype-exact token."""
for f in files:
data = peeks.get(f.id) or {}
context.roles[f.id] = role_from_metadata(data.get('metadata'), context)
if data.get('quant'):
context.variants[f.id] = data['quant']
continue
probe = data.get('probe')
if not probe:
continue
+138 -31
View File
@@ -1,4 +1,4 @@
"""Remote safetensors header probe for CivitAI files, cached by file id."""
"""Remote safetensors and GGUF header probes for CivitAI files, cached by file id."""
import hashlib
import json
@@ -7,6 +7,7 @@ from modules.logger import log
peek_cache = None
GGUF_READ_LIMIT = 1024 * 1024 # KV block plus tensor infos of a diffusion GGUF fit well inside 1 MB
def url_hash(url: str) -> str:
@@ -40,7 +41,34 @@ def peek_cache_put(file_id: int, url: str, response: dict):
def normalize_url(url: str) -> str:
# civitai.red serves the same download service and rewrites downloadUrl to
# its own host; normalize so the guard, cache hash and fetch host agree.
return url.replace('https://civitai.red/', 'https://civitai.com/', 1)
url = url.replace('https://civitai.red/', 'https://civitai.com/', 1)
if not url.startswith('https://civitai.com/'):
raise ValueError('only civitai urls are allowed')
return url
def read_range(url: str, start: int, end: int) -> bytes:
"""Bytes start..end of a remote file; a 200 reply means the server ignored the range and sends from byte 0."""
from modules import shared
headers = {'Range': f'bytes={start}-{end}'}
token = getattr(shared.opts, 'civitai_token', '') or ''
if token:
headers['Authorization'] = f'Bearer {token}'
r = shared.req(url, headers=headers, stream=True)
status = getattr(r, 'status_code', 500)
if status not in (200, 206):
raise RuntimeError(f'HTTP {status}')
want = end - start + 1
need = end + 1 if status == 200 else want
buf = b''
try:
for chunk in r.iter_content(chunk_size=65536):
buf += chunk
if len(buf) >= need:
break
finally:
r.close()
return buf[start:start + want] if status == 200 else buf[:want]
def peek_header(url: str, file_id: int = 0) -> dict:
@@ -48,45 +76,19 @@ def peek_header(url: str, file_id: int = 0) -> dict:
__metadata__ block plus a full model_probe analysis (architecture,
dtypes, quant scheme) without downloading the file. Failures come back
as {"metadata": None, "error": reason} and are not cached."""
from modules import shared
url = normalize_url(url)
if not url.startswith('https://civitai.com/'):
raise ValueError('only civitai urls are allowed')
if file_id:
cached = peek_cache_get(file_id, url)
if cached is not None:
return cached
base_headers = {}
token = getattr(shared.opts, 'civitai_token', '') or ''
if token:
base_headers['Authorization'] = f'Bearer {token}'
def read_range(start: int, end: int) -> bytes:
r = shared.req(url, headers={**base_headers, 'Range': f'bytes={start}-{end}'}, stream=True)
status = getattr(r, 'status_code', 500)
if status not in (200, 206):
raise RuntimeError(f'HTTP {status}')
want = end - start + 1
# A 200 reply means the server ignored the range and sends from byte 0.
need = end + 1 if status == 200 else want
buf = b''
try:
for chunk in r.iter_content(chunk_size=65536):
buf += chunk
if len(buf) >= need:
break
finally:
r.close()
return buf[start:start + want] if status == 200 else buf[:want]
try:
prefix = read_range(0, 7)
prefix = read_range(url, 0, 7)
if len(prefix) < 8:
return {"metadata": None, "error": "short read"}
header_len = struct.unpack('<Q', prefix)[0]
if header_len <= 0 or header_len > 16 * 1024 * 1024:
return {"metadata": None, "error": f"implausible header length: {header_len}"}
header = json.loads(read_range(8, 7 + header_len).decode('utf-8'))
header = json.loads(read_range(url, 8, 7 + header_len).decode('utf-8'))
except Exception as e:
return {"metadata": None, "error": str(e)}
from modules import model_probe
@@ -100,6 +102,111 @@ def peek_header(url: str, file_id: int = 0) -> dict:
return response
class GGUFReader:
"""Minimal GGUF v2/v3 header parser over an in-memory prefix of the file."""
SCALARS = {0: 'B', 1: 'b', 2: 'H', 3: 'h', 4: 'I', 5: 'i', 6: 'f', 7: '?', 10: 'Q', 11: 'q', 12: 'd'}
def __init__(self, buf: bytes):
self.buf = buf
self.pos = 0
def take(self, fmt: str):
size = struct.calcsize('<' + fmt)
if self.pos + size > len(self.buf):
raise EOFError('header exceeds the read window')
value = struct.unpack_from('<' + fmt, self.buf, self.pos)[0]
self.pos += size
return value
def string(self) -> str:
n = self.take('Q')
if self.pos + n > len(self.buf):
raise EOFError('header exceeds the read window')
value = self.buf[self.pos:self.pos + n].decode('utf-8', 'replace')
self.pos += n
return value
def value(self, kind: int):
if kind == 8:
return self.string()
if kind == 9:
element = self.take('I')
count = self.take('Q')
return [self.value(element) for _ in range(count)]
return self.take(self.SCALARS[kind])
def parse(self) -> tuple[dict, dict]:
"""(kv, tensors) where tensors maps name to dtype and shape; tensor
infos past the read window are dropped, the dominant type survives."""
if self.buf[:4] != b'GGUF':
raise ValueError('not a gguf file')
self.pos = 4
version = self.take('I')
if version not in (2, 3):
raise ValueError(f'unsupported gguf version {version}')
n_tensors = self.take('Q')
n_kv = self.take('Q')
kv = {}
for _ in range(n_kv):
key = self.string()
kv[key] = self.value(self.take('I'))
tensors = {}
try:
for _ in range(n_tensors):
name = self.string()
dims = [self.take('Q') for _ in range(self.take('I'))]
ggml_type = self.take('I')
self.take('Q')
tensors[name] = {'dtype': ggml_type, 'shape': list(reversed(dims))}
except EOFError:
pass
return kv, tensors
def gguf_quant_token(kv: dict, probe: dict) -> str | None:
"""general.file_type names the quant exactly (Q4_K_M); files without it fall back to the dominant tensor type (Q4_K)."""
import gguf
file_type = kv.get('general.file_type')
if isinstance(file_type, int):
try:
return gguf.LlamaFileType(file_type).name.removeprefix('MOSTLY_').removeprefix('ALL_')
except ValueError:
pass
return (probe.get('quant') or {}).get('format')
def peek_gguf(url: str, file_id: int = 0) -> dict:
"""GGUF counterpart of peek_header: the general.* metadata, a model_probe
analysis over the tensor types and the quant token in "quant"."""
url = normalize_url(url)
if file_id:
cached = peek_cache_get(file_id, url)
if cached is not None:
return cached
try:
import gguf
kv, tensors = GGUFReader(read_range(url, 0, GGUF_READ_LIMIT - 1)).parse()
header = {name: {'dtype': gguf.GGMLQuantizationType(t['dtype']).name, 'shape': t['shape']} for name, t in tensors.items()}
except Exception as e:
return {"metadata": None, "error": str(e)}
from modules import model_probe
metadata = {k: v for k, v in kv.items() if k.startswith('general.') and not isinstance(v, list)}
probe = model_probe.analyze_header(header, container='gguf', arch_metadata={k: v for k, v in metadata.items() if isinstance(v, str)})
response = {
"metadata": metadata,
"tensors": len(header),
"probe": probe,
"quant": gguf_quant_token(kv, probe),
}
if file_id:
peek_cache_put(file_id, url, response)
return response
def peek_file(name: str, url: str, file_id: int = 0) -> dict:
return peek_gguf(url, file_id) if name.lower().endswith('.gguf') else peek_header(url, file_id)
def peek_headers(files: list) -> dict[int, dict]:
"""Probe several CivitFile entries concurrently; failed probes log and yield an error dict."""
from concurrent.futures import ThreadPoolExecutor
@@ -108,7 +215,7 @@ def peek_headers(files: list) -> dict[int, dict]:
def probe(f):
try:
return peek_header(f.download_url, f.id)
return peek_file(f.name, f.download_url, f.id)
except Exception as e:
return {"metadata": None, "error": str(e)}
+187 -35
View File
@@ -570,6 +570,92 @@
}
]
},
{
"source": "GET /model-versions/3011173",
"id": 3011173,
"modelId": 2681714,
"name": "Q4",
"baseModel": "Ideogram 4.0",
"model": {
"name": "Ideogram 4 GGUF",
"type": "Checkpoint",
"nsfw": false,
"poi": false
},
"files": [
{
"id": 2890211,
"name": "ideogram4GGUF_q4.gguf",
"type": "Model",
"metadata": {
"format": "GGUF",
"size": "full"
},
"downloadUrl": "https://civitai.com/api/download/models/3011173?fileId=2890211",
"hashes": {
"SHA256": "08431D93824E56513CCCC84A4D10D327DF992CFF5B5A0DCFDC739F4D50C2CC9E"
}
}
]
},
{
"source": "GET /model-versions/2500306",
"id": 2500306,
"modelId": 2053259,
"name": "NSFW FAST MOVE V2 Q4KM H",
"baseModel": "Wan Video 2.2 I2V-A14B",
"model": {
"name": "WAN 2.2 Enhanced NSFW | SVI | camera prompt adherence (Lightning Edition) I2V and T2V fp8 GGUF",
"type": "Checkpoint",
"nsfw": true,
"poi": false
},
"files": [
{
"id": 2388352,
"name": "wan22EnhancedNSFWSVICamera_nsfwFASTMOVEV2Q4KMH.gguf",
"type": "Model",
"metadata": {
"format": "GGUF",
"size": "full",
"fp": "fp8"
},
"downloadUrl": "https://civitai.com/api/download/models/2500306?fileId=2388352",
"hashes": {
"SHA256": "90EBBD34A858C0B6EACDFE259A83F5EB93D623FCD66BBFE5F36076AAE70C2826"
}
}
]
},
{
"source": "GET /model-versions/2192278",
"id": 2192278,
"modelId": 1933000,
"name": "HIGH Q5_0",
"baseModel": "Wan Video 2.2 T2V-A14B",
"model": {
"name": "Wan 2.2 T2V A14B GGUF",
"type": "Checkpoint",
"nsfw": false,
"poi": false
},
"files": [
{
"id": 2085422,
"name": "wan22T2VA14BGGUF_highQ50.gguf",
"type": "Model",
"metadata": {
"format": "GGUF",
"size": "full",
"fp": "fp16"
},
"downloadUrl": "https://civitai.com/api/download/models/2192278?fileId=2085422",
"hashes": {
"SHA256": "694902F318737151C6599BE284605E94B101843A42252803A5F46321E836C1ED"
}
}
]
},
{
"source": "GET /models?ids=2832886 (synthetic: quantType stripped from metadata, kept in url)",
"id": 93196857,
@@ -896,6 +982,54 @@
"dominant_dtype": "I8"
}
},
"3077947": {
"metadata": {
"general.architecture": "wan",
"general.quantization_version": 2,
"general.file_type": 7
},
"probe": {
"ok": true,
"dominant_dtype": "Q8_0",
"quant": {
"scheme": "gguf",
"format": "Q8_0"
}
},
"quant": "Q8_0"
},
"2890211": {
"metadata": {},
"probe": {
"ok": true,
"dominant_dtype": "Q4_0",
"quant": {
"scheme": "gguf",
"format": "Q4_0"
}
},
"quant": "Q4_0"
},
"2388352": {
"metadata": {
"general.architecture": "wan",
"general.quantization_version": 2,
"general.file_type": 15
},
"probe": {
"ok": true,
"dominant_dtype": "Q4_K",
"quant": {
"scheme": "gguf",
"format": "Q4_K"
}
},
"quant": "Q4_K_M"
},
"2085422": {
"metadata": null,
"error": "HTTP 401"
},
"3246945": {
"metadata": null,
"error": "HTTP 401"
@@ -919,14 +1053,14 @@
},
"expected": {
"precision_on": {
"3003586": {
"2949563": "ideogram4INT8_ideogram4UncondINT8-int8.safetensors"
"3357716": {
"3246005": "krea2Anime_v20_3246005-fp8_e4m3fn.safetensors",
"3246945": "krea2Anime_v20_3246945-bf16.safetensors",
"3245419": "krea2Anime_v20_3245419-int4.safetensors",
"3245462": "krea2Anime_v20_3245462-int8.safetensors"
},
"3005491": {
"2884783": "ideogram4_fp8Scaled-fp8_e4m3fn.safetensors"
},
"3005513": {
"2884823": "ideogram4_fp8Uncond-fp8_e4m3fn-uncond.safetensors"
"3140650": {
"3020814": "krea2_anime2real-pre-bf16.safetensors"
},
"3009287": {
"2888419": "wan22I2VA14BMoeDistill_i2vHigh-nf4.safetensors"
@@ -934,43 +1068,59 @@
"3009384": {
"2888503": "wan22I2VA14BMoeDistill_i2vLow-nf4.safetensors"
},
"3025163": {
"2903936": "log.txt",
"2903951": "1nnie3.safetensors"
"3005491": {
"2884783": "ideogram4_fp8Scaled-fp8_e4m3fn.safetensors"
},
"3140650": {
"3020814": "krea2_anime2real-pre-bf16.safetensors"
"3005513": {
"2884823": "ideogram4_fp8Uncond-fp8_e4m3fn-uncond.safetensors"
},
"3196857": {
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077945": "gingerlabsWan22T2V_v10_3077945-fp8.safetensors",
"3077947": "gingerlabsWan22T2V_v10_3077947.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948.gguf",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors"
"3003586": {
"2949563": "ideogram4INT8_ideogram4UncondINT8-int8.safetensors"
},
"3246186": {
"3128937": "ideogram4_3128937-fp32.safetensors",
"3128943": "ideogram4_txt_3128943-fp8.safetensors",
"3128946": "ideogram4_txt_3128946-fp8.safetensors",
"3128958": "ideogram4_3128958-fp8.safetensors",
"3128937": "ideogram4_3128937-fp32.safetensors",
"3128943": "ideogram4_txt_3128943-fp8.safetensors",
"3128969": "ideogram4_3128969-int8.safetensors",
"3128980": "ideogram4_3128980-nvfp4.safetensors",
"3129012": "ideogram4_3129012-fp8.safetensors",
"3129014": "ideogram4_3129014-nvfp4.safetensors",
"3129032": "ideogram4_3129032-int8.safetensors"
},
"3357716": {
"3245419": "krea2Anime_v20_3245419-int4.safetensors",
"3245462": "krea2Anime_v20_3245462-int8.safetensors",
"3246005": "krea2Anime_v20_3246005-fp8_e4m3fn.safetensors",
"3246945": "krea2Anime_v20_3246945-bf16.safetensors"
"3025163": {
"2903951": "1nnie3.safetensors",
"2903936": "log.txt"
},
"83196857": {
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077945": "gingerlabsWan22T2V_v10_3077945-fp8.safetensors",
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors",
"3077947": "gingerlabsWan22T2V_v10_3077947-Q8_0.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948-Q8_0.gguf",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors"
"3077948": "gingerlabsWan22T2V_v10_3077948-Q8_0.gguf"
},
"3196857": {
"3077945": "gingerlabsWan22T2V_v10_3077945-fp8.safetensors",
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors",
"3077947": "gingerlabsWan22T2V_v10_3077947-Q8_0.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948.gguf"
},
"3011173": {
"2890211": "ideogram4GGUF_q4-Q4_0.gguf"
},
"2500306": {
"2388352": "wan22EnhancedNSFWSVICamera_nsfwFASTMOVEV2Q4KMH-Q4_K_M.gguf"
},
"2192278": {
"2085422": "wan22T2VA14BGGUF_highQ50.gguf"
},
"93196857": {
"3077945": "gingerlabsWan22T2V_v10_3077945-fp8.safetensors",
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors",
"3077947": "gingerlabsWan22T2V_v10_3077947-Q8_0.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948-Q8_0.gguf"
},
"90000001": {
"1": "synthWan_v10_1-fp8_e5m2-low-noise.safetensors",
@@ -988,13 +1138,6 @@
"25": "collide.vae-25.safetensors",
"26": "collide.vae-26.safetensors",
"27": "noext-fp16"
},
"93196857": {
"3077905": "gingerlabsWan22T2V_v10_txt-bf16.safetensors",
"3077945": "gingerlabsWan22T2V_v10_3077945-fp8.safetensors",
"3077947": "gingerlabsWan22T2V_v10_3077947-Q8_0.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948-Q8_0.gguf",
"3077956": "gingerlabsWan22T2V_v10_3077956-fp8.safetensors"
}
},
"precision_off": {
@@ -1051,6 +1194,15 @@
"3077947": "gingerlabsWan22T2V_v10_3077947.gguf",
"3077948": "gingerlabsWan22T2V_v10_3077948.gguf"
},
"3011173": {
"2890211": "ideogram4GGUF_q4.gguf"
},
"2500306": {
"2388352": "wan22EnhancedNSFWSVICamera_nsfwFASTMOVEV2Q4KMH.gguf"
},
"2192278": {
"2085422": "wan22T2VA14BGGUF_highQ50.gguf"
},
"93196857": {
"3077945": "gingerlabsWan22T2V_v10_3077945.safetensors",
"3077905": "gingerlabsWan22T2V_v10_txt.safetensors",
+56 -1
View File
@@ -12,6 +12,7 @@ Covers:
- companion files routing to their own type folder
- a failed remote header read returning an error and never entering the probe cache
- a successful header read entering the probe cache
- a GGUF header read naming the quant from general.file_type, or from the dominant tensor type without it
- resolve_file reporting the version fetch reason and an unknown file id
No running server required.
@@ -78,6 +79,9 @@ def test_suffix_and_variant():
check('variant quant', names.file_variant(make_file(1, 'a.gguf', quant='Q8_0')), 'Q8_0')
check('variant url', names.file_variant(make_file(1, 'a.gguf', url='https://civitai.com/api/download/models/1?format=GGUF&quantType=Q4_K_M')), 'Q4_K_M')
check('variant none', names.file_variant(make_file(1, 'a.gguf', url='https://civitai.com/api/download/models/1?fileId=1')), None)
check('variant gguf ignores fp', names.file_variant(make_file(1, 'a.gguf', fp='fp16')), None)
check('variant gguf format ignores fp', names.file_variant(CivitFile.parse_obj({'id': 1, 'name': 'a.bin', 'metadata': {'format': 'GGUF', 'fp': 'fp16', 'quantType': 'Q5_0'}})), 'Q5_0')
check('variant safetensors quant fallback', names.file_variant(make_file(1, 'a.safetensors', quant='Q8_0')), 'Q8_0')
check('variant bad url', names.file_variant(make_file(1, 'a.gguf', url='')), None)
@@ -105,6 +109,12 @@ def test_peek_upgrade():
check('failed peek name', names.save_name(f, [f], context), 'm-fp8.safetensors')
context = names.apply_peeks(names.NameContext(), [f], {7: {'metadata': None, 'probe': {'ok': False, 'dominant_dtype': None}}})
check('upgrade bad probe', context.variants.get(7), None)
g = make_file(8, 'm.gguf', fp='fp8', quant='Q8_0')
context = names.apply_peeks(names.NameContext(), [g], {8: {'metadata': {}, 'probe': {'ok': True}, 'quant': 'Q4_K_M'}})
check('gguf header quant wins', names.save_name(g, [g], context), 'm-Q4_K_M.gguf')
context = names.apply_peeks(names.NameContext(), [g], {8: {'metadata': None, 'error': 'HTTP 401'}})
check('gguf failed peek uses metadata quant', names.save_name(g, [g], context), 'm-Q8_0.gguf')
check('peek targets', [x.id for x in names.peek_targets([f, g, make_file(9, 'c.zip')])], [7, 8])
def test_route_type():
@@ -170,6 +180,51 @@ def test_peek_cache():
sys.modules.pop('modules.paths', None)
def gguf_bytes(kv: dict, tensors: list) -> bytes:
def string(s):
b = s.encode('utf-8')
return struct.pack('<Q', len(b)) + b
out = b'GGUF' + struct.pack('<IQQ', 3, len(tensors), len(kv))
for key, value in kv.items():
out += string(key)
if isinstance(value, str):
out += struct.pack('<I', 8) + string(value)
else:
out += struct.pack('<II', 4, value)
for name, dims, ggml_type in tensors:
out += string(name) + struct.pack('<I', len(dims)) + b''.join(struct.pack('<Q', d) for d in dims) + struct.pack('<IQ', ggml_type, 0)
return out
def test_peek_gguf():
from modules.civitai import peek_civitai
url = 'https://civitai.com/api/download/models/2?fileId=12'
with tempfile.TemporaryDirectory() as tmpdir:
peek_civitai.peek_cache = {}
install_stubs(tmpdir, gguf_bytes({'general.architecture': 'wan', 'general.file_type': 15}, [('blocks.0.attn.q.weight', [5120, 5120], 12), ('blocks.0.norm.weight', [5120], 0)]))
result = peek_civitai.peek_file('m.gguf', url, 12)
check('gguf file_type quant', result.get('quant'), 'Q4_K_M')
check('gguf metadata', result.get('metadata'), {'general.architecture': 'wan', 'general.file_type': 15})
check('gguf probe format', result['probe']['quant']['format'], 'Q4_K')
check('gguf tensors', result.get('tensors'), 2)
check('gguf cached', peek_civitai.peek_cache_get(12, url), result)
peek_civitai.peek_cache = {}
install_stubs(tmpdir, gguf_bytes({}, [('w', [16, 16], 8), ('b', [16], 1)]))
check('gguf tensor quant', peek_civitai.peek_file('m.gguf', url, 12).get('quant'), 'Q8_0')
peek_civitai.peek_cache = {}
install_stubs(tmpdir, gguf_bytes({'general.file_type': 999}, [('w', [16, 16], 2)]))
check('gguf unknown file_type falls back', peek_civitai.peek_file('m.gguf', url, 12).get('quant'), 'Q4_0')
peek_civitai.peek_cache = {}
install_stubs(tmpdir, b'GGUF' + struct.pack('<IQQ', 3, 1, 1) + b'\x05')
result = peek_civitai.peek_file('m.gguf', url, 12)
check('gguf truncated error', 'read window' in (result.get('error') or ''), True)
check('gguf truncated not cached', peek_civitai.peek_cache, {})
install_stubs(tmpdir, b'\x08' + b'\0' * 40)
check('gguf bad magic', peek_civitai.peek_file('m.gguf', url, 12).get('error'), 'not a gguf file')
sys.modules.pop('modules.shared', None)
sys.modules.pop('modules.paths', None)
def test_resolve_file():
fixture = json.load(open(FIXTURE, encoding='utf-8'))
raw = next(v for v in fixture['versions'] if v['id'] == 3357716)
@@ -208,7 +263,7 @@ def test_resolve_file():
if __name__ == '__main__':
for test in [test_fixture_names, test_suffix_and_variant, test_roles, test_peek_upgrade, test_route_type, test_peek_cache, test_resolve_file]:
for test in [test_fixture_names, test_suffix_and_variant, test_roles, test_peek_upgrade, test_route_type, test_peek_cache, test_peek_gguf, test_resolve_file]:
test()
log.warning(f'Total: {passed} passed, {failed} failed')
sys.exit(1 if failed else 0)