← back to Wallco Ai
scripts/build-tif.py
589 lines
#!/usr/bin/env python3
"""
build-tif.py — TIF + SVG print-master generator for wallco.ai designs.
Steve's rule: every published design needs a 150-DPI TIF on disk before publish.
Browser surfaces the thumb; the real file must exist. If the design is "crisp"
(≤8 dominant colors, hard edges), also emit an SVG.
Disk policy: Mac2 is at 95%+. Do NOT bulk-generate. Run one design at a time.
Usage:
python3 scripts/build-tif.py <design_id>
python3 scripts/build-tif.py --check <design_id> # report state, no work
python3 scripts/build-tif.py --force <design_id> # rebuild even if exists
For seamless tiles, the source PNG IS the master (one repeat). The TIF is
saved at the source's native resolution + 150 DPI metadata. tif_max_print_w_in
is set to 240 (20 ft) since the printer tiles infinitely.
For murals (non-tiling single images), the source is upscaled toward the
requested print size at build time — chained Real-ESRGAN x4 via Replicate
(same model as the digital-download route), then an exact LANCZOS resize, with
a pure-LANCZOS fallback when no Replicate token is configured. A 1024px mural
otherwise caps at ~7" of true 150-DPI print; the upscale stage lets us fulfill
large-format scenic-mural orders. Patterns are NOT upscaled — they tile
infinitely so the native repeat IS the 150-DPI master. (DTD verdict 2026-06-03,
Option A: wallco's flat screen-print art reconstructs cleanly under ESRGAN and
murals are viewed at 4-8 ft where 100-150 DPI is standard; premium/soft results
route to native panelized re-gen — the mural_panel path — as the Option-B
upgrade.) Disable per-build with --no-mural-upscale; size with --mural-print-ft.
"""
import argparse
import io
import os
import re
import sys
import time
import base64
import shutil
import subprocess
from pathlib import Path
from collections import Counter
from urllib.parse import urlparse
import psycopg2
from psycopg2.extras import RealDictCursor
def _pg_password():
"""Resolve Postgres password: PGPASSWORD env var takes precedence, then
parse it out of DATABASE_URL in .env. Never hard-code the secret."""
if os.environ.get("PGPASSWORD"):
return os.environ["PGPASSWORD"]
env_file = ROOT / ".env"
if env_file.exists():
for line in env_file.read_text().splitlines():
if line.startswith("DATABASE_URL="):
url = line.split("=", 1)[1].strip().strip('"')
parsed = urlparse(url)
if parsed.password:
return parsed.password
return ""
from PIL import Image, ImageStat
ROOT = Path(__file__).resolve().parents[1]
# TIFs are large (multi-MB) print archives. Mac2's boot volume runs at ~93%, so
# the canonical archive lives on the Henry external volume alongside the existing
# 10k-file archive. Fall back to the repo dir when Henry isn't mounted, and allow
# an explicit override via WALLCO_TIF_DIR. SVGs are small and keep their
# established home in the repo (data/svg) where the server resolver looks.
_HENRY_TIF = Path("/Volumes/Henry/wallco-ai-archive/tif")
TIF_DIR = Path(os.environ.get("WALLCO_TIF_DIR")
or (_HENRY_TIF if _HENRY_TIF.is_dir() else ROOT / "data" / "tif"))
SVG_DIR = Path(os.environ.get("WALLCO_SVG_DIR") or (ROOT / "data" / "svg"))
TIF_DIR.mkdir(parents=True, exist_ok=True)
SVG_DIR.mkdir(parents=True, exist_ok=True)
DPI = 150
MURAL_MAX_W_FT = 20
MURAL_MAX_H_FT = 11
MIN_FREE_GB = 5 # refuse to build if < 5 GB free
VECTOR_COLOR_THRESHOLD = 8 # ≤ this many dominant colors → crisp
VECTOR_COVERAGE_THRESHOLD = 0.95 # those colors must cover ≥ this fraction
VTRACER_BIN = shutil.which("vtracer") # optional real vectorizer
def db():
return psycopg2.connect(
host="127.0.0.1", port=5432,
user="dw_admin", password=_pg_password(),
dbname="dw_unified",
)
def free_gb(path=None):
# Guard the volume we actually write the (large) TIFs to — Henry when mounted.
s = shutil.disk_usage(path or TIF_DIR)
return s.free / (1024**3)
def fetch_design(conn, design_id):
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("""
SELECT id, kind, width_in, height_in, local_path, image_url,
tif_path, tif_built_at, svg_path, dig_number
FROM all_designs WHERE id = %s
""", (design_id,))
return cur.fetchone()
def resolve_source(d):
"""Return a real on-disk path for the design's source PNG, or None."""
lp = d.get("local_path")
if lp and Path(lp).exists():
return Path(lp)
# Some rows store relative paths or just filenames; try a few guesses.
if lp:
for candidate in (
ROOT / lp.lstrip("/"),
ROOT / "data" / "generated" / Path(lp).name,
):
if candidate.exists():
return candidate
return None
def assess_crispness(img):
"""
Quantize to 16 colors with PIL.quantize(method=MEDIANCUT), check the top-N
bucket coverage. Return (color_count, coverage_pct, eligible_bool).
"""
# Downsample for speed; crispness is invariant to scale at this granularity.
thumb = img.copy()
thumb.thumbnail((600, 600))
if thumb.mode != "RGB":
thumb = thumb.convert("RGB")
q = thumb.quantize(colors=16, method=Image.MEDIANCUT)
pal = q.getpalette()
counts = Counter(q.getdata())
total = sum(counts.values())
ranked = counts.most_common()
cumulative = 0
used = 0
for idx, n in ranked:
used += 1
cumulative += n
if cumulative / total >= VECTOR_COVERAGE_THRESHOLD:
break
coverage = cumulative / total
eligible = used <= VECTOR_COLOR_THRESHOLD and coverage >= VECTOR_COVERAGE_THRESHOLD
return used, coverage, eligible
def write_tif(img, out_path):
"""Write a TIF at 150 DPI with LZW compression."""
# Convert to RGB if needed (no alpha for print masters).
if img.mode in ("RGBA", "LA", "P"):
bg = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "P":
img = img.convert("RGBA")
bg.paste(img, mask=img.split()[-1] if img.mode == "RGBA" else None)
img = bg
elif img.mode != "RGB":
img = img.convert("RGB")
img.save(
out_path,
format="TIFF",
compression="tiff_lzw",
dpi=(DPI, DPI),
)
return out_path.stat().st_size
def write_svg(img, out_path, src_path, vector_attempted):
"""
Try real vectorization with vtracer if installed. Otherwise embed the
source PNG as a base64 raster inside the SVG so the file IS an SVG
(renders + scales in any viewer) — but mark svg_is_vector=False so the DB
knows the truth.
"""
if VTRACER_BIN and vector_attempted:
try:
subprocess.run(
[VTRACER_BIN, "--input", str(src_path), "--output", str(out_path),
"--colormode", "color", "--mode", "polygon",
"--filter_speckle", "4", "--color_precision", "6"],
check=True, capture_output=True, timeout=120,
)
if out_path.exists() and out_path.stat().st_size > 0:
return out_path.stat().st_size, True
except Exception as e:
print(f" vtracer failed: {e}", file=sys.stderr)
# Fallback: SVG with embedded raster. Still scalable in browsers, just not vector.
w, h = img.size
buf = io.BytesIO()
img.save(buf, format="PNG", optimize=True)
b64 = base64.b64encode(buf.getvalue()).decode("ascii")
svg = (
f'<?xml version="1.0" encoding="UTF-8"?>\n'
f'<svg xmlns="http://www.w3.org/2000/svg" width="{w}" height="{h}" '
f'viewBox="0 0 {w} {h}" preserveAspectRatio="xMidYMid meet">\n'
f' <image href="data:image/png;base64,{b64}" width="{w}" height="{h}"/>\n'
f'</svg>\n'
)
out_path.write_text(svg, encoding="utf-8")
return out_path.stat().st_size, False
# ── Real-ESRGAN mural upscaler ──────────────────────────────────────────────
# Patterns tile infinitely (native repeat = master). Murals are single,
# non-repeating images: at 1024px a mural caps at ~7" of true 150-DPI print, so
# to fulfill large-format orders we upscale the source toward the target print
# size. Mechanism mirrors the digital-download route (server.js
# /api/design/:id/download): Replicate nightmareai/real-esrgan x4, chained, then
# an exact LANCZOS resize. Pure-LANCZOS fallback (honest — no invented detail)
# when no Replicate token is set.
REPLICATE_ESRGAN_VERSION = (
"f121d640bd286e1fdc67f9799164c1d5be36ff74576ee11c803ae5b665dd46aa")
MURAL_KINDS = ("mural", "mural_panel") # non-tiling, single-image designs
TILE_KINDS = ("seamless_tile", "pattern_repeat", "best_seller_seed")
MURAL_DEFAULT_PRINT_FT = float(os.environ.get("WALLCO_MURAL_PRINT_FT", "12"))
MURAL_MAX_LONG_PX = 36000 # hard cap (20 ft @ 150 DPI)
ESRGAN_MAX_PASSES = int(os.environ.get("WALLCO_ESRGAN_PASSES", "2"))
def _replicate_token():
t = (os.environ.get("REPLICATE_API_TOKEN") or "").strip()
if t:
return t
envf = Path.home() / "Projects/animate-museum-posts/.env"
if envf.exists():
m = re.search(r"^REPLICATE_API_TOKEN=(\S+)", envf.read_text(), re.M)
if m:
return m.group(1)
return ""
def _esrgan_x4(img, token):
"""One Real-ESRGAN x4 pass via Replicate. Returns a PIL RGB image or raises."""
import urllib.request
import json as _json
buf = io.BytesIO()
img.save(buf, format="PNG")
b64 = "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()
req = urllib.request.Request(
"https://api.replicate.com/v1/predictions",
data=_json.dumps({
"version": REPLICATE_ESRGAN_VERSION,
"input": {"image": b64, "scale": 4, "face_enhance": False},
}).encode(),
headers={"Authorization": "Bearer " + token,
"Content-Type": "application/json"})
pred = _json.loads(urllib.request.urlopen(req, timeout=60).read())
pid = pred.get("id")
if not pid:
raise RuntimeError("esrgan start failed: " + str(pred)[:200])
t0 = time.time()
while time.time() - t0 < 600:
time.sleep(4)
poll = urllib.request.Request(
"https://api.replicate.com/v1/predictions/" + pid,
headers={"Authorization": "Bearer " + token})
pr = _json.loads(urllib.request.urlopen(poll, timeout=60).read())
st = pr.get("status")
if st == "succeeded":
out = pr.get("output")
url = out[0] if isinstance(out, list) else out
data = urllib.request.urlopen(url, timeout=180).read()
return Image.open(io.BytesIO(data)).convert("RGB")
if st in ("failed", "canceled"):
raise RuntimeError("esrgan " + st + ": " + str(pr.get("error") or "")[:200])
raise RuntimeError("esrgan timeout")
def upscale_mural(img, target_long_px):
"""
Upscale a mural source toward target_long_px on its long edge: chained
Real-ESRGAN x4 (≤ ESRGAN_MAX_PASSES) until ≥ target, then an exact LANCZOS
resize to the target. Pure-LANCZOS fallback when Replicate is unavailable.
Returns (upscaled_img, method_str).
"""
img = img.convert("RGB")
if max(img.size) >= target_long_px:
return img, "native (already ≥ target)"
token = _replicate_token()
method_parts = []
cur = img
if token:
passes = 0
while max(cur.size) < target_long_px and passes < ESRGAN_MAX_PASSES:
try:
cur = _esrgan_x4(cur, token)
passes += 1
method_parts.append("esrgan-x4")
print(f" esrgan pass {passes} → {cur.size[0]}×{cur.size[1]} px",
flush=True)
except Exception as e:
print(f" esrgan pass {passes+1} failed ({e}); "
f"finishing with LANCZOS", file=sys.stderr)
break
else:
print(" no REPLICATE_API_TOKEN → pure-LANCZOS upscale "
"(interpolated, no ESRGAN-reconstructed detail)", file=sys.stderr)
cw, ch = cur.size
scale = target_long_px / max(cw, ch)
if abs(scale - 1.0) > 1e-3:
cur = cur.resize((round(cw * scale), round(ch * scale)), Image.LANCZOS)
method_parts.append(f"lanczos→{max(cur.size)}px")
return cur, (" + ".join(method_parts) if method_parts else "lanczos")
def compute_max_print(kind, w_px, h_px):
"""Max printable size at 150 DPI."""
if kind in TILE_KINDS:
# Tiles infinitely — printer can fill any wall.
return MURAL_MAX_W_FT * 12, MURAL_MAX_H_FT * 12 # 240" × 132"
# Mural: bound by (post-upscale) source resolution / DPI.
return round(w_px / DPI, 2), round(h_px / DPI, 2)
def build_one(design_id, force=False, check_only=False, force_svg=None,
mural_upscale=None, mural_print_ft=None):
# force_svg: write an SVG even when the design isn't crisp/vector-eligible
# (raster-embed fallback). Steve's rule: every SKU gets an SVG file. Default
# pulls from WALLCO_FORCE_SVG so the generator hook can opt every new pattern
# in without changing the signature for unrelated callers.
if force_svg is None:
force_svg = os.environ.get("WALLCO_FORCE_SVG") == "1"
# mural_upscale: for mural-kind rows, upscale toward the print size at build
# time (default ON; disable per-build with --no-mural-upscale or
# WALLCO_MURAL_UPSCALE=0). mural_print_ft sizes the target long edge.
if mural_upscale is None:
mural_upscale = os.environ.get("WALLCO_MURAL_UPSCALE", "1") != "0"
if mural_print_ft is None:
mural_print_ft = MURAL_DEFAULT_PRINT_FT
if free_gb() < MIN_FREE_GB:
print(f"REFUSE: only {free_gb():.1f} GB free (need ≥ {MIN_FREE_GB}).",
file=sys.stderr)
return 2
conn = db()
d = fetch_design(conn, design_id)
if not d:
print(f"design {design_id} not found", file=sys.stderr)
return 3
src = resolve_source(d)
if not src:
print(f"design {design_id}: source PNG not found "
f"(local_path={d.get('local_path')!r})", file=sys.stderr)
return 4
tif_path = TIF_DIR / f"{design_id}.tif"
svg_path = SVG_DIR / f"{design_id}.svg"
state = {
"id": design_id,
"dig": d.get("dig_number"),
"kind": d.get("kind"),
"source": str(src),
"tif_exists": tif_path.exists(),
"svg_exists": svg_path.exists(),
"free_gb": round(free_gb(), 1),
}
if check_only:
print(state)
return 0
if tif_path.exists() and not force:
print(f"design {design_id}: TIF already at {tif_path} "
f"({tif_path.stat().st_size:,} bytes) — use --force to rebuild")
return 0
t0 = time.time()
img = Image.open(src)
w_px, h_px = img.size
kind = d.get("kind")
# Mural upscale stage — single non-repeating images only. Patterns tile, so
# their native repeat is already the 150-DPI master and are never upscaled.
if kind in MURAL_KINDS and mural_upscale:
target_long = min(MURAL_MAX_LONG_PX, round(mural_print_ft * 12 * DPI))
if max(w_px, h_px) < target_long:
# Rough headroom: final RGB buffer + an in-flight working copy.
need_gb = (target_long * target_long * 3 * 2) / (1024**3)
if free_gb() < max(MIN_FREE_GB, need_gb):
print(f"REFUSE mural-upscale: need ~{need_gb:.1f} GB headroom, "
f"only {free_gb():.1f} GB free on {TIF_DIR}", file=sys.stderr)
conn.close()
return 2
print(f" mural upscale → target {target_long}px long edge "
f"(~{mural_print_ft:.0f} ft @ {DPI} DPI)…")
img, mural_method = upscale_mural(img, target_long)
w_px, h_px = img.size
print(f" ✓ upscaled to {w_px}×{h_px} px via {mural_method}")
max_w_in, max_h_in = compute_max_print(kind, w_px, h_px)
print(f"design {design_id} ({kind}) → {w_px}×{h_px} px → "
f"max print {max_w_in}\"×{max_h_in}\" at {DPI} DPI")
tif_bytes = write_tif(img, tif_path)
print(f" ✓ TIF {tif_path} {tif_bytes:,} bytes")
color_count, coverage, eligible = assess_crispness(img)
print(f" crispness: {color_count} dominant colors, "
f"{coverage*100:.1f}% coverage → "
f"{'CRISP (vector-eligible)' if eligible else 'fuzzy (raster only)'}")
svg_bytes = None
svg_is_vector = False
if (eligible or force_svg) and kind not in MURAL_KINDS:
# Attempt real vectorization only when the design is crisp enough to be
# worth it; otherwise (force_svg on a fuzzy design) go straight to the
# raster-embed SVG so every SKU still gets a scalable .svg file.
# Murals are skipped: scenic art isn't vector-eligible and an SVG
# embedding the upscaled (up to 36k px) raster would be pathological.
svg_bytes, svg_is_vector = write_svg(img, svg_path, src, vector_attempted=eligible)
print(f" ✓ SVG {svg_path} {svg_bytes:,} bytes "
f"({'real vector' if svg_is_vector else 'embedded raster — install vtracer for real vectors'})")
with conn.cursor() as cur:
cur.execute("""
UPDATE all_designs SET
tif_path = %s,
tif_bytes = %s,
tif_w_px = %s,
tif_h_px = %s,
tif_dpi = %s,
tif_max_print_w_in = %s,
tif_max_print_h_in = %s,
tif_built_at = now(),
svg_path = %s,
svg_bytes = %s,
svg_is_vector = %s,
svg_built_at = CASE WHEN %s IS NULL THEN svg_built_at ELSE now() END,
vector_eligible = %s,
vector_color_count = %s
WHERE id = %s
""", (
str(tif_path), tif_bytes, w_px, h_px, DPI,
max_w_in, max_h_in,
str(svg_path) if svg_bytes else None,
svg_bytes, svg_is_vector,
svg_bytes,
eligible, color_count,
design_id,
))
conn.commit()
conn.close()
print(f" done in {time.time()-t0:.1f}s · {free_gb():.1f} GB free now")
return 0
def scan_and_maybe_build(design_id):
"""
Bulk-eligible-only worker: open the source, compute crispness, ALWAYS
record vector_color_count + vector_eligible. Only write TIF + SVG when
eligible (≤ VECTOR_COLOR_THRESHOLD dominant colors).
"""
if free_gb() < MIN_FREE_GB:
return (design_id, "halt", f"low disk ({free_gb():.1f} GB)")
conn = db()
try:
d = fetch_design(conn, design_id)
if not d:
return (design_id, "skip", "not found")
src = resolve_source(d)
if not src:
return (design_id, "skip", f"source PNG missing ({d.get('local_path')!r})")
try:
img = Image.open(src)
except Exception as e:
return (design_id, "err", f"open: {e}")
w_px, h_px = img.size
color_count, coverage, eligible = assess_crispness(img)
with conn.cursor() as cur:
if not eligible:
cur.execute(
"UPDATE all_designs SET vector_color_count=%s, vector_eligible=FALSE WHERE id=%s",
(color_count, design_id),
)
conn.commit()
return (design_id, "fuzzy", f"{color_count} colors")
# Eligible → write both TIF and SVG
tif_path = TIF_DIR / f"{design_id}.tif"
svg_path = SVG_DIR / f"{design_id}.svg"
tif_bytes = write_tif(img, tif_path)
svg_bytes, svg_is_vector = write_svg(img, svg_path, src, vector_attempted=True)
max_w_in, max_h_in = compute_max_print(d.get("kind"), w_px, h_px)
cur.execute("""
UPDATE all_designs SET
tif_path=%s, tif_bytes=%s, tif_w_px=%s, tif_h_px=%s, tif_dpi=%s,
tif_max_print_w_in=%s, tif_max_print_h_in=%s, tif_built_at=now(),
svg_path=%s, svg_bytes=%s, svg_is_vector=%s, svg_built_at=now(),
vector_eligible=TRUE, vector_color_count=%s
WHERE id=%s
""", (
str(tif_path), tif_bytes, w_px, h_px, DPI,
max_w_in, max_h_in,
str(svg_path), svg_bytes, svg_is_vector,
color_count, design_id,
))
conn.commit()
return (design_id, "built", f"{color_count}c · tif {tif_bytes//1024}KB · svg {svg_bytes//1024}KB")
finally:
conn.close()
def bulk_eligible_only(rebuild=False, concurrency=4):
"""Scan every published design; write TIF+SVG only for ≤8-color crisp ones."""
if free_gb() < 8:
print(f"REFUSE bulk: only {free_gb():.1f} GB free (need ≥ 8 to start).", file=sys.stderr)
return 2
conn = db()
with conn.cursor() as cur:
if rebuild:
cur.execute("SELECT id FROM all_designs WHERE is_published=TRUE AND local_path IS NOT NULL ORDER BY id")
else:
cur.execute("""SELECT id FROM all_designs
WHERE is_published=TRUE AND local_path IS NOT NULL
AND (vector_color_count IS NULL OR (vector_eligible=TRUE AND tif_path IS NULL))
ORDER BY id""")
ids = [r[0] for r in cur.fetchall()]
conn.close()
total = len(ids)
print(f"bulk-eligible-only: {total:,} designs to scan · {concurrency} workers · {free_gb():.1f} GB free")
if not total:
return 0
from multiprocessing import Pool
counts = {"built": 0, "fuzzy": 0, "skip": 0, "err": 0, "halt": 0}
t0 = time.time()
halted = False
with Pool(processes=concurrency) as pool_:
for i, (did, status, note) in enumerate(pool_.imap_unordered(scan_and_maybe_build, ids, chunksize=20), 1):
counts[status] = counts.get(status, 0) + 1
if status == "halt":
print(f"\nHALT @ #{i}: {note}", file=sys.stderr)
halted = True
pool_.terminate()
break
if status == "built":
print(f" ✓ {did} {note}")
if i % 200 == 0 or i == total:
rate = i / max(time.time()-t0, 0.001)
eta = (total - i) / max(rate, 0.001)
print(f" [{i:,}/{total:,}] built={counts['built']} fuzzy={counts['fuzzy']} skip={counts['skip']} err={counts.get('err',0)} "
f"· {rate:.1f}/s · ETA {eta/60:.1f}m · free {free_gb():.1f} GB", flush=True)
print(f"\nDONE in {(time.time()-t0)/60:.1f}m · "
f"built={counts['built']} fuzzy={counts['fuzzy']} skip={counts['skip']} err={counts.get('err',0)}"
+ (f" HALTED={counts['halt']}" if halted else ""))
return 0 if not halted else 2
def main():
p = argparse.ArgumentParser()
p.add_argument("design_id", type=int, nargs="?")
p.add_argument("--force", action="store_true")
p.add_argument("--check", action="store_true")
p.add_argument("--force-svg", dest="force_svg", action="store_true",
help="Always emit an SVG (raster-embed fallback) even for fuzzy/non-crisp designs")
p.add_argument("--no-mural-upscale", dest="mural_upscale", action="store_false",
default=None,
help="Skip the Real-ESRGAN mural upscale (mural-kind rows are upscaled by default)")
p.add_argument("--mural-print-ft", dest="mural_print_ft", type=float, default=None,
help=f"Target mural print width in feet (default {MURAL_DEFAULT_PRINT_FT:.0f}, "
f"capped at {MURAL_MAX_LONG_PX//DPI//12} ft / {MURAL_MAX_LONG_PX}px @ {DPI} DPI)")
p.add_argument("--bulk-eligible-only", action="store_true",
help="Scan all published designs; write TIF+SVG only for those ≤8 dominant colors")
p.add_argument("--rebuild", action="store_true",
help="With --bulk-eligible-only: re-scan even rows that already have vector_color_count")
p.add_argument("--concurrency", type=int, default=4)
args = p.parse_args()
if args.bulk_eligible_only:
sys.exit(bulk_eligible_only(rebuild=args.rebuild, concurrency=args.concurrency))
if args.design_id is None:
p.error("design_id is required (or pass --bulk-eligible-only)")
sys.exit(build_one(args.design_id, force=args.force, check_only=args.check,
force_svg=(args.force_svg or None),
mural_upscale=args.mural_upscale,
mural_print_ft=args.mural_print_ft))
if __name__ == "__main__":
main()