Buckets:
| import hashlib | |
| import math | |
| import json | |
| import io | |
| import argparse | |
| import tarfile | |
| import threading | |
| from collections import Counter | |
| from pathlib import Path, PurePosixPath | |
| import numpy as np | |
| import rasterio | |
| from PIL import Image | |
| from shapely.geometry import Polygon, box | |
| def starts(length: int) -> list[int]: | |
| if length <= 0: | |
| raise ValueError("image dimension must be positive") | |
| if length <= 1024: | |
| return [0] | |
| return sorted(set([*range(0, length - 1024 + 1, 819), length - 1024])) | |
| def clip_object(vertices: list[float], bounds: tuple[int, int, int, int]) -> dict | None: | |
| if len(vertices) < 6 or len(vertices) % 2 or not all(math.isfinite(v) for v in vertices): | |
| raise ValueError("invalid polygon coordinates") | |
| polygon = Polygon(np.asarray(vertices).reshape(-1, 2)) | |
| if not polygon.is_valid or polygon.area <= 0: | |
| raise ValueError("invalid or zero-area source polygon") | |
| clipped = polygon.intersection(box(*bounds)) | |
| if clipped.is_empty or clipped.area == 0: | |
| return None | |
| if clipped.geom_type != "Polygon": | |
| raise ValueError("clipping produced unsupported multipart geometry") | |
| x, y, right, bottom = clipped.bounds | |
| fraction = clipped.area / polygon.area | |
| local = [value for px, py in list(clipped.exterior.coords)[:-1] for value in (px - bounds[0], py - bounds[1])] | |
| return { | |
| "bbox": [x - bounds[0], y - bounds[1], right - x, bottom - y], | |
| "original_polygon": vertices, | |
| "clipped_polygon": local, | |
| "polygon_area": clipped.area, | |
| "visibility": fraction, | |
| "ignore": fraction < 0.7 or min(right - x, bottom - y) < 2, | |
| } | |
| def map_oem_mask(mask: np.ndarray) -> np.ndarray: | |
| if mask.ndim != 2 or not np.issubdtype(mask.dtype, np.integer): | |
| raise ValueError("expected integer single-channel mask") | |
| if mask.size == 0 or mask.min() < 0 or mask.max() > 8: | |
| raise ValueError("unknown OpenEarthMap mask class") | |
| return np.array([255, 15, 13, 16, 2, 13, 12, 14, 1], dtype=np.uint8)[mask] | |
| def split_groups(groups: list[str]) -> dict[str, str]: | |
| ordered = sorted(set(groups), key=lambda key: hashlib.sha256(("rgb-corpus-v1:" + key).encode()).hexdigest()) | |
| if len(ordered) < 3: | |
| raise ValueError("at least three independent groups required") | |
| held = max(1, round(len(ordered) * 0.1)) | |
| return {key: ("test" if i < held else "val" if i < 2 * held else "train") for i, key in enumerate(ordered)} | |
| def stratified_scene_splits(labels: dict[str, list[dict]]) -> dict[str, str]: | |
| groups = sorted(labels, key=lambda scene: hashlib.sha256(("dota:" + scene).encode()).digest()) | |
| if len(groups) < 3: | |
| raise ValueError("at least three independent scenes required") | |
| target = max(1, round(len(groups) * 0.1)) | |
| available = { | |
| scene: {row["canonical_class"] for row in labels[scene] if row["canonical_class"] and not row["difficult"]} | |
| for scene in groups | |
| } | |
| frequencies = Counter(label for classes in available.values() for label in classes) | |
| remaining = set(groups) | |
| assigned = {} | |
| for split in ("test", "val"): | |
| chosen = set() | |
| for name in sorted(frequencies, key=lambda name: (frequencies[name], name)): | |
| if any(name in available[scene] for scene in chosen): | |
| continue | |
| candidates = [scene for scene in remaining if name in available[scene]] | |
| if not candidates: | |
| continue | |
| scene = min(candidates, key=lambda candidate: ( | |
| -sum(1 / frequencies[label] for label in available[candidate] if not any(label in available[previous] for previous in chosen)), | |
| hashlib.sha256(f"{split}:{candidate}".encode()).digest(), | |
| )) | |
| chosen.add(scene) | |
| if len(chosen) > target: | |
| target = len(chosen) | |
| while len(chosen) < target and remaining - chosen: | |
| candidates = remaining - chosen | |
| chosen.add(min(candidates, key=lambda scene: hashlib.sha256(f"{split}:{scene}".encode()).digest())) | |
| for scene in chosen: | |
| assigned[scene] = split | |
| remaining.difference_update(chosen) | |
| assigned.update({scene: "train" for scene in remaining}) | |
| if len(assigned) != len(groups): | |
| raise ValueError("failed to assign every scene") | |
| return assigned | |
| def select_detection_tiles(size: tuple[int, int], labels: list[dict], limit: int) -> list[tuple[int, int, int, int]]: | |
| if limit <= 0: | |
| raise ValueError("tile limit must be positive") | |
| width, height = size | |
| candidates = [(x, y, min(x + 1024, width), min(y + 1024, height)) for y in starts(height) for x in starts(width)] | |
| covered = {} | |
| class_tiles = {} | |
| for bounds in candidates: | |
| tile_classes = set() | |
| for label in labels: | |
| name = label["canonical_class"] | |
| if name and not label["difficult"] and clip_object(label["vertices"], bounds) is not None: | |
| tile_classes.add(name) | |
| covered[bounds] = tile_classes | |
| for name in tile_classes: | |
| class_tiles.setdefault(name, []).append(bounds) | |
| selected = [] | |
| for name in sorted(class_tiles, key=lambda name: (len(class_tiles[name]), name)): | |
| if any(name in covered[bounds] for bounds in selected): | |
| continue | |
| choices = [bounds for bounds in class_tiles[name] if bounds not in selected] | |
| if choices: | |
| selected.append(min(choices, key=lambda bounds: hashlib.sha256(f"{name}:{bounds}".encode()).digest())) | |
| for bounds in sorted(candidates, key=lambda bounds: hashlib.sha256(str(bounds).encode()).digest()): | |
| if len(selected) >= limit: | |
| break | |
| if bounds not in selected: | |
| selected.append(bounds) | |
| return selected | |
| def parse_dota(text: str) -> tuple[list[dict], dict[str, str]]: | |
| mapping = {"plane": "aircraft", "ship": "vessel", "storage-tank": "storage_tank", "bridge": "bridge", "large-vehicle": "large_vehicle", "small-vehicle": "vehicle", "helicopter": "aircraft"} | |
| known = set(mapping) | {"baseball-diamond", "tennis-court", "basketball-court", "ground-track-field", "harbor", "roundabout", "soccer-ball-field", "swimming-pool", "container-crane"} | |
| result, meta = [], {} | |
| seen = set() | |
| for line_number, line in enumerate(text.splitlines(), 1): | |
| if not line.strip(): | |
| continue | |
| if ":" in line and len(line.split()) < 10: | |
| key, value = line.split(":", 1) | |
| meta[key.strip().lower()] = value.strip() | |
| continue | |
| fields = line.split() | |
| if len(fields) != 10 or fields[8] not in known or fields[9] not in {"0", "1", "2"}: | |
| raise ValueError(f"invalid DOTA annotation at line {line_number}") | |
| coordinates = [float(v) for v in fields[:8]] | |
| polygon = Polygon(np.asarray(coordinates).reshape(4, 2)) | |
| if not all(math.isfinite(v) for v in coordinates) or not polygon.is_valid or polygon.area <= 0: | |
| raise ValueError(f"invalid polygon at line {line_number}") | |
| key = (*coordinates, fields[8]) | |
| if key in seen: | |
| raise ValueError(f"duplicate annotation at line {line_number}") | |
| seen.add(key) | |
| result.append({"source_line": line_number, "original_class": fields[8], "canonical_class": mapping.get(fields[8]), "vertices": coordinates, "difficult": fields[9] != "0"}) | |
| return result, meta | |
| def select_dota(labels: dict[str, list[dict]], limit: int) -> list[str]: | |
| if limit <= 0: | |
| raise ValueError("scene limit must be positive") | |
| by_class = {} | |
| for scene, objects in labels.items(): | |
| for name in {item["canonical_class"] for item in objects} - {None}: | |
| by_class.setdefault(name, []).append(scene) | |
| selected = [] | |
| for name in sorted(by_class, key=lambda name: (len(by_class[name]), name)): | |
| candidates = sorted(by_class[name], key=lambda scene: hashlib.sha256(scene.encode()).digest()) | |
| for scene in candidates[:max(1, limit // max(1, len(by_class)))]: | |
| if scene not in selected and len(selected) < limit: | |
| selected.append(scene) | |
| for scene in sorted(labels, key=lambda scene: hashlib.sha256(scene.encode()).digest()): | |
| if len(selected) == limit: | |
| break | |
| if scene not in selected: | |
| selected.append(scene) | |
| return selected | |
| class storage_guard: | |
| def __init__(self, workspace: Path, maximum_bytes: int): | |
| self.used = sum(path.stat().st_size for path in workspace.rglob("*") if path.is_file()) | |
| self.maximum_bytes = maximum_bytes | |
| self.lock = threading.Lock() | |
| self.reserve(0) | |
| def reserve(self, size: int) -> None: | |
| with self.lock: | |
| if size < 0 or self.used + size > self.maximum_bytes: | |
| raise OSError("20 GB workspace storage budget exceeded; existing and partial files count") | |
| self.used += size | |
| class corpus: | |
| def __init__(self, root: Path, maximum_bytes: int): | |
| self.root = root | |
| self.maximum_bytes = maximum_bytes | |
| self.bytes_written = 0 | |
| self.workspace_budget = storage_guard(Path.cwd(), 20000000000) | |
| self.samples, self.scenes, self.rejected = [], [], [] | |
| self.hashes = {} | |
| self.documents = {split: {"images": [], "annotations": [], "ignored_annotations": []} for split in ("train", "val", "test")} | |
| if (root / "metadata" / "samples.jsonl").exists(): | |
| raise FileExistsError("output already exists; choose a new output version rather than overwrite it") | |
| root.mkdir(parents=True, exist_ok=True) | |
| def reject(self, meta: dict, reason: str) -> None: | |
| self.rejected.append({"source_dataset": meta.get("source_dataset"), "source_image_id": meta.get("source_image_id"), "reason": reason}) | |
| def accept(self, image: Image.Image, meta: dict, task: str) -> bool: | |
| if meta.get("source_dataset") == "dota" and meta.get("source_metadata", {}).get("imagesource") in {"GF", "JL", "GF-2", "JL-1"}: | |
| self.reject(meta, "panchromatic_source_not_rgb") | |
| return False | |
| if image.mode != "RGB": | |
| self.reject(meta, "not_native_rgb") | |
| return False | |
| pixels = np.asarray(image) | |
| if float(np.mean(np.all(pixels == 0, axis=2))) > 0.98: | |
| self.reject(meta, "more_than_98_percent_black") | |
| return False | |
| digest = hashlib.sha256(str(image.size).encode() + image.tobytes()).hexdigest() | |
| if digest in self.hashes: | |
| self.reject(meta, "duplicate_source_pixels") | |
| return False | |
| self.hashes[digest] = meta["id"] | |
| self.scenes.append(dict(meta, task=task, pixel_sha256=digest, original_dimensions=list(image.size))) | |
| return True | |
| def save(self, image: Image.Image, path: str) -> int: | |
| buffer = io.BytesIO() | |
| image.save(buffer, format="PNG", compress_level=3) | |
| payload = buffer.getvalue() | |
| if self.bytes_written + len(payload) > self.maximum_bytes: | |
| raise OSError("derived dataset storage budget reached") | |
| self.workspace_budget.reserve(len(payload)) | |
| destination = self.root / path | |
| destination.parent.mkdir(parents=True, exist_ok=True) | |
| destination.write_bytes(payload) | |
| self.bytes_written += len(payload) | |
| return len(payload) | |
| def tile(self, image: Image.Image, meta: dict, task: str, x: int, y: int) -> tuple[dict, tuple[int, int, int, int]] | None: | |
| bounds = (x, y, min(x + 1024, image.width), min(y + 1024, image.height)) | |
| cropped = image.crop(bounds) | |
| if np.mean(np.all(np.asarray(cropped) == 0, axis=2)) > 0.98: | |
| self.reject(dict(meta, source_image_id=f"{meta['source_image_id']}:{x}:{y}"), "mostly_black_tile") | |
| return None | |
| tile = Image.new("RGB", (1024, 1024)) | |
| tile.paste(cropped) | |
| name = f"{meta['id']}_{x}_{y}" | |
| relative = f"{task}/{meta['split']}/images/{name}.png" | |
| size = self.save(tile, relative) | |
| return dict(meta, id=name, task=task, image=relative, original_dimensions=list(image.size), tile_coordinates=list(bounds), width=1024, height=1024, valid_width=bounds[2] - x, valid_height=bounds[3] - y, image_bytes=size), bounds | |
| def segmentation(self, image: Image.Image, mask: np.ndarray, meta: dict, validity: np.ndarray | None = None) -> None: | |
| mapped = map_oem_mask(mask) | |
| if mask.shape != (image.height, image.width): | |
| raise ValueError("image/mask shape mismatch") | |
| valid = ~np.all(np.asarray(image) == 0, axis=2) | |
| if validity is not None: | |
| if validity.shape != mask.shape or validity.dtype != bool: | |
| raise ValueError("invalid source validity mask") | |
| valid &= validity | |
| mapped[~valid] = 255 | |
| if np.all(mapped == 255): | |
| self.reject(meta, "entire_mask_unlabeled") | |
| return | |
| if not self.accept(image, meta, "segmentation"): | |
| return | |
| for y in starts(image.height): | |
| for x in starts(image.width): | |
| result = self.tile(image, meta, "segmentation", x, y) | |
| if result is None: | |
| continue | |
| sample, bounds = result | |
| target = np.full((1024, 1024), 255, dtype=np.uint8) | |
| target[:sample["valid_height"], :sample["valid_width"]] = mapped[y:bounds[3], x:bounds[2]] | |
| relative = f"segmentation/{meta['split']}/masks/{sample['id']}.png" | |
| self.save(Image.fromarray(target), relative) | |
| ids, counts = np.unique(target, return_counts=True) | |
| sample.update(mask=relative, pixels={str(int(k)): int(v) for k, v in zip(ids, counts)}, annotated_classes=[1, 2, 12, 13, 14, 15, 16]) | |
| self.samples.append(sample) | |
| def detection(self, image: Image.Image, labels: list[dict], meta: dict, tile_limit: int = 100000) -> None: | |
| if not self.accept(image, meta, "detection"): | |
| return | |
| document = self.documents[meta["split"]] | |
| candidates = [(x, y, min(x + 1024, image.width), min(y + 1024, image.height)) for y in starts(image.height) for x in starts(image.width)] | |
| chosen_tiles = select_detection_tiles((image.width, image.height), labels, tile_limit) | |
| for bounds in chosen_tiles: | |
| x, y = bounds[:2] | |
| local, ignored = [], [] | |
| for label in labels: | |
| clipped = clip_object(label["vertices"], bounds) | |
| if clipped is None: | |
| continue | |
| if label["canonical_class"] is None: | |
| ignored.append({ | |
| "category_id": None, | |
| "canonical_class": None, | |
| "original_class": label["original_class"], | |
| "source_dataset": meta["source_dataset"], | |
| "source_image_id": meta["source_image_id"], | |
| "source_annotation_id": f"{meta['source_image_id']}:{label['source_line']}", | |
| "bbox": clipped["bbox"], | |
| "area": clipped["bbox"][2] * clipped["bbox"][3], | |
| "polygon_area": clipped["polygon_area"], | |
| "obb_polygon": label["vertices"], | |
| "clipped_polygon": clipped["clipped_polygon"], | |
| "visible_fraction": clipped["visibility"], | |
| "source_difficult": label["difficult"], | |
| "ignore": True, | |
| "iscrowd": 0, | |
| }) | |
| continue | |
| identifier = {"building": 1, "bridge": 3, "vehicle": 4, "large_vehicle": 5, "aircraft": 6, "vessel": 7, "storage_tank": 8}[label["canonical_class"]] | |
| annotation = { | |
| "category_id": identifier, | |
| "canonical_class": label["canonical_class"], | |
| "original_class": label["original_class"], | |
| "source_dataset": meta["source_dataset"], | |
| "source_image_id": meta["source_image_id"], | |
| "source_annotation_id": f"{meta['source_image_id']}:{label['source_line']}", | |
| "bbox": clipped["bbox"], | |
| "area": clipped["bbox"][2] * clipped["bbox"][3], | |
| "polygon_area": clipped["polygon_area"], | |
| "obb_polygon": label["vertices"], | |
| "clipped_polygon": clipped["clipped_polygon"], | |
| "visible_fraction": clipped["visibility"], | |
| "source_difficult": label["difficult"], | |
| "iscrowd": 0, | |
| } | |
| if clipped["ignore"] or label["difficult"]: | |
| ignored.append(annotation) | |
| else: | |
| local.append(annotation) | |
| result = self.tile(image, meta, "detection", x, y) | |
| if result is None: | |
| continue | |
| sample, _ = result | |
| image_id = len(document["images"]) + 1 | |
| sample.update(annotated_classes=[3, 4, 5, 6, 7, 8], unknown_classes=[1, 2, 9, 10, 11, 12, 13, 14, 15, 16], detection_image_id=image_id, positive_instances=len(local), ignored_instances=len(ignored), negative_for_annotated_classes=not local and not ignored) | |
| for item in local: | |
| item.update(id=len(document["annotations"]) + 1, image_id=image_id) | |
| document["annotations"].append(item) | |
| for item in ignored: | |
| item.update(image_id=image_id, ignore=True) | |
| document["ignored_annotations"].append(item) | |
| label_path = f"detection/{meta['split']}/labels/{sample['id']}.json" | |
| self.json_file(label_path, {"annotations": local, "ignored_annotations": ignored}) | |
| sample["labels"] = label_path | |
| document["images"].append({"id": image_id, "file_name": sample["image"], "width": 1024, "height": 1024, "source_image_id": meta["source_image_id"], "license_group": meta["license_group"]}) | |
| self.samples.append(sample) | |
| if len(candidates) > len(chosen_tiles): | |
| self.reject(meta, f"selection_limit: retained {len(chosen_tiles)} of {len(candidates)} tiles; label-aware sampling; not a quality rejection") | |
| def json_file(self, path: str, value: dict | list) -> None: | |
| destination = self.root / path | |
| destination.parent.mkdir(parents=True, exist_ok=True) | |
| payload = json.dumps(value, indent=2, allow_nan=False).encode("utf-8") | |
| self.workspace_budget.reserve(len(payload)) | |
| destination.write_bytes(payload) | |
| def finish(self) -> None: | |
| classes = ["building", "road", "bridge", "vehicle", "large_vehicle", "aircraft", "vessel", "storage_tank", "railway", "parking_area", "construction", "water", "vegetation", "agricultural_area", "bare_ground", "developed_area"] | |
| for split, document in self.documents.items(): | |
| document["categories"] = [{"id": i + 1, "name": name} for i, name in enumerate(classes)] | |
| document["info"] = {"ontology_version": "1.0.0", "bbox_format": "xywh", "obb_coordinates": "original source pixels", "clipped_polygon_coordinates": "tile pixels", "warning": "Use supplied loader: ignored regions, padding and partial class supervision are not handled by a naive COCO loader."} | |
| self.json_file(f"detection/{split}/annotations.json", document) | |
| for name, rows in (("metadata/samples.jsonl", self.samples), ("metadata/scenes.jsonl", self.scenes), ("rejected.jsonl", self.rejected)): | |
| path = self.root / name | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| payload = "".join(json.dumps(row, allow_nan=False) + "\n" for row in rows).encode("utf-8") | |
| self.workspace_budget.reserve(len(payload)) | |
| path.write_bytes(payload) | |
| counts = Counter(a["category_id"] for doc in self.documents.values() for a in doc["annotations"]) | |
| pixels = Counter() | |
| for sample in self.samples: | |
| pixels.update({int(k): v for k, v in sample.get("pixels", {}).items()}) | |
| size_stats = {} | |
| for i, name in enumerate(classes, 1): | |
| annotations = [a for doc in self.documents.values() for a in doc["annotations"] if a["category_id"] == i] | |
| dims = np.array([[a["bbox"][2], a["bbox"][3], a["area"]] for a in annotations]) | |
| size_stats[name] = None if not annotations else { | |
| "percentiles": [0, 1, 5, 25, 50, 75, 95, 99, 100], | |
| "width_height_area": np.percentile(dims, [0, 1, 5, 25, 50, 75, 95, 99, 100], axis=0).tolist(), | |
| "short_side_below_4px": int(np.sum(dims[:, :2].min(axis=1) < 4)), | |
| } | |
| stats = { | |
| "training_ready": False, | |
| "status": "built_awaiting_independent_validation_and_visual_inspection", | |
| "derived_image_mask_bytes": self.bytes_written, | |
| "original_images": len(self.scenes), | |
| "tiles": len(self.samples), | |
| "images_per_source": dict(Counter(s["source_dataset"] for s in self.scenes)), | |
| "tiles_per_source": dict(Counter(s["source_dataset"] for s in self.samples)), | |
| "split_counts": dict(Counter(f"{s['task']}/{s['split']}" for s in self.samples)), | |
| "detection_instances": {name: counts[i] for i, name in enumerate(classes, 1)}, | |
| "segmentation_pixels": {name: pixels[i] for i, name in enumerate(classes, 1)}, | |
| "ignore_pixels": pixels[255], | |
| "object_size": size_stats, | |
| "positive_class_imbalance_ratio": max(counts.values()) / min(counts.values()) if counts else None, | |
| "objects_per_tile_percentiles": np.percentile([s["positive_instances"] for s in self.samples if s["task"] == "detection"], [0, 25, 50, 75, 95, 100]).tolist() if any(s["task"] == "detection" for s in self.samples) else None, | |
| "source_scene_and_tile_counts_are_distinct": True, | |
| "rejection_records": len(self.rejected), | |
| "unique_source_detection_instances": len({a["source_annotation_id"] for doc in self.documents.values() for a in doc["annotations"]}), | |
| "geographic_split_evidence": {"openearthmap": "whole attributed city/region groups", "dota": "original scene identity only; absolute geolocation unavailable"}, | |
| "cross_source_geographic_overlap": "not fully verifiable because DOTA scenes lack absolute geolocation", | |
| } | |
| self.json_file("stats.json", stats) | |
| frequencies = {str(i): counts[i] for i in range(1, 17)} | |
| self.json_file("metadata/sampling.json", {"detection_instances": frequencies, "recommendation": "Use class-aware sampling or capped inverse-square-root class-frequency weights; no duplicated image files.", "augmentation": ["horizontal and vertical flips", "90-degree rotations", "label-aware arbitrary rotations", "moderate brightness/contrast and color changes", "moderate blur/compression", "scale jitter constrained by measured tiny-object sizes"], "avoid": ["rotating boxes without polygons", "turning ignored regions into background", "heavy downsampling of small vehicles", "applying bilinear interpolation to masks"]}) | |
| def build(output: Path) -> None: | |
| config = json.loads(Path("datasets/build_config.json").read_text()) | |
| oem_manifest = Path("datasets/metadata/oem_downloaded.json") | |
| if not oem_manifest.exists(): | |
| raise FileNotFoundError("OpenEarthMap acquisition has not completed; run acquire.py first") | |
| oem = json.loads(oem_manifest.read_text()) | |
| writer = corpus(output, maximum_bytes=config["maximum_derived_bytes"]) | |
| labels_by_scene, label_metadata = {}, {} | |
| with tarfile.open("datasets/raw/dotav1.5_annotations_val.tar.gz") as archive: | |
| for member in archive: | |
| if member.isfile() and member.name.endswith(".txt"): | |
| scene = PurePosixPath(member.name).stem | |
| source_file = archive.extractfile(member) | |
| if source_file is None: | |
| raise ValueError(f"unreadable archive member: {member.name}") | |
| try: | |
| labels, meta = parse_dota(source_file.read().decode("utf-8-sig")) | |
| except ValueError as error: | |
| writer.reject({"source_dataset": "dota", "source_image_id": scene}, str(error)) | |
| continue | |
| if meta.get("imagesource") in {"GF", "JL", "GF-2", "JL-1"}: | |
| writer.reject({"source_dataset": "dota", "source_image_id": scene}, "panchromatic_source_not_rgb") | |
| continue | |
| labels_by_scene[scene] = labels | |
| label_metadata[scene] = dict(meta, annotation_member=member.name) | |
| selected = select_dota(labels_by_scene, config["dota_scene_limit"]) | |
| splits = stratified_scene_splits({scene: labels_by_scene[scene] for scene in selected}) | |
| writer.json_file("metadata/dota_selection.json", {"selected_scenes": selected, "selection": "rare-class scene coverage followed by deterministic hash fill", "original_split": "val", "split_assignment": splits, "tile_selection": "deterministic hash within each source scene; at most floor(tile_limit / selected_scenes) tiles per scene"}) | |
| per_scene = max(1, config["dota_tile_limit"] // len(selected)) | |
| processed = set() | |
| with tarfile.open("datasets/raw/dotav1.0_images_val.tar.gz", "r|gz") as archive: | |
| for member in archive: | |
| scene = PurePosixPath(member.name).stem | |
| if not member.isfile() or not member.name.endswith(".png") or scene not in splits: | |
| continue | |
| metadata = { | |
| "id": "dota_" + scene, "source_dataset": "dota", "source_image_id": scene, | |
| "group": "dota:" + scene, "split": splits[scene], "original_split": "val", | |
| "license": "academic purposes only; commercial use prohibited", | |
| "license_group": "academic_only", "source_member": member.name, | |
| "source_metadata": label_metadata[scene], "source_object_count": len(labels_by_scene[scene]), | |
| "geographic_coordinates": None, "geographic_evidence": "source scene only", | |
| } | |
| source_file = archive.extractfile(member) | |
| if source_file is None: | |
| raise ValueError(f"unreadable image member: {member.name}") | |
| payload = source_file.read() | |
| metadata["source_sha256"] = hashlib.sha256(payload).hexdigest() | |
| try: | |
| with Image.open(io.BytesIO(payload)) as image: | |
| image.load() | |
| writer.detection(image, labels_by_scene[scene], metadata, tile_limit=per_scene) | |
| except (Image.UnidentifiedImageError, Image.DecompressionBombError, ValueError) as error: | |
| writer.reject(metadata, f"invalid_image_or_annotation: {error}") | |
| processed.add(scene) | |
| if len(processed) % 20 == 0: | |
| print(f"DOTA: {len(processed)}/{len(selected)} scenes", flush=True) | |
| for scene in set(selected) - processed: | |
| writer.reject({"source_dataset": "dota", "source_image_id": scene}, "source_image_missing_from_archive") | |
| for index, row in enumerate(oem): | |
| source_id = PurePosixPath(row["image_member"]).stem | |
| metadata = { | |
| "id": "oem_" + source_id, "source_dataset": "openearthmap", "source_image_id": source_id, | |
| "group": row["group"], "split": row["split"], "original_split": row["original_split"], | |
| "license": row["license"], "image_license": row["image_license"], | |
| "license_group": row["license_group"], "country": row["country"], "region": row["region"], | |
| "providers": row["providers"], "source_files": row["files"], | |
| } | |
| try: | |
| with rasterio.open(row["image_path"]) as src: | |
| if src.count != 3 or any(dtype != "uint8" for dtype in src.dtypes): | |
| writer.reject(metadata, "not_native_8bit_three_channel_rgb") | |
| continue | |
| pixels = src.read() | |
| validity = src.dataset_mask() > 0 | |
| if src.crs: | |
| metadata.update(crs=src.crs.to_string(), source_transform=list(src.transform), geographic_bounds=list(src.bounds)) | |
| else: | |
| metadata.update(crs=None, geographic_bounds=None) | |
| with rasterio.open(row["mask_path"]) as src: | |
| if src.count != 1: | |
| raise ValueError("mask must have one indexed channel") | |
| mask = src.read(1) | |
| validity &= src.dataset_mask() > 0 | |
| writer.segmentation(Image.fromarray(np.moveaxis(pixels, 0, -1)), mask, metadata, validity=validity) | |
| except (rasterio.errors.RasterioError, ValueError) as error: | |
| writer.reject(metadata, f"invalid_image_or_mask: {error}") | |
| if (index + 1) % 25 == 0: | |
| print(f"OpenEarthMap: {index + 1}/{len(oem)} images", flush=True) | |
| writer.finish() | |
| print(f"built {len(writer.samples)} samples from {len(writer.scenes)} source images; validation still required", flush=True) | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--output", type=Path, default=Path("datasets/corpus")) | |
| args = parser.parse_args() | |
| build(args.output) | |
Xet Storage Details
- Size:
- 29.9 kB
- Xet hash:
- 01fb916bab46db1d7ca13e56810e208873e14fd3e8fd19a0777223e29030abf9
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.