Buckets:
| import argparse | |
| import json | |
| import hashlib | |
| import io | |
| from pathlib import Path | |
| import numpy as np | |
| from PIL import Image, ImageDraw | |
| def load_sample(root: Path, sample: dict, expected_hashes: dict[str, str] | None = None) -> dict: | |
| def read(relative: str) -> bytes: | |
| payload = (root / relative).read_bytes() | |
| if expected_hashes is not None: | |
| hashes = {hashlib.sha256(payload).hexdigest()} | |
| # Git's Windows checkout may change JSON line endings, never image bytes. | |
| if Path(relative).suffix.lower() in {".json", ".jsonl"}: | |
| hashes.add(hashlib.sha256(payload.replace(b"\r\n", b"\n")).hexdigest()) | |
| if expected_hashes.get(relative) not in hashes: | |
| raise ValueError(f"file differs from approved validation: {relative}") | |
| return payload | |
| with Image.open(io.BytesIO(read(sample["image"]))) as image: | |
| pixels = np.array(image) | |
| valid = np.zeros(pixels.shape[:2], dtype=bool) | |
| valid[:sample["valid_height"], :sample["valid_width"]] = True | |
| valid &= ~np.all(pixels == 0, axis=2) | |
| result = {"image": pixels, "metadata": sample, "annotated_classes": sample["annotated_classes"]} | |
| if sample["task"] == "segmentation": | |
| with Image.open(io.BytesIO(read(sample["mask"]))) as image: | |
| mask = np.array(image) | |
| result.update(mask=mask, valid_mask=valid & (mask != 255)) | |
| return result | |
| labels = json.loads(read(sample["labels"])) | |
| ignored = Image.new("1", (pixels.shape[1], pixels.shape[0])) | |
| drawing = ImageDraw.Draw(ignored) | |
| for annotation in labels["ignored_annotations"]: | |
| vertices = annotation["clipped_polygon"] | |
| drawing.polygon(list(zip(vertices[::2], vertices[1::2])), fill=1) | |
| boxes = np.array([a["bbox"] for a in labels["annotations"]], dtype=np.float32).reshape(-1, 4) | |
| result.update( | |
| boxes=boxes, box_format="xywh", | |
| class_ids=np.array([a["category_id"] for a in labels["annotations"]], dtype=np.int64), | |
| annotations=labels["annotations"], ignored_annotations=labels["ignored_annotations"], | |
| valid_mask=valid & ~np.asarray(ignored, dtype=bool), | |
| ) | |
| return result | |
| class dataset: | |
| def __init__(self, root: str | Path, task: str, split: str, license_scope: str = "commercial_compatible"): | |
| self.root = Path(root) | |
| if task not in {"detection", "segmentation"} or split not in {"train", "val", "test"}: | |
| raise ValueError("invalid task or split") | |
| if license_scope not in {"all", "commercial_compatible", "academic_only"}: | |
| raise ValueError("invalid license scope") | |
| stats = json.loads((self.root / "stats.json").read_text()) | |
| if not stats.get("training_ready"): | |
| raise ValueError("corpus has not passed validation and visual review; consult stats.json") | |
| validation = json.loads((self.root / "validation.json").read_text()) | |
| sample_bytes = (self.root / "metadata" / "samples.jsonl").read_bytes() | |
| normalized_sample_bytes = sample_bytes.replace(b"\r\n", b"\n") | |
| sample_hashes = { | |
| hashlib.sha256(sample_bytes).hexdigest(), | |
| hashlib.sha256(normalized_sample_bytes).hexdigest(), | |
| } | |
| if not validation.get("automated_passed") or validation["content_sha256"] != stats["validation_content_sha256"] or validation["samples_sha256"] not in sample_hashes: | |
| raise ValueError("validation approval no longer matches the corpus") | |
| self.expected_hashes = validation["file_sha256"] | |
| rows = [json.loads(line) for line in sample_bytes.decode().splitlines()] | |
| self.samples = [row for row in rows if row["task"] == task and row["split"] == split and (license_scope == "all" or row["license_group"] == license_scope)] | |
| if not self.samples: | |
| raise ValueError("no samples match task, split and license scope; academic data requires explicit license_scope='all' or 'academic_only'") | |
| def __len__(self) -> int: | |
| return len(self.samples) | |
| def __getitem__(self, index: int) -> dict: | |
| return load_sample(self.root, self.samples[index], expected_hashes=self.expected_hashes) | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--root", default="datasets/corpus") | |
| parser.add_argument("--task", choices=["detection", "segmentation"], required=True) | |
| parser.add_argument("--split", choices=["train", "val", "test"], default="train") | |
| parser.add_argument("--license-scope", choices=["all", "commercial_compatible", "academic_only"], default="commercial_compatible") | |
| args = parser.parse_args() | |
| data = dataset(args.root, args.task, args.split, args.license_scope) | |
| sample = data[0] | |
| print(json.dumps({"samples": len(data), "image_shape": list(sample["image"].shape), "task": args.task, "valid_pixels": int(sample["valid_mask"].sum())})) | |
Xet Storage Details
- Size:
- 4.98 kB
- Xet hash:
- 6bd1ccfb009e4234ac53ca2791ebcb96a3943e75ace6019331d08f29a9171b79
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.