#!/usr/bin/env python3 """Capture Obscura and Chromium concurrently with the same real settle delay. Every run uses a new output directory, checks process status and non-empty screenshots, records browser versions and timings, and reports raw full-canvas pixel diagnostics plus background-tolerant structural-edge diagnostics from check.py. It deliberately emits no aggregate parity verdict. An optional pre-change Obscura binary can be captured concurrently so regressions are compared against the same live-page moment. """ import argparse import concurrent.futures import json import os import re import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path import numpy as np from PIL import Image from playwright.sync_api import sync_playwright from check import pair_metrics def slug(url): value = re.sub(r"[^a-z0-9]+", "-", url.lower()).strip("-") return value[:80] or "page" def binary_version(binary): result = subprocess.run( [binary, "--version"], capture_output=True, text=True, timeout=10 ) text = (result.stdout or result.stderr).strip() return {"status": result.returncode, "text": text} def diagnostic_text(value): if value is None: return "" if isinstance(value, bytes): return value.decode(errors="replace") return value def capture_obscura(binary, url, screenshot, log, width, height, settle_ms): env = dict( os.environ, OBSCURA_SHOT_W=str(width), OBSCURA_SHOT_H=str(height), OBSCURA_ALLOW_PRIVATE_NETWORK="1", ) command = [ binary, "fetch", url, "--screenshot", str(screenshot), "--timeout", "50000", "--wait", f"{settle_ms / 1000:g}", ] started = time.time() try: result = subprocess.run( command, capture_output=True, text=True, timeout=75, env=env ) log.write_text(result.stdout + result.stderr) ok = result.returncode == 0 and screenshot.is_file() and screenshot.stat().st_size > 0 return { "ok": ok, "status": result.returncode, "elapsed_s": round(time.time() - started, 3), } except subprocess.TimeoutExpired as error: log.write_text( diagnostic_text(error.stdout) + diagnostic_text(error.stderr) ) return {"ok": False, "status": "timeout", "elapsed_s": round(time.time() - started, 3)} def load_rgb(path): return np.asarray(Image.open(path).convert("RGB")) def write_results(path, manifest): path.write_text(json.dumps(manifest, indent=2) + "\n") def main(): parser = argparse.ArgumentParser() parser.add_argument("urls", help="one URL per line; # comments allowed") parser.add_argument("--obscura-bin", required=True) parser.add_argument("--baseline-bin") parser.add_argument( "--chromium-bin", help="Chromium executable (default: Playwright's pinned Chromium)", ) parser.add_argument("--out", required=True, help="must not already exist") parser.add_argument("--width", type=int, default=1280) parser.add_argument("--height", type=int, default=1400) parser.add_argument("--settle-ms", type=int, default=3000) args = parser.parse_args() out = Path(args.out) out.mkdir(parents=True, exist_ok=False) urls = [ line.strip() for line in Path(args.urls).read_text().splitlines() if line.strip() and not line.lstrip().startswith("#") ] manifest = { "started_utc": datetime.now(timezone.utc).isoformat(), "viewport": {"width": args.width, "height": args.height, "dpr": 1}, "settle_ms_after_load": args.settle_ms, "obscura": binary_version(args.obscura_bin), "baseline": binary_version(args.baseline_bin) if args.baseline_bin else None, "pages": [], } results_path = out / "results.json" write_results(results_path, manifest) with sync_playwright() as playwright: chromium_executable = args.chromium_bin or playwright.chromium.executable_path manifest["chromium_executable"] = chromium_executable browser = playwright.chromium.launch( executable_path=chromium_executable, headless=True, args=[ "--no-sandbox", "--hide-scrollbars", "--disable-background-networking", "--force-device-scale-factor=1", ], ) manifest["chromium_version"] = browser.version with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: for index, url in enumerate(urls): name = f"{index:03d}-{slug(url)}" ours_path = out / f"{name}.obscura.png" chrome_path = out / f"{name}.chrome.png" baseline_path = out / f"{name}.baseline.png" page_result = {"url": url, "name": name} context = browser.new_context( viewport={"width": args.width, "height": args.height}, device_scale_factor=1, color_scheme="light", reduced_motion="no-preference", locale="en-US", timezone_id="UTC", ) page = context.new_page() chrome_messages = [] page.on("console", lambda message: chrome_messages.append(f"console {message.type}: {message.text}")) page.on("pageerror", lambda error: chrome_messages.append(f"pageerror: {error}")) ours_future = executor.submit( capture_obscura, args.obscura_bin, url, ours_path, out / f"{name}.obscura.log", args.width, args.height, args.settle_ms, ) baseline_future = None if args.baseline_bin: baseline_future = executor.submit( capture_obscura, args.baseline_bin, url, baseline_path, out / f"{name}.baseline.log", args.width, args.height, args.settle_ms, ) chrome_started = time.time() try: page.goto(url, wait_until="load", timeout=50000) page.wait_for_timeout(args.settle_ms) page.screenshot( path=str(chrome_path), full_page=False, timeout=50000 ) chrome_ok = chrome_path.is_file() and chrome_path.stat().st_size > 0 chrome_status = 0 except Exception as error: chrome_messages.append(f"capture error: {error}") chrome_ok = False chrome_status = "error" (out / f"{name}.chrome.log").write_text("\n".join(chrome_messages) + "\n") page_result["chromium"] = { "ok": chrome_ok, "status": chrome_status, "elapsed_s": round(time.time() - chrome_started, 3), "title": page.title() if chrome_ok else None, } context.close() page_result["obscura"] = ours_future.result() if baseline_future: page_result["baseline"] = baseline_future.result() if chrome_ok and page_result["obscura"]["ok"]: chrome_rgb = load_rgb(chrome_path) current_metrics = pair_metrics(load_rgb(ours_path), chrome_rgb) page_result["metrics"] = current_metrics if baseline_future and page_result["baseline"]["ok"]: baseline_metrics = pair_metrics(load_rgb(baseline_path), chrome_rgb) page_result["baseline_metrics"] = baseline_metrics for key in ( "rgb_mae", "pixels_gt_10", "pixels_gt_50", "edge_bbox_max_delta", "edge_row_projection_delta", "edge_column_projection_delta", ): if key in current_metrics and key in baseline_metrics: if current_metrics[key] is None or baseline_metrics[key] is None: continue page_result.setdefault("delta_vs_baseline", {})[key] = round( current_metrics[key] - baseline_metrics[key], 6 ) manifest["pages"].append(page_result) write_results(results_path, manifest) metric = page_result.get("metrics", {}).get("pixels_gt_50") edge_bbox = page_result.get("metrics", {}).get("edge_bbox_max_delta") edge_row = page_result.get("metrics", {}).get("edge_row_projection_delta") edge_col = page_result.get("metrics", {}).get("edge_column_projection_delta") edge_delta = page_result.get("delta_vs_baseline", {}).get( "edge_column_projection_delta" ) print( f"{name:84} " f"p>50={metric if metric is not None else 'capture-fail'} " f"edge_bbox={edge_bbox if edge_bbox is not None else '-'} " f"edge_row={edge_row if edge_row is not None else '-'} " f"edge_col={edge_col if edge_col is not None else '-'} " f"edge_col_delta={edge_delta if edge_delta is not None else '-'}", flush=True, ) browser.close() manifest["finished_utc"] = datetime.now(timezone.utc).isoformat() write_results(results_path, manifest) failed = [ page["name"] for page in manifest["pages"] if not page.get("chromium", {}).get("ok") or not page.get("obscura", {}).get("ok") ] if failed: print(f"capture failures: {', '.join(failed)}", file=sys.stderr) raise SystemExit(1) print(f"paired captures and raw diagnostics: {results_path}") if __name__ == "__main__": main()