Buckets:
| """Aggregate a Visual-Jigsaw visual-quality campaign without GPU work.""" | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| from pathlib import Path | |
| from .visual_quality import atomic_write_json, campaign_row, read_json | |
| HEADLINE = ( | |
| "native.bit_accuracy", | |
| "native.hamming_fraction", | |
| "native.token_agreement", | |
| "pixel_perceptual.psnr_db", | |
| "pixel_perceptual.ssim", | |
| "pixel_perceptual.ms_ssim", | |
| "pixel_perceptual.lpips", | |
| "pixel_perceptual.dists", | |
| "semantic.dinov2_cosine", | |
| "semantic.clip_cosine", | |
| ) | |
| def summarize(campaign_dir: Path) -> dict: | |
| manifest_path = campaign_dir / "campaign.json" | |
| manifest = read_json(manifest_path) if manifest_path.exists() else None | |
| expected = ( | |
| [Path(row["output_dir"]) for row in manifest["evals"]] | |
| if manifest is not None | |
| else sorted(path for path in campaign_dir.iterdir() if path.is_dir()) | |
| ) | |
| rows = [] | |
| incomplete = [] | |
| for output in expected: | |
| summary_path = output / "summary.json" | |
| complete = output / "COMPLETE" | |
| if not (summary_path.exists() and complete.exists()): | |
| incomplete.append(str(output)) | |
| continue | |
| row = campaign_row(read_json(summary_path)) | |
| row["output_dir"] = str(output) | |
| rows.append(row) | |
| rows.sort(key=lambda row: Path(row["eval_run"]).name) | |
| result = { | |
| "schema": "zebra_visual_quality_campaign_summary_v1", | |
| "campaign_dir": str(campaign_dir.resolve()), | |
| "expected_runs": len(expected), | |
| "completed_runs": len(rows), | |
| "incomplete_runs": incomplete, | |
| "rows": rows, | |
| } | |
| atomic_write_json(campaign_dir / "comparison.json", result) | |
| columns = [ | |
| "eval_run", | |
| "samples", | |
| "count_matched_samples", | |
| "generated_images", | |
| "paired_images", | |
| "exact_resolution_pairs", | |
| "geometry_comparable_pairs", | |
| "blank_fraction", | |
| "near_duplicate_fraction", | |
| *HEADLINE, | |
| "output_dir", | |
| ] | |
| with (campaign_dir / "comparison.csv").open( | |
| "w", encoding="utf-8", newline="" | |
| ) as handle: | |
| writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore") | |
| writer.writeheader() | |
| writer.writerows(rows) | |
| markdown = [ | |
| "# Visual Quality Campaign", | |
| "", | |
| f"Completed {len(rows)}/{len(expected)} eval runs.", | |
| "", | |
| "| eval | count match | bit acc. | SSIM | LPIPS | DISTS | DINOv2 | CLIP | blank | duplicate |", | |
| "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|", | |
| ] | |
| def value(row, key): | |
| item = row.get(key) | |
| return "NA" if item is None else f"{item:.5f}" | |
| for row in rows: | |
| markdown.append( | |
| f"| {Path(row['eval_run']).name} | {row['count_matched_samples']}/{row['samples']} | " | |
| f"{value(row, 'native.bit_accuracy')} | " | |
| f"{value(row, 'pixel_perceptual.ssim')} | " | |
| f"{value(row, 'pixel_perceptual.lpips')} | " | |
| f"{value(row, 'pixel_perceptual.dists')} | " | |
| f"{value(row, 'semantic.dinov2_cosine')} | " | |
| f"{value(row, 'semantic.clip_cosine')} | " | |
| f"{value(row, 'blank_fraction')} | " | |
| f"{value(row, 'near_duplicate_fraction')} |" | |
| ) | |
| (campaign_dir / "comparison.md").write_text( | |
| "\n".join(markdown) + "\n", encoding="utf-8" | |
| ) | |
| return result | |
| def main() -> None: | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument("--campaign-dir", type=Path, required=True) | |
| args = parser.parse_args() | |
| result = summarize(args.campaign_dir.resolve()) | |
| print(json.dumps({key: result[key] for key in ("expected_runs", "completed_runs")})) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 3.8 kB
- Xet hash:
- c81b39758afdfff608b18c48327b0cc8ecf1eeb70eb5089a789a15b891732183
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.