`visual-search/train/robustness_test.py`

74 lines
#!/usr/bin/env python3
# Real-world-ish validation of the fine-tuned visual-ID: does a DEGRADED "phone-like" photo of a
# catalog item still retrieve the CORRECT product from the live FT index — and do the confidence
# thresholds (>=0.92 high, >=0.85 medium) mean anything on that query distribution?
# We can't shoot real samples autonomously, so we simulate handheld capture: downscale, JPEG-recompress,
# blur, brightness/glare, small rotate+crop — then query the LIVE /api/identify-multi endpoint.
import os, io, sys, json, base64, subprocess, urllib.request, urllib.parse, ssl
from PIL import Image, ImageFilter, ImageEnhance

APP = "https://photo.designerwallcoverings.com/api/identify-multi"
AUTH = base64.b64encode(b"admin:DW2024!").decode()
N = 40
ctx = ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE

# pull N random rows that ARE embedded on KAMATERA (the live FT index) — so the correct answer is in it
remote_sql = ("select e.dw_sku, e.image_url from image_embeddings e where e.embedding is not null "
              "and left(e.image_url,4)='http' order by random() limit " + str(N))
cmd = ('DB="$(grep ^DW_UNIFIED_DB= /root/public-projects/dwphoto/.env|cut -d= -f2-)"; '
       'psql "$DB" -F "|" -tA -c "' + remote_sql + '"')
out = subprocess.check_output(['ssh','root@45.61.58.125', cmd]).decode()
rows = [l.split('|',1) for l in out.splitlines() if '|' in l]
print(f"pulled {len(rows)} live-indexed rows", flush=True)

def small(u):
    if 'shopify' in u:
        pr=urllib.parse.urlparse(u);q=urllib.parse.parse_qs(pr.query);q['width']=['512']
        return urllib.parse.urlunparse(pr._replace(query=urllib.parse.urlencode(q,doseq=True)))
    return u

def degrade(im):
    # mimic a handheld iPhone shot of a physical sample
    w,h = im.size
    im = im.rotate(4, expand=False, fillcolor=(240,240,240))        # slight tilt
    im = im.crop((int(w*0.06), int(h*0.06), int(w*0.94), int(h*0.94)))  # off-center crop
    im = im.resize((max(64,im.size[0]//2), max(64,im.size[1]//2)))   # lower res (distance)
    im = im.filter(ImageFilter.GaussianBlur(1.2))                    # hand shake / focus
    im = ImageEnhance.Brightness(im).enhance(1.18)                   # glare/overexposure
    im = ImageEnhance.Contrast(im).enhance(0.92)
    b=io.BytesIO(); im.save(b,'JPEG',quality=55); b.seek(0)          # phone JPEG compression
    return Image.open(b).convert('RGB')

def query(im):
    bio=io.BytesIO(); im.save(bio,'JPEG',quality=80)
    b64='data:image/jpeg;base64,'+base64.b64encode(bio.getvalue()).decode()
    req=urllib.request.Request(APP, data=json.dumps({'front':b64}).encode(),
        headers={'Content-Type':'application/json','Authorization':'Basic '+AUTH,'User-Agent':'rt/1'})
    return json.load(urllib.request.urlopen(req, timeout=30, context=ctx))

top1=0; top5=0; conf={'high':0,'medium':0,'low':0}; scored=[]; n=0
for dw_sku,url in rows:
    try:
        data=urllib.request.urlopen(urllib.request.Request(small(url),headers={'User-Agent':'rt/1'}),timeout=20).read()
        q=degrade(Image.open(io.BytesIO(data)).convert('RGB'))
        r=query(q); vis=r.get('visual') or []
        if not vis: continue
        n+=1
        skus=[str(v.get('dw_sku')) for v in vis]
        if skus and skus[0]==dw_sku: top1+=1
        if dw_sku in skus[:5]: top5+=1
        conf[r.get('confidence','low')] = conf.get(r.get('confidence','low'),0)+1
        scored.append((dw_sku, round(vis[0].get('score',0),3), skus[0]==dw_sku, r.get('confidence')))
    except Exception as e:
        print('skip',e, flush=True)

print(f"\n=== DEGRADED (phone-like) image→image retrieval on the LIVE FT index, n={n} ===")
print(f"top-1 correct: {top1}/{n} ({100*top1/max(1,n):.0f}%)   top-5: {top5}/{n} ({100*top5/max(1,n):.0f}%)")
print(f"confidence buckets: {conf}")
correct_scores=[s for _,s,c,_ in scored if c]; wrong_scores=[s for _,s,c,_ in scored if not c]
if correct_scores: print(f"top score when CORRECT: min={min(correct_scores):.3f} avg={sum(correct_scores)/len(correct_scores):.3f}")
if wrong_scores:   print(f"top score when WRONG:   max={max(wrong_scores):.3f} avg={sum(wrong_scores)/len(wrong_scores):.3f}")
# threshold sanity: of matches labeled high (>=0.92), how many were actually correct?
high=[c for _,s,c,cf in scored if cf=='high'];
if high: print(f"of 'high'-confidence results, {sum(high)}/{len(high)} were the CORRECT product ({100*sum(high)/len(high):.0f}%)")