File size: 5,632 Bytes
535fb25 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | #!/usr/bin/env python3
from __future__ import annotations
import argparse
import hashlib
import json
import os
import sys
from pathlib import Path
from typing import Any
import numpy as np
from PIL import Image, ImageDraw
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from flow_grpo.dataset_paths import DatasetPathResolver
from flow_grpo.server_profiles import apply_server_profile_defaults
apply_server_profile_defaults()
OUT_DIR = REPO_ROOT / "analysis_outputs" / "h20_eval_corruption"
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def load_rows(path: Path, n: int) -> list[dict[str, Any]]:
rows = []
with path.open("r", encoding="utf-8") as handle:
for line in handle:
if line.strip():
rows.append(json.loads(line))
if len(rows) >= n:
break
return rows
def image_stats(path: Path) -> dict[str, Any]:
image = Image.open(path)
arr = np.asarray(image.convert("L"), dtype=np.float32)
return {
"path": str(path),
"exists": True,
"mode": image.mode,
"size": list(image.size),
"min": float(arr.min()),
"max": float(arr.max()),
"mean": float(arr.mean()),
"std": float(arr.std()),
"sha256": sha256_file(path),
}
def thumb(path: Path, label: str, size: tuple[int, int] = (192, 192)) -> Image.Image:
image = Image.open(path).convert("RGB")
image.thumbnail((size[0], size[1] - 28), Image.Resampling.BILINEAR)
canvas = Image.new("RGB", size, "white")
canvas.paste(image, ((size[0] - image.width) // 2, 24 + (size[1] - 28 - image.height) // 2))
draw = ImageDraw.Draw(canvas)
draw.text((4, 4), label[:28], fill=(0, 0, 0))
return canvas
def make_contact_sheet(pairs: list[dict[str, Any]], out_path: Path) -> None:
cell_w, cell_h = 192, 192
sheet = Image.new("RGB", (cell_w * 2, cell_h * len(pairs)), "white")
for row, item in enumerate(pairs):
sheet.paste(thumb(Path(item["resolved_input"]), f"{row} input"), (0, row * cell_h))
sheet.paste(thumb(Path(item["resolved_gt"]), f"{row} gt"), (cell_w, row * cell_h))
sheet.save(out_path)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--jsonl", default=os.environ.get("TEST_JSONL"))
parser.add_argument("--num_samples", type=int, default=8)
parser.add_argument("--output_dir", default=str(OUT_DIR))
args = parser.parse_args()
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
resolver = DatasetPathResolver(os.environ.get("DATASET_ROOT"), os.environ.get("DATASET_PATH_REMAP_FROM"), os.environ.get("DATASET_PATH_REMAP_TO"))
rows = load_rows(Path(args.jsonl).expanduser().resolve(), args.num_samples)
audit = {
"jsonl": args.jsonl,
"dataset_root": str(resolver.dataset_root),
"remap_from": resolver.remap_from,
"remap_to": resolver.remap_to,
"samples": [],
}
contact_items = []
missing = []
for index, sample in enumerate(rows):
raw_input = (sample.get("input_images") or [None])[0]
raw_gt = sample.get("output_image") or sample.get("gt_image")
raw_mask = sample.get("output_mask") or sample.get("gt_mask") or sample.get("mask")
resolved_input = resolver.resolve(raw_input, label=f"sample {index} input")
resolved_gt = resolver.resolve(raw_gt, label=f"sample {index} gt")
resolved_mask = resolver.resolve(raw_mask, label=f"sample {index} mask") if raw_mask else None
item = {
"index": index,
"prompt": sample.get("prompt") or sample.get("instruction"),
"original_input": raw_input,
"original_gt": raw_gt,
"original_mask": raw_mask,
"resolved_input": str(resolved_input),
"resolved_gt": str(resolved_gt),
"resolved_mask": str(resolved_mask) if resolved_mask else None,
"input_stats": image_stats(resolved_input),
"gt_stats": image_stats(resolved_gt),
}
if resolved_mask and not resolved_mask.exists():
missing.append(str(resolved_mask))
audit["samples"].append(item)
contact_items.append(item)
make_contact_sheet(contact_items, out_dir / "eval_input_contact_sheet.png")
(out_dir / "eval_input_stats.json").write_text(json.dumps(audit, indent=2, sort_keys=True) + "\n", encoding="utf-8")
md_lines = ["# H20 Eval Input Audit", "", f"- jsonl: `{args.jsonl}`", f"- samples: `{len(rows)}`", f"- missing: `{len(missing)}`", ""]
for item in audit["samples"]:
md_lines.extend([
f"## sample {item['index']}",
f"- input: `{item['resolved_input']}`",
f"- gt: `{item['resolved_gt']}`",
f"- input mean/std: `{item['input_stats']['mean']:.3f}` / `{item['input_stats']['std']:.3f}`",
f"- gt mean/std: `{item['gt_stats']['mean']:.3f}` / `{item['gt_stats']['std']:.3f}`",
"",
])
(out_dir / "eval_input_audit.md").write_text("\n".join(md_lines), encoding="utf-8")
print(json.dumps(audit, indent=2, sort_keys=True))
if missing:
raise RuntimeError(f"Missing resolved mask paths: {missing[:5]}")
print(f"[eval-inputs] wrote outputs under {out_dir}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|