CLaMP3-SAAS-GGUF / validation /validate-release.py
zhenghaoz's picture
Replace all CLaMP3 formats with mandatory bundled-MERT GGUFs
960c419
Raw History Blame Contribute Delete
4.6 kB
import collections, hashlib, json, math, os, struct, subprocess, sys
from pathlib import Path
import numpy as np
R=Path('/home/openclaw/.hermes/cache/scratch/clamp3-bundled-release')
O=R/'validation'; O.mkdir(exist_ok=True)
B=Path('/home/openclaw/.hermes/cache/scratch/clamp3-embedded-production')
W=Path('/home/openclaw/.hermes/cache/scratch/clamp3-embedded-published-oracle')
def sha(p):
h=hashlib.sha256()
with open(p,'rb') as f:
for b in iter(lambda:f.read(8*1024*1024),b''):h.update(b)
return h.hexdigest()
def inventory(p):
with open(p,'rb') as f:
def u(t):return struct.unpack('<'+t,f.read(struct.calcsize('<'+t)))
def s():return f.read(u('Q')[0]).decode()
def v(t):
if t==8:return s()
if t==9:
t,n=u('IQ');return [v(t) for _ in range(n)]
return u({0:'B',1:'b',2:'H',3:'h',4:'I',5:'i',6:'f',7:'?',10:'Q',11:'q',12:'d'}[t])[0]
assert f.read(4)==b'GGUF';version,count,keys=u('IQQ');assert version==3
md={}
for _ in range(keys):
key=s();md[key]=v(u('I')[0])
entries=[]
for _ in range(count):
name=s();shape=u('Q'*u('I')[0]);typ,offset=u('IQ');n=math.prod(shape)
nb={0:n*4,1:n*2,8:n//32*34,12:n//256*144,14:n//256*210}[typ]
entries.append(dict(name=name,shape=shape,type=typ,offset=offset,bytes=nb))
align=md.get('general.alignment',32);base=(f.tell()+align-1)//align*align
for e in entries:
f.seek(base+e['offset']);h=hashlib.sha256();left=e['bytes']
while left:
b=f.read(min(left,8*1024*1024));assert b;h.update(b);left-=len(b)
e['sha256']=h.hexdigest()
return md,entries
def metrics(a,b):
a=np.array(a,dtype=np.float64);b=np.array(b,dtype=np.float64)
assert a.shape==b.shape==(768,) and np.isfinite(a).all() and np.isfinite(b).all()
d=a-b
return dict(max_abs=float(abs(d).max()),rmse=float(np.sqrt(np.mean(d*d))),relative_l2=float(np.linalg.norm(d)/np.linalg.norm(b)),cosine=float(a@b/(np.linalg.norm(a)*np.linalg.norm(b))),norm=float(np.linalg.norm(a)))
formats=['f32','f16','q8_0','q4_k_m'];invs={};manifest={}
for fmt in formats:
p=R/f'clamp3-saas-{fmt}.gguf';md,es=inventory(p);assert len(es)==606
assert md['clamp3.audio_frontend']=='mert-v1'
assert md['clamp3.mert.source_revision']=='12af15fef9d0ac838c3f475bfbbf26d2060dd4f5'
mert=[e for e in es if e['name'].startswith('mert.')];assert len(mert)==209
invs[fmt]={e['name']:e for e in es}
manifest[p.name]=dict(size=p.stat().st_size,sha256=sha(p),tensors=len(es),mert_tensors=len(mert),tensor_types=dict(collections.Counter(str(e['type']) for e in es)))
(O/f'tensors-{fmt}.json').write_text(json.dumps(es,indent=2)+'\n')
print(fmt,manifest[p.name],flush=True)
for fmt in ['q8_0','q4_k_m']:
for name,e in invs['f32'].items():
if name.startswith('mert.'):
assert (e['type'],e['shape'],e['sha256'])==(invs[fmt][name]['type'],invs[fmt][name]['shape'],invs[fmt][name]['sha256'])
# Existing independent source-tensor audit attested the original CLaMP3 branch.
for fmt in ['f32','f16']:
_,old=inventory(Path('/mnt/openclaw/clamp3-gguf-work/publication')/f'clamp3-saas-{fmt}.gguf')
assert len(old)==397
for e in old:
new=invs[fmt][e['name']];assert (e['shape'],e['type'],e['sha256'])==(new['shape'],new['type'],new['sha256'])
report=dict(scope='CPU CLI smoke/conformance, synthetic waveforms through real pretrained MERT; not retrieval evaluation',mert_preserved_in_quantized=True,clamp3_397_source_audited_tensors_preserved=True,cases={})
refs=json.load(open('/mnt/openclaw/clamp3-gguf-work/publication/evidence/reference-f32.json'))
texts=refs['text_cases'][:2]
baseline={}
for fmt in formats:
cases={}
for label,args,reference in [(c['name'],['--text',c['utf8']],c['expected']) for c in texts]+[(c,['--audio',str(W/(c+'.wav'))],None) for c in ['broadband','two_chunks','resampled','resampled-input']]:
run=subprocess.run([str(B/'clap-model'),'--model',str(R/f'clamp3-saas-{fmt}.gguf'),'--threads','2',*args],capture_output=True,text=True)
assert run.returncode==0,(fmt,label,run.stderr)
a=json.loads(run.stdout);assert len(a)==768 and np.isfinite(a).all()
(O/f'{fmt}-{label}.json').write_text(json.dumps(a)+'\n')
if fmt=='f32':baseline[label]=a
m=metrics(a,baseline[label]);m['finite_768']=True
if reference is not None:
m['vs_independent_original_f32_reference']=metrics(a,reference)
if fmt=='f32':assert m['vs_independent_original_f32_reference']['max_abs']<5e-4
cases[label]=m;print(fmt,label,m,flush=True)
report['cases'][fmt]=cases
(R/'manifest.json').write_text(json.dumps(manifest,indent=2)+'\n')
(R/'validation.json').write_text(json.dumps(report,indent=2)+'\n')
print('All four bundled variants validated',flush=True)