258 lines
10 KiB
Python
Executable File
258 lines
10 KiB
Python
Executable File
#!/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 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:
|
|
text = (error.stdout or "") + (error.stderr or "")
|
|
log.write_text(text if isinstance(text, str) else text.decode(errors="replace"))
|
|
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()
|