Buckets:
| """Source-complete DOTA targets without rewriting the approved corpus.""" | |
| from dataclasses import dataclass, replace | |
| from pathlib import Path | |
| import json | |
| import cv2 | |
| import numpy as np | |
| from PIL import Image, ImageDraw | |
| from shapely.geometry import Polygon | |
| from load_dataset import dataset, load_sample | |
| from training.model_factory import DOTA_NAMES | |
| SOURCE_CLASSES = {name: index for index, name in enumerate(DOTA_NAMES)} | |
| class DetectorExample: | |
| image: np.ndarray | |
| class_ids: np.ndarray | |
| polygons: np.ndarray | |
| boxes: np.ndarray # pixel xywhr, radians | |
| valid_mask: np.ndarray | |
| ignore_polygons: tuple | |
| source_annotation_ids: tuple | |
| metadata: dict | |
| def rotated_boxes(polygons): | |
| result = [] | |
| for points in polygons: | |
| (x, y), (w, h), angle = cv2.minAreaRect(np.asarray(points, np.float32)) | |
| if angle >= 90: | |
| w, h, angle = h, w, angle - 90 | |
| if angle < 0: | |
| w, h, angle = h, w, angle + 90 | |
| result.append([x, y, w, h, np.deg2rad(angle)]) | |
| return np.asarray(result, np.float32).reshape(-1, 5) | |
| def split_detection_annotations(sample: dict) -> tuple[list[tuple[np.ndarray, int, str]], list[np.ndarray]]: | |
| """Apply the detector's target-eligibility rules without decoding an image. | |
| Sampling, evaluation-target construction, and training must agree about | |
| which tile fragments are positive supervision. Keeping this geometry-only | |
| portion separate makes that agreement testable without treating a source | |
| scene's category list as if every one of its tiles contained that class. | |
| """ | |
| meta = sample["metadata"] | |
| if meta["task"] != "detection": | |
| raise ValueError("detection sample required") | |
| positives, ignored = [], [] | |
| origin = np.array(meta["tile_coordinates"][:2]) | |
| for a in sample["annotations"] + sample["ignored_annotations"]: | |
| name = a["original_class"].replace(" ", "-") | |
| local = np.asarray(a["clipped_polygon"], dtype=float).reshape(-1, 2) | |
| geometry = Polygon(local) | |
| if not geometry.is_valid or geometry.area <= 0: | |
| raise ValueError("invalid clipped polygon") | |
| if a.get("training_ignore", False): | |
| if meta.get("source_dataset") not in {"dior-r", "dior-voc", "fair1m", "soda-a"}: | |
| raise ValueError("explicit source ignore is only defined for reviewed partial-label sources") | |
| ignored.append(local) | |
| continue | |
| if name not in SOURCE_CLASSES: | |
| raise ValueError(f"unrecognized source class: {name}") | |
| fragment = a["visible_fraction"] < .7 or min(a["bbox"][2:]) < 2 | |
| # The corpus's generic ignore flag also marks healthy auxiliary classes. | |
| if a["source_difficult"] or fragment: | |
| ignored.append(local) | |
| continue | |
| source = np.asarray(a["obb_polygon"], dtype=float).reshape(-1, 2) - origin | |
| if a["visible_fraction"] >= 1 - 1e-8: | |
| points = source | |
| else: | |
| points = local if len(local) == 4 else np.asarray(geometry.minimum_rotated_rectangle.exterior.coords)[:-1] | |
| if points.shape != (4, 2): | |
| raise ValueError("OBB target must have four vertices") | |
| if min(rotated_boxes([points])[0, 2:4]) < 2: | |
| # The pinned loss drops thinner rotated targets; keep their regions ignored. | |
| ignored.append(local) | |
| continue | |
| positives.append((points, SOURCE_CLASSES[name], a["source_annotation_id"])) | |
| return positives, ignored | |
| def supervised_class_ids(sample: dict) -> set[int]: | |
| """Detector-head classes with at least one eligible positive in this tile.""" | |
| return {class_id for _, class_id, _ in split_detection_annotations(sample)[0]} | |
| def convert_detection_sample(sample: dict) -> DetectorExample: | |
| meta, image = sample["metadata"], sample["image"] | |
| height, width = image.shape[:2] | |
| valid = np.zeros((height, width), dtype=bool) | |
| valid[:meta["valid_height"], :meta["valid_width"]] = True | |
| valid &= ~np.all(image == 0, axis=2) | |
| positives, ignored = split_detection_annotations(sample) | |
| polygons = [points for points, _, _ in positives] | |
| classes = [class_id for _, class_id, _ in positives] | |
| identifiers = [identifier for _, _, identifier in positives] | |
| ignore_mask = Image.new("1", (width, height)) | |
| drawing = ImageDraw.Draw(ignore_mask) | |
| for points in ignored: | |
| drawing.polygon([tuple(p) for p in points], fill=1) | |
| ignored_pixels = np.asarray(ignore_mask, dtype=bool).copy() | |
| if meta.get("source_dataset") == "dior-r" and polygons: | |
| # Keep genuine supervised positives inside broad ignored scene annotations. | |
| positives = Image.new("1", (width, height)) | |
| draw = ImageDraw.Draw(positives) | |
| for points in polygons: | |
| draw.polygon([tuple(p) for p in points], fill=1) | |
| ignored_pixels &= ~np.asarray(positives, dtype=bool) | |
| valid &= ~ignored_pixels | |
| polygons = np.asarray(polygons, np.float32).reshape(-1, 4, 2) | |
| return DetectorExample(image, np.asarray(classes, np.int64), polygons, | |
| rotated_boxes(polygons), valid, tuple(ignored), tuple(identifiers), dict(meta)) | |
| def load_detection_tile(root: Path, sample: dict) -> DetectorExample: | |
| approved = dataset(root, "detection", sample["split"], license_scope="all") | |
| by_id = {row["id"]: row for row in approved.samples} | |
| if by_id.get(sample["id"]) != sample: | |
| raise ValueError("sample differs from approved manifest") | |
| return convert_detection_sample(load_sample(root, sample, approved.expected_hashes)) | |
| def transform_example(example: DetectorExample, *, horizontal=False, vertical=False, rotations=0): | |
| """Exact flips/right-angle rotations; polygons use continuous edge coordinates.""" | |
| image, mask = example.image.copy(), example.valid_mask.copy() | |
| polygons = example.polygons.copy() | |
| ignored = [p.copy() for p in example.ignore_polygons] | |
| height, width = image.shape[:2] | |
| geometries = [polygons, *ignored] | |
| if horizontal: | |
| image, mask = image[:, ::-1], mask[:, ::-1] | |
| for points in geometries: | |
| points[..., 0] = width - points[..., 0] | |
| if vertical: | |
| image, mask = image[::-1], mask[::-1] | |
| for points in geometries: | |
| points[..., 1] = height - points[..., 1] | |
| for _ in range(rotations % 4): | |
| image, mask = np.rot90(image), np.rot90(mask) | |
| for points in geometries: | |
| x = points[..., 0].copy() | |
| points[..., 0] = points[..., 1] | |
| points[..., 1] = width - x | |
| width, height = height, width | |
| return replace(example, image=np.ascontiguousarray(image), valid_mask=np.ascontiguousarray(mask), | |
| polygons=polygons, boxes=rotated_boxes(polygons), ignore_polygons=tuple(ignored)) | |
| def detection_batch(examples): | |
| import torch | |
| boxes, classes, indices = [], [], [] | |
| for i, e in enumerate(examples): | |
| height, width = e.image.shape[:2] | |
| boxes.append(e.boxes / np.array([width, height, width, height, 1], np.float32)) | |
| classes.append(e.class_ids) | |
| indices.extend([i] * len(e.class_ids)) | |
| coverage = [] | |
| for example in examples: | |
| ids = example.metadata.get("annotated_source_classes", list(range(len(SOURCE_CLASSES)))) | |
| if any(type(c) is not int or c not in range(len(SOURCE_CLASSES)) for c in ids): | |
| raise ValueError("invalid annotated source class IDs") | |
| if not set(example.class_ids.tolist()) <= set(ids): | |
| raise ValueError("positive source class lacks supervision coverage") | |
| coverage.append([c in ids for c in range(len(SOURCE_CLASSES))]) | |
| return {"class_valid_mask": torch.tensor(coverage, dtype=torch.bool), | |
| "img": torch.from_numpy(np.stack([e.image.transpose(2, 0, 1) for e in examples])).float() / 255, | |
| "bboxes": torch.from_numpy(np.concatenate(boxes)), | |
| "cls": torch.from_numpy(np.concatenate(classes)).reshape(-1, 1).float(), | |
| "batch_idx": torch.tensor(indices, dtype=torch.float32), | |
| "valid_mask": torch.from_numpy(np.stack([e.valid_mask for e in examples]))} | |
| def build_detection_cache(root: Path, output: Path) -> Path: | |
| """Write a train-only metadata index, never duplicate image/label assets.""" | |
| repo = Path(__file__).resolve().parents[1] | |
| output = output.resolve() | |
| if not output.is_relative_to(repo / "training"): | |
| raise ValueError("cache must be under training/") | |
| data = dataset(root, "detection", "train", license_scope="all") | |
| output.parent.mkdir(parents=True, exist_ok=True) | |
| with output.open("x", encoding="utf-8") as handle: | |
| json.dump({"corpus_content_sha256": json.loads((Path(root) / "stats.json").read_text())["validation_content_sha256"], | |
| "source_classes": SOURCE_CLASSES, "samples": data.samples}, handle) | |
| return output | |
Xet Storage Details
- Size:
- 8.9 kB
- Xet hash:
- fddaac8215655513f8eb4b2fbdf85c8e30b021c9edf9c17f6b88d4927205532a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.