Buckets:
| import argparse | |
| import hashlib | |
| import json | |
| import random | |
| import time | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| from PIL import Image, ImageDraw, ImageFont | |
| from shapely.geometry import Polygon | |
| def check_groups(rows: list[dict]) -> None: | |
| splits = {} | |
| for row in rows: | |
| group = row["group"] | |
| if group in splits and splits[group] != row["split"]: | |
| raise ValueError(f"split leakage in group {group}") | |
| splits[group] = row["split"] | |
| def check_annotation(annotation: dict, width: int, height: int) -> None: | |
| bbox = np.array(annotation["bbox"], dtype=float) | |
| if bbox.shape != (4,) or not np.isfinite(bbox).all(): | |
| raise ValueError("invalid bounding box shape or coordinates") | |
| x, y, w, h = bbox | |
| if min(x, y) < -1e-6 or min(w, h) <= 0 or x + w > width + 1e-6 or y + h > height + 1e-6: | |
| raise ValueError("zero-area or out-of-bounds box") | |
| if annotation["category_id"] not in range(1, 17): | |
| raise ValueError("unknown canonical category") | |
| vertices = np.array(annotation["clipped_polygon"], dtype=float).reshape(-1, 2) | |
| polygon = Polygon(vertices) | |
| if not polygon.is_valid or polygon.area <= 0 or not np.isfinite(vertices).all(): | |
| raise ValueError("invalid clipped polygon") | |
| if np.any(vertices < -1e-6) or np.any(vertices[:, 0] > width + 1e-6) or np.any(vertices[:, 1] > height + 1e-6): | |
| raise ValueError("polygon outside image") | |
| palette = np.array([ | |
| [0, 0, 0], [220, 20, 60], [255, 255, 255], [255, 128, 0], | |
| [255, 255, 0], [255, 0, 255], [0, 255, 255], [0, 180, 255], | |
| [200, 120, 255], [150, 150, 150], [200, 200, 0], [160, 80, 30], | |
| [0, 80, 255], [40, 190, 50], [210, 200, 70], [160, 90, 70], | |
| [150, 150, 150], | |
| ], dtype=np.uint8) | |
| def previews(root: Path, samples: list[dict]) -> list[str]: | |
| rng = random.Random(20261022) | |
| font = ImageFont.load_default(size=15) | |
| by_class = defaultdict(list) | |
| for sample in samples: | |
| if sample["task"] == "detection": | |
| labels = json.loads((root / sample["labels"]).read_text()) | |
| for annotation in labels["annotations"]: | |
| by_class[annotation["canonical_class"]].append((sample, annotation)) | |
| paths = [] | |
| for name, candidates in sorted(by_class.items()): | |
| rng.shuffle(candidates) | |
| selected, seen = [], set() | |
| for candidate in candidates: | |
| if candidate[0]["source_image_id"] not in seen: | |
| selected.append(candidate) | |
| seen.add(candidate[0]["source_image_id"]) | |
| if len(selected) == 6: | |
| break | |
| sheet = Image.new("RGB", (1152, 832), (25, 25, 25)) | |
| for index, (sample, annotation) in enumerate(selected): | |
| with Image.open(root / sample["image"]) as image: | |
| x, y, w, h = annotation["bbox"] | |
| side = min(1024, max(384, int(max(w, h) * 1.5))) | |
| left = max(0, min(1024 - side, int(x + w / 2 - side / 2))) | |
| top = max(0, min(1024 - side, int(y + h / 2 - side / 2))) | |
| crop = image.crop((left, top, left + side, top + side)).resize((384, 384)) | |
| drawing = ImageDraw.Draw(crop) | |
| vertices = annotation["clipped_polygon"] | |
| points = [((px - left) * 384 / side, (py - top) * 384 / side) for px, py in zip(vertices[::2], vertices[1::2])] | |
| drawing.line(points + points[:1], fill=(255, 255, 0), width=2) | |
| col, row = index % 3, index // 3 | |
| sheet.paste(crop, (col * 384, row * 416)) | |
| ImageDraw.Draw(sheet).text((col * 384 + 4, row * 416 + 386), f"{name}: {sample['source_image_id']} {sample['split']}", fill="white", font=font) | |
| path = f"previews/detection/{name}.jpg" | |
| (root / path).parent.mkdir(parents=True, exist_ok=True) | |
| sheet.save(root / path, quality=92) | |
| paths.append(path) | |
| for split in ("train", "val", "test"): | |
| candidates = [s for s in samples if s["task"] == "segmentation" and s["split"] == split] | |
| rng.shuffle(candidates) | |
| sheet = Image.new("RGB", (1536, 1248), (25, 25, 25)) | |
| for index, sample in enumerate(candidates[:6]): | |
| with Image.open(root / sample["image"]) as image: | |
| pixels = np.array(image) | |
| with Image.open(root / sample["mask"]) as image: | |
| mask = np.array(image) | |
| colored = palette[np.where(mask == 255, 0, mask)] | |
| overlay = (pixels.astype(float) * 0.55 + colored * 0.45).astype(np.uint8) | |
| overlay[mask == 255] = pixels[mask == 255] | |
| plain = Image.fromarray(pixels).resize((384, 384)) | |
| tinted = Image.fromarray(overlay).resize((384, 384)) | |
| col, row = index % 2, index // 2 | |
| sheet.paste(plain, (col * 768, row * 416)) | |
| sheet.paste(tinted, (col * 768 + 384, row * 416)) | |
| ImageDraw.Draw(sheet).text((col * 768 + 4, row * 416 + 386), sample["source_image_id"], fill="white", font=font) | |
| path = f"previews/segmentation/{split}.jpg" | |
| (root / path).parent.mkdir(parents=True, exist_ok=True) | |
| sheet.save(root / path, quality=90) | |
| paths.append(path) | |
| return paths | |
| def audit(root: Path, seconds: float = 280) -> dict: | |
| invalidate(root) | |
| start = time.monotonic() | |
| samples_path = root / "metadata" / "samples.jsonl" | |
| samples = [json.loads(line) for line in samples_path.read_text().splitlines()] | |
| scenes = [json.loads(line) for line in (root / "metadata" / "scenes.jsonl").read_text().splitlines()] | |
| check_groups(samples) | |
| errors, near_duplicates, black_tiles = [], [], [] | |
| digests, fingerprints = {}, [] | |
| manifest_hash = hashlib.sha256(samples_path.read_bytes()) | |
| file_hashes = {} | |
| class_counts, pixel_counts = Counter(), Counter() | |
| split_classes = defaultdict(Counter) | |
| for index, sample in enumerate(samples): | |
| if time.monotonic() - start > seconds: | |
| raise TimeoutError(f"sanity-check limit reached after {index} samples; run a smaller shard") | |
| try: | |
| payload = (root / sample["image"]).read_bytes() | |
| manifest_hash.update(hashlib.sha256(payload).digest()) | |
| file_hashes[sample["image"]] = hashlib.sha256(payload).hexdigest() | |
| with Image.open(root / sample["image"]) as image: | |
| image.load() | |
| if image.mode != "RGB" or image.size != (1024, 1024): | |
| raise ValueError("image must be a 1024x1024 RGB PNG") | |
| pixels = np.array(image) | |
| gray = np.array(image.convert("L").resize((17, 16)), dtype=np.int16) | |
| valid_pixels = pixels[:sample["valid_height"], :sample["valid_width"]] | |
| if np.mean(np.all(valid_pixels == 0, axis=2)) > 0.98: | |
| black_tiles.append(sample["id"]) | |
| digest = hashlib.sha256(pixels.tobytes()).hexdigest() | |
| if digest in digests: | |
| previous = digests[digest] | |
| errors.append({"sample": sample["id"], "reason": "duplicate_tile_pixels", "other": previous["id"], "cross_split": previous["split"] != sample["split"]}) | |
| digests[digest] = sample | |
| signature = int.from_bytes(np.packbits(gray[:, 1:] > gray[:, :-1]).tobytes()) | |
| if gray.std() > 12: | |
| for old_signature, old_gray, old_sample in fingerprints: | |
| if old_sample["split"] != sample["split"] and old_sample["group"] != sample["group"] and (old_signature ^ signature).bit_count() <= 4 and np.abs(old_gray - gray).mean() < 4: | |
| near_duplicates.append([old_sample["id"], sample["id"]]) | |
| fingerprints.append((signature, gray, sample)) | |
| if sample["task"] == "segmentation": | |
| data = (root / sample["mask"]).read_bytes() | |
| manifest_hash.update(hashlib.sha256(data).digest()) | |
| file_hashes[sample["mask"]] = hashlib.sha256(data).hexdigest() | |
| with Image.open(root / sample["mask"]) as image: | |
| mask = np.array(image) | |
| if mask.shape != (1024, 1024) or not set(np.unique(mask)) <= set(range(1, 17)) | {255}: | |
| raise ValueError("invalid mask shape or canonical class") | |
| if np.any(mask[sample["valid_height"]:, :] != 255) or np.any(mask[:, sample["valid_width"]:] != 255): | |
| raise ValueError("padding is not ignored") | |
| ids, counts = np.unique(mask, return_counts=True) | |
| pixel_counts.update({int(k): int(v) for k, v in zip(ids, counts)}) | |
| split_classes["segmentation/" + sample["split"]].update({int(k): int(v) for k, v in zip(ids, counts)}) | |
| else: | |
| data = (root / sample["labels"]).read_bytes() | |
| manifest_hash.update(hashlib.sha256(data).digest()) | |
| file_hashes[sample["labels"]] = hashlib.sha256(data).hexdigest() | |
| labels = json.loads(data) | |
| seen = set() | |
| for annotation in labels["annotations"] + labels["ignored_annotations"]: | |
| if annotation["category_id"] is not None: | |
| check_annotation(annotation, sample["valid_width"], sample["valid_height"]) | |
| else: | |
| if not annotation.get("ignore") or not annotation.get("original_class"): | |
| raise ValueError("unmapped source objects must be preserved as ignored annotations") | |
| annotation = dict(annotation, category_id=1) | |
| check_annotation(annotation, sample["valid_width"], sample["valid_height"]) | |
| key = (annotation["source_annotation_id"], tuple(annotation["bbox"])) | |
| if key in seen: | |
| raise ValueError("duplicate annotation") | |
| seen.add(key) | |
| class_counts.update(a["category_id"] for a in labels["annotations"]) | |
| split_classes["detection/" + sample["split"]].update(a["category_id"] for a in labels["annotations"]) | |
| except (OSError, ValueError, KeyError) as error: | |
| errors.append({"sample": sample.get("id"), "reason": str(error)}) | |
| for task, expected in (("detection", {3, 4, 5, 6, 7, 8}), ("segmentation", {1, 2, 12, 13, 14, 15, 16})): | |
| for split in ("train", "val", "test"): | |
| missing = expected - set(split_classes[task + "/" + split]) | |
| if missing: | |
| errors.append({"split": task + "/" + split, "reason": "selected_source_classes_missing", "class_ids": sorted(missing)}) | |
| if black_tiles: | |
| errors.append({"reason": "mostly_black_tiles", "samples": black_tiles}) | |
| if near_duplicates: | |
| errors.append({"reason": "cross_split_near_duplicates_need_review", "pairs": near_duplicates}) | |
| report = { | |
| "automated_passed": not errors, | |
| "samples_checked": len(samples), | |
| "scene_groups": len({s["group"] for s in scenes}), | |
| "errors": errors, | |
| "class_counts": dict(class_counts), | |
| "pixel_counts": dict(pixel_counts), | |
| "split_class_counts": {key: dict(value) for key, value in split_classes.items()}, | |
| "content_sha256": manifest_hash.hexdigest(), | |
| "samples_sha256": hashlib.sha256(samples_path.read_bytes()).hexdigest(), | |
| "file_sha256": file_hashes, | |
| "seconds": time.monotonic() - start, | |
| "scope": "Every derived image and label; exact pixels, near-duplicate candidates, scene groups and coordinates. Unknown DOTA absolute geography is not certified.", | |
| } | |
| (root / "validation.json").write_text(json.dumps(report, indent=2), encoding="utf-8") | |
| return report | |
| def invalidate(root: Path) -> None: | |
| path = root / "stats.json" | |
| if path.exists(): | |
| stats = json.loads(path.read_text()) | |
| stats.update(training_ready=False, status="validation_or_review_pending") | |
| path.write_text(json.dumps(stats, indent=2), encoding="utf-8") | |
| def mark_ready(root: Path) -> None: | |
| invalidate(root) | |
| review_path = root / "visual_review.json" | |
| if not review_path.exists(): | |
| raise ValueError("visual review is missing") | |
| review = json.loads(review_path.read_text()) | |
| if review.get("passed") is not True or not review.get("inspected_files"): | |
| raise ValueError("visual review has not passed") | |
| report = audit(root) | |
| if not report["automated_passed"] or review.get("content_sha256") != report["content_sha256"]: | |
| raise ValueError("automated validation failed or reviewed content has changed") | |
| required = {str(path.relative_to(root)).replace("\\", "/") for path in (root / "previews").rglob("*.jpg")} | |
| if not required or not required <= set(review["inspected_files"]): | |
| raise ValueError("not all generated preview sheets have been inspected") | |
| stats_path = root / "stats.json" | |
| stats = json.loads(stats_path.read_text()) | |
| stats.update(training_ready=True, status="ready_for_source_aware_training_with_documented_coverage_and_geography_limitations", validation_content_sha256=report["content_sha256"], visual_review="visual_review.json") | |
| stats_path.write_text(json.dumps(stats, indent=2), encoding="utf-8") | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--root", type=Path, default=Path("datasets/corpus")) | |
| parser.add_argument("--previews-only", action="store_true") | |
| parser.add_argument("--finalize", action="store_true") | |
| args = parser.parse_args() | |
| if args.finalize: | |
| mark_ready(args.root) | |
| print("corpus passed automated and recorded visual checks") | |
| elif args.previews_only: | |
| samples = [json.loads(line) for line in (args.root / "metadata" / "samples.jsonl").read_text().splitlines()] | |
| print(json.dumps({"preview_files": previews(args.root, samples)})) | |
| else: | |
| report = audit(args.root) | |
| print(json.dumps({key: value for key, value in report.items() if key != "file_sha256"}, indent=2)) | |
| if not report["automated_passed"]: | |
| raise SystemExit(1) | |
Xet Storage Details
- Size:
- 14.3 kB
- Xet hash:
- 1babf172a3c991a283a02cbc764cf979b93c15fc04a2924aa9da53bf3df7516a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.