`scripts/cross_match_pd.py`

125 lines
#!/usr/bin/env python3
"""
For each Sister Parish pattern, find the 3 nearest public-domain pattern
images (from wallco.ai's pd_source_designs) by color distance in CIE Lab space.

Lab distance is perceptually meaningful where raw RGB or HSL is not —
a sage green and an olive green are close in Lab, far apart in RGB.

Output: dw_unified.sp_pd_color_matches join table.
Idempotent: TRUNCATE + re-insert each run.
"""
import json
import subprocess
from typing import List, Tuple

def psql_json(sql: str):
    out = subprocess.run(['psql', 'dw_unified', '-At', '-q', '-c', sql],
                         capture_output=True, text=True, check=True).stdout.strip()
    return json.loads(out) if out else []

def hex_to_rgb(h: str):
    if not h or not h.startswith('#'): return None
    h = h.lstrip('#')
    if len(h) != 6: return None
    return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)

# sRGB → linear → XYZ → Lab (D65). Standard formulas.
def srgb_to_linear(c):
    c /= 255.0
    return c / 12.92 if c <= 0.04045 else ((c + 0.055) / 1.055) ** 2.4

def rgb_to_lab(rgb):
    r, g, b = (srgb_to_linear(v) for v in rgb)
    # sRGB D65 matrix
    x = r * 0.4124564 + g * 0.3575761 + b * 0.1804375
    y = r * 0.2126729 + g * 0.7151522 + b * 0.0721750
    z = r * 0.0193339 + g * 0.1191920 + b * 0.9503041
    # D65 reference white
    xn, yn, zn = 0.95047, 1.00000, 1.08883
    def f(t):
        return t ** (1/3) if t > 0.008856 else (7.787 * t + 16/116)
    L = 116 * f(y / yn) - 16
    a = 500 * (f(x / xn) - f(y / yn))
    b2 = 200 * (f(y / yn) - f(z / zn))
    return (L, a, b2)

def lab_dist(a, b):
    return ((a[0]-b[0])**2 + (a[1]-b[1])**2 + (a[2]-b[2])**2) ** 0.5

# Pull data
sp = psql_json("SELECT COALESCE(json_agg(t),'[]'::json) FROM (SELECT sp_product_id, title, dominant_hex FROM sisterparish_catalog WHERE dominant_hex IS NOT NULL ORDER BY sp_product_id) t;")
pd = psql_json("SELECT COALESCE(json_agg(t),'[]'::json) FROM (SELECT id, source, title, dominant_hex, source_url, local_path FROM pd_source_designs WHERE dominant_hex IS NOT NULL) t;")

print(f"SP patterns with hex: {len(sp)}")
print(f"PD images  with hex: {len(pd)}")
if not sp or not pd:
    print("Need both corpora populated; exiting")
    raise SystemExit(0)

# Pre-compute PD Lab values
pd_lab = []
for p in pd:
    rgb = hex_to_rgb(p['dominant_hex'])
    if rgb is None: continue
    pd_lab.append((p, rgb_to_lab(rgb)))

# Match each SP
matches = []
for s in sp:
    rgb = hex_to_rgb(s['dominant_hex'])
    if rgb is None: continue
    sl = rgb_to_lab(rgb)
    scored = [(lab_dist(sl, p_lab), p) for (p, p_lab) in pd_lab]
    scored.sort(key=lambda x: x[0])
    for rank, (dist, pmatch) in enumerate(scored[:3], 1):
        matches.append((s['sp_product_id'], s['title'], s['dominant_hex'],
                        pmatch['id'], pmatch['source'], pmatch['title'],
                        pmatch['dominant_hex'], pmatch['source_url'], pmatch['local_path'],
                        rank, round(dist, 2)))

# Write to PG
ddl = """
CREATE TABLE IF NOT EXISTS sp_pd_color_matches (
  sp_product_id   BIGINT NOT NULL,
  sp_title        TEXT,
  sp_hex          TEXT,
  pd_id           BIGINT NOT NULL,
  pd_source       TEXT,
  pd_title        TEXT,
  pd_hex          TEXT,
  pd_source_url   TEXT,
  pd_local_path   TEXT,
  rank            INTEGER,
  lab_distance    NUMERIC(8,2),
  computed_at     TIMESTAMPTZ DEFAULT NOW(),
  PRIMARY KEY (sp_product_id, pd_id)
);
TRUNCATE sp_pd_color_matches;
"""
subprocess.run(['psql', 'dw_unified', '-q', '-c', ddl], check=True)

def esc(v):
    if v is None: return 'NULL'
    return "'" + str(v).replace("'", "''") + "'"

with open('/tmp/sp_pd_matches.sql', 'w') as f:
    f.write("BEGIN;\n")
    for m in matches:
        sp_id, sp_t, sp_h, pd_id, pd_src, pd_t, pd_h, pd_u, pd_lp, rank, dist = m
        f.write(f"INSERT INTO sp_pd_color_matches (sp_product_id,sp_title,sp_hex,pd_id,pd_source,pd_title,pd_hex,pd_source_url,pd_local_path,rank,lab_distance) "
                f"VALUES ({sp_id},{esc(sp_t)},{esc(sp_h)},{pd_id},{esc(pd_src)},{esc(pd_t)},{esc(pd_h)},{esc(pd_u)},{esc(pd_lp)},{rank},{dist});\n")
    f.write("COMMIT;\n")
subprocess.run(['psql', 'dw_unified', '-q', '-f', '/tmp/sp_pd_matches.sql'], check=True)
import os; os.unlink('/tmp/sp_pd_matches.sql')

print(f"\nWrote {len(matches)} matches ({len(sp)} SP × 3 nearest PD)")

# Spot-check 5
print("\nSpot-check (5 SP patterns + nearest PD):")
out = subprocess.run(['psql', 'dw_unified', '-c',
    "SELECT sp_title, sp_hex, pd_title, pd_hex, pd_source, lab_distance "
    "FROM sp_pd_color_matches WHERE rank=1 ORDER BY lab_distance LIMIT 5;"],
    capture_output=True, text=True)
print(out.stdout)