Data API Reference

oriented_det.data

Data loading, tiling, transforms, and evaluation for oriented detection.

APCalculator

Calculate Average Precision (AP) for oriented object detection.

Source code in oriented_det/data/evaluation.py

class APCalculator:
    """Calculate Average Precision (AP) for oriented object detection."""

    def __init__(
        self,
        iou_threshold: float = 0.5,
        class_names: Optional[Sequence[str]] = None,
        device: DeviceType = None,
        *,
        use_exact_rotated_iou: bool = True,
    ):
        if not 0 < iou_threshold <= 1:
            raise ValueError("IoU threshold must be in (0, 1]")
        self.iou_threshold = iou_threshold
        self.class_names = class_names or []
        self.class_to_id = {name: i for i, name in enumerate(self.class_names)} if self.class_names else {}
        self.use_exact_rotated_iou = bool(use_exact_rotated_iou)
        self.device = None if self.use_exact_rotated_iou else device
        self._exact_iou_backend: Optional[str] = (
            resolve_exact_polygon_iou_backend() if self.use_exact_rotated_iou else None
        )

    def compute_ap(
        self,
        detections: Sequence[Detection],
        ground_truths: Sequence[GroundTruth],
        *,
        class_name: Optional[str] = None,
        show_progress: bool = False,
        progress_stream: Optional[Any] = None,
        max_iou_calculations: Optional[int] = None,
        min_box_size: float = 1.0,
    ) -> Tuple[float, ClassEvalMetrics]:
        """Compute Average Precision for a single class.

        Args:
            detections: List of detections
            ground_truths: List of ground truths
            class_name: Optional class name
            show_progress: Whether to show progress bar
            max_iou_calculations: Maximum number of IoU calculations to allow.
                                 If exceeded, will use approximate method or return 0.0.
            min_box_size: Minimum box size (width/height) in pixels to consider valid.
                         Boxes smaller than this will be filtered out (default: 1.0).
        """
        if class_name:
            detections = [d for d in detections if d.class_name == class_name]
            ground_truths = [gt for gt in ground_truths if gt.class_name == class_name]

        # Filter out degenerated boxes
        original_det_count = len(detections)
        original_gt_count = len(ground_truths)
        detections = [d for d in detections if d.rbox.is_valid(min_size=min_box_size)]
        ground_truths = [gt for gt in ground_truths if gt.rbox.is_valid(min_size=min_box_size)]

        if show_progress:
            filtered_dets = original_det_count - len(detections)
            filtered_gts = original_gt_count - len(ground_truths)
            if filtered_dets > 0 or filtered_gts > 0:
                filter_msg = f"  Filtered {filtered_dets} degenerated detections, {filtered_gts} degenerated ground truths"
                print(filter_msg)

        if not ground_truths:
            ap = 0.0 if detections else 1.0
            return ap, ClassEvalMetrics(
                num_gts=0,
                num_dets=len(detections),
                recall=0.0,
                ap=ap,
            )

        # Check if we should optimize due to too many calculations
        num_dets = len(detections)
        num_gts = len(ground_truths)
        total_calculations = num_dets * num_gts

        # Use batch IoU if available and calculations are too many
        use_batch_iou = max_iou_calculations and total_calculations > max_iou_calculations

        # Exact polygon mAP (final_nms_use_cpu): never use GPU sampling IoU
        from ..utils import is_gpu_device
        use_gpu = (
            not self.use_exact_rotated_iou
            and torch is not None
            and self.device is not None
            and is_gpu_device(self.device)
        )

        sorted_dets = sorted(detections, key=lambda d: d.score, reverse=True)
        tp_fp_from_batch = False
        tp: List[int] = []
        fp: List[int] = []

        if use_batch_iou:
            # Try using batch IoU computation which is much faster
            try:
                det_rboxes = [d.rbox for d in sorted_dets]
                gt_rboxes = [gt.rbox for gt in ground_truths]

                # Use chunking for very large matrices to avoid memory issues
                # Process detections in chunks to limit memory usage.
                # GPU IoU builds [chunk*S, num_gt] tensors with S ~= default oriented IoU samples
                # (typically 100; tier/env can raise it). When num_gt is large (e.g. 72k)
                # a fixed 5000-detect chunk can OOM. Cap chunk so chunk*S*num_gt <= ~500M elements.
                num_gt = len(ground_truths)
                max_elements = 500_000_000  # ~500M elements for bool tensor (~500 MB)
                samples_per_box = 100  # conservative match to oriented_box_iou_gpu default envelope
                adaptive_chunk = max(1, max_elements // (samples_per_box * max(1, num_gt)))
                CHUNK_SIZE = min(5000, adaptive_chunk)
                num_dets = len(sorted_dets)

                if show_progress and tqdm is not None:
                    if self.use_exact_rotated_iou:
                        iou_type = f"CPU exact ({self._exact_iou_backend})"
                    else:
                        iou_type = "GPU-accelerated" if use_gpu else "CPU"
                    print(f"    Using chunked batch IoU computation ({iou_type}) for {class_name} ({total_calculations:,} calculations)")
                    if num_dets > CHUNK_SIZE:
                        print(f"    Processing {num_dets:,} detections in chunks of {CHUNK_SIZE:,}")

                # Process detections in chunks
                gt_matched = [False] * len(ground_truths)
                tp = []
                fp = []

                # Process detections in chunks; tqdm to progress_stream so console sees it but train.log does not
                tqdm_file = progress_stream if progress_stream is not None else sys.stderr
                det_iter = (
                    tqdm(range(0, num_dets, CHUNK_SIZE), desc="    Processing chunks", unit="chunk", leave=False, file=tqdm_file)
                    if (show_progress and tqdm is not None and num_dets > CHUNK_SIZE)
                    else range(0, num_dets, CHUNK_SIZE)
                )

                for chunk_start in det_iter:
                    chunk_end = min(chunk_start + CHUNK_SIZE, num_dets)
                    chunk_dets = sorted_dets[chunk_start:chunk_end]
                    chunk_det_rboxes = det_rboxes[chunk_start:chunk_end]

                    # Compute IoU matrix for this chunk
                    chunk_iou_matrix = None
                    try:
                        if use_gpu:
                            # Use GPU-accelerated IoU (much faster)
                            from ..ops.gpu_ops import oriented_box_iou_gpu
                            from ..ops.utils import rboxes_to_tensor

                            # Convert RBox objects to tensors
                            chunk_det_tensors = rboxes_to_tensor(chunk_det_rboxes, device=self.device)
                            gt_tensors = rboxes_to_tensor(gt_rboxes, device=self.device)

                            # Compute IoU matrix on GPU
                            iou_matrix_gpu = oriented_box_iou_gpu(chunk_det_tensors, gt_tensors)

                            # Convert to CPU list of lists for compatibility
                            chunk_iou_matrix = iou_matrix_gpu.cpu().tolist()
                        else:
                            from ..ops.iou import batch_rbox_iou
                            chunk_iou_matrix = batch_rbox_iou(
                                chunk_det_rboxes,
                                gt_rboxes,
                                device=self.device,
                                intersection_backend=(
                                    self._exact_iou_backend
                                    if self.use_exact_rotated_iou
                                    else "auto"
                                ),
                            )
                    except Exception as chunk_e:
                        # If chunk batch fails, fall back to per-detection computation for this chunk
                        if show_progress:
                            print(f"    Chunk batch IoU failed, using per-detection for chunk {chunk_start}-{chunk_end}: {chunk_e}")
                        chunk_iou_matrix = None

                    # Process each detection in this chunk
                    for local_idx, det in enumerate(chunk_dets):
                        det_idx = chunk_start + local_idx
                        best_iou = 0.0
                        best_gt_idx = -1

                        if chunk_iou_matrix is not None:
                            # Use pre-computed IoU matrix
                            for gt_idx, gt in enumerate(ground_truths):
                                if gt_matched[gt_idx] or gt.class_name != det.class_name:
                                    continue
                                if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                                    continue
                                det_iou = chunk_iou_matrix[local_idx][gt_idx]
                                if det_iou > best_iou:
                                    best_iou = det_iou
                                    best_gt_idx = gt_idx
                        else:
                            # Fall back to per-detection IoU computation
                            for gt_idx, gt in enumerate(ground_truths):
                                if gt_matched[gt_idx] or gt.class_name != det.class_name:
                                    continue
                                if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                                    continue
                                try:
                                    from ..ops.iou import rbox_iou
                                    det_iou = rbox_iou(
                                        det.rbox,
                                        gt.rbox,
                                        intersection_backend=(
                                            self._exact_iou_backend
                                            if self.use_exact_rotated_iou
                                            else "auto"
                                        ),
                                    )
                                    if det_iou > best_iou:
                                        best_iou = det_iou
                                        best_gt_idx = gt_idx
                                except Exception:
                                    continue

                        # Check if match is valid
                        if best_iou >= self.iou_threshold and best_gt_idx >= 0:
                            gt = ground_truths[best_gt_idx]
                            if gt.difficult == 0:
                                gt_matched[best_gt_idx] = True
                                tp.append(1)
                                fp.append(0)
                            else:
                                tp.append(0)
                                fp.append(0)
                        else:
                            tp.append(0)
                            fp.append(1)

                    # Clear chunk matrix from memory
                    del chunk_iou_matrix
                    if torch is not None and torch.cuda.is_available():
                        torch.cuda.empty_cache()
                    elif torch is not None and getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
                        torch.mps.empty_cache()

                tp_fp_from_batch = True

            except Exception as e:
                # Fall back to regular computation if batch fails
                if show_progress:
                    print(f"    Batch IoU failed for {class_name}, using regular method (will be slow): {e}")
                use_batch_iou = False

        if not tp_fp_from_batch:
            # Track which ground truths have been matched (standard per-detection path)
            gt_matched = [False] * len(ground_truths)
            tp = []
            fp = []

            # Use progress bar if requested and many detections; write to progress_stream so console sees it but train.log does not
            tqdm_file = progress_stream if progress_stream is not None else sys.stderr
            use_pbar = show_progress and tqdm is not None and len(sorted_dets) > 5000
            det_iter = tqdm(sorted_dets, desc=f"  {class_name or 'All'}", unit="det", leave=False, ncols=100, file=tqdm_file) if use_pbar else sorted_dets

            for det_idx, det in enumerate(det_iter):
                best_iou = 0.0
                best_gt_idx = -1

                # Find best matching ground truth (only from same image when image_id is set)
                for gt_idx, gt in enumerate(ground_truths):
                    if gt_matched[gt_idx] or gt.class_name != det.class_name:
                        continue
                    if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                        continue
                    try:
                        det_iou = iou.rbox_iou(
                            det.rbox,
                            gt.rbox,
                            intersection_backend=(
                                self._exact_iou_backend
                                if self.use_exact_rotated_iou
                                else "auto"
                            ),
                        )
                        if det_iou > best_iou:
                            best_iou = det_iou
                            best_gt_idx = gt_idx
                    except Exception:
                        # Skip problematic IoU calculations and continue with next GT
                        continue

                # Check if match is valid
                if best_iou >= self.iou_threshold and best_gt_idx >= 0:
                    gt = ground_truths[best_gt_idx]
                    if gt.difficult == 0:  # Only count non-difficult as TP
                        gt_matched[best_gt_idx] = True
                        tp.append(1)
                        fp.append(0)
                    else:
                        tp.append(0)
                        fp.append(0)  # Difficult GTs don't count as FP either
                else:
                    tp.append(0)
                    fp.append(1)

        # Compute precision and recall
        tp_cumsum = []
        fp_cumsum = []
        cumsum_tp = 0
        cumsum_fp = 0

        for t, f in zip(tp, fp):
            cumsum_tp += t
            cumsum_fp += f
            tp_cumsum.append(cumsum_tp)
            fp_cumsum.append(cumsum_fp)

        num_positives = sum(1 for gt in ground_truths if gt.difficult == 0)
        num_dets = len(sorted_dets)
        if num_positives == 0:
            return 0.0, ClassEvalMetrics(
                num_gts=0,
                num_dets=num_dets,
                recall=0.0,
                ap=0.0,
            )

        recalls = [t / num_positives for t in tp_cumsum]
        precisions = [
            tp_cumsum[i] / (tp_cumsum[i] + fp_cumsum[i]) if (tp_cumsum[i] + fp_cumsum[i]) > 0 else 0.0
            for i in range(len(tp_cumsum))
        ]

        final_recall = recalls[-1] if recalls else 0.0

        # Compute AP using 11-point interpolation
        ap = 0.0
        for t in [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]:
            precisions_at_recall = [p for r, p in zip(recalls, precisions) if r >= t]
            if precisions_at_recall:
                ap += max(precisions_at_recall) / 11.0

        return ap, ClassEvalMetrics(
            num_gts=num_positives,
            num_dets=num_dets,
            recall=final_recall,
            ap=ap,
        )

    def compute_map(
        self,
        all_detections: Dict[str, List[Detection]],  # image_id -> detections
        all_ground_truths: Dict[str, List[GroundTruth]],  # image_id -> ground_truths
        show_progress: bool = True,
        progress_stream: Optional[Any] = None,
        max_iou_calculations_per_class: int = 10_000_000,  # 10M IoU calculations max per class
        min_box_size: float = 1.0,
    ) -> Tuple[Dict[str, float], Dict[str, ClassEvalMetrics]]:
        """Compute mAP across all classes.

        Args:
            all_detections: Dictionary mapping image_id -> list of Detection objects
            all_ground_truths: Dictionary mapping image_id -> list of GroundTruth objects
            show_progress: Whether to show progress bars
            max_iou_calculations_per_class: Maximum IoU calculations per class before optimization
            min_box_size: Minimum box size (width/height) in pixels to consider valid.
                         Boxes smaller than this will be filtered out (default: 1.0).

        Returns:
            ``(class_aps, class_metrics)`` — AP and MMRotate-style stats per class.
        """
        # Collect all unique class names
        all_classes = set()
        for dets in all_detections.values():
            all_classes.update(d.class_name for d in dets)
        for gts in all_ground_truths.values():
            all_classes.update(gt.class_name for gt in gts)

        if self.class_names:
            all_classes = all_classes.intersection(set(self.class_names))

        # Aggregate detections and ground truths by class
        class_aps: Dict[str, float] = {}
        class_metrics: Dict[str, ClassEvalMetrics] = {}
        sorted_classes = sorted(all_classes)
        num_classes = len(sorted_classes)

        # Use progress bar for classes; write to progress_stream so console sees it but train.log does not
        tqdm_file = progress_stream if progress_stream is not None else sys.stderr
        use_pbar = show_progress and tqdm is not None and num_classes > 1
        class_iter = tqdm(sorted_classes, desc="Computing AP per class", unit="class", ncols=120, file=tqdm_file) if use_pbar else sorted_classes

        for class_name in class_iter:
            try:
                class_detections = []
                class_ground_truths = []

                for image_id in set(all_detections.keys()) | set(all_ground_truths.keys()):
                    dets = all_detections.get(image_id, [])
                    gts = all_ground_truths.get(image_id, [])

                    class_detections.extend([d for d in dets if d.class_name == class_name])
                    class_ground_truths.extend([gt for gt in gts if gt.class_name == class_name])

                # Log class statistics
                num_dets = len(class_detections)
                num_gts = len(class_ground_truths)

                # Compute AP for this class (show progress if many detections or ground truths)
                # Show progress if computation will be slow (many IoU calculations)
                total_iou_calculations = num_dets * num_gts
                show_det_progress = total_iou_calculations > 100000 or num_dets > 5000 or num_gts > 1000

                # Warn if too many calculations
                if total_iou_calculations > max_iou_calculations_per_class:
                    warning_msg = (f"  WARNING: {class_name} has {total_iou_calculations:,} IoU calculations "
                                  f"({num_dets:,} dets × {num_gts:,} GTs). This will be slow!")
                    if use_pbar and isinstance(class_iter, tqdm):
                        class_iter.write(warning_msg)
                    else:
                        print(warning_msg)

                # Update progress bar with class info
                if use_pbar and isinstance(class_iter, tqdm):
                    if show_det_progress:
                        class_iter.set_postfix_str(f"{class_name}: {num_dets} dets × {num_gts} GTs = {total_iou_calculations:,} IoUs")
                        class_iter.write(f"  Computing AP for {class_name}: {num_dets:,} dets, {num_gts:,} GTs ({total_iou_calculations:,} IoU calculations)")
                    else:
                        class_iter.set_postfix_str(f"{class_name}: {num_dets} dets, {num_gts} GTs")

                ap, metrics = self.compute_ap(
                    class_detections,
                    class_ground_truths,
                    class_name=class_name,
                    show_progress=show_det_progress,
                    progress_stream=progress_stream,
                    max_iou_calculations=max_iou_calculations_per_class,
                    min_box_size=min_box_size,
                )
                class_aps[class_name] = ap
                class_metrics[class_name] = metrics

                # Update progress bar with current AP if using tqdm
                if use_pbar and isinstance(class_iter, tqdm):
                    class_iter.set_postfix_str(f"{class_name}: AP={ap:.4f}")

            except Exception as e:
                # Log error and continue with other classes
                error_msg = f"Error computing AP for class '{class_name}': {e}"
                if use_pbar and isinstance(class_iter, tqdm):
                    class_iter.write(f"  ERROR: {error_msg}")
                else:
                    print(f"  ERROR: {error_msg}")
                import traceback
                traceback.print_exc()
                # Set AP to 0.0 for failed classes
                class_aps[class_name] = 0.0
                class_metrics[class_name] = ClassEvalMetrics(
                    num_gts=0,
                    num_dets=0,
                    recall=0.0,
                    ap=0.0,
                )
                if use_pbar and isinstance(class_iter, tqdm):
                    class_iter.set_postfix_str(f"{class_name}: ERROR")

        return class_aps, class_metrics

compute_ap(detections, ground_truths, *, class_name=None, show_progress=False, progress_stream=None, max_iou_calculations=None, min_box_size=1.0)

Compute Average Precision for a single class.

Parameters:

Name	Type	Description	Default
detections	Sequence[Detection]	List of detections	required
ground_truths	Sequence[GroundTruth]	List of ground truths	required
class_name	Optional[str]	Optional class name	None
show_progress	bool	Whether to show progress bar	False
max_iou_calculations	Optional[int]	Maximum number of IoU calculations to allow. If exceeded, will use approximate method or return 0.0.	None
min_box_size	float	Minimum box size (width/height) in pixels to consider valid. Boxes smaller than this will be filtered out (default: 1.0).	1.0

Source code in oriented_det/data/evaluation.py

def compute_ap(
    self,
    detections: Sequence[Detection],
    ground_truths: Sequence[GroundTruth],
    *,
    class_name: Optional[str] = None,
    show_progress: bool = False,
    progress_stream: Optional[Any] = None,
    max_iou_calculations: Optional[int] = None,
    min_box_size: float = 1.0,
) -> Tuple[float, ClassEvalMetrics]:
    """Compute Average Precision for a single class.

    Args:
        detections: List of detections
        ground_truths: List of ground truths
        class_name: Optional class name
        show_progress: Whether to show progress bar
        max_iou_calculations: Maximum number of IoU calculations to allow.
                             If exceeded, will use approximate method or return 0.0.
        min_box_size: Minimum box size (width/height) in pixels to consider valid.
                     Boxes smaller than this will be filtered out (default: 1.0).
    """
    if class_name:
        detections = [d for d in detections if d.class_name == class_name]
        ground_truths = [gt for gt in ground_truths if gt.class_name == class_name]

    # Filter out degenerated boxes
    original_det_count = len(detections)
    original_gt_count = len(ground_truths)
    detections = [d for d in detections if d.rbox.is_valid(min_size=min_box_size)]
    ground_truths = [gt for gt in ground_truths if gt.rbox.is_valid(min_size=min_box_size)]

    if show_progress:
        filtered_dets = original_det_count - len(detections)
        filtered_gts = original_gt_count - len(ground_truths)
        if filtered_dets > 0 or filtered_gts > 0:
            filter_msg = f"  Filtered {filtered_dets} degenerated detections, {filtered_gts} degenerated ground truths"
            print(filter_msg)

    if not ground_truths:
        ap = 0.0 if detections else 1.0
        return ap, ClassEvalMetrics(
            num_gts=0,
            num_dets=len(detections),
            recall=0.0,
            ap=ap,
        )

    # Check if we should optimize due to too many calculations
    num_dets = len(detections)
    num_gts = len(ground_truths)
    total_calculations = num_dets * num_gts

    # Use batch IoU if available and calculations are too many
    use_batch_iou = max_iou_calculations and total_calculations > max_iou_calculations

    # Exact polygon mAP (final_nms_use_cpu): never use GPU sampling IoU
    from ..utils import is_gpu_device
    use_gpu = (
        not self.use_exact_rotated_iou
        and torch is not None
        and self.device is not None
        and is_gpu_device(self.device)
    )

    sorted_dets = sorted(detections, key=lambda d: d.score, reverse=True)
    tp_fp_from_batch = False
    tp: List[int] = []
    fp: List[int] = []

    if use_batch_iou:
        # Try using batch IoU computation which is much faster
        try:
            det_rboxes = [d.rbox for d in sorted_dets]
            gt_rboxes = [gt.rbox for gt in ground_truths]

            # Use chunking for very large matrices to avoid memory issues
            # Process detections in chunks to limit memory usage.
            # GPU IoU builds [chunk*S, num_gt] tensors with S ~= default oriented IoU samples
            # (typically 100; tier/env can raise it). When num_gt is large (e.g. 72k)
            # a fixed 5000-detect chunk can OOM. Cap chunk so chunk*S*num_gt <= ~500M elements.
            num_gt = len(ground_truths)
            max_elements = 500_000_000  # ~500M elements for bool tensor (~500 MB)
            samples_per_box = 100  # conservative match to oriented_box_iou_gpu default envelope
            adaptive_chunk = max(1, max_elements // (samples_per_box * max(1, num_gt)))
            CHUNK_SIZE = min(5000, adaptive_chunk)
            num_dets = len(sorted_dets)

            if show_progress and tqdm is not None:
                if self.use_exact_rotated_iou:
                    iou_type = f"CPU exact ({self._exact_iou_backend})"
                else:
                    iou_type = "GPU-accelerated" if use_gpu else "CPU"
                print(f"    Using chunked batch IoU computation ({iou_type}) for {class_name} ({total_calculations:,} calculations)")
                if num_dets > CHUNK_SIZE:
                    print(f"    Processing {num_dets:,} detections in chunks of {CHUNK_SIZE:,}")

            # Process detections in chunks
            gt_matched = [False] * len(ground_truths)
            tp = []
            fp = []

            # Process detections in chunks; tqdm to progress_stream so console sees it but train.log does not
            tqdm_file = progress_stream if progress_stream is not None else sys.stderr
            det_iter = (
                tqdm(range(0, num_dets, CHUNK_SIZE), desc="    Processing chunks", unit="chunk", leave=False, file=tqdm_file)
                if (show_progress and tqdm is not None and num_dets > CHUNK_SIZE)
                else range(0, num_dets, CHUNK_SIZE)
            )

            for chunk_start in det_iter:
                chunk_end = min(chunk_start + CHUNK_SIZE, num_dets)
                chunk_dets = sorted_dets[chunk_start:chunk_end]
                chunk_det_rboxes = det_rboxes[chunk_start:chunk_end]

                # Compute IoU matrix for this chunk
                chunk_iou_matrix = None
                try:
                    if use_gpu:
                        # Use GPU-accelerated IoU (much faster)
                        from ..ops.gpu_ops import oriented_box_iou_gpu
                        from ..ops.utils import rboxes_to_tensor

                        # Convert RBox objects to tensors
                        chunk_det_tensors = rboxes_to_tensor(chunk_det_rboxes, device=self.device)
                        gt_tensors = rboxes_to_tensor(gt_rboxes, device=self.device)

                        # Compute IoU matrix on GPU
                        iou_matrix_gpu = oriented_box_iou_gpu(chunk_det_tensors, gt_tensors)

                        # Convert to CPU list of lists for compatibility
                        chunk_iou_matrix = iou_matrix_gpu.cpu().tolist()
                    else:
                        from ..ops.iou import batch_rbox_iou
                        chunk_iou_matrix = batch_rbox_iou(
                            chunk_det_rboxes,
                            gt_rboxes,
                            device=self.device,
                            intersection_backend=(
                                self._exact_iou_backend
                                if self.use_exact_rotated_iou
                                else "auto"
                            ),
                        )
                except Exception as chunk_e:
                    # If chunk batch fails, fall back to per-detection computation for this chunk
                    if show_progress:
                        print(f"    Chunk batch IoU failed, using per-detection for chunk {chunk_start}-{chunk_end}: {chunk_e}")
                    chunk_iou_matrix = None

                # Process each detection in this chunk
                for local_idx, det in enumerate(chunk_dets):
                    det_idx = chunk_start + local_idx
                    best_iou = 0.0
                    best_gt_idx = -1

                    if chunk_iou_matrix is not None:
                        # Use pre-computed IoU matrix
                        for gt_idx, gt in enumerate(ground_truths):
                            if gt_matched[gt_idx] or gt.class_name != det.class_name:
                                continue
                            if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                                continue
                            det_iou = chunk_iou_matrix[local_idx][gt_idx]
                            if det_iou > best_iou:
                                best_iou = det_iou
                                best_gt_idx = gt_idx
                    else:
                        # Fall back to per-detection IoU computation
                        for gt_idx, gt in enumerate(ground_truths):
                            if gt_matched[gt_idx] or gt.class_name != det.class_name:
                                continue
                            if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                                continue
                            try:
                                from ..ops.iou import rbox_iou
                                det_iou = rbox_iou(
                                    det.rbox,
                                    gt.rbox,
                                    intersection_backend=(
                                        self._exact_iou_backend
                                        if self.use_exact_rotated_iou
                                        else "auto"
                                    ),
                                )
                                if det_iou > best_iou:
                                    best_iou = det_iou
                                    best_gt_idx = gt_idx
                            except Exception:
                                continue

                    # Check if match is valid
                    if best_iou >= self.iou_threshold and best_gt_idx >= 0:
                        gt = ground_truths[best_gt_idx]
                        if gt.difficult == 0:
                            gt_matched[best_gt_idx] = True
                            tp.append(1)
                            fp.append(0)
                        else:
                            tp.append(0)
                            fp.append(0)
                    else:
                        tp.append(0)
                        fp.append(1)

                # Clear chunk matrix from memory
                del chunk_iou_matrix
                if torch is not None and torch.cuda.is_available():
                    torch.cuda.empty_cache()
                elif torch is not None and getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
                    torch.mps.empty_cache()

            tp_fp_from_batch = True

        except Exception as e:
            # Fall back to regular computation if batch fails
            if show_progress:
                print(f"    Batch IoU failed for {class_name}, using regular method (will be slow): {e}")
            use_batch_iou = False

    if not tp_fp_from_batch:
        # Track which ground truths have been matched (standard per-detection path)
        gt_matched = [False] * len(ground_truths)
        tp = []
        fp = []

        # Use progress bar if requested and many detections; write to progress_stream so console sees it but train.log does not
        tqdm_file = progress_stream if progress_stream is not None else sys.stderr
        use_pbar = show_progress and tqdm is not None and len(sorted_dets) > 5000
        det_iter = tqdm(sorted_dets, desc=f"  {class_name or 'All'}", unit="det", leave=False, ncols=100, file=tqdm_file) if use_pbar else sorted_dets

        for det_idx, det in enumerate(det_iter):
            best_iou = 0.0
            best_gt_idx = -1

            # Find best matching ground truth (only from same image when image_id is set)
            for gt_idx, gt in enumerate(ground_truths):
                if gt_matched[gt_idx] or gt.class_name != det.class_name:
                    continue
                if det.image_id is not None and gt.image_id is not None and det.image_id != gt.image_id:
                    continue
                try:
                    det_iou = iou.rbox_iou(
                        det.rbox,
                        gt.rbox,
                        intersection_backend=(
                            self._exact_iou_backend
                            if self.use_exact_rotated_iou
                            else "auto"
                        ),
                    )
                    if det_iou > best_iou:
                        best_iou = det_iou
                        best_gt_idx = gt_idx
                except Exception:
                    # Skip problematic IoU calculations and continue with next GT
                    continue

            # Check if match is valid
            if best_iou >= self.iou_threshold and best_gt_idx >= 0:
                gt = ground_truths[best_gt_idx]
                if gt.difficult == 0:  # Only count non-difficult as TP
                    gt_matched[best_gt_idx] = True
                    tp.append(1)
                    fp.append(0)
                else:
                    tp.append(0)
                    fp.append(0)  # Difficult GTs don't count as FP either
            else:
                tp.append(0)
                fp.append(1)

    # Compute precision and recall
    tp_cumsum = []
    fp_cumsum = []
    cumsum_tp = 0
    cumsum_fp = 0

    for t, f in zip(tp, fp):
        cumsum_tp += t
        cumsum_fp += f
        tp_cumsum.append(cumsum_tp)
        fp_cumsum.append(cumsum_fp)

    num_positives = sum(1 for gt in ground_truths if gt.difficult == 0)
    num_dets = len(sorted_dets)
    if num_positives == 0:
        return 0.0, ClassEvalMetrics(
            num_gts=0,
            num_dets=num_dets,
            recall=0.0,
            ap=0.0,
        )

    recalls = [t / num_positives for t in tp_cumsum]
    precisions = [
        tp_cumsum[i] / (tp_cumsum[i] + fp_cumsum[i]) if (tp_cumsum[i] + fp_cumsum[i]) > 0 else 0.0
        for i in range(len(tp_cumsum))
    ]

    final_recall = recalls[-1] if recalls else 0.0

    # Compute AP using 11-point interpolation
    ap = 0.0
    for t in [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]:
        precisions_at_recall = [p for r, p in zip(recalls, precisions) if r >= t]
        if precisions_at_recall:
            ap += max(precisions_at_recall) / 11.0

    return ap, ClassEvalMetrics(
        num_gts=num_positives,
        num_dets=num_dets,
        recall=final_recall,
        ap=ap,
    )

compute_map(all_detections, all_ground_truths, show_progress=True, progress_stream=None, max_iou_calculations_per_class=10000000, min_box_size=1.0)

Compute mAP across all classes.

Parameters:

Name	Type	Description	Default
all_detections	Dict[str, List[Detection]]	Dictionary mapping image_id -> list of Detection objects	required
all_ground_truths	Dict[str, List[GroundTruth]]	Dictionary mapping image_id -> list of GroundTruth objects	required
show_progress	bool	Whether to show progress bars	True
max_iou_calculations_per_class	int	Maximum IoU calculations per class before optimization	10000000
min_box_size	float	Minimum box size (width/height) in pixels to consider valid. Boxes smaller than this will be filtered out (default: 1.0).	1.0

Returns:

Type	Description
Tuple[Dict[str, float], Dict[str, ClassEvalMetrics]]	(class_aps, class_metrics) — AP and MMRotate-style stats per class.

Source code in oriented_det/data/evaluation.py

def compute_map(
    self,
    all_detections: Dict[str, List[Detection]],  # image_id -> detections
    all_ground_truths: Dict[str, List[GroundTruth]],  # image_id -> ground_truths
    show_progress: bool = True,
    progress_stream: Optional[Any] = None,
    max_iou_calculations_per_class: int = 10_000_000,  # 10M IoU calculations max per class
    min_box_size: float = 1.0,
) -> Tuple[Dict[str, float], Dict[str, ClassEvalMetrics]]:
    """Compute mAP across all classes.

    Args:
        all_detections: Dictionary mapping image_id -> list of Detection objects
        all_ground_truths: Dictionary mapping image_id -> list of GroundTruth objects
        show_progress: Whether to show progress bars
        max_iou_calculations_per_class: Maximum IoU calculations per class before optimization
        min_box_size: Minimum box size (width/height) in pixels to consider valid.
                     Boxes smaller than this will be filtered out (default: 1.0).

    Returns:
        ``(class_aps, class_metrics)`` — AP and MMRotate-style stats per class.
    """
    # Collect all unique class names
    all_classes = set()
    for dets in all_detections.values():
        all_classes.update(d.class_name for d in dets)
    for gts in all_ground_truths.values():
        all_classes.update(gt.class_name for gt in gts)

    if self.class_names:
        all_classes = all_classes.intersection(set(self.class_names))

    # Aggregate detections and ground truths by class
    class_aps: Dict[str, float] = {}
    class_metrics: Dict[str, ClassEvalMetrics] = {}
    sorted_classes = sorted(all_classes)
    num_classes = len(sorted_classes)

    # Use progress bar for classes; write to progress_stream so console sees it but train.log does not
    tqdm_file = progress_stream if progress_stream is not None else sys.stderr
    use_pbar = show_progress and tqdm is not None and num_classes > 1
    class_iter = tqdm(sorted_classes, desc="Computing AP per class", unit="class", ncols=120, file=tqdm_file) if use_pbar else sorted_classes

    for class_name in class_iter:
        try:
            class_detections = []
            class_ground_truths = []

            for image_id in set(all_detections.keys()) | set(all_ground_truths.keys()):
                dets = all_detections.get(image_id, [])
                gts = all_ground_truths.get(image_id, [])

                class_detections.extend([d for d in dets if d.class_name == class_name])
                class_ground_truths.extend([gt for gt in gts if gt.class_name == class_name])

            # Log class statistics
            num_dets = len(class_detections)
            num_gts = len(class_ground_truths)

            # Compute AP for this class (show progress if many detections or ground truths)
            # Show progress if computation will be slow (many IoU calculations)
            total_iou_calculations = num_dets * num_gts
            show_det_progress = total_iou_calculations > 100000 or num_dets > 5000 or num_gts > 1000

            # Warn if too many calculations
            if total_iou_calculations > max_iou_calculations_per_class:
                warning_msg = (f"  WARNING: {class_name} has {total_iou_calculations:,} IoU calculations "
                              f"({num_dets:,} dets × {num_gts:,} GTs). This will be slow!")
                if use_pbar and isinstance(class_iter, tqdm):
                    class_iter.write(warning_msg)
                else:
                    print(warning_msg)

            # Update progress bar with class info
            if use_pbar and isinstance(class_iter, tqdm):
                if show_det_progress:
                    class_iter.set_postfix_str(f"{class_name}: {num_dets} dets × {num_gts} GTs = {total_iou_calculations:,} IoUs")
                    class_iter.write(f"  Computing AP for {class_name}: {num_dets:,} dets, {num_gts:,} GTs ({total_iou_calculations:,} IoU calculations)")
                else:
                    class_iter.set_postfix_str(f"{class_name}: {num_dets} dets, {num_gts} GTs")

            ap, metrics = self.compute_ap(
                class_detections,
                class_ground_truths,
                class_name=class_name,
                show_progress=show_det_progress,
                progress_stream=progress_stream,
                max_iou_calculations=max_iou_calculations_per_class,
                min_box_size=min_box_size,
            )
            class_aps[class_name] = ap
            class_metrics[class_name] = metrics

            # Update progress bar with current AP if using tqdm
            if use_pbar and isinstance(class_iter, tqdm):
                class_iter.set_postfix_str(f"{class_name}: AP={ap:.4f}")

        except Exception as e:
            # Log error and continue with other classes
            error_msg = f"Error computing AP for class '{class_name}': {e}"
            if use_pbar and isinstance(class_iter, tqdm):
                class_iter.write(f"  ERROR: {error_msg}")
            else:
                print(f"  ERROR: {error_msg}")
            import traceback
            traceback.print_exc()
            # Set AP to 0.0 for failed classes
            class_aps[class_name] = 0.0
            class_metrics[class_name] = ClassEvalMetrics(
                num_gts=0,
                num_dets=0,
                recall=0.0,
                ap=0.0,
            )
            if use_pbar and isinstance(class_iter, tqdm):
                class_iter.set_postfix_str(f"{class_name}: ERROR")

    return class_aps, class_metrics

AirbusPlaygroundCSVDataset

CSV-backed Airbus Playground dataset yielding DOTASample objects.

Source code in oriented_det/data/airbus_playground.py

class AirbusPlaygroundCSVDataset:
    """CSV-backed Airbus Playground dataset yielding DOTASample objects."""

    def __init__(
        self,
        data_root: str | Path,
        *,
        split: str,
        annotations_file: str | Path = "annotations.csv",
        split_file: str | Path = "split.csv",
        val_split_id: int = 0,
        train_includes_val: bool = False,
        allowed_classes: Optional[Sequence[str]] = None,
        difficult_strategy: str = "drop",
        ignore_labels: Optional[Sequence[str]] = None,
        map_labels: Optional[Dict[str, str]] = None,
        filter_empty_gt: bool = False,
    ):
        if split not in {"train", "val"}:
            raise ValueError(f"Unsupported split '{split}'. Expected 'train' or 'val'.")

        if val_split_id < 0:
            raise ValueError(f"val_split_id must be non-negative; got {val_split_id}")

        ds = (difficult_strategy or "drop").strip().lower()
        if ds not in {"drop", "ignore", "keep"}:
            raise ValueError(
                f"Invalid difficult_strategy={difficult_strategy!r}; expected 'drop', 'ignore', or 'keep'."
            )

        self.data_root = Path(data_root)
        self.split = split
        self.val_split_id = int(val_split_id)
        self.train_includes_val = bool(train_includes_val)
        self.allowed_classes = set(allowed_classes) if allowed_classes is not None else None
        self.difficult_strategy = ds
        self.ignore_labels = set(ignore_labels or [])
        self.map_labels = dict(map_labels or {})
        self.filter_empty_gt = bool(filter_empty_gt)

        self.annotations_path = _resolve_csv_path(self.data_root, annotations_file)
        self.split_path = _resolve_csv_path(self.data_root, split_file)
        if not self.annotations_path.exists():
            raise FileNotFoundError(f"Annotations CSV not found: {self.annotations_path}")
        if not self.split_path.exists():
            raise FileNotFoundError(f"Split CSV not found: {self.split_path}")

        self._samples_meta = self._load_split_rows()
        self._tiles_before_empty_filter = len(self._samples_meta)
        self._annotations_by_tile = self._load_annotations_rows()
        if self.filter_empty_gt:
            self._samples_meta = [
                row
                for row in self._samples_meta
                if len(self._annotations_by_tile.get(row.get("tile_relpath", ""), [])) > 0
            ]
        self._empty_gt_filtered_count = self._tiles_before_empty_filter - len(self._samples_meta)

    @property
    def tiles_discovered_count(self) -> int:
        """Tiles in split CSV before ``filter_empty_gt`` (if enabled)."""
        return self._tiles_before_empty_filter

    @property
    def empty_gt_filtered_count(self) -> int:
        """Tiles removed by ``filter_empty_gt`` at init."""
        return self._empty_gt_filtered_count

    def _load_split_rows(self) -> List[Dict[str, str]]:
        with self.split_path.open("r", encoding="utf-8", newline="") as f:
            reader = csv.DictReader(f)
            all_rows = list(reader)
        if not all_rows:
            return []

        mode = detect_airbus_split_csv_format([r.get("split", "") for r in all_rows])
        want_val = self.split == "val"
        include_all_for_train = self.split == "train" and self.train_includes_val
        rows: List[Dict[str, str]] = []
        for row in all_rows:
            raw = row.get("split", "")
            if include_all_for_train:
                rows.append(row)
                continue
            if mode == "train_val":
                if raw.strip().lower() == self.split:
                    rows.append(row)
            else:
                try:
                    sid = int(str(raw).strip())
                except ValueError:
                    continue
                is_val = sid == self.val_split_id
                if is_val == want_val:
                    rows.append(row)

        # Stable order for reproducibility.
        rows.sort(key=lambda row: row.get("tile_relpath", ""))
        return rows

    def _load_annotations_rows(self) -> Dict[str, List[DOTAAnnotation]]:
        selected_tiles = {row.get("tile_relpath", "") for row in self._samples_meta}
        annotations_by_tile: Dict[str, List[DOTAAnnotation]] = defaultdict(list)

        with self.annotations_path.open("r", encoding="utf-8", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                tile_relpath = row.get("tile_relpath", "")
                if tile_relpath not in selected_tiles:
                    continue

                class_name = str(row.get("class_name", "")).strip()
                if not class_name:
                    continue
                if class_name in self.ignore_labels:
                    continue
                class_name = self.map_labels.get(class_name, class_name)
                if self.allowed_classes is not None and class_name not in self.allowed_classes:
                    continue

                difficult = int(str(row.get("difficult", "0")).strip() or "0")
                if self.difficult_strategy == "drop" and difficult == 1:
                    continue

                coords = str(row.get("dota_coords", "")).strip()
                if not coords:
                    continue
                line = f"{coords} {class_name} {difficult}"
                try:
                    ann = DOTAAnnotation.from_line(line)
                except Exception:
                    continue
                annotations_by_tile[tile_relpath].append(ann)

        return annotations_by_tile

    def get_raw_class_names_from_csv(self) -> List[str]:
        """Return unique class_name values in the CSV for the current split (before map_labels/ignore_labels).
        Useful to verify CSV content and that map_labels keys match."""
        selected_tiles = {row.get("tile_relpath", "") for row in self._samples_meta}
        raw: Set[str] = set()
        with self.annotations_path.open("r", encoding="utf-8", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                if row.get("tile_relpath", "") not in selected_tiles:
                    continue
                cn = str(row.get("class_name", "")).strip()
                if cn:
                    raw.add(cn)
        return sorted(raw)

    def _load_image_size(self, image_path: Path) -> Tuple[int, int]:
        _require_pillow()
        with Image.open(image_path) as img:
            return img.size

    def __len__(self) -> int:
        return len(self._samples_meta)

    def __getitem__(self, idx: int) -> DOTASample:
        if idx < 0 or idx >= len(self):
            raise IndexError(f"Index {idx} out of range for dataset of size {len(self)}")

        row = self._samples_meta[idx]
        tile_relpath = row.get("tile_relpath", "")
        if not tile_relpath:
            raise ValueError("split.csv row is missing tile_relpath")

        image_path = self.data_root / tile_relpath
        if not image_path.exists():
            raise FileNotFoundError(f"Image not found: {image_path}")

        width, height = self._load_image_size(image_path)
        annotations = tuple(self._annotations_by_tile.get(tile_relpath, []))
        return DOTASample(
            image_path=image_path,
            width=width,
            height=height,
            annotations=annotations,
        )

    def __iter__(self) -> Iterable[DOTASample]:
        for idx in range(len(self)):
            yield self[idx]

    def get_class_names(self) -> List[str]:
        classes = set()
        for annotations in self._annotations_by_tile.values():
            for ann in annotations:
                classes.add(ann.class_name)
        return sorted(classes)

empty_gt_filtered_count property

Tiles removed by filter_empty_gt at init.

tiles_discovered_count property

Tiles in split CSV before filter_empty_gt (if enabled).

get_raw_class_names_from_csv()

Return unique class_name values in the CSV for the current split (before map_labels/ignore_labels). Useful to verify CSV content and that map_labels keys match.

Source code in oriented_det/data/airbus_playground.py

def get_raw_class_names_from_csv(self) -> List[str]:
    """Return unique class_name values in the CSV for the current split (before map_labels/ignore_labels).
    Useful to verify CSV content and that map_labels keys match."""
    selected_tiles = {row.get("tile_relpath", "") for row in self._samples_meta}
    raw: Set[str] = set()
    with self.annotations_path.open("r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row.get("tile_relpath", "") not in selected_tiles:
                continue
            cn = str(row.get("class_name", "")).strip()
            if cn:
                raw.add(cn)
    return sorted(raw)

AlbumentationsTransform

Wrapper for albumentations transforms that only apply non-geometric augmentations.

This transform applies albumentations augmentations that do not modify the spatial layout of the image (no rotation, scaling, translation, etc.) since albumentations does not support oriented bounding boxes. Only color, contrast, blur, noise, and similar augmentations are applied.

Parameters:

Name	Type	Description	Default
transform		An albumentations Compose transform containing only non-geometric augmentations. The transform should work on numpy arrays (H, W, C).	required
p	float	Probability of applying the transform (default: 1.0)	1.0

Example

import albumentations as A from oriented_det.data import AlbumentationsTransform

Create non-geometric augmentation pipeline

aug = A.Compose([ ... A.RandomBrightnessContrast(p=0.5), ... A.RandomGamma(p=0.3), ... A.GaussNoise(p=0.2), ... A.GaussianBlur(p=0.2), ... A.CLAHE(p=0.3), ... ])

transform = AlbumentationsTransform(aug) augmented_image = transform(image) # PIL Image or numpy array

Source code in oriented_det/data/transforms.py

class AlbumentationsTransform:
    """Wrapper for albumentations transforms that only apply non-geometric augmentations.

    This transform applies albumentations augmentations that do not modify the spatial
    layout of the image (no rotation, scaling, translation, etc.) since albumentations
    does not support oriented bounding boxes. Only color, contrast, blur, noise, and
    similar augmentations are applied.

    Args:
        transform: An albumentations Compose transform containing only non-geometric
            augmentations. The transform should work on numpy arrays (H, W, C).
        p: Probability of applying the transform (default: 1.0)

    Example:
        >>> import albumentations as A
        >>> from oriented_det.data import AlbumentationsTransform
        >>> 
        >>> # Create non-geometric augmentation pipeline
        >>> aug = A.Compose([
        ...     A.RandomBrightnessContrast(p=0.5),
        ...     A.RandomGamma(p=0.3),
        ...     A.GaussNoise(p=0.2),
        ...     A.GaussianBlur(p=0.2),
        ...     A.CLAHE(p=0.3),
        ... ])
        >>> 
        >>> transform = AlbumentationsTransform(aug)
        >>> augmented_image = transform(image)  # PIL Image or numpy array
    """

    def __init__(self, transform, p: float = 1.0):
        if A is None:
            raise RuntimeError("albumentations is required. Install with: pip install albumentations")
        if not 0 <= p <= 1:
            raise ValueError("Probability p must be in [0, 1]")
        self.transform = transform
        self.p = p

    def __call__(self, image) -> any:  # type: ignore
        """Apply albumentations transform to image.

        Args:
            image: PIL Image or numpy array (H, W, C) in RGB format

        Returns:
            Transformed image in the same format as input
        """
        if random.random() >= self.p:
            return image

        # Convert PIL Image to numpy array if needed
        is_pil = False
        if Image is not None and isinstance(image, Image.Image):
            is_pil = True
            import numpy as np
            # Convert to RGB if RGBA or other format
            if image.mode != "RGB":
                image = image.convert("RGB")
            image_array = np.array(image)
        else:
            import numpy as np
            image_array = np.asarray(image)
            # Handle RGBA numpy arrays (4 channels -> 3 channels)
            if image_array.ndim == 3 and image_array.shape[2] == 4:
                # Drop alpha channel
                image_array = image_array[:, :, :3]

        # Ensure image is in (H, W, C) format with uint8 dtype
        if image_array.dtype != np.uint8:
            # Normalize to [0, 255] if in [0, 1] range
            if image_array.max() <= 1.0:
                image_array = (image_array * 255).astype(np.uint8)
            else:
                image_array = image_array.astype(np.uint8)

        # Apply albumentations transform
        # Note: albumentations expects numpy array in (H, W, C) format
        transformed = self.transform(image=image_array)["image"]

        # Convert back to PIL Image if input was PIL
        if is_pil:
            return Image.fromarray(transformed, "RGB")

        return transformed

__call__(image)

Apply albumentations transform to image.

Parameters:

Name	Type	Description	Default
image		PIL Image or numpy array (H, W, C) in RGB format	required

Returns:

Type	Description
any	Transformed image in the same format as input

Source code in oriented_det/data/transforms.py

def __call__(self, image) -> any:  # type: ignore
    """Apply albumentations transform to image.

    Args:
        image: PIL Image or numpy array (H, W, C) in RGB format

    Returns:
        Transformed image in the same format as input
    """
    if random.random() >= self.p:
        return image

    # Convert PIL Image to numpy array if needed
    is_pil = False
    if Image is not None and isinstance(image, Image.Image):
        is_pil = True
        import numpy as np
        # Convert to RGB if RGBA or other format
        if image.mode != "RGB":
            image = image.convert("RGB")
        image_array = np.array(image)
    else:
        import numpy as np
        image_array = np.asarray(image)
        # Handle RGBA numpy arrays (4 channels -> 3 channels)
        if image_array.ndim == 3 and image_array.shape[2] == 4:
            # Drop alpha channel
            image_array = image_array[:, :, :3]

    # Ensure image is in (H, W, C) format with uint8 dtype
    if image_array.dtype != np.uint8:
        # Normalize to [0, 255] if in [0, 1] range
        if image_array.max() <= 1.0:
            image_array = (image_array * 255).astype(np.uint8)
        else:
            image_array = image_array.astype(np.uint8)

    # Apply albumentations transform
    # Note: albumentations expects numpy array in (H, W, C) format
    transformed = self.transform(image=image_array)["image"]

    # Convert back to PIL Image if input was PIL
    if is_pil:
        return Image.fromarray(transformed, "RGB")

    return transformed

ClassEvalMetrics dataclass

Per-class detection stats (MMRotate-style eval table).

Source code in oriented_det/data/evaluation.py

@dataclass(frozen=True)
class ClassEvalMetrics:
    """Per-class detection stats (MMRotate-style eval table)."""

    num_gts: int
    num_dets: int
    recall: float
    ap: float

ClassGtBestIouMetrics dataclass

Per-class GT alignment: max IoU vs raw detections (before score filter).

Source code in oriented_det/data/evaluation.py

@dataclass(frozen=True)
class ClassGtBestIouMetrics:
    """Per-class GT alignment: max IoU vs raw detections (before score filter)."""

    num_gts: int
    mean_best_iou_any: float
    mean_best_iou_same_class: float
    median_best_iou_same_class: float

Compose

Compose multiple transforms.

Source code in oriented_det/data/transforms.py

class Compose:
    """Compose multiple transforms."""

    def __init__(self, transforms: Sequence[OrientedTransform]):
        self.transforms = transforms

    def __call__(self, image, rboxes: Sequence[RBox], image_width: int, image_height: int) -> Tuple[any, List[RBox]]:  # type: ignore
        current_image = image
        current_boxes = list(rboxes)

        for transform in self.transforms:
            current_image, current_boxes = transform(
                current_image, current_boxes, image_width, image_height
            )

        return current_image, current_boxes

DOTAAnnotation dataclass

Single DOTA annotation entry.

Source code in oriented_det/data/dota.py

@dataclass(frozen=True)
class DOTAAnnotation:
    """Single DOTA annotation entry."""

    class_name: str
    difficult: int  # 0 or 1
    polygon: Polygon
    rbox: RBox

    @classmethod
    def from_line(cls, line: str, *, class_map: Optional[Dict[str, int]] = None) -> "DOTAAnnotation":
        """Parse a single DOTA annotation line.

        DOTA Polygon Format (official default):
        ---------------------------------------
        Official DOTA (captain-whu.github.io) uses comma-separated:
          "x1, y1, x2, y2, x3, y3, x4, y4, category, difficult"
        We produce this format by default (see format_dota_line, to_line).
        For reading, we also accept space-separated:
          "x1 y1 x2 y2 x3 y3 x4 y4 class_name difficult"
        (class = parts[8:-1] for multi-word names).

        Corner Order Convention:
        - The 4 corners are ordered sequentially around the polygon perimeter
        - Typically, corners follow a consistent winding order (clockwise or counter-clockwise)
        - The first corner (x1, y1) is usually the top-left or top-most point
        - Subsequent corners follow the polygon boundary

        Conversion to QBox/RBox:
        - The polygon is first converted to a QBox, which normalizes the point order
        - QBox ensures counter-clockwise orientation and orders points starting from top-most
        - RBox is then derived from QBox, computing center, dimensions, and angle

        Args:
            line: DOTA annotation line
            class_map: Optional mapping from class names to IDs

        Returns:
            DOTAAnnotation with parsed polygon and RBox

        Example:
            >>> line = "100 200 300 200 300 400 100 400 plane 0"
            >>> ann = DOTAAnnotation.from_line(line)
            >>> # Creates a rectangle with corners at (100,200), (300,200), (300,400), (100,400)
        """
        raw = line.strip()
        if "," in raw:
            # Official DOTA: x1, y1, x2, y2, x3, y3, x4, y4, category, difficult (10 fields)
            parts = [p.strip() for p in raw.split(",")]
            if len(parts) < 10:
                raise ValueError(f"Invalid DOTA annotation line (comma): {line}")
            coords = [float(parts[i]) for i in range(8)]
            class_name = parts[8].strip()
            difficult = int(parts[9])
        else:
            # Space-separated: allows multi-word class (e.g. "General Cargo")
            parts = raw.split()
            if len(parts) < 9:
                raise ValueError(f"Invalid DOTA annotation line: {line}")
            coords = [float(parts[i]) for i in range(8)]
            class_name = " ".join(parts[8:-1]) if len(parts) > 9 else parts[8]
            difficult = int(parts[-1])

        # Extract 4 corner points: (x1,y1), (x2,y2), (x3,y3), (x4,y4)
        points = [(coords[i], coords[i + 1]) for i in range(0, 8, 2)]

        # Convert to Polygon (validates and normalizes orientation)
        polygon = Polygon(points)

        # Convert to RBox via QBox (QBox normalizes point order)
        # QBox ensures counter-clockwise order and starts from top-most point
        rbox = transforms.polygon_to_rbox(polygon)

        return cls(class_name=class_name, difficult=difficult, polygon=polygon, rbox=rbox)

    def to_line(self) -> str:
        """Serialize to official DOTA format (comma-separated)."""
        coords = []
        for p in self.polygon:
            coords.extend(p)
        return format_dota_line(*coords, self.class_name, self.difficult)

from_line(line, *, class_map=None) classmethod

Parse a single DOTA annotation line.

DOTA Polygon Format (official default):

Official DOTA (captain-whu.github.io) uses comma-separated: "x1, y1, x2, y2, x3, y3, x4, y4, category, difficult" We produce this format by default (see format_dota_line, to_line). For reading, we also accept space-separated: "x1 y1 x2 y2 x3 y3 x4 y4 class_name difficult" (class = parts[8:-1] for multi-word names).

Corner Order Convention: - The 4 corners are ordered sequentially around the polygon perimeter - Typically, corners follow a consistent winding order (clockwise or counter-clockwise) - The first corner (x1, y1) is usually the top-left or top-most point - Subsequent corners follow the polygon boundary

Conversion to QBox/RBox: - The polygon is first converted to a QBox, which normalizes the point order - QBox ensures counter-clockwise orientation and orders points starting from top-most - RBox is then derived from QBox, computing center, dimensions, and angle

Parameters:

Name	Type	Description	Default
line	str	DOTA annotation line	required
class_map	Optional[Dict[str, int]]	Optional mapping from class names to IDs	None

Returns:

Type	Description
'DOTAAnnotation'	DOTAAnnotation with parsed polygon and RBox

Example

line = "100 200 300 200 300 400 100 400 plane 0" ann = DOTAAnnotation.from_line(line)

Creates a rectangle with corners at (100,200), (300,200), (300,400), (100,400)

Source code in oriented_det/data/dota.py

@classmethod
def from_line(cls, line: str, *, class_map: Optional[Dict[str, int]] = None) -> "DOTAAnnotation":
    """Parse a single DOTA annotation line.

    DOTA Polygon Format (official default):
    ---------------------------------------
    Official DOTA (captain-whu.github.io) uses comma-separated:
      "x1, y1, x2, y2, x3, y3, x4, y4, category, difficult"
    We produce this format by default (see format_dota_line, to_line).
    For reading, we also accept space-separated:
      "x1 y1 x2 y2 x3 y3 x4 y4 class_name difficult"
    (class = parts[8:-1] for multi-word names).

    Corner Order Convention:
    - The 4 corners are ordered sequentially around the polygon perimeter
    - Typically, corners follow a consistent winding order (clockwise or counter-clockwise)
    - The first corner (x1, y1) is usually the top-left or top-most point
    - Subsequent corners follow the polygon boundary

    Conversion to QBox/RBox:
    - The polygon is first converted to a QBox, which normalizes the point order
    - QBox ensures counter-clockwise orientation and orders points starting from top-most
    - RBox is then derived from QBox, computing center, dimensions, and angle

    Args:
        line: DOTA annotation line
        class_map: Optional mapping from class names to IDs

    Returns:
        DOTAAnnotation with parsed polygon and RBox

    Example:
        >>> line = "100 200 300 200 300 400 100 400 plane 0"
        >>> ann = DOTAAnnotation.from_line(line)
        >>> # Creates a rectangle with corners at (100,200), (300,200), (300,400), (100,400)
    """
    raw = line.strip()
    if "," in raw:
        # Official DOTA: x1, y1, x2, y2, x3, y3, x4, y4, category, difficult (10 fields)
        parts = [p.strip() for p in raw.split(",")]
        if len(parts) < 10:
            raise ValueError(f"Invalid DOTA annotation line (comma): {line}")
        coords = [float(parts[i]) for i in range(8)]
        class_name = parts[8].strip()
        difficult = int(parts[9])
    else:
        # Space-separated: allows multi-word class (e.g. "General Cargo")
        parts = raw.split()
        if len(parts) < 9:
            raise ValueError(f"Invalid DOTA annotation line: {line}")
        coords = [float(parts[i]) for i in range(8)]
        class_name = " ".join(parts[8:-1]) if len(parts) > 9 else parts[8]
        difficult = int(parts[-1])

    # Extract 4 corner points: (x1,y1), (x2,y2), (x3,y3), (x4,y4)
    points = [(coords[i], coords[i + 1]) for i in range(0, 8, 2)]

    # Convert to Polygon (validates and normalizes orientation)
    polygon = Polygon(points)

    # Convert to RBox via QBox (QBox normalizes point order)
    # QBox ensures counter-clockwise order and starts from top-most point
    rbox = transforms.polygon_to_rbox(polygon)

    return cls(class_name=class_name, difficult=difficult, polygon=polygon, rbox=rbox)

to_line()

Serialize to official DOTA format (comma-separated).

Source code in oriented_det/data/dota.py

def to_line(self) -> str:
    """Serialize to official DOTA format (comma-separated)."""
    coords = []
    for p in self.polygon:
        coords.extend(p)
    return format_dota_line(*coords, self.class_name, self.difficult)

DOTADataset

Efficient DOTA dataset loader with lazy parsing.

Source code in oriented_det/data/dota.py

class DOTADataset:
    """Efficient DOTA dataset loader with lazy parsing."""

    def __init__(
        self,
        root_dir: str | Path,
        *,
        split: str = "train",
        split_file: Optional[str | Path] = None,
        label_dir: Optional[str | Path] = None,
        image_dir: Optional[str | Path] = None,
        class_map: Optional[Dict[str, int]] = None,
        allowed_classes: Optional[Sequence[str]] = None,
        ignore_labels: Optional[Sequence[str]] = None,
        # "drop" -> remove at read-time, "ignore"/"keep" -> keep in sample.annotations.
        difficult_strategy: str = "drop",
        filter_empty_gt: bool = False,
    ):
        """Initialize DOTA dataset.

        Args:
            root_dir: Root directory of DOTA dataset (used as base if label_dir/image_dir not specified)
            split: Split name ("train", "val", "test") - used for pattern matching
            split_file: Optional path to a file listing image names (one per line).
                       If provided, only images listed in this file will be used.
                       This follows the official DOTA dataset convention where
                       train.txt, val.txt, test.txt list the image names.
            label_dir: Optional custom path to label directory. If None, uses root_dir/labelTxt.
                      This allows having train/val/test in separate folders.
                      Can be the same as image_dir to read images and annotations from one folder.
            image_dir: Optional custom path to image directory. If None, uses root_dir/images.
                      This allows having train/val/test in separate folders.
                      Can be the same as label_dir to read images and annotations from one folder.
            class_map: Optional mapping from class names to numeric IDs
            allowed_classes: Optional list of allowed class names (whitelist)
            ignore_labels: Optional list of class names to exclude (blacklist)
            difficult_strategy: How to handle difficult annotations: drop | ignore | keep.
            filter_empty_gt: If True, drop tiles whose effective GT count is zero after
                difficult_strategy, allowed_classes, and ignore_labels (MMRotate DOTADataset parity).
        """
        self.root_dir = Path(root_dir)
        self.filter_empty_gt = bool(filter_empty_gt)
        self.split = split
        self.split_file = Path(split_file) if split_file else None
        self.class_map = class_map or {}
        self.allowed_classes = allowed_classes
        self.ignore_labels = list(ignore_labels) if ignore_labels else None
        # Normalize difficult strategy
        ds = (difficult_strategy or "drop").strip().lower()
        if ds not in {"drop", "ignore", "keep"}:
            raise ValueError(f"Invalid difficult_strategy={difficult_strategy!r}; expected 'drop', 'ignore', or 'keep'.")
        self.difficult_strategy = ds
        # Read-time drop of difficult=1 (only when strategy is "drop")
        self._drop_difficult = self.difficult_strategy == "drop"

        # Support custom label and image directories for split-specific folders
        if label_dir is not None:
            self.label_dir = Path(label_dir)
        else:
            self.label_dir = self.root_dir / "labelTxt"

        if image_dir is not None:
            self.image_dir = Path(image_dir)
        else:
            self.image_dir = self.root_dir / "images"

        if not self.label_dir.exists():
            raise FileNotFoundError(f"DOTA label directory not found: {self.label_dir}")
        if not self.image_dir.exists():
            raise FileNotFoundError(f"DOTA image directory not found: {self.image_dir}")

        discovered = self._discover_annotation_files()
        self._annotation_files_discovered_count = len(discovered)
        if self.filter_empty_gt:
            self._annotation_files = [
                ann_path
                for ann_path in discovered
                if self._effective_gt_count(ann_path) > 0
            ]
            self._empty_gt_filtered_count = (
                self._annotation_files_discovered_count - len(self._annotation_files)
            )
        else:
            self._annotation_files = discovered
            self._empty_gt_filtered_count = 0

    @property
    def annotation_files_discovered_count(self) -> int:
        """Label files found before ``filter_empty_gt`` (if enabled)."""
        return self._annotation_files_discovered_count

    @property
    def empty_gt_filtered_count(self) -> int:
        """Tiles removed by ``filter_empty_gt`` at init."""
        return self._empty_gt_filtered_count

    def _discover_annotation_files(self) -> List[Path]:
        """Discover annotation files for the split.

        Two modes:
        1. If split_file is provided: Read image names from file and find corresponding
           annotation files (e.g., train.txt lists "P0001", finds "P0001_train.txt")
        2. Otherwise: Pattern match on annotation filenames (e.g., "*_train.txt")

        Returns:
            Sorted list of annotation file paths
        """
        if self.split_file is not None:
            # Mode 1: Use split file (official DOTA convention)
            split_path = Path(self.split_file)
            if not split_path.is_absolute():
                # Try relative to root_dir
                split_path = self.root_dir / split_path

            if not split_path.exists():
                raise FileNotFoundError(
                    f"Split file not found: {self.split_file} "
                    f"(tried {split_path})"
                )

            # Read image names from split file
            image_names = set()
            with split_path.open("r", encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if line:
                        # Remove extension if present
                        name = Path(line).stem
                        image_names.add(name)

            # Find corresponding annotation files
            annotation_files = []
            for img_name in image_names:
                # Try different annotation file naming conventions
                for suffix in [f"_{self.split}.txt", ".txt"]:
                    ann_file = self.label_dir / f"{img_name}{suffix}"
                    if ann_file.exists():
                        annotation_files.append(ann_file)
                        break

            return sorted(annotation_files)
        else:
            # Mode 2: Pattern matching (backward compatible)
            # First try pattern with split suffix (e.g., "*_train.txt")
            pattern = re.compile(rf".*_{self.split}\.txt$", re.IGNORECASE)
            files = [f for f in self.label_dir.iterdir() if f.is_file() and pattern.match(f.name)]

            # If no files found with split suffix, fall back to all .txt files
            # This supports cases where annotation files have the same name as images
            if not files:
                files = [f for f in self.label_dir.iterdir() if f.is_file() and f.suffix.lower() == ".txt"]

            return sorted(files)

    def _load_image_size(self, image_path: Path) -> Tuple[int, int]:
        """Load image dimensions efficiently."""
        if Image is None:
            raise RuntimeError("PIL/Pillow is required to load image dimensions.")
        with Image.open(image_path) as img:
            return img.size  # Returns (width, height)

    def _parse_annotation_lines(self, ann_path: Path) -> Tuple[DOTAAnnotation, ...]:
        """Parse object lines from a DOTA label file (no image I/O)."""
        annotations: List[DOTAAnnotation] = []
        with ann_path.open("r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("imagesource") or line.startswith("gsd"):
                    continue
                try:
                    annotations.append(
                        DOTAAnnotation.from_line(line, class_map=self.class_map)
                    )
                except (ValueError, Exception):
                    continue
        return tuple(annotations)

    def _effective_gt_count(self, ann_path: Path) -> int:
        """GT count after the same filters applied when loading a sample."""
        annotations = self._parse_annotation_lines(ann_path)
        if not annotations:
            return 0
        if (
            self.allowed_classes is not None
            or self.ignore_labels is not None
            or self._drop_difficult
        ):
            sample = DOTASample(
                image_path=ann_path,
                width=0,
                height=0,
                annotations=annotations,
            )
            sample = sample.filter_by_class(
                allowed_classes=self.allowed_classes,
                ignore_labels=self.ignore_labels,
                drop_difficult=self._drop_difficult,
            )
            return len(sample.annotations)
        return len(annotations)

    def _parse_annotation_file(self, ann_path: Path) -> Optional[DOTASample]:
        """Parse a single DOTA annotation file.

        Returns:
            DOTASample if image exists, None if image is missing (with warning printed)
        """
        image_name = ann_path.stem + ".png"
        image_path = self.image_dir / image_name

        if not image_path.exists():
            image_name = ann_path.stem + ".jpg"
            image_path = self.image_dir / image_name

        if not image_path.exists():
            warnings.warn(f"Image not found for annotation: {ann_path}. Skipping this sample.", UserWarning)
            return None

        width, height = self._load_image_size(image_path)
        annotations = self._parse_annotation_lines(ann_path)

        sample = DOTASample(
            image_path=image_path,
            width=width,
            height=height,
            annotations=annotations,
        )

        if self.allowed_classes is not None or self.ignore_labels is not None or self._drop_difficult:
            sample = sample.filter_by_class(
                allowed_classes=self.allowed_classes,
                ignore_labels=self.ignore_labels,
                drop_difficult=self._drop_difficult,
            )

        return sample

    def __len__(self) -> int:
        return len(self._annotation_files)

    def __getitem__(self, idx: int) -> DOTASample:
        if idx < 0 or idx >= len(self._annotation_files):
            raise IndexError(f"Index {idx} out of range for dataset of size {len(self)}")
        ann_path = self._annotation_files[idx]
        sample = self._parse_annotation_file(ann_path)
        if sample is None:
            # Image was missing (warning already printed)
            # Raise error for indexed access since we can't return the requested item
            raise FileNotFoundError(
                f"Image not found for annotation at index {idx}: {ann_path}. "
                f"Use iteration (for sample in dataset) to automatically skip missing images."
            )
        return sample

    def __iter__(self) -> Iterator[DOTASample]:
        for ann_path in self._annotation_files:
            sample = self._parse_annotation_file(ann_path)
            if sample is not None:
                yield sample

    def get_class_names(self) -> List[str]:
        """Extract unique class names from all annotations."""
        classes = set()
        for sample in self:
            for ann in sample.annotations:
                classes.add(ann.class_name)
        return sorted(classes)

annotation_files_discovered_count property

Label files found before filter_empty_gt (if enabled).

empty_gt_filtered_count property

Tiles removed by filter_empty_gt at init.

__init__(root_dir, *, split='train', split_file=None, label_dir=None, image_dir=None, class_map=None, allowed_classes=None, ignore_labels=None, difficult_strategy='drop', filter_empty_gt=False)

Initialize DOTA dataset.

Parameters:

Name	Type	Description	Default
root_dir	str | Path	Root directory of DOTA dataset (used as base if label_dir/image_dir not specified)	required
split	str	Split name ("train", "val", "test") - used for pattern matching	'train'
split_file	Optional[str | Path]	Optional path to a file listing image names (one per line). If provided, only images listed in this file will be used. This follows the official DOTA dataset convention where train.txt, val.txt, test.txt list the image names.	None
label_dir	Optional[str | Path]	Optional custom path to label directory. If None, uses root_dir/labelTxt. This allows having train/val/test in separate folders. Can be the same as image_dir to read images and annotations from one folder.	None
image_dir	Optional[str | Path]	Optional custom path to image directory. If None, uses root_dir/images. This allows having train/val/test in separate folders. Can be the same as label_dir to read images and annotations from one folder.	None
class_map	Optional[Dict[str, int]]	Optional mapping from class names to numeric IDs	None
allowed_classes	Optional[Sequence[str]]	Optional list of allowed class names (whitelist)	None
ignore_labels	Optional[Sequence[str]]	Optional list of class names to exclude (blacklist)	None
difficult_strategy	str	How to handle difficult annotations: drop | ignore | keep.	'drop'
filter_empty_gt	bool	If True, drop tiles whose effective GT count is zero after difficult_strategy, allowed_classes, and ignore_labels (MMRotate DOTADataset parity).	False

Source code in oriented_det/data/dota.py

def __init__(
    self,
    root_dir: str | Path,
    *,
    split: str = "train",
    split_file: Optional[str | Path] = None,
    label_dir: Optional[str | Path] = None,
    image_dir: Optional[str | Path] = None,
    class_map: Optional[Dict[str, int]] = None,
    allowed_classes: Optional[Sequence[str]] = None,
    ignore_labels: Optional[Sequence[str]] = None,
    # "drop" -> remove at read-time, "ignore"/"keep" -> keep in sample.annotations.
    difficult_strategy: str = "drop",
    filter_empty_gt: bool = False,
):
    """Initialize DOTA dataset.

    Args:
        root_dir: Root directory of DOTA dataset (used as base if label_dir/image_dir not specified)
        split: Split name ("train", "val", "test") - used for pattern matching
        split_file: Optional path to a file listing image names (one per line).
                   If provided, only images listed in this file will be used.
                   This follows the official DOTA dataset convention where
                   train.txt, val.txt, test.txt list the image names.
        label_dir: Optional custom path to label directory. If None, uses root_dir/labelTxt.
                  This allows having train/val/test in separate folders.
                  Can be the same as image_dir to read images and annotations from one folder.
        image_dir: Optional custom path to image directory. If None, uses root_dir/images.
                  This allows having train/val/test in separate folders.
                  Can be the same as label_dir to read images and annotations from one folder.
        class_map: Optional mapping from class names to numeric IDs
        allowed_classes: Optional list of allowed class names (whitelist)
        ignore_labels: Optional list of class names to exclude (blacklist)
        difficult_strategy: How to handle difficult annotations: drop | ignore | keep.
        filter_empty_gt: If True, drop tiles whose effective GT count is zero after
            difficult_strategy, allowed_classes, and ignore_labels (MMRotate DOTADataset parity).
    """
    self.root_dir = Path(root_dir)
    self.filter_empty_gt = bool(filter_empty_gt)
    self.split = split
    self.split_file = Path(split_file) if split_file else None
    self.class_map = class_map or {}
    self.allowed_classes = allowed_classes
    self.ignore_labels = list(ignore_labels) if ignore_labels else None
    # Normalize difficult strategy
    ds = (difficult_strategy or "drop").strip().lower()
    if ds not in {"drop", "ignore", "keep"}:
        raise ValueError(f"Invalid difficult_strategy={difficult_strategy!r}; expected 'drop', 'ignore', or 'keep'.")
    self.difficult_strategy = ds
    # Read-time drop of difficult=1 (only when strategy is "drop")
    self._drop_difficult = self.difficult_strategy == "drop"

    # Support custom label and image directories for split-specific folders
    if label_dir is not None:
        self.label_dir = Path(label_dir)
    else:
        self.label_dir = self.root_dir / "labelTxt"

    if image_dir is not None:
        self.image_dir = Path(image_dir)
    else:
        self.image_dir = self.root_dir / "images"

    if not self.label_dir.exists():
        raise FileNotFoundError(f"DOTA label directory not found: {self.label_dir}")
    if not self.image_dir.exists():
        raise FileNotFoundError(f"DOTA image directory not found: {self.image_dir}")

    discovered = self._discover_annotation_files()
    self._annotation_files_discovered_count = len(discovered)
    if self.filter_empty_gt:
        self._annotation_files = [
            ann_path
            for ann_path in discovered
            if self._effective_gt_count(ann_path) > 0
        ]
        self._empty_gt_filtered_count = (
            self._annotation_files_discovered_count - len(self._annotation_files)
        )
    else:
        self._annotation_files = discovered
        self._empty_gt_filtered_count = 0

get_class_names()

Extract unique class names from all annotations.

Source code in oriented_det/data/dota.py

def get_class_names(self) -> List[str]:
    """Extract unique class names from all annotations."""
    classes = set()
    for sample in self:
        for ann in sample.annotations:
            classes.add(ann.class_name)
    return sorted(classes)

DOTASample dataclass

Single DOTA image sample with annotations.

Source code in oriented_det/data/dota.py

@dataclass(frozen=True)
class DOTASample:
    """Single DOTA image sample with annotations."""

    image_path: Path
    width: int
    height: int
    annotations: Tuple[DOTAAnnotation, ...]

    @property
    def num_objects(self) -> int:
        return len(self.annotations)

    def filter_by_class(
        self,
        allowed_classes: Optional[Sequence[str]] = None,
        ignore_labels: Optional[Sequence[str]] = None,
        drop_difficult: bool = False,
    ) -> "DOTASample":
        """Return a filtered copy of this sample."""
        filtered = list(self.annotations)

        if drop_difficult:
            filtered = [ann for ann in filtered if ann.difficult == 0]

        if allowed_classes is not None:
            allowed_set = set(allowed_classes)
            filtered = [ann for ann in filtered if ann.class_name in allowed_set]

        if ignore_labels is not None:
            ignore_set = set(ignore_labels)
            filtered = [ann for ann in filtered if ann.class_name not in ignore_set]

        return DOTASample(
            image_path=self.image_path,
            width=self.width,
            height=self.height,
            annotations=tuple(filtered)
        )

filter_by_class(allowed_classes=None, ignore_labels=None, drop_difficult=False)

Return a filtered copy of this sample.

Source code in oriented_det/data/dota.py

def filter_by_class(
    self,
    allowed_classes: Optional[Sequence[str]] = None,
    ignore_labels: Optional[Sequence[str]] = None,
    drop_difficult: bool = False,
) -> "DOTASample":
    """Return a filtered copy of this sample."""
    filtered = list(self.annotations)

    if drop_difficult:
        filtered = [ann for ann in filtered if ann.difficult == 0]

    if allowed_classes is not None:
        allowed_set = set(allowed_classes)
        filtered = [ann for ann in filtered if ann.class_name in allowed_set]

    if ignore_labels is not None:
        ignore_set = set(ignore_labels)
        filtered = [ann for ann in filtered if ann.class_name not in ignore_set]

    return DOTASample(
        image_path=self.image_path,
        width=self.width,
        height=self.height,
        annotations=tuple(filtered)
    )

Detection dataclass

Single detection result.

Source code in oriented_det/data/evaluation.py

@dataclass(frozen=True)
class Detection:
    """Single detection result."""

    rbox: RBox
    score: float
    class_id: int
    class_name: str
    image_id: Optional[str] = None  # Required for correct mAP: match only to GTs from same image

DiagonalFlip

Bases: OrientedTransform

Diagonal flip (MMRotate direction='diagonal', le90 angle π − θ).

Source code in oriented_det/data/transforms.py

class DiagonalFlip(OrientedTransform):
    """Diagonal flip (MMRotate ``direction='diagonal'``, le90 angle π − θ)."""

    def __init__(self, p: float = 0.5):
        if not 0 <= p <= 1:
            raise ValueError("Probability p must be in [0, 1]")
        self.p = p

    def apply_to_image(self, image) -> any:  # type: ignore
        if Image is None:
            raise RuntimeError("PIL/Pillow is required for image transforms.")
        from PIL import Image as PILImage
        if isinstance(image, PILImage.Image):
            return image.transpose(PILImage.FLIP_LEFT_RIGHT).transpose(PILImage.FLIP_TOP_BOTTOM)
        return image

    def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
        return transforms.flip_diagonal(rbox, float(image_width), float(image_height))

    def __call__(self, image, rboxes: Sequence[RBox], image_width: int, image_height: int) -> Tuple[any, List[RBox]]:  # type: ignore
        if random.random() < self.p:
            flipped_image = self.apply_to_image(image)
            flipped_boxes = [self.apply_to_rbox(box, image_width, image_height) for box in rboxes]
            return flipped_image, flipped_boxes
        return image, list(rboxes)

GroundTruth dataclass

Single ground truth annotation.

Source code in oriented_det/data/evaluation.py

@dataclass(frozen=True)
class GroundTruth:
    """Single ground truth annotation."""

    rbox: RBox
    class_id: int
    class_name: str
    difficult: int = 0
    image_id: Optional[str] = None  # Required for correct mAP: match only to detections from same image

GtBestIouAlignmentMetrics dataclass

Global and per-class mean best IoU (per-GT max over raw detections).

Source code in oriented_det/data/evaluation.py

@dataclass(frozen=True)
class GtBestIouAlignmentMetrics:
    """Global and per-class mean best IoU (per-GT max over raw detections)."""

    num_gts: int
    mean_best_iou_any: float
    mean_best_iou_same_class: float
    median_best_iou_any: float
    median_best_iou_same_class: float
    per_class: Dict[str, ClassGtBestIouMetrics]

HorizontalFlip

Bases: OrientedTransform

Horizontal flip transform for oriented boxes.

Source code in oriented_det/data/transforms.py

class HorizontalFlip(OrientedTransform):
    """Horizontal flip transform for oriented boxes."""

    def __init__(self, p: float = 0.5):
        if not 0 <= p <= 1:
            raise ValueError("Probability p must be in [0, 1]")
        self.p = p

    def apply_to_image(self, image) -> any:  # type: ignore
        if Image is None:
            raise RuntimeError("PIL/Pillow is required for image transforms.")
        from PIL import Image as PILImage
        if isinstance(image, PILImage.Image):
            return image.transpose(PILImage.FLIP_LEFT_RIGHT)
        return image

    def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
        """Flip RBox horizontally."""
        return transforms.flip_horizontal(rbox, float(image_width))

    def __call__(self, image, rboxes: Sequence[RBox], image_width: int, image_height: int) -> Tuple[any, List[RBox]]:  # type: ignore
        if random.random() < self.p:
            flipped_image = self.apply_to_image(image)
            flipped_boxes = [self.apply_to_rbox(box, image_width, image_height) for box in rboxes]
            return flipped_image, flipped_boxes
        return image, list(rboxes)

apply_to_rbox(rbox, image_width, image_height)

Flip RBox horizontally.

Source code in oriented_det/data/transforms.py

def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
    """Flip RBox horizontally."""
    return transforms.flip_horizontal(rbox, float(image_width))

ImageTiler

Efficient tiler for splitting large images into overlapping patches.

The tiler supports configurable overlap and minimum overlap thresholds for keeping annotations that cross tile boundaries.

Source code in oriented_det/data/tiling.py

class ImageTiler:
    """Efficient tiler for splitting large images into overlapping patches.

    The tiler supports configurable overlap and minimum overlap thresholds
    for keeping annotations that cross tile boundaries.
    """

    def __init__(
        self,
        tile_size: int = 1024,
        overlap: float = 0.2,
        *,
        min_box_area: float = 16.0,
        min_overlap_ratio: float = 0.0,
        edge_handling: str = "clip",  # "clip", "ignore", "keep"
    ):
        """Initialize ImageTiler.

        Args:
            tile_size: Size of each tile (width and height in pixels)
            overlap: Overlap ratio between adjacent tiles [0, 1).
                    overlap=0.2 means 20% overlap, so stride = tile_size * 0.8
            min_box_area: Minimum area (in pixels²) for a box to be kept in a tile.
                         Boxes smaller than this are filtered out.
            min_overlap_ratio: Minimum ratio of box area that must overlap with tile
                              to keep the box. Range [0, 1].
                              - 0.0: Keep box if any part overlaps (default)
                              - 0.5: Keep box only if >= 50% of its area is in tile
                              - 1.0: Keep box only if fully inside tile
            edge_handling: Strategy for boxes crossing tile boundaries:
                          - "clip": Keep boxes that overlap tile (default)
                          - "ignore": Only keep boxes fully inside tile
                          - "keep": Same as "clip" (kept for backward compatibility)

        Example:
            >>> tiler = ImageTiler(
            ...     tile_size=1024,
            ...     overlap=0.2,  # 20% overlap between tiles
            ...     min_box_area=64,  # Filter boxes < 64 pixels²
            ...     min_overlap_ratio=0.3,  # Keep box if >= 30% overlaps tile
            ... )
        """
        if tile_size <= 0:
            raise ValueError("Tile size must be positive.")
        if not 0 <= overlap < 1:
            raise ValueError("Overlap must be in [0, 1).")
        if not 0 <= min_overlap_ratio <= 1:
            raise ValueError("min_overlap_ratio must be in [0, 1].")
        if edge_handling not in {"clip", "ignore", "keep"}:
            raise ValueError(f"Unknown edge_handling: {edge_handling}")

        self.tile_size = tile_size
        self.overlap = overlap
        self.min_box_area = min_box_area
        self.min_overlap_ratio = min_overlap_ratio
        self.edge_handling = edge_handling
        self.stride = int(tile_size * (1 - overlap))

    def generate_tiles(
        self,
        image_width: int,
        image_height: int,
        *,
        prefix: str = "tile"
    ) -> List[Tile]:
        """Generate tile grid for an image of given dimensions."""
        tiles = []
        tile_id = 0

        y = 0
        while y < image_height:
            x = 0
            while x < image_width:
                width = min(self.tile_size, image_width - x)
                height = min(self.tile_size, image_height - y)

                tiles.append(Tile(
                    x=x,
                    y=y,
                    width=width,
                    height=height,
                    tile_id=f"{prefix}_{tile_id:04d}"
                ))

                tile_id += 1
                x += self.stride
                if x >= image_width:
                    break

            y += self.stride
            if y >= image_height:
                break

        return tiles

    def assign_boxes_to_tile(
        self,
        tile: Tile,
        rboxes: Sequence[RBox],
        class_names: Sequence[str],
        difficult_flags: Optional[Sequence[int]] = None,
    ) -> TiledSample:
        """Assign boxes to a tile, applying edge handling and overlap thresholds.

        Args:
            tile: Target tile
            rboxes: Sequence of RBox annotations
            class_names: Class names corresponding to rboxes
            difficult_flags: Optional difficulty flags

        Returns:
            TiledSample with boxes assigned to the tile
        """
        if difficult_flags is None:
            difficult_flags = [0] * len(rboxes)

        if len(class_names) != len(rboxes):
            raise ValueError("class_names and rboxes must have same length.")
        if len(difficult_flags) != len(rboxes):
            raise ValueError("difficult_flags and rboxes must have same length.")

        assigned_boxes = []
        assigned_classes = []
        assigned_difficult = []

        # Create tile polygon for overlap calculation
        tile_poly = Polygon.rectangle(
            tile.x + tile.width / 2.0,
            tile.y + tile.height / 2.0,
            tile.width,
            tile.height
        )

        for rbox, cls_name, diff in zip(rboxes, class_names, difficult_flags):
            rbox_poly = rbox.to_polygon()

            # Check overlap based on edge handling strategy
            if self.edge_handling == "ignore":
                # Only keep boxes fully inside tile
                corners = rbox.corners()
                if not all(tile.contains_point(x, y) for x, y in corners):
                    continue
                clipped = rbox
            else:  # "clip" or "keep"
                # Check if box overlaps tile
                clipped = tile.clip_box(rbox)
                if clipped is None:
                    continue

                # Check minimum overlap ratio if specified
                if self.min_overlap_ratio > 0:
                    # Compute intersection area
                    from ..ops.utils import polygon_intersection_area
                    intersection_area = polygon_intersection_area(rbox_poly, tile_poly)
                    box_area = rbox.area

                    if box_area > 0:
                        overlap_ratio = intersection_area / box_area
                        if overlap_ratio < self.min_overlap_ratio:
                            continue

            # Translate to tile-local coordinates
            local_rbox = RBox(
                cx=clipped.cx - tile.x,
                cy=clipped.cy - tile.y,
                width=clipped.width,
                height=clipped.height,
                angle=clipped.angle
            )

            # Filter by minimum area
            if local_rbox.area >= self.min_box_area:
                assigned_boxes.append(local_rbox)
                assigned_classes.append(cls_name)
                assigned_difficult.append(diff)

        return TiledSample(
            tile=tile,
            image_path=Path(""),  # Will be set by caller
            annotations=tuple(assigned_boxes),
            class_names=tuple(assigned_classes),
            difficult_flags=tuple(assigned_difficult)
        )

    def tile_image(
        self,
        image_path: Path,
        image_width: int,
        image_height: int,
        rboxes: Sequence[RBox],
        class_names: Sequence[str],
        difficult_flags: Optional[Sequence[int]] = None,
    ) -> Iterator[TiledSample]:
        """Generate tiled samples from an image and its annotations."""
        tiles = self.generate_tiles(image_width, image_height)

        for tile in tiles:
            tiled = self.assign_boxes_to_tile(
                tile=tile,
                rboxes=rboxes,
                class_names=class_names,
                difficult_flags=difficult_flags,
            )
            # Set image path
            tiled = TiledSample(
                tile=tiled.tile,
                image_path=image_path,
                annotations=tiled.annotations,
                class_names=tiled.class_names,
                difficult_flags=tiled.difficult_flags,
            )
            yield tiled

__init__(tile_size=1024, overlap=0.2, *, min_box_area=16.0, min_overlap_ratio=0.0, edge_handling='clip')

Initialize ImageTiler.

Parameters:

Name	Type	Description	Default
tile_size	int	Size of each tile (width and height in pixels)	1024
overlap	float	Overlap ratio between adjacent tiles [0, 1). overlap=0.2 means 20% overlap, so stride = tile_size * 0.8	0.2
min_box_area	float	Minimum area (in pixels²) for a box to be kept in a tile. Boxes smaller than this are filtered out.	16.0
min_overlap_ratio	float	Minimum ratio of box area that must overlap with tile to keep the box. Range [0, 1]. - 0.0: Keep box if any part overlaps (default) - 0.5: Keep box only if >= 50% of its area is in tile - 1.0: Keep box only if fully inside tile	0.0
edge_handling	str	Strategy for boxes crossing tile boundaries: - "clip": Keep boxes that overlap tile (default) - "ignore": Only keep boxes fully inside tile - "keep": Same as "clip" (kept for backward compatibility)	'clip'

Example

tiler = ImageTiler( ... tile_size=1024, ... overlap=0.2, # 20% overlap between tiles ... min_box_area=64, # Filter boxes < 64 pixels² ... min_overlap_ratio=0.3, # Keep box if >= 30% overlaps tile ... )

Source code in oriented_det/data/tiling.py

def __init__(
    self,
    tile_size: int = 1024,
    overlap: float = 0.2,
    *,
    min_box_area: float = 16.0,
    min_overlap_ratio: float = 0.0,
    edge_handling: str = "clip",  # "clip", "ignore", "keep"
):
    """Initialize ImageTiler.

    Args:
        tile_size: Size of each tile (width and height in pixels)
        overlap: Overlap ratio between adjacent tiles [0, 1).
                overlap=0.2 means 20% overlap, so stride = tile_size * 0.8
        min_box_area: Minimum area (in pixels²) for a box to be kept in a tile.
                     Boxes smaller than this are filtered out.
        min_overlap_ratio: Minimum ratio of box area that must overlap with tile
                          to keep the box. Range [0, 1].
                          - 0.0: Keep box if any part overlaps (default)
                          - 0.5: Keep box only if >= 50% of its area is in tile
                          - 1.0: Keep box only if fully inside tile
        edge_handling: Strategy for boxes crossing tile boundaries:
                      - "clip": Keep boxes that overlap tile (default)
                      - "ignore": Only keep boxes fully inside tile
                      - "keep": Same as "clip" (kept for backward compatibility)

    Example:
        >>> tiler = ImageTiler(
        ...     tile_size=1024,
        ...     overlap=0.2,  # 20% overlap between tiles
        ...     min_box_area=64,  # Filter boxes < 64 pixels²
        ...     min_overlap_ratio=0.3,  # Keep box if >= 30% overlaps tile
        ... )
    """
    if tile_size <= 0:
        raise ValueError("Tile size must be positive.")
    if not 0 <= overlap < 1:
        raise ValueError("Overlap must be in [0, 1).")
    if not 0 <= min_overlap_ratio <= 1:
        raise ValueError("min_overlap_ratio must be in [0, 1].")
    if edge_handling not in {"clip", "ignore", "keep"}:
        raise ValueError(f"Unknown edge_handling: {edge_handling}")

    self.tile_size = tile_size
    self.overlap = overlap
    self.min_box_area = min_box_area
    self.min_overlap_ratio = min_overlap_ratio
    self.edge_handling = edge_handling
    self.stride = int(tile_size * (1 - overlap))

assign_boxes_to_tile(tile, rboxes, class_names, difficult_flags=None)

Assign boxes to a tile, applying edge handling and overlap thresholds.

Parameters:

Name	Type	Description	Default
tile	Tile	Target tile	required
rboxes	Sequence[RBox]	Sequence of RBox annotations	required
class_names	Sequence[str]	Class names corresponding to rboxes	required
difficult_flags	Optional[Sequence[int]]	Optional difficulty flags	None

Returns:

Type	Description
TiledSample	TiledSample with boxes assigned to the tile

Source code in oriented_det/data/tiling.py

def assign_boxes_to_tile(
    self,
    tile: Tile,
    rboxes: Sequence[RBox],
    class_names: Sequence[str],
    difficult_flags: Optional[Sequence[int]] = None,
) -> TiledSample:
    """Assign boxes to a tile, applying edge handling and overlap thresholds.

    Args:
        tile: Target tile
        rboxes: Sequence of RBox annotations
        class_names: Class names corresponding to rboxes
        difficult_flags: Optional difficulty flags

    Returns:
        TiledSample with boxes assigned to the tile
    """
    if difficult_flags is None:
        difficult_flags = [0] * len(rboxes)

    if len(class_names) != len(rboxes):
        raise ValueError("class_names and rboxes must have same length.")
    if len(difficult_flags) != len(rboxes):
        raise ValueError("difficult_flags and rboxes must have same length.")

    assigned_boxes = []
    assigned_classes = []
    assigned_difficult = []

    # Create tile polygon for overlap calculation
    tile_poly = Polygon.rectangle(
        tile.x + tile.width / 2.0,
        tile.y + tile.height / 2.0,
        tile.width,
        tile.height
    )

    for rbox, cls_name, diff in zip(rboxes, class_names, difficult_flags):
        rbox_poly = rbox.to_polygon()

        # Check overlap based on edge handling strategy
        if self.edge_handling == "ignore":
            # Only keep boxes fully inside tile
            corners = rbox.corners()
            if not all(tile.contains_point(x, y) for x, y in corners):
                continue
            clipped = rbox
        else:  # "clip" or "keep"
            # Check if box overlaps tile
            clipped = tile.clip_box(rbox)
            if clipped is None:
                continue

            # Check minimum overlap ratio if specified
            if self.min_overlap_ratio > 0:
                # Compute intersection area
                from ..ops.utils import polygon_intersection_area
                intersection_area = polygon_intersection_area(rbox_poly, tile_poly)
                box_area = rbox.area

                if box_area > 0:
                    overlap_ratio = intersection_area / box_area
                    if overlap_ratio < self.min_overlap_ratio:
                        continue

        # Translate to tile-local coordinates
        local_rbox = RBox(
            cx=clipped.cx - tile.x,
            cy=clipped.cy - tile.y,
            width=clipped.width,
            height=clipped.height,
            angle=clipped.angle
        )

        # Filter by minimum area
        if local_rbox.area >= self.min_box_area:
            assigned_boxes.append(local_rbox)
            assigned_classes.append(cls_name)
            assigned_difficult.append(diff)

    return TiledSample(
        tile=tile,
        image_path=Path(""),  # Will be set by caller
        annotations=tuple(assigned_boxes),
        class_names=tuple(assigned_classes),
        difficult_flags=tuple(assigned_difficult)
    )

generate_tiles(image_width, image_height, *, prefix='tile')

Generate tile grid for an image of given dimensions.

Source code in oriented_det/data/tiling.py

def generate_tiles(
    self,
    image_width: int,
    image_height: int,
    *,
    prefix: str = "tile"
) -> List[Tile]:
    """Generate tile grid for an image of given dimensions."""
    tiles = []
    tile_id = 0

    y = 0
    while y < image_height:
        x = 0
        while x < image_width:
            width = min(self.tile_size, image_width - x)
            height = min(self.tile_size, image_height - y)

            tiles.append(Tile(
                x=x,
                y=y,
                width=width,
                height=height,
                tile_id=f"{prefix}_{tile_id:04d}"
            ))

            tile_id += 1
            x += self.stride
            if x >= image_width:
                break

        y += self.stride
        if y >= image_height:
            break

    return tiles

tile_image(image_path, image_width, image_height, rboxes, class_names, difficult_flags=None)

Generate tiled samples from an image and its annotations.

Source code in oriented_det/data/tiling.py

def tile_image(
    self,
    image_path: Path,
    image_width: int,
    image_height: int,
    rboxes: Sequence[RBox],
    class_names: Sequence[str],
    difficult_flags: Optional[Sequence[int]] = None,
) -> Iterator[TiledSample]:
    """Generate tiled samples from an image and its annotations."""
    tiles = self.generate_tiles(image_width, image_height)

    for tile in tiles:
        tiled = self.assign_boxes_to_tile(
            tile=tile,
            rboxes=rboxes,
            class_names=class_names,
            difficult_flags=difficult_flags,
        )
        # Set image path
        tiled = TiledSample(
            tile=tiled.tile,
            image_path=image_path,
            annotations=tiled.annotations,
            class_names=tiled.class_names,
            difficult_flags=tiled.difficult_flags,
        )
        yield tiled

OrientedTransform dataclass

Base class for transforms that modify both image and oriented boxes.

Source code in oriented_det/data/transforms.py

@dataclass
class OrientedTransform:
    """Base class for transforms that modify both image and oriented boxes."""

    def apply_to_image(self, image) -> any:  # type: ignore
        """Apply transform to image. Returns transformed image."""
        raise NotImplementedError

    def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
        """Apply transform to RBox. Returns transformed RBox."""
        raise NotImplementedError

apply_to_image(image)

Apply transform to image. Returns transformed image.

Source code in oriented_det/data/transforms.py

def apply_to_image(self, image) -> any:  # type: ignore
    """Apply transform to image. Returns transformed image."""
    raise NotImplementedError

apply_to_rbox(rbox, image_width, image_height)

Apply transform to RBox. Returns transformed RBox.

Source code in oriented_det/data/transforms.py

def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
    """Apply transform to RBox. Returns transformed RBox."""
    raise NotImplementedError

Rotate

Bases: OrientedTransform

Rotation transform for oriented boxes.

Source code in oriented_det/data/transforms.py

class Rotate(OrientedTransform):
    """Rotation transform for oriented boxes."""

    def __init__(self, degrees: float, p: float = 1.0):
        if not 0 <= p <= 1:
            raise ValueError("Probability p must be in [0, 1]")
        self.degrees = degrees
        self.radians = math.radians(degrees)
        self.p = p

    def apply_to_image(self, image) -> any:  # type: ignore
        if Image is None:
            raise RuntimeError("PIL/Pillow is required for image transforms.")
        from PIL import Image as PILImage
        if isinstance(image, PILImage.Image):
            return image.rotate(self.degrees, expand=False)
        return image

    def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
        """Rotate RBox around image center."""
        center_x, center_y = image_width / 2.0, image_height / 2.0

        # Translate to origin
        dx = rbox.cx - center_x
        dy = rbox.cy - center_y

        # Rotate
        cos_a = math.cos(self.radians)
        sin_a = math.sin(self.radians)
        new_dx = dx * cos_a - dy * sin_a
        new_dy = dx * sin_a + dy * cos_a

        # Translate back
        new_cx = center_x + new_dx
        new_cy = center_y + new_dy
        new_angle = rbox.angle + self.radians

        return RBox(new_cx, new_cy, rbox.width, rbox.height, new_angle)

    def __call__(self, image, rboxes: Sequence[RBox], image_width: int, image_height: int) -> Tuple[any, List[RBox]]:  # type: ignore
        if random.random() < self.p:
            rotated_image = self.apply_to_image(image)
            rotated_boxes = [self.apply_to_rbox(box, image_width, image_height) for box in rboxes]
            return rotated_image, rotated_boxes
        return image, list(rboxes)

apply_to_rbox(rbox, image_width, image_height)

Rotate RBox around image center.

Source code in oriented_det/data/transforms.py

def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
    """Rotate RBox around image center."""
    center_x, center_y = image_width / 2.0, image_height / 2.0

    # Translate to origin
    dx = rbox.cx - center_x
    dy = rbox.cy - center_y

    # Rotate
    cos_a = math.cos(self.radians)
    sin_a = math.sin(self.radians)
    new_dx = dx * cos_a - dy * sin_a
    new_dy = dx * sin_a + dy * cos_a

    # Translate back
    new_cx = center_x + new_dx
    new_cy = center_y + new_dy
    new_angle = rbox.angle + self.radians

    return RBox(new_cx, new_cy, rbox.width, rbox.height, new_angle)

Tile dataclass

Represents a single tile/patch from a larger image.

Source code in oriented_det/data/tiling.py

@dataclass(frozen=True)
class Tile:
    """Represents a single tile/patch from a larger image."""

    x: int  # Left coordinate in original image
    y: int  # Top coordinate in original image
    width: int
    height: int
    tile_id: str

    @property
    def bounds(self) -> Tuple[int, int, int, int]:
        """Returns (x, y, x + width, y + height)."""
        return (self.x, self.y, self.x + self.width, self.y + self.height)

    def contains_point(self, px: float, py: float) -> bool:
        """Check if a point lies within this tile."""
        return self.x <= px < self.x + self.width and self.y <= py < self.y + self.height

    def clip_box(self, rbox: RBox) -> Optional[RBox]:
        """Clip an RBox to this tile's bounds, returning None if fully outside."""
        poly = rbox.to_polygon()
        tile_poly = Polygon.rectangle(
            self.x + self.width / 2.0,
            self.y + self.height / 2.0,
            self.width,
            self.height
        )

        # Quick AABB check
        rbox_bounds = poly.bounds
        tile_bounds = tile_poly.bounds

        if (rbox_bounds[2] < tile_bounds[0] or rbox_bounds[0] > tile_bounds[2] or
            rbox_bounds[3] < tile_bounds[1] or rbox_bounds[1] > tile_bounds[3]):
            return None

        # Check if any corner of rbox is inside tile
        corners = rbox.corners()
        has_inside = any(self.contains_point(x, y) for x, y in corners)

        if not has_inside:
            return None

        # For simplicity, return the original box if any part overlaps
        # More precise clipping would require polygon intersection
        return rbox

bounds property

Returns (x, y, x + width, y + height).

clip_box(rbox)

Clip an RBox to this tile's bounds, returning None if fully outside.

Source code in oriented_det/data/tiling.py

def clip_box(self, rbox: RBox) -> Optional[RBox]:
    """Clip an RBox to this tile's bounds, returning None if fully outside."""
    poly = rbox.to_polygon()
    tile_poly = Polygon.rectangle(
        self.x + self.width / 2.0,
        self.y + self.height / 2.0,
        self.width,
        self.height
    )

    # Quick AABB check
    rbox_bounds = poly.bounds
    tile_bounds = tile_poly.bounds

    if (rbox_bounds[2] < tile_bounds[0] or rbox_bounds[0] > tile_bounds[2] or
        rbox_bounds[3] < tile_bounds[1] or rbox_bounds[1] > tile_bounds[3]):
        return None

    # Check if any corner of rbox is inside tile
    corners = rbox.corners()
    has_inside = any(self.contains_point(x, y) for x, y in corners)

    if not has_inside:
        return None

    # For simplicity, return the original box if any part overlaps
    # More precise clipping would require polygon intersection
    return rbox

contains_point(px, py)

Check if a point lies within this tile.

Source code in oriented_det/data/tiling.py

def contains_point(self, px: float, py: float) -> bool:
    """Check if a point lies within this tile."""
    return self.x <= px < self.x + self.width and self.y <= py < self.y + self.height

TiledSample dataclass

A sample with its annotations clipped to a specific tile.

Source code in oriented_det/data/tiling.py

@dataclass(frozen=True)
class TiledSample:
    """A sample with its annotations clipped to a specific tile."""

    tile: Tile
    image_path: Path
    annotations: Tuple[RBox, ...]  # RBoxes clipped to tile coordinates
    class_names: Tuple[str, ...]
    difficult_flags: Tuple[int, ...]

VerticalFlip

Bases: OrientedTransform

Vertical flip transform for oriented boxes.

Source code in oriented_det/data/transforms.py

class VerticalFlip(OrientedTransform):
    """Vertical flip transform for oriented boxes."""

    def __init__(self, p: float = 0.5):
        if not 0 <= p <= 1:
            raise ValueError("Probability p must be in [0, 1]")
        self.p = p

    def apply_to_image(self, image) -> any:  # type: ignore
        if Image is None:
            raise RuntimeError("PIL/Pillow is required for image transforms.")
        from PIL import Image as PILImage
        if isinstance(image, PILImage.Image):
            return image.transpose(PILImage.FLIP_TOP_BOTTOM)
        return image

    def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
        """Flip RBox vertically."""
        return transforms.flip_vertical(rbox, float(image_height))

    def __call__(self, image, rboxes: Sequence[RBox], image_width: int, image_height: int) -> Tuple[any, List[RBox]]:  # type: ignore
        if random.random() < self.p:
            flipped_image = self.apply_to_image(image)
            flipped_boxes = [self.apply_to_rbox(box, image_width, image_height) for box in rboxes]
            return flipped_image, flipped_boxes
        return image, list(rboxes)

apply_to_rbox(rbox, image_width, image_height)

Flip RBox vertically.

Source code in oriented_det/data/transforms.py

def apply_to_rbox(self, rbox: RBox, image_width: int, image_height: int) -> RBox:
    """Flip RBox vertically."""
    return transforms.flip_vertical(rbox, float(image_height))

apply_random_train_flips(image, rboxes, *, image_width, image_height, enable_horizontal=True, enable_vertical=True, enable_diagonal=False)

Pick at most one random flip (MMRotate DOTA-style 25% per enabled mode).

When horizontal, vertical, and diagonal are all enabled, each has probability 0.25 and there is a 0.25 chance of no flip — matching RRandomFlip(flip_ratio=[0.25, 0.25, 0.25], direction=['horizontal', 'vertical', 'diagonal']).

With only horizontal and vertical enabled (no diagonal), uses 0.5 per enabled mode and no no-flip bucket (legacy two-flip behavior).

Source code in oriented_det/data/flips.py

def apply_random_train_flips(
    image,
    rboxes: Sequence[RBox],
    *,
    image_width: float,
    image_height: float,
    enable_horizontal: bool = True,
    enable_vertical: bool = True,
    enable_diagonal: bool = False,
) -> Tuple[any, List[RBox]]:
    """Pick at most one random flip (MMRotate DOTA-style 25% per enabled mode).

    When horizontal, vertical, and diagonal are all enabled, each has probability
  0.25 and there is a 0.25 chance of no flip — matching
  ``RRandomFlip(flip_ratio=[0.25, 0.25, 0.25],
  direction=['horizontal', 'vertical', 'diagonal'])``.

    With only horizontal and vertical enabled (no diagonal), uses 0.5 per enabled
  mode and no no-flip bucket (legacy two-flip behavior).
    """
    modes: List[FlipMode] = []
    if enable_horizontal:
        modes.append("horizontal")
    if enable_vertical:
        modes.append("vertical")
    if enable_diagonal:
        modes.append("diagonal")
    if not modes:
        return image, list(rboxes)

    rboxes_list = list(rboxes)
    if len(modes) == 1:
        if random.random() >= 0.5:
            return image, rboxes_list
        mode = modes[0]
    elif enable_diagonal and enable_horizontal and enable_vertical:
        # MMRotate: three flip types + implicit no-flip (25% each).
        bucket = random.random()
        if bucket >= 0.75:
            return image, rboxes_list
        mode = modes[int(bucket / 0.25)]
    else:
        # Two modes only (e.g. H+V without diagonal): 50% each.
        mode = modes[0] if random.random() < 0.5 else modes[1]

    return (
        apply_flip_to_image(image, mode),
        apply_flip_to_rboxes(
            rboxes_list, mode, image_width=image_width, image_height=image_height
        ),
    )

build_dota_loader(root_dir, *, split='train', split_file=None, label_dir=None, image_dir=None, batch_size=1, shuffle=False, num_workers=0, class_map=None, allowed_classes=None, ignore_labels=None, difficult_strategy='drop', filter_empty_gt=False, collate_fn=None)

Build a PyTorch DataLoader for DOTA dataset.

Parameters:

Name	Type	Description	Default
root_dir	str | Path	Root directory of DOTA dataset (used as base if label_dir/image_dir not specified)	required
split	str	Split name ("train", "val", "test")	'train'
split_file	Optional[str | Path]	Optional path to a file listing image names (one per line). If provided, only images listed in this file will be used. This follows the official DOTA dataset convention.	None
label_dir	Optional[str | Path]	Optional custom path to label directory. If None, uses root_dir/labelTxt. This allows having train/val/test in separate folders.	None
image_dir	Optional[str | Path]	Optional custom path to image directory. If None, uses root_dir/images. This allows having train/val/test in separate folders.	None
batch_size	int	Batch size for DataLoader	1
shuffle	bool	Whether to shuffle the dataset	False
num_workers	int	Number of worker processes for data loading	0
class_map	Optional[Dict[str, int]]	Optional mapping from class names to numeric IDs	None
allowed_classes	Optional[Sequence[str]]	Optional list of allowed class names (whitelist)	None
ignore_labels	Optional[Sequence[str]]	Optional list of class names to exclude (blacklist)	None
difficult_strategy	str	drop | ignore | keep (same as DOTADataset)	'drop'
filter_empty_gt	bool	Drop tiles with no effective GT (same as DOTADataset)	False
collate_fn	Optional[Callable]	Optional custom collate function. If None, uses collate_dota_samples from oriented_det.train.utils which converts DOTASample objects to (images, targets) format expected by training engines.	None

Returns:

Type	Description
'DataLoader'	DataLoader that yields batches in format determined by collate_fn.
'DataLoader'	Default collate_fn returns (images, targets) tuple for training.

Source code in oriented_det/data/dota.py

def build_dota_loader(
    root_dir: str | Path,
    *,
    split: str = "train",
    split_file: Optional[str | Path] = None,
    label_dir: Optional[str | Path] = None,
    image_dir: Optional[str | Path] = None,
    batch_size: int = 1,
    shuffle: bool = False,
    num_workers: int = 0,
    class_map: Optional[Dict[str, int]] = None,
    allowed_classes: Optional[Sequence[str]] = None,
    ignore_labels: Optional[Sequence[str]] = None,
    difficult_strategy: str = "drop",
    filter_empty_gt: bool = False,
    collate_fn: Optional[Callable] = None,
) -> "DataLoader":
    """Build a PyTorch DataLoader for DOTA dataset.

    Args:
        root_dir: Root directory of DOTA dataset (used as base if label_dir/image_dir not specified)
        split: Split name ("train", "val", "test")
        split_file: Optional path to a file listing image names (one per line).
                   If provided, only images listed in this file will be used.
                   This follows the official DOTA dataset convention.
        label_dir: Optional custom path to label directory. If None, uses root_dir/labelTxt.
                  This allows having train/val/test in separate folders.
        image_dir: Optional custom path to image directory. If None, uses root_dir/images.
                  This allows having train/val/test in separate folders.
        batch_size: Batch size for DataLoader
        shuffle: Whether to shuffle the dataset
        num_workers: Number of worker processes for data loading
        class_map: Optional mapping from class names to numeric IDs
        allowed_classes: Optional list of allowed class names (whitelist)
        ignore_labels: Optional list of class names to exclude (blacklist)
        difficult_strategy: drop | ignore | keep (same as DOTADataset)
        filter_empty_gt: Drop tiles with no effective GT (same as DOTADataset)
        collate_fn: Optional custom collate function. If None, uses collate_dota_samples
                   from oriented_det.train.utils which converts DOTASample objects
                   to (images, targets) format expected by training engines.

    Returns:
        DataLoader that yields batches in format determined by collate_fn.
        Default collate_fn returns (images, targets) tuple for training.
    """
    try:
        from torch.utils.data import DataLoader
    except ImportError:
        raise RuntimeError("PyTorch is required to build DataLoader.")

    dataset = DOTADataset(
        root_dir=root_dir,
        split=split,
        split_file=split_file,
        label_dir=label_dir,
        image_dir=image_dir,
        class_map=class_map,
        allowed_classes=allowed_classes,
        ignore_labels=ignore_labels,
        difficult_strategy=difficult_strategy,
        filter_empty_gt=filter_empty_gt,
    )

    # Use provided collate_fn or default to collate_dota_samples
    if collate_fn is None:
        # Lazy import to avoid circular dependencies
        try:
            from ..train.utils import collate_dota_samples
            collate_fn = collate_dota_samples
        except ImportError:
            # Fallback if train module not available
            collate_fn = _default_collate_fn

    return DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=shuffle,
        num_workers=num_workers,
        collate_fn=collate_fn,
    )

build_dota_split_dataset(tile_roots, *, split, same_folder=False, difficult_strategy='drop', allowed_classes=None, ignore_labels=None, filter_empty_gt=False)

Build one DOTADataset or ConcatDataset over multiple tile roots.

Source code in oriented_det/data/dota.py

def build_dota_split_dataset(
    tile_roots: Sequence[str | Path],
    *,
    split: str,
    same_folder: bool = False,
    difficult_strategy: str = "drop",
    allowed_classes: Optional[Sequence[str]] = None,
    ignore_labels: Optional[Sequence[str]] = None,
    filter_empty_gt: bool = False,
):
    """Build one ``DOTADataset`` or ``ConcatDataset`` over multiple tile roots."""
    from torch.utils.data import ConcatDataset

    roots = [Path(r) for r in tile_roots]
    if not roots:
        raise ValueError("tile_roots must be non-empty")

    datasets = [
        DOTADataset(
            root_dir=root,
            split=split,
            label_dir=label_dir,
            image_dir=image_dir,
            difficult_strategy=difficult_strategy,
            allowed_classes=allowed_classes,
            ignore_labels=ignore_labels,
            filter_empty_gt=filter_empty_gt,
        )
        for root, label_dir, image_dir in (
            _dota_dirs_for_root(r, same_folder=same_folder) for r in roots
        )
    ]
    if len(datasets) == 1:
        return datasets[0]
    return ConcatDataset(datasets)

collect_dota_image_paths(tile_roots, *, same_folder=False)

Collect image paths from one or more DOTA tile directories.

Source code in oriented_det/data/dota.py

def collect_dota_image_paths(
    tile_roots: Sequence[str | Path],
    *,
    same_folder: bool = False,
) -> List[Path]:
    """Collect image paths from one or more DOTA tile directories."""
    paths: List[Path] = []
    for root in tile_roots:
        _, _, image_dir = _dota_dirs_for_root(Path(root), same_folder=same_folder)
        if not image_dir.exists():
            raise FileNotFoundError(f"DOTA image directory not found: {image_dir}")
        paths.extend(sorted(image_dir.glob("*.jpg")))
        paths.extend(sorted(image_dir.glob("*.png")))
    return paths

collect_dota_split_image_paths(tile_roots, *, split, same_folder=False, difficult_strategy='drop', allowed_classes=None, ignore_labels=None, filter_empty_gt=False, log_filter_empty_gt=False)

Collect image paths using the same tile set as :func:build_dota_split_dataset.

When filter_empty_gt is True, skips tiles with no effective GT after difficult_strategy, allowed_classes, and ignore_labels (training parity). Set log_filter_empty_gt to print the same summary as training startup.

Source code in oriented_det/data/dota.py

def collect_dota_split_image_paths(
    tile_roots: Sequence[str | Path],
    *,
    split: str,
    same_folder: bool = False,
    difficult_strategy: str = "drop",
    allowed_classes: Optional[Sequence[str]] = None,
    ignore_labels: Optional[Sequence[str]] = None,
    filter_empty_gt: bool = False,
    log_filter_empty_gt: bool = False,
) -> List[Path]:
    """Collect image paths using the same tile set as :func:`build_dota_split_dataset`.

    When ``filter_empty_gt`` is True, skips tiles with no effective GT after
    ``difficult_strategy``, ``allowed_classes``, and ``ignore_labels`` (training parity).
    Set ``log_filter_empty_gt`` to print the same summary as training startup.
    """
    dataset = build_dota_split_dataset(
        tile_roots,
        split=split,
        same_folder=same_folder,
        difficult_strategy=difficult_strategy,
        allowed_classes=allowed_classes,
        ignore_labels=ignore_labels,
        filter_empty_gt=filter_empty_gt,
    )
    if log_filter_empty_gt and filter_empty_gt:
        print("DOTA filter_empty_gt (preds/metrics, same as training):")
        print(format_dota_empty_gt_filter_log(dataset, split=split))
    paths: List[Path] = []
    for ds in iter_dota_datasets(dataset):
        for ann_path in ds._annotation_files:
            image_path = _image_path_for_annotation(ann_path, ds.image_dir)
            if image_path is not None:
                paths.append(image_path)
    return sorted(paths)

compute_gt_best_iou_alignment_metrics(all_detections, all_ground_truths, *, class_names=None, skip_difficult=False, use_exact_rotated_iou=True, device=None, show_progress=False, progress_stream=None)

Per-GT max rotated IoU vs raw detections (global + per-class means).

For each non-skipped GT, best_any is the max IoU over all detections on the same image; best_same restricts to same-class detections. Uses exact polygon IoU on CPU when use_exact_rotated_iou is true (same as mAP matching).

Source code in oriented_det/data/evaluation.py

def compute_gt_best_iou_alignment_metrics(
    all_detections: Dict[str, List[Detection]],
    all_ground_truths: Dict[str, List[GroundTruth]],
    *,
    class_names: Optional[Sequence[str]] = None,
    skip_difficult: bool = False,
    use_exact_rotated_iou: bool = True,
    device: DeviceType = None,
    show_progress: bool = False,
    progress_stream: Optional[Any] = None,
) -> GtBestIouAlignmentMetrics:
    """Per-GT max rotated IoU vs raw detections (global + per-class means).

    For each non-skipped GT, ``best_any`` is the max IoU over all detections on the
    same image; ``best_same`` restricts to same-class detections. Uses exact polygon
    IoU on CPU when ``use_exact_rotated_iou`` is true (same as mAP matching).
    """
    best_any_by_class: Dict[str, List[float]] = defaultdict(list)
    best_same_by_class: Dict[str, List[float]] = defaultdict(list)

    image_ids = sorted(all_ground_truths.keys())
    tqdm_file = progress_stream if progress_stream is not None else sys.stderr
    use_pbar = show_progress and tqdm is not None and len(image_ids) > 200
    img_iter: Any = (
        tqdm(image_ids, desc="GT alignment IoU", unit="img", leave=False, file=tqdm_file)
        if use_pbar
        else image_ids
    )

    for image_id in img_iter:
        ground_truths = all_ground_truths.get(image_id, [])
        raw_detections = all_detections.get(image_id, [])
        if not ground_truths:
            continue

        raw_iou: Optional[list[list[float]]] = None
        if raw_detections:
            gt_rboxes = [g.rbox for g in ground_truths]
            if use_exact_rotated_iou:
                raw_iou = iou.batch_rbox_iou(
                    [d.rbox for d in raw_detections],
                    gt_rboxes,
                    intersection_backend=resolve_exact_polygon_iou_backend(),
                )
            else:
                if torch is None:
                    raw_iou = iou.batch_rbox_iou(
                        [d.rbox for d in raw_detections],
                        gt_rboxes,
                    )
                else:
                    from ..ops.gpu_ops import oriented_box_iou_gpu
                    from ..ops.utils import rboxes_to_tensor

                    dev = device
                    if dev is None:
                        dev = torch.device("cuda" if torch.cuda.is_available() else "cpu")
                    ta = rboxes_to_tensor([d.rbox for d in raw_detections], device=dev)
                    tb = rboxes_to_tensor(gt_rboxes, device=dev)
                    raw_iou = oriented_box_iou_gpu(ta, tb).detach().cpu().tolist()

        for gt_idx, gt in enumerate(ground_truths):
            if skip_difficult and gt.difficult != 0:
                continue
            best_any = 0.0
            best_same = 0.0
            if raw_iou is not None:
                for det_idx, det in enumerate(raw_detections):
                    iou_val = raw_iou[det_idx][gt_idx]
                    if iou_val > best_any:
                        best_any = iou_val
                    if detection_matches_ground_truth_class(det, gt) and iou_val > best_same:
                        best_same = iou_val
            cname = gt.class_name or "unknown"
            best_any_by_class[cname].append(best_any)
            best_same_by_class[cname].append(best_same)

    return _aggregate_gt_best_iou_samples(
        best_any_by_class,
        best_same_by_class,
        class_names=class_names,
    )

create_albumentations_augmentation(brightness_limit=0.2, contrast_limit=0.2, gamma_limit=(80, 120), gauss_noise_var_limit=(10.0, 50.0), blur_limit=3, clahe_clip_limit=4.0, p_brightness_contrast=0.5, p_gamma=0.3, p_noise=0.2, p_blur=0.2, p_clahe=0.3)

Create a default albumentations augmentation pipeline with non-geometric transforms.

This function creates a sensible default augmentation pipeline using only non-geometric transforms that are safe for oriented bounding boxes.

Parameters:

Name	Type	Description	Default
brightness_limit	float	Range for brightness adjustment (-limit to +limit)	0.2
contrast_limit	float	Range for contrast adjustment (-limit to +limit)	0.2
gamma_limit	tuple	Range for gamma correction (as percentage, e.g., (80, 120))	(80, 120)
gauss_noise_var_limit	tuple	Variance range for Gaussian noise (will be converted to std_range internally)	(10.0, 50.0)
blur_limit	int	Maximum kernel size for Gaussian blur (must be odd)	3
clahe_clip_limit	float	Clip limit for CLAHE (Contrast Limited Adaptive Histogram Equalization)	4.0
p_brightness_contrast	float	Probability of applying brightness/contrast	0.5
p_gamma	float	Probability of applying gamma correction	0.3
p_noise	float	Probability of adding Gaussian noise	0.2
p_blur	float	Probability of applying Gaussian blur	0.2
p_clahe	float	Probability of applying CLAHE	0.3

Returns:

Type	Description
AlbumentationsTransform	AlbumentationsTransform instance with the configured augmentation pipeline

Example

from oriented_det.data import create_albumentations_augmentation

Create default augmentation

aug = create_albumentations_augmentation()

Apply to image

augmented_image = aug(image)

Source code in oriented_det/data/transforms.py

def create_albumentations_augmentation(
    brightness_limit: float = 0.2,
    contrast_limit: float = 0.2,
    gamma_limit: tuple = (80, 120),
    gauss_noise_var_limit: tuple = (10.0, 50.0),
    blur_limit: int = 3,
    clahe_clip_limit: float = 4.0,
    p_brightness_contrast: float = 0.5,
    p_gamma: float = 0.3,
    p_noise: float = 0.2,
    p_blur: float = 0.2,
    p_clahe: float = 0.3,
) -> AlbumentationsTransform:
    """Create a default albumentations augmentation pipeline with non-geometric transforms.

    This function creates a sensible default augmentation pipeline using only
    non-geometric transforms that are safe for oriented bounding boxes.

    Args:
        brightness_limit: Range for brightness adjustment (-limit to +limit)
        contrast_limit: Range for contrast adjustment (-limit to +limit)
        gamma_limit: Range for gamma correction (as percentage, e.g., (80, 120))
        gauss_noise_var_limit: Variance range for Gaussian noise (will be converted to std_range internally)
        blur_limit: Maximum kernel size for Gaussian blur (must be odd)
        clahe_clip_limit: Clip limit for CLAHE (Contrast Limited Adaptive Histogram Equalization)
        p_brightness_contrast: Probability of applying brightness/contrast
        p_gamma: Probability of applying gamma correction
        p_noise: Probability of adding Gaussian noise
        p_blur: Probability of applying Gaussian blur
        p_clahe: Probability of applying CLAHE

    Returns:
        AlbumentationsTransform instance with the configured augmentation pipeline

    Example:
        >>> from oriented_det.data import create_albumentations_augmentation
        >>> 
        >>> # Create default augmentation
        >>> aug = create_albumentations_augmentation()
        >>> 
        >>> # Apply to image
        >>> augmented_image = aug(image)
    """
    if A is None:
        raise RuntimeError("albumentations is required. Install with: pip install albumentations")

    # Ensure blur_limit is odd
    if blur_limit % 2 == 0:
        blur_limit += 1

    transforms = []

    # Color and contrast augmentations
    if p_brightness_contrast > 0:
        transforms.append(
            A.RandomBrightnessContrast(
                brightness_limit=brightness_limit,
                contrast_limit=contrast_limit,
                p=p_brightness_contrast,
            )
        )

    if p_gamma > 0:
        transforms.append(
            A.RandomGamma(gamma_limit=gamma_limit, p=p_gamma)
        )

    if p_clahe > 0:
        transforms.append(
            A.CLAHE(clip_limit=clahe_clip_limit, p=p_clahe)
        )

    # Noise and blur augmentations
    if p_noise > 0:
        # Convert variance to normalized standard deviation: std = sqrt(var) / 255
        # std_range expects values in [0, 1] as fraction of max value (255 for uint8)
        std_range = (
            math.sqrt(gauss_noise_var_limit[0]) / 255.0,
            math.sqrt(gauss_noise_var_limit[1]) / 255.0
        )
        transforms.append(
            A.GaussNoise(std_range=std_range, p=p_noise)
        )

    if p_blur > 0:
        transforms.append(
            A.GaussianBlur(blur_limit=(3, blur_limit), p=p_blur)
        )

    # Compose all transforms
    aug_pipeline = A.Compose(transforms)

    return AlbumentationsTransform(aug_pipeline, p=1.0)

detect_airbus_split_csv_format(split_values)

Infer whether split.csv uses legacy train/val strings or integer fold ids.

Fold mode: every non-empty value must parse as a non-negative int. Training code treats fold val_split_id (default 0) as validation and all other folds as training, unless train_includes_val is enabled (train on all folds; val fold for monitoring only).

Source code in oriented_det/data/airbus_playground.py

def detect_airbus_split_csv_format(split_values: Sequence[str]) -> Literal["train_val", "fold_ids"]:
    """Infer whether split.csv uses legacy train/val strings or integer fold ids.

    Fold mode: every non-empty value must parse as a non-negative int. Training code treats
    fold ``val_split_id`` (default 0) as validation and all other folds as training, unless
    ``train_includes_val`` is enabled (train on all folds; val fold for monitoring only).
    """
    non_empty = [str(s).strip() for s in split_values if s is not None and str(s).strip() != ""]
    if not non_empty:
        raise ValueError("split.csv contains no non-empty split values.")
    lowered = {s.lower() for s in non_empty}
    if lowered <= {"train", "val"}:
        return "train_val"
    for s in non_empty:
        try:
            v = int(s)
        except ValueError as exc:
            raise ValueError(
                f"split.csv: expected integer fold id or 'train'/'val'; got {s!r}"
            ) from exc
        if v < 0:
            raise ValueError(f"split.csv: fold id must be non-negative; got {v}")
    return "fold_ids"

dota_dataset_class_names(dataset)

Union class names from DOTADataset, ConcatDataset, or Subset.

Source code in oriented_det/data/dota.py

def dota_dataset_class_names(dataset) -> List[str]:
    """Union class names from ``DOTADataset``, ``ConcatDataset``, or ``Subset``."""
    from torch.utils.data import ConcatDataset, Subset

    if isinstance(dataset, Subset):
        return dota_dataset_class_names(dataset.dataset)
    if isinstance(dataset, ConcatDataset):
        seen: set[str] = set()
        names: List[str] = []
        for sub in dataset.datasets:
            for name in dota_dataset_class_names(sub):
                if name not in seen:
                    seen.add(name)
                    names.append(name)
        return sorted(names)
    if isinstance(dataset, DOTADataset):
        return dataset.get_class_names()
    raise TypeError(f"Unsupported dataset type for class discovery: {type(dataset)!r}")

dota_label_path_for_image(image_path, *, same_folder=False)

Resolve the DOTA annotation .txt path for an image file.

Source code in oriented_det/data/dota.py

def dota_label_path_for_image(image_path: Path, *, same_folder: bool = False) -> Path:
    """Resolve the DOTA annotation ``.txt`` path for an image file."""
    image_path = Path(image_path)
    if same_folder:
        return image_path.with_suffix(".txt")
    parent = image_path.parent
    if parent.name == "images":
        root = parent.parent
        label_dir = root / "labels"
        if not label_dir.exists():
            label_dir = root / "labelTxt"
        return label_dir / f"{image_path.stem}.txt"
    label_dir = parent / "labels"
    if label_dir.exists():
        return label_dir / f"{image_path.stem}.txt"
    label_txt = parent / "labelTxt"
    if label_txt.exists():
        return label_txt / f"{image_path.stem}.txt"
    return parent / f"{image_path.stem}.txt"

format_airbus_empty_gt_filter_log(dataset, *, split)

One-line summary for train logs when filter_empty_gt is enabled.

Source code in oriented_det/data/airbus_playground.py

def format_airbus_empty_gt_filter_log(dataset: AirbusPlaygroundCSVDataset, *, split: str) -> str:
    """One-line summary for train logs when ``filter_empty_gt`` is enabled."""
    discovered = dataset.tiles_discovered_count
    filtered = dataset.empty_gt_filtered_count
    kept = discovered - filtered
    return (
        f"  {split}: filter_empty_gt dropped {filtered} / {discovered} tiles "
        f"({kept} kept)"
    )

format_dota_empty_gt_filter_log(dataset, *, split)

One-line summary for train logs when filter_empty_gt is enabled.

Source code in oriented_det/data/dota.py

def format_dota_empty_gt_filter_log(dataset, *, split: str) -> str:
    """One-line summary for train logs when ``filter_empty_gt`` is enabled."""
    discovered, kept, filtered = dota_empty_gt_filter_summary(dataset)
    return (
        f"  {split}: filter_empty_gt dropped {filtered} / {discovered} tiles "
        f"({kept} kept)"
    )

format_dota_line(x1, y1, x2, y2, x3, y3, x4, y4, category, difficult)

Format a single annotation line in official DOTA format (comma-separated).

See https://captain-whu.github.io/DOTA/dataset.html

Source code in oriented_det/data/dota.py

def format_dota_line(
    x1: float, y1: float, x2: float, y2: float,
    x3: float, y3: float, x4: float, y4: float,
    category: str,
    difficult: int,
) -> str:
    """Format a single annotation line in official DOTA format (comma-separated).

    See https://captain-whu.github.io/DOTA/dataset.html
    """
    def _fmt(v: float) -> str:
        # DOTA commonly stores integer pixel coords, but floats are also valid.
        # Keep floats as-is, but drop the trailing ".0" when the value is integral.
        try:
            fv = float(v)
        except Exception:
            return str(v)
        if fv.is_integer():
            return str(int(fv))
        return str(fv)

    return (
        f"{_fmt(x1)}, {_fmt(y1)}, {_fmt(x2)}, {_fmt(y2)}, "
        f"{_fmt(x3)}, {_fmt(y3)}, {_fmt(x4)}, {_fmt(y4)}, {category}, {int(difficult)}"
    )

format_gt_best_iou_alignment_table(metrics, class_names=None, *, markdown=False)

Format per-class mean best IoU (same class) table.

Source code in oriented_det/data/evaluation.py

def format_gt_best_iou_alignment_table(
    metrics: GtBestIouAlignmentMetrics,
    class_names: Optional[Sequence[str]] = None,
    *,
    markdown: bool = False,
) -> str:
    """Format per-class mean best IoU (same class) table."""
    if not metrics.per_class:
        return ""

    if class_names:
        ordered = [c for c in class_names if c in metrics.per_class]
        ordered.extend(c for c in sorted(metrics.per_class) if c not in ordered)
    else:
        ordered = sorted(metrics.per_class.keys())

    rows = [(name, metrics.per_class[name]) for name in ordered]
    class_w = max(len("class"), *(len(name) for name, _ in rows))
    col_n = max(len("gts"), 3)
    col_any = max(len("mean_any"), 8)
    col_same = max(len("mean_same"), 9)
    col_med = max(len("med_same"), 8)

    if markdown:
        lines = [
            "| Class | gts | mean_any | mean_same | med_same |",
            "| --- | ---: | ---: | ---: | ---: |",
        ]
        for name, row in rows:
            lines.append(
                f"| `{name}` | {row.num_gts} | {row.mean_best_iou_any:.4f} | "
                f"{row.mean_best_iou_same_class:.4f} | {row.median_best_iou_same_class:.4f} |"
            )
        lines.append(
            f"| **global** | {metrics.num_gts} | {metrics.mean_best_iou_any:.4f} | "
            f"{metrics.mean_best_iou_same_class:.4f} | {metrics.median_best_iou_same_class:.4f} |"
        )
        return "\n".join(lines)

    def _sep() -> str:
        return (
            f"|{'-' * (class_w + 2)}|{'-' * (col_n + 2)}:|"
            f"{'-' * (col_any + 2)}:|{'-' * (col_same + 2)}:|{'-' * (col_med + 2)}:|"
        )

    lines = [
        f"| {'class'.ljust(class_w)} | {'gts'.rjust(col_n)} | "
        f"{'mean_any'.rjust(col_any)} | {'mean_same'.rjust(col_same)} | "
        f"{'med_same'.rjust(col_med)} |",
        _sep(),
    ]
    for name, row in rows:
        lines.append(
            f"| {name.ljust(class_w)} | {row.num_gts:>{col_n}d} | "
            f"{row.mean_best_iou_any:>{col_any}.4f} | {row.mean_best_iou_same_class:>{col_same}.4f} | "
            f"{row.median_best_iou_same_class:>{col_med}.4f} |"
        )
    lines.append(
        f"| {'global'.ljust(class_w)} | {metrics.num_gts:>{col_n}d} | "
        f"{metrics.mean_best_iou_any:>{col_any}.4f} | {metrics.mean_best_iou_same_class:>{col_same}.4f} | "
        f"{metrics.median_best_iou_same_class:>{col_med}.4f} |"
    )
    return "\n".join(lines)

format_gt_best_iou_alignment_table_from_dict(data, class_names=None, *, markdown=False)

Format table from :func:gt_best_iou_alignment_metrics_to_dict output.

Source code in oriented_det/data/evaluation.py

def format_gt_best_iou_alignment_table_from_dict(
    data: Dict[str, Any],
    class_names: Optional[Sequence[str]] = None,
    *,
    markdown: bool = False,
) -> str:
    """Format table from :func:`gt_best_iou_alignment_metrics_to_dict` output."""
    per_class_raw = data.get("per_class") or {}
    if not per_class_raw:
        return ""
    per_class = {
        name: ClassGtBestIouMetrics(
            num_gts=int(row.get("num_gts", 0)),
            mean_best_iou_any=float(row.get("mean_best_iou_any", 0.0)),
            mean_best_iou_same_class=float(row.get("mean_best_iou_same_class", 0.0)),
            median_best_iou_same_class=float(row.get("median_best_iou_same_class", 0.0)),
        )
        for name, row in per_class_raw.items()
    }
    metrics = GtBestIouAlignmentMetrics(
        num_gts=int(data.get("num_gts", 0)),
        mean_best_iou_any=float(data.get("mean_best_iou_any", 0.0)),
        mean_best_iou_same_class=float(data.get("mean_best_iou_same_class", 0.0)),
        median_best_iou_any=float(data.get("median_best_iou_any", 0.0)),
        median_best_iou_same_class=float(data.get("median_best_iou_same_class", 0.0)),
        per_class=per_class,
    )
    return format_gt_best_iou_alignment_table(metrics, class_names, markdown=markdown)

generate_airbus_playground_csvs(data_root, *, annotations_file='annotations.csv', split_file='split.csv', num_splits=10, seed=42, ignore_labels=None, map_labels=None, include_stats=False)

Generate annotations.csv and split.csv from Airbus export folders.

Each image group (dataset_id, zone_id, image_id) is assigned an integer fold id in 0 .. num_splits-1 (balanced round-robin on a shuffled group list). The split column stores that integer as text. By convention fold 0 is the default validation fold; set dataset.val_split_id in the training config to use another fold as validation.

Source code in oriented_det/data/airbus_playground.py

def generate_airbus_playground_csvs(
    data_root: str | Path,
    *,
    annotations_file: str = "annotations.csv",
    split_file: str = "split.csv",
    num_splits: int = 10,
    seed: int = 42,
    ignore_labels: Optional[Sequence[str]] = None,
    map_labels: Optional[Dict[str, str]] = None,
    include_stats: bool = False,
) -> Tuple[Path, Path] | Tuple[Path, Path, Dict[str, Any]]:
    """Generate annotations.csv and split.csv from Airbus export folders.

    Each image group ``(dataset_id, zone_id, image_id)`` is assigned an integer fold id in
    ``0 .. num_splits-1`` (balanced round-robin on a shuffled group list). The ``split`` column
    stores that integer as text. By convention fold ``0`` is the default validation fold; set
    ``dataset.val_split_id`` in the training config to use another fold as validation.
    """
    if num_splits < 2:
        raise ValueError("num_splits must be at least 2.")

    root = Path(data_root)
    tiles = discover_airbus_tiles(root)
    if not tiles:
        raise FileNotFoundError(f"No Airbus Playground tiles found under {root}")

    groups: Dict[Tuple[str, str, str], List[TileRecord]] = defaultdict(list)
    for tile in tiles:
        groups[_image_group_key(tile)].append(tile)

    group_keys = sorted(groups.keys())
    rng = random.Random(seed)
    rng.shuffle(group_keys)
    group_to_fold = {gk: str(i % num_splits) for i, gk in enumerate(group_keys)}

    ignore_set = set(ignore_labels or [])
    map_dict = dict(map_labels or {})
    stats: Dict[str, Any] = {
        "raw_label_counts": Counter(),
        "ignored_label_counts": Counter(),
        "final_label_counts": Counter(),
        "mapping_pairs": Counter(),
        "ignored_objects": 0,
        "mapped_objects": 0,
        "kept_objects": 0,
    }

    split_rows: List[Dict[str, str]] = []
    annotation_rows: List[Dict[str, str]] = []
    for tile in sorted(tiles, key=lambda x: x.tile_relpath):
        split_value = group_to_fold[_image_group_key(tile)]
        split_rows.append(
            {
                "tile_relpath": tile.tile_relpath,
                "dataset_id": tile.dataset_id,
                "zone_id": tile.zone_id,
                "image_id": tile.image_id,
                "tile_id": tile.tile_id,
                "split": split_value,
            }
        )

        label_rows = _load_airbus_label_rows(
            root / tile.label_relpath,
            ignore_labels=ignore_set,
            map_labels=map_dict,
            stats=stats,
        )
        for dota_coords, class_name, difficult in label_rows:
            annotation_rows.append(
                {
                    "tile_relpath": tile.tile_relpath,
                    "dataset_id": tile.dataset_id,
                    "zone_id": tile.zone_id,
                    "image_id": tile.image_id,
                    "tile_id": tile.tile_id,
                    "dota_coords": dota_coords,
                    "class_name": class_name,
                    "difficult": str(difficult),
                }
            )

    annotations_path = _resolve_csv_path(root, annotations_file)
    split_path = _resolve_csv_path(root, split_file)
    annotations_path.parent.mkdir(parents=True, exist_ok=True)
    split_path.parent.mkdir(parents=True, exist_ok=True)

    with annotations_path.open("w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=[
                "tile_relpath",
                "dataset_id",
                "zone_id",
                "image_id",
                "tile_id",
                "dota_coords",
                "class_name",
                "difficult",
            ],
        )
        writer.writeheader()
        writer.writerows(annotation_rows)

    with split_path.open("w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=["tile_relpath", "dataset_id", "zone_id", "image_id", "tile_id", "split"],
        )
        writer.writeheader()
        writer.writerows(split_rows)

    if include_stats:
        split_counts = Counter(row["split"] for row in split_rows)
        tiles_per_split = {k: int(split_counts[k]) for k in sorted(split_counts.keys(), key=int)}
        summary = {
            "total_tiles": len(split_rows),
            "total_groups": len(group_keys),
            "num_splits": num_splits,
            "tiles_per_split": tiles_per_split,
            "default_val_fold_id": 0,
            "seed": seed,
            "ignore_labels": sorted(ignore_set),
            "map_labels": map_dict,
            "raw_objects": int(sum(stats["raw_label_counts"].values())),
            "ignored_objects": int(stats["ignored_objects"]),
            "mapped_objects": int(stats["mapped_objects"]),
            "kept_objects": int(stats["kept_objects"]),
            "raw_label_counts": dict(sorted(stats["raw_label_counts"].items())),
            "ignored_label_counts": dict(sorted(stats["ignored_label_counts"].items())),
            "final_label_counts": dict(sorted(stats["final_label_counts"].items())),
            "mapping_pairs": dict(sorted(stats["mapping_pairs"].items())),
        }
        return annotations_path, split_path, summary

    return annotations_path, split_path

gt_best_iou_alignment_metrics_to_dict(metrics)

JSON-serializable view of :class:GtBestIouAlignmentMetrics.

Source code in oriented_det/data/evaluation.py

def gt_best_iou_alignment_metrics_to_dict(
    metrics: GtBestIouAlignmentMetrics,
) -> Dict[str, Any]:
    """JSON-serializable view of :class:`GtBestIouAlignmentMetrics`."""
    return {
        "num_gts": metrics.num_gts,
        "mean_best_iou_any": metrics.mean_best_iou_any,
        "mean_best_iou_same_class": metrics.mean_best_iou_same_class,
        "median_best_iou_any": metrics.median_best_iou_any,
        "median_best_iou_same_class": metrics.median_best_iou_same_class,
        "per_class": {
            name: {
                "num_gts": row.num_gts,
                "mean_best_iou_any": row.mean_best_iou_any,
                "mean_best_iou_same_class": row.mean_best_iou_same_class,
                "median_best_iou_same_class": row.median_best_iou_same_class,
            }
            for name, row in metrics.per_class.items()
        },
    }

resolve_dota_tile_roots(*, tiles_dirs=None, tiles_dir=None, split_label='split')

Resolve one or more DOTA tile roots from singular/plural config fields.

Source code in oriented_det/data/dota.py

def resolve_dota_tile_roots(
    *,
    tiles_dirs: Optional[Sequence[str | Path]] = None,
    tiles_dir: Optional[str | Path] = None,
    split_label: str = "split",
) -> List[Path]:
    """Resolve one or more DOTA tile roots from singular/plural config fields."""
    if tiles_dirs is not None:
        roots = [Path(p) for p in tiles_dirs]
        if not roots:
            raise ValueError(f"dataset.*_tiles_dirs for {split_label} must be non-empty when set")
        return roots
    if tiles_dir is not None:
        return [Path(tiles_dir)]
    raise ValueError(
        f"DOTA {split_label} requires dataset.*_tiles_dir or dataset.*_tiles_dirs"
    )

visualize_tiles(image_path, image_width, image_height, tiles, rboxes=None, class_names=None, *, output_path=None, tile_color=(255, 0, 0), box_color=(0, 255, 0), tile_width=2, box_width=2)

Visualize tiles and annotations for debugging tiling operations.

This function helps debug tiling by drawing: - Tile boundaries in red (by default) - Annotations in green (by default) - Tile IDs as labels

Parameters:

Name	Type	Description	Default
image_path	Path	Path to the image file	required
image_width	int	Width of the image	required
image_height	int	Height of the image	required
tiles	Sequence[Tile]	Sequence of Tile objects to visualize	required
rboxes	Optional[Sequence[RBox]]	Optional sequence of RBox annotations	None
class_names	Optional[Sequence[str]]	Optional class names for annotations	None
output_path	Optional[Path]	Optional path to save visualization	None
tile_color	Tuple[int, int, int]	RGB color for tile boundaries (default: red)	(255, 0, 0)
box_color	Tuple[int, int, int]	RGB color for annotation boxes (default: green)	(0, 255, 0)
tile_width	int	Line width for tile boundaries	2
box_width	int	Line width for annotation boxes	2

Returns:

Type	Description
any	PIL Image with visualization (or numpy array if PIL unavailable)

Example

tiler = ImageTiler(tile_size=512, overlap=0.2) tiles = tiler.generate_tiles(2000, 2000) visualize_tiles( ... image_path=Path("image.png"), ... image_width=2000, ... image_height=2000, ... tiles=tiles, ... rboxes=annotations, ... output_path=Path("tiles_vis.png"), ... )

Source code in oriented_det/data/tiling.py

def visualize_tiles(
    image_path: Path,
    image_width: int,
    image_height: int,
    tiles: Sequence[Tile],
    rboxes: Optional[Sequence[RBox]] = None,
    class_names: Optional[Sequence[str]] = None,
    *,
    output_path: Optional[Path] = None,
    tile_color: Tuple[int, int, int] = (255, 0, 0),
    box_color: Tuple[int, int, int] = (0, 255, 0),
    tile_width: int = 2,
    box_width: int = 2,
) -> any:  # type: ignore
    """Visualize tiles and annotations for debugging tiling operations.

    This function helps debug tiling by drawing:
    - Tile boundaries in red (by default)
    - Annotations in green (by default)
    - Tile IDs as labels

    Args:
        image_path: Path to the image file
        image_width: Width of the image
        image_height: Height of the image
        tiles: Sequence of Tile objects to visualize
        rboxes: Optional sequence of RBox annotations
        class_names: Optional class names for annotations
        output_path: Optional path to save visualization
        tile_color: RGB color for tile boundaries (default: red)
        box_color: RGB color for annotation boxes (default: green)
        tile_width: Line width for tile boundaries
        box_width: Line width for annotation boxes

    Returns:
        PIL Image with visualization (or numpy array if PIL unavailable)

    Example:
        >>> tiler = ImageTiler(tile_size=512, overlap=0.2)
        >>> tiles = tiler.generate_tiles(2000, 2000)
        >>> visualize_tiles(
        ...     image_path=Path("image.png"),
        ...     image_width=2000,
        ...     image_height=2000,
        ...     tiles=tiles,
        ...     rboxes=annotations,
        ...     output_path=Path("tiles_vis.png"),
        ... )
    """
    try:
        from PIL import Image
        from ..utils import viz
    except ImportError:
        raise RuntimeError("PIL/Pillow is required for tile visualization.")

    # Load or create image
    if image_path.exists():
        image = Image.open(image_path).convert("RGB")
    else:
        # Create blank image for visualization
        image = Image.new("RGB", (image_width, image_height), color="white")

    # Draw tile boundaries
    tile_polygons = []
    tile_specs = []
    for tile in tiles:
        tile_poly = Polygon.rectangle(
            tile.x + tile.width / 2.0,
            tile.y + tile.height / 2.0,
            tile.width,
            tile.height
        )
        tile_polygons.append(tile_poly.points)
        tile_specs.append(viz.DrawingSpec(outline=tile_color, width=tile_width))

    # Draw tiles
    image = viz.draw_polygons(
        image,
        tile_polygons,
        specs=tile_specs,
    )

    # Draw annotations if provided
    if rboxes:
        annotation_polygons = [rbox.to_polygon().points for rbox in rboxes]
        annotation_specs = [
            viz.DrawingSpec(outline=box_color, width=box_width)
            for _ in annotation_polygons
        ]
        image = viz.draw_polygons(
            image,
            annotation_polygons,
            specs=annotation_specs,
        )

    if output_path:
        image.save(output_path)

    return image

Important Notes

DOTA Dataset Loading Modes

The DOTA loader supports three modes for organizing your dataset:

1. Pattern matching (default): Matches annotation files by split pattern (e.g., *_train.txt)
2. Split file (official DOTA convention): Uses a file listing image names (e.g., train.txt)
3. Separate folders: Train/val/test in different directories

DOTA Polygon Format

• DOTA uses 8 coordinates: x1 y1 x2 y2 x3 y3 x4 y4 class_name difficult
• Corners are ordered sequentially around the polygon perimeter
• The loader converts polygons to QBox (which normalizes point order) and then to RBox
• QBox ensures counter-clockwise orientation and orders points starting from top-most

Data Augmentation

OrientedDet supports two types of data augmentation:

1. Geometric Transforms (Oriented Bounding Box Aware): HorizontalFlip, VerticalFlip, Rotate, Compose
2. These transforms modify both the image and oriented bounding boxes

3. Angle information is preserved correctly

4. Albumentations (Non-Geometric Only): create_albumentations_augmentation, AlbumentationsTransform

5. Only non-geometric augmentations are supported (color, contrast, blur, noise, etc.)
6. Note: Albumentations does not support oriented bounding boxes, so only non-geometric transforms can be used

Examples

DOTA Dataset Loading

from oriented_det.data import DOTADataset, build_dota_loader

# Mode 1: Pattern matching (backward compatible)
dataset = DOTADataset(
    root_dir="/path/to/dota",
    split="train",
    allowed_classes=["plane", "ship", "vehicle"],
    difficult_strategy="drop"
)

# Mode 2: Using split file (official DOTA convention)
dataset = DOTADataset(
    root_dir="/path/to/dota",
    split="train",
    split_file="train.txt",  # Lists image names, one per line
    allowed_classes=["plane", "ship", "vehicle"],
    difficult_strategy="drop"
)

# Mode 3: Separate folders for each split
train_dataset = DOTADataset(
    root_dir="/path/to/data_root",
    split="train",
    label_dir="/path/to/data_root/train/labelTxt",
    image_dir="/path/to/data_root/train/images",
    difficult_strategy="drop"
)

# Or use PyTorch DataLoader
loader = build_dota_loader(
    root_dir="/path/to/dota",
    split="train",
    split_file="train.txt",  # Optional
    batch_size=4,
    shuffle=True
)

Image Tiling

from oriented_det.data import ImageTiler, visualize_tiles
from pathlib import Path

# Configure tiler with overlap and filtering options
tiler = ImageTiler(
    tile_size=1024,
    overlap=0.2,  # 20% overlap between tiles
    min_box_area=64,  # Filter boxes < 64 pixels²
    min_overlap_ratio=0.3,  # Keep box only if >= 30% overlaps tile
    edge_handling="clip"  # "clip", "ignore", or "keep"
)

# Generate tiles and process
for tiled_sample in tiler.tile_image(
    image_path=Path("large_image.png"),
    image_width=4000,
    image_height=4000,
    rboxes=annotations,
    class_names=classes
):
    # Process each tile
    process_tile(tiled_sample)

# Visualize tiles for debugging
tiles = tiler.generate_tiles(4000, 4000)
visualize_tiles(
    image_path=Path("large_image.png"),
    image_width=4000,
    image_height=4000,
    tiles=tiles,
    rboxes=annotations,
    class_names=classes,
    output_path=Path("tiles_vis.png")
)

Data Augmentation

from oriented_det.data import HorizontalFlip, Rotate, Compose, create_albumentations_augmentation

# Geometric transforms (oriented bounding box aware)
aug = Compose([
    HorizontalFlip(p=0.5),
    Rotate(degrees=90, p=0.3),
])

augmented_image, augmented_boxes = aug(image, rboxes, image_width, image_height)

# Albumentations (non-geometric only)
aug = create_albumentations_augmentation(
    brightness_limit=0.2,
    contrast_limit=0.2,
    gamma_limit=(80, 120),
    gauss_noise_var_limit=(10.0, 50.0),
    blur_limit=3,
    clahe_clip_limit=4.0,
    p_brightness_contrast=0.5,
    p_gamma=0.3,
    p_noise=0.2,
    p_blur=0.2,
    p_clahe=0.3,
)

augmented_image = aug(image)  # Returns PIL Image

Oriented mAP Evaluation

from oriented_det.data import Detection, GroundTruth, compute_oriented_map
from oriented_det.geometry import RBox

detections = {
    "img1": [Detection(rbox=..., score=0.9, class_id=0, class_name="plane")],
}
ground_truths = {
    "img1": [GroundTruth(rbox=..., class_id=0, class_name="plane")],
}

mean_ap, class_aps = compute_oriented_map(
    detections, ground_truths, iou_threshold=0.5
)