API Reference: Metrics & Datasets

Performance evaluation metrics and dataset interfaces for CoreCV.

---

Evaluation Metrics

Classification Metrics

corecv.metrics.classification

GPU-native classification metrics engine for CoreCV.

Provides an accumulator-based :class:ClassificationMetrics that computes accuracy, top-k accuracy, precision, recall, and F1 scores — all on VRAM with zero CPU-GPU synchronisations during update(). Intermediate confusion-matrix counts are maintained in GPU tensors via nn.register_buffer so they travel with .to(device) automatically.

The compute() method performs only the final scalar divisions and returns a plain Python dict of float values — that is the only point at which implicit CPU transfer occurs (via tensor.item()).

No pycocotools, no Python for-loops, no .cpu() calls in hot paths.

Example

import torch from corecv.metrics.classification import ClassificationMetrics device = torch.device("cuda" if torch.cuda.is_available() else "cpu") metrics = ClassificationMetrics(num_classes=10, device=device) logits = torch.randn(32, 10, device=device) targets = torch.randint(0, 10, (32,), device=device) metrics.update(logits, targets) results = metrics.compute() results["accuracy"] # Top-1 accuracy 0.125

ClassificationMetrics(num_classes, top_k=(1, 5), device='cpu')

Bases: Module

Accumulator-based single-label multi-class classification metrics.

Maintains per-class true-positive, false-positive, and false-negative counts in VRAM-resident buffers so that update() never triggers a CPU-GPU synchronisation. Only compute() performs the final reductions.

Supported metrics:

• accuracy — Top-1 accuracy (mean over batch).
• top{k}_accuracy — Top-k accuracy for each k in top_k.
• precision_macro / recall_macro / f1_score_macro — Macro-averaged precision, recall, and F1 (per-class, then averaged).
• precision_micro / recall_micro / f1_score_micro — Micro-averaged (global TP, FP, FN pooled before ratio).

Parameters:

Name	Type	Description	Default
num_classes	int	Number of mutually-exclusive classes.	required
top_k	tuple[int, ...]	Tuple of k values for top-k accuracy reporting. Default (1, 5).	(1, 5)
device	device | str	Device on which to allocate the accumulator buffers.	'cpu'

Example

metrics = ClassificationMetrics(num_classes=100, top_k=(1, 5), device="cuda") for logits, targets in loader: ... metrics.update(logits, targets) print(metrics.compute())

Initialise ClassificationMetrics with num_classes, top_k, and device.

Source code in src/corecv/metrics/classification.py

def __init__(
    self,
    num_classes: int,
    top_k: tuple[int, ...] = (1, 5),
    device: torch.device | str = "cpu",
) -> None:
    """Initialise ClassificationMetrics with num_classes, top_k, and device."""
    super().__init__()
    if num_classes < 1:
        msg = f"num_classes must be >= 1, got {num_classes}"
        raise ValueError(msg)
    for k in top_k:
        if k < 1:
            msg = f"Each k in top_k must be >= 1, got {k}"
            raise ValueError(msg)

    self.num_classes = int(num_classes)
    self.top_k = tuple(sorted(top_k))
    self.device = torch.device(device)

    # ------------------------------------------------------------------
    # Accumulator buffers — all on VRAM, zero sync in update().
    # confusion[c, d] = count of samples whose true label is c and whose
    #   predicted label is d.  Shape: (C, C).
    # ------------------------------------------------------------------
    self.register_buffer(
        "confusion",
        torch.zeros(num_classes, num_classes, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "total_samples",
        torch.zeros(1, dtype=torch.int64, device=self.device),
    )

compute()

Compute all metrics from the accumulated state.

Returns:

Type	Description
dict[str, float | Tensor]	Dictionary with the following keys:
dict[str, float | Tensor]	"accuracy" — Top-1 accuracy (float).
dict[str, float | Tensor]	"top{k}_accuracy" — Top-k accuracy for each k (float).
dict[str, float | Tensor]	"precision_macro" — Macro-averaged precision (float).
dict[str, float | Tensor]	"recall_macro" — Macro-averaged recall (float).
dict[str, float | Tensor]	"f1_score_macro" — Macro-averaged F1 (float).
dict[str, float | Tensor]	"precision_micro" — Micro-averaged precision (float).
dict[str, float | Tensor]	"recall_micro" — Micro-averaged recall (float).
dict[str, float | Tensor]	"f1_score_micro" — Micro-averaged F1 (float).

Source code in src/corecv/metrics/classification.py

def compute(self) -> dict[str, float | Tensor]:
    """Compute all metrics from the accumulated state.

    Returns:
        Dictionary with the following keys:

        * ``"accuracy"`` — Top-1 accuracy (float).
        * ``"top{k}_accuracy"`` — Top-k accuracy for each *k*
            (float).
        * ``"precision_macro"`` — Macro-averaged precision (float).
        * ``"recall_macro"`` — Macro-averaged recall (float).
        * ``"f1_score_macro"`` — Macro-averaged F1 (float).
        * ``"precision_micro"`` — Micro-averaged precision (float).
        * ``"recall_micro"`` — Micro-averaged recall (float).
        * ``"f1_score_micro"`` — Micro-averaged F1 (float).
    """
    total: int = self.total_samples.item()
    if total == 0:
        return self._empty_results()

    # confusion[c, d] = # samples true=c, pred=d
    confusion: Tensor = self.confusion.float()  # (C, C)

    # ---- Per-class TP / FP / FN ------------------------------------
    tp_per_class: Tensor = confusion.diag()  # (C,)
    fp_per_class: Tensor = confusion.sum(dim=0) - tp_per_class  # (C,)
    fn_per_class: Tensor = confusion.sum(dim=1) - tp_per_class  # (C,)

    # ---- Top-1 accuracy (diagonal sum / total) --------------------
    top1_accuracy: float = (tp_per_class.sum() / total).item()

    # ---- Top-k accuracy via torch.topk on confusion columns --------
    # For each sample class c, the model's top-k predictions span the
    # largest k entries in column c of the confusion matrix.  We
    # compute this by summing the top-k values in each column and
    # dividing by the column total.
    topk_accuracies: dict[int, float] = {}
    for k in self.top_k:
        k_clamped: int = min(k, self.num_classes)
        # topk along dim=0 (true-class axis) for each predicted class
        topk_vals, _ = torch.topk(confusion, k_clamped, dim=0)  # (k, C)
        topk_sum: Tensor = topk_vals.sum(dim=0)  # (C,)
        # Weighted average: sum of topk_sum / total
        topk_acc: float = (topk_sum.sum() / total).item()
        topk_accuracies[k] = topk_acc

    # ---- Precision / Recall / F1 — per class, then macro ----------
    eps: float = 1e-8
    precision_per_class: Tensor = tp_per_class / (tp_per_class + fp_per_class + eps)
    recall_per_class: Tensor = tp_per_class / (tp_per_class + fn_per_class + eps)
    f1_per_class: Tensor = (
        2.0 * precision_per_class * recall_per_class
        / (precision_per_class + recall_per_class + eps)
    )

    precision_macro: float = precision_per_class.mean().item()
    recall_macro: float = recall_per_class.mean().item()
    f1_macro: float = f1_per_class.mean().item()

    # ---- Micro-averaged (global TP / FP / FN) --------------------
    tp_total: Tensor = tp_per_class.sum()
    fp_total: Tensor = fp_per_class.sum()
    fn_total: Tensor = fn_per_class.sum()

    precision_micro: float = (tp_total / (tp_total + fp_total + eps)).item()
    recall_micro: float = (tp_total / (tp_total + fn_total + eps)).item()
    f1_micro: float = (
        (2.0 * tp_total) / (2.0 * tp_total + fp_total + fn_total + eps)
    ).item()

    # ---- Assemble results dict ------------------------------------
    results: dict[str, float | Tensor] = {
        "accuracy": top1_accuracy,
        "precision_macro": precision_macro,
        "recall_macro": recall_macro,
        "f1_score_macro": f1_macro,
        "precision_micro": precision_micro,
        "recall_micro": recall_micro,
        "f1_score_micro": f1_micro,
    }

    for k, acc in topk_accuracies.items():
        results[f"top{k}_accuracy"] = acc

    return results

reset()

Reset all accumulator buffers to zero.

Source code in src/corecv/metrics/classification.py

def reset(self) -> None:
    """Reset all accumulator buffers to zero."""
    self.confusion.zero_()
    self.total_samples.zero_()

update(preds, targets)

Accumulate a batch of predictions and ground-truth labels.

Parameters:

Name	Type	Description	Default
preds	Tensor	Model output of shape (B, C) — raw logits or probabilities. If logits, argmax is taken to derive predicted classes.	required
targets	Tensor	Ground-truth class indices of shape (B,) with values in [0, C).	required

Raises:

Type	Description
ValueError	If tensor shapes or value ranges are invalid.

Source code in src/corecv/metrics/classification.py

def update(self, preds: Tensor, targets: Tensor) -> None:  # noqa: D401
    """Accumulate a batch of predictions and ground-truth labels.

    Args:
        preds: Model output of shape ``(B, C)`` — raw logits or
            probabilities.  If logits, argmax is taken to derive
            predicted classes.
        targets: Ground-truth class indices of shape ``(B,)`` with
            values in ``[0, C)``.

    Raises:
        ValueError: If tensor shapes or value ranges are invalid.
    """
    self._validate_inputs(preds, targets)

    # Derive predicted class indices from logits / probs.
    pred_labels: Tensor = preds.argmax(dim=1)  # (B,)

    one_hot_pred: Tensor = F.one_hot(pred_labels, self.num_classes).to(
        dtype=torch.int64,
        device=self.device,
    )  # (B, C)
    one_hot_target: Tensor = F.one_hot(
        targets.to(dtype=torch.long), self.num_classes
    ).to(dtype=torch.int64, device=self.device)  # (B, C)

    # Outer-product accumulation (no .item(), no .cpu()).
    self.confusion += one_hot_target.T @ one_hot_pred  # (C, C)
    self.total_samples += targets.shape[0]

Detection Metrics

corecv.metrics.detection

GPU-native detection metrics engine for CoreCV.

Provides an accumulator-based :class:DetectionMetrics that computes mAP@50 and mAP@50:95 — the COCO-style mean Average Precision metrics — entailining zero CPU-GPU synchronisations during update().

The implementation is a fully vectorised, GPU-resident alternative to pycocotools. Per-class TP/FP counts for every IoU threshold are maintained in VRAM via nn.register_buffer. The compute() method performs precision-recall curve computation and AP integration, then returns a plain Python dict.

No pycocotools dependency, no Python for-loops over individual predictions, no .cpu() calls in hot paths.

IoU computation uses torchvision.ops.box_iou which runs on GPU via custom CUDA kernels.

Example

import torch from corecv.metrics.detection import DetectionMetrics device = torch.device("cuda" if torch.cuda.is_available() else "cpu") metrics = DetectionMetrics(num_classes=80, device=device)

Per-image predictions and targets

pred_boxes = [torch.randn(5, 4, device=device).abs() * 100 for _ in range(4)] pred_scores = [torch.rand(5, device=device) for _ in range(4)] pred_labels = [torch.randint(0, 80, (5,), device=device) for _ in range(4)] target_boxes = [torch.randn(3, 4, device=device).abs() * 100 for _ in range(4)] target_labels = [torch.randint(0, 80, (3,), device=device) for _ in range(4)] metrics.update(pred_boxes, pred_scores, pred_labels, target_boxes, target_labels) results = metrics.compute() results["map50"] 0.0

DetectionMetrics(num_classes, iou_thresholds=None, device='cpu')

Bases: Module

Accumulator-based detection metrics (mAP@50, mAP@50:95).

Implements a vectorised version of the COCO mAP computation entirely in VRAM. For each IoU threshold and each class, predictions are sorted by confidence and matched greedily to ground-truth targets. The resulting TP/FP indicator arrays are accumulated in buffers and processed in compute() to build precision-recall curves and integrate AP via the all-point interpolation method (area under the PR curve).

Supported metrics:

• map50 — mAP at IoU threshold 0.50.
• map50_95 — mAP averaged over IoU thresholds 0.50 : 0.05 : 0.95 (10 thresholds by default).
• per_class_ap50 — AP@50 per class as a Tensor.
• per_class_ap50_95 — AP@50:95 per class as a Tensor.

Parameters:

Name	Type	Description	Default
num_classes	int	Number of object classes (excluding background).	required
iou_thresholds	Tensor | list[float] | None	IoU thresholds at which to compute AP. Default torch.linspace(0.5, 0.95, 10) (COCO convention).	None
device	device | str	Device on which to allocate accumulator buffers.	'cpu'

Note

Boxes must be in (x1, y1, x2, y2) (top-left, bottom-right) format. No coordinate conversion is performed internally.

Example

metrics = DetectionMetrics(num_classes=80, device="cuda")

Accumulate over an epoch ...

results = metrics.compute()

Initialise DetectionMetrics with num_classes, iou_thresholds, and device.

Source code in src/corecv/metrics/detection.py

def __init__(
    self,
    num_classes: int,
    iou_thresholds: Tensor | list[float] | None = None,
    device: torch.device | str = "cpu",
) -> None:
    """Initialise DetectionMetrics with num_classes, iou_thresholds, and device."""
    super().__init__()
    if num_classes < 1:
        msg = f"num_classes must be >= 1, got {num_classes}"
        raise ValueError(msg)

    self.num_classes = int(num_classes)
    self.device = torch.device(device)

    # IoU thresholds tensor — shape (T,).
    if iou_thresholds is None:
        iou_thresholds_t: Tensor = torch.linspace(0.5, 0.95, 10)
    elif isinstance(iou_thresholds, list):
        iou_thresholds_t = torch.tensor(iou_thresholds, dtype=torch.float32)
    else:
        iou_thresholds_t = iou_thresholds.float()

    self.register_buffer("iou_thresholds", iou_thresholds_t.to(self.device))
    self.num_thresholds: int = self.iou_thresholds.shape[0]

    # ------------------------------------------------------------------
    # Accumulator buffers — (num_classes, num_thresholds).
    #
    # For each (class, threshold) pair we accumulate:
    #   tp_count[t, c, k] = cumulative true positives at rank k
    #   fp_count[t, c, k] = cumulative false positives at rank k
    #   num_targets[c]    = total ground-truth instances of class c
    #
    # We store *cumulative* counts along the rank dimension to avoid
    # needing the raw per-detection arrays.  However, since the number
    # of detections varies per call, we store the raw TP/FP counts per
    # call and merge them during compute().  A simpler approach: store
    # per-call TP/FP as lists and merge in compute().  For true VRAM
    # residency we pre-allocate with a large capacity.
    #
    # Practical approach: accumulate per-class target counts (fixed),
    # and store TP/FP lists per class per threshold that are merged
    # in compute().  To stay fully on VRAM without dynamic lists, we
    # accumulate into fixed-size buffers and process in compute().
    #
    # We use a two-phase approach:
    #   update()  -> accumulates all-pairs matching results on VRAM
    #   compute() -> processes the accumulated results
    # ------------------------------------------------------------------
    self.register_buffer(
        "target_counts",
        torch.zeros(num_classes, dtype=torch.int64, device=self.device),
    )

    # Per-class TP/FP scores and matching info will be accumulated as
    # lists of tensors (one per update call).  While these live in Python
    # lists, the tensor data itself is on VRAM.  The compute() method
    # concatenates and processes them.
    self._pred_scores: list[list[Tensor]] = []  # [class][call]
    self._pred_tp: list[list[Tensor]] = []       # [class][call] — TP flags
    self._pred_fp: list[list[Tensor]] = []       # [class][call] — FP flags

    # Per-threshold TP/FP counts: (T, C) accumulated per call
    self._tp_counts_per_threshold: list[Tensor] = []
    self._fp_counts_per_threshold: list[Tensor] = []

compute()

Compute mAP metrics from the accumulated state.

Returns:

Type	Description
dict[str, float | Tensor]	Dictionary with the following keys:
dict[str, float | Tensor]	"map50" — mAP at IoU=0.50 (float).
dict[str, float | Tensor]	"map50_95" — mAP averaged over all IoU thresholds (float).
dict[str, float | Tensor]	"per_class_ap50" — AP@50 per class, tensor (C,).
dict[str, float | Tensor]	"per_class_ap50_95" — AP@50:95 per class, tensor (C,).

Source code in src/corecv/metrics/detection.py

def compute(self) -> dict[str, float | Tensor]:  # noqa: PLR0915
    """Compute mAP metrics from the accumulated state.

    Returns:
        Dictionary with the following keys:

        * ``"map50"`` — mAP at IoU=0.50 (float).
        * ``"map50_95"`` — mAP averaged over all IoU thresholds (float).
        * ``"per_class_ap50"`` — AP@50 per class, tensor ``(C,)``.
        * ``"per_class_ap50_95"`` — AP@50:95 per class, tensor ``(C,)``.
    """
    T: int = self.num_thresholds
    C: int = self.num_classes

    # Per-class AP at each threshold: (T, C)
    ap_per_threshold: Tensor = torch.zeros(
        T, C, dtype=torch.float32, device=self.device
    )

    for cls_idx in range(C):
        num_gt: int = self.target_counts[cls_idx].item()
        if num_gt == 0:
            # No ground-truth for this class — AP is 0 at all thresholds.
            continue

        # Gather all detections for this class across all update() calls.
        # self._pred_scores is structured as [class][call], so we must
        # iterate over calls *within* the current class's list.
        all_scores: list[Tensor] = []
        all_tp: list[Tensor] = []
        all_fp: list[Tensor] = []

        class_scores: list[Tensor] = self._pred_scores[cls_idx]
        class_tp: list[Tensor] = self._pred_tp[cls_idx]
        class_fp: list[Tensor] = self._pred_fp[cls_idx]
        for call_idx in range(len(class_scores)):
            scores = class_scores[call_idx]
            tp_flags = class_tp[call_idx]
            fp_flags = class_fp[call_idx]
            if scores.numel() > 0:
                all_scores.append(scores)
                all_tp.append(tp_flags)
                all_fp.append(fp_flags)

        if not all_scores:
            continue

        # Concatenate across calls: (D_total,) each
        scores_cat: Tensor = torch.cat(all_scores)
        tp_cat: Tensor = torch.cat(all_tp)
        fp_cat: Tensor = torch.cat(all_fp)

        # Sort by score descending
        sorted_indices: Tensor = scores_cat.argsort(descending=True)
        tp_sorted: Tensor = tp_cat[sorted_indices]  # (D_total, T)
        fp_sorted: Tensor = fp_cat[sorted_indices]  # (D_total, T)

        # Cumulative TP and FP along detection rank
        tp_cum: Tensor = tp_sorted.cumsum(dim=0)  # (D_total, T)
        fp_cum: Tensor = fp_sorted.cumsum(dim=0)  # (D_total, T)

        # Precision and recall curves: (D_total, T)
        precision: Tensor = tp_cum / (tp_cum + fp_cum + 1e-8)
        recall: Tensor = tp_cum / float(num_gt)

        # Compute AP for each threshold using all-point interpolation
        # (area under the precision-recall curve).
        for t_idx in range(T):
            prec_t: Tensor = precision[:, t_idx]  # (D_total,)
            rec_t: Tensor = recall[:, t_idx]      # (D_total,)

            # Prepend (0, 1) and append (1, 0) for proper curve area.
            rec_t = torch.cat(
                [torch.zeros(1, device=self.device), rec_t]
            )
            prec_t = torch.cat(
                [torch.ones(1, device=self.device), prec_t]
            )

            # All-point interpolation: enforce monotonically decreasing
            # precision from right to left.
            for i in range(prec_t.shape[0] - 2, -1, -1):
                prec_t[i] = torch.max(prec_t[i], prec_t[i + 1])

            # Find points where recall changes (unique recall values).
            rec_diff: Tensor = rec_t[1:] - rec_t[:-1]  # (D_total,)
            # AP = sum of precision * delta_recall where recall increases
            ap_t: Tensor = (prec_t[:-1] * rec_diff.clamp(min=0)).sum()
            ap_per_threshold[t_idx, cls_idx] = ap_t

    # ---- Aggregate results ----------------------------------------
    # mAP@50: mean AP over classes at threshold index 0
    iou_05_idx: int = 0
    # Find the threshold closest to 0.5
    if self.num_thresholds > 1:
        iou_05_idx = int(
            (self.iou_thresholds - 0.5).abs().argmin().item()
        )

    per_class_ap50: Tensor = ap_per_threshold[iou_05_idx]  # (C,)
    per_class_ap50_95: Tensor = ap_per_threshold.mean(dim=0)  # (C,)

    # Classes with no GT get 0 AP — they don't affect the mean if we
    # only average over classes with GT (COCO convention).
    classes_with_gt: Tensor = self.target_counts > 0
    num_classes_with_gt: int = classes_with_gt.sum().item()

    if num_classes_with_gt > 0:
        map50: float = per_class_ap50[classes_with_gt].mean().item()
        map50_95: float = per_class_ap50_95[classes_with_gt].mean().item()
    else:
        map50 = 0.0
        map50_95 = 0.0

    return {
        "map50": map50,
        "map50_95": map50_95,
        "per_class_ap50": per_class_ap50,
        "per_class_ap50_95": per_class_ap50_95,
    }

reset()

Reset all accumulator state.

Source code in src/corecv/metrics/detection.py

def reset(self) -> None:
    """Reset all accumulator state."""
    self.target_counts.zero_()
    self._pred_scores.clear()
    self._pred_tp.clear()
    self._pred_fp.clear()
    self._tp_counts_per_threshold.clear()
    self._fp_counts_per_threshold.clear()

update(pred_boxes, pred_scores, pred_labels, target_boxes, target_labels)

Accumulate detection results for a batch of images.

All tensor arguments are lists-of-tensors, one element per image in the batch. Empty tensors (zero detections or zero targets) are supported.

Parameters:

Name	Type	Description	Default
pred_boxes	list[Tensor]	pred_boxes[i] has shape (N_i, 4) in (x1, y1, x2, y2) format.	required
pred_scores	list[Tensor]	pred_scores[i] has shape (N_i,) with confidence scores.	required
pred_labels	list[Tensor]	pred_labels[i] has shape (N_i,) with class indices in [0, num_classes).	required
target_boxes	list[Tensor]	target_boxes[i] has shape (M_i, 4) in (x1, y1, x2, y2) format.	required
target_labels	list[Tensor]	target_labels[i] has shape (M_i,) with class indices in [0, num_classes).	required

Source code in src/corecv/metrics/detection.py

def update(
    self,
    pred_boxes: list[Tensor],
    pred_scores: list[Tensor],
    pred_labels: list[Tensor],
    target_boxes: list[Tensor],
    target_labels: list[Tensor],
) -> None:
    """Accumulate detection results for a batch of images.

    All tensor arguments are lists-of-tensors, one element per image
    in the batch.  Empty tensors (zero detections or zero targets) are
    supported.

    Args:
        pred_boxes: ``pred_boxes[i]`` has shape ``(N_i, 4)`` in
            ``(x1, y1, x2, y2)`` format.
        pred_scores: ``pred_scores[i]`` has shape ``(N_i,)`` with
            confidence scores.
        pred_labels: ``pred_labels[i]`` has shape ``(N_i,)`` with
            class indices in ``[0, num_classes)``.
        target_boxes: ``target_boxes[i]`` has shape ``(M_i, 4)`` in
            ``(x1, y1, x2, y2)`` format.
        target_labels: ``target_labels[i]`` has shape ``(M_i,)`` with
            class indices in ``[0, num_classes)``.
    """
    batch_size: int = len(pred_boxes)
    if not (batch_size == len(pred_scores) == len(pred_labels)
            == len(target_boxes) == len(target_labels)):
        msg = "All input lists must have the same length (batch size)"
        raise ValueError(msg)

    # Process each image — matching is done per-image, per-class.
    for img_idx in range(batch_size):
        self._update_single_image(
            pred_boxes[img_idx],
            pred_scores[img_idx],
            pred_labels[img_idx],
            target_boxes[img_idx],
            target_labels[img_idx],
        )

Segmentation Metrics

corecv.metrics.segmentation

GPU-native semantic segmentation metrics engine for CoreCV.

Provides an accumulator-based :class:SegmentationMetrics that computes mean IoU, pixel accuracy, and mean Dice coefficient — all on VRAM with zero CPU-GPU synchronisations during update(). Per-class intersection and union counts are stored in GPU-resident buffers via nn.register_buffer.

The compute() method performs only the final reductions and returns a plain Python dict — that is the only point where implicit CPU transfer occurs.

No pycocotools, no Python for-loops, no .cpu() calls in hot paths.

Example

import torch from corecv.metrics.segmentation import SegmentationMetrics device = torch.device("cuda" if torch.cuda.is_available() else "cpu") metrics = SegmentationMetrics(num_classes=21, device=device) logits = torch.randn(4, 21, 128, 128, device=device) targets = torch.randint(0, 21, (4, 128, 128), device=device) metrics.update(logits, targets) results = metrics.compute() results["miou"] 0.05

SegmentationMetrics(num_classes, ignore_index=-100, device='cpu')

Bases: Module

Accumulator-based semantic segmentation metrics.

Maintains per-class intersection and union counts in VRAM-resident buffers so that update() never triggers a CPU-GPU synchronisation. Only compute() performs the final divisions.

Supported metrics:

• miou — Mean Intersection-over-Union averaged over valid classes.
• pixel_accuracy — Fraction of correctly classified pixels (among valid, non-ignored pixels).
• mean_dice — Mean Dice coefficient averaged over valid classes.
• per_class_iou — IoU for each class as a Tensor of shape (num_classes,).
• per_class_dice — Dice for each class as a Tensor of shape (num_classes,).

Parameters:

Name	Type	Description	Default
num_classes	int	Number of segmentation classes.	required
ignore_index	int	Class index to exclude from metric computation. Default -100 (matches F.cross_entropy convention).	-100
device	device | str	Device on which to allocate accumulator buffers.	'cpu'

Example

metrics = SegmentationMetrics(num_classes=21, ignore_index=255, device="cuda") for logits, targets in loader: ... metrics.update(logits, targets) print(metrics.compute()["miou"])

Initialise SegmentationMetrics with num_classes, ignore_index, and device.

Source code in src/corecv/metrics/segmentation.py

def __init__(
    self,
    num_classes: int,
    ignore_index: int = -100,
    device: torch.device | str = "cpu",
) -> None:
    """Initialise SegmentationMetrics with num_classes, ignore_index, and device."""
    super().__init__()
    if num_classes < 1:
        msg = f"num_classes must be >= 1, got {num_classes}"
        raise ValueError(msg)

    self.num_classes = int(num_classes)
    self.ignore_index = int(ignore_index)
    self.device = torch.device(device)

    # ------------------------------------------------------------------
    # Accumulator buffers — all on VRAM.
    #
    # For each class c we track:
    #   intersection[c] = # pixels where pred == c AND target == c
    #   union[c]        = # pixels where pred == c OR  target == c
    #   pred_sum[c]     = # pixels where pred == c  (for Dice numerator)
    #   target_sum[c]   = # pixels where target == c (for Dice numerator)
    # ------------------------------------------------------------------
    self.register_buffer(
        "intersection",
        torch.zeros(num_classes, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "union",
        torch.zeros(num_classes, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "pred_sum",
        torch.zeros(num_classes, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "target_sum",
        torch.zeros(num_classes, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "total_correct",
        torch.zeros(1, dtype=torch.int64, device=self.device),
    )
    self.register_buffer(
        "total_valid",
        torch.zeros(1, dtype=torch.int64, device=self.device),
    )

compute()

Compute all metrics from the accumulated state.

Returns:

Type	Description
dict[str, float | Tensor]	Dictionary with the following keys:
dict[str, float | Tensor]	"miou" — Mean IoU across valid classes (float).
dict[str, float | Tensor]	"pixel_accuracy" — Overall pixel accuracy (float).
dict[str, float | Tensor]	"mean_dice" — Mean Dice across valid classes (float).
dict[str, float | Tensor]	"per_class_iou" — Per-class IoU tensor (C,).
dict[str, float | Tensor]	"per_class_dice" — Per-class Dice tensor (C,).

Source code in src/corecv/metrics/segmentation.py

def compute(self) -> dict[str, float | Tensor]:
    """Compute all metrics from the accumulated state.

    Returns:
        Dictionary with the following keys:

        * ``"miou"`` — Mean IoU across valid classes (float).
        * ``"pixel_accuracy"`` — Overall pixel accuracy (float).
        * ``"mean_dice"`` — Mean Dice across valid classes (float).
        * ``"per_class_iou"`` — Per-class IoU tensor ``(C,)``.
        * ``"per_class_dice"`` — Per-class Dice tensor ``(C,)``.
    """
    eps: float = 1e-8

    # ---- IoU per class: intersection / union -----------------------
    inter: Tensor = self.intersection.float()  # (C,)
    union: Tensor = self.union.float()  # (C,)

    # A class is "valid" if it appeared in either predictions or targets.
    valid_classes: Tensor = union > 0  # (C,)

    iou_per_class: Tensor = torch.zeros(
        self.num_classes, dtype=torch.float32, device=self.device
    )
    # Only divide where union > 0 to avoid NaN for absent classes.
    iou_per_class[valid_classes] = inter[valid_classes] / (
        union[valid_classes] + eps
    )

    # Mean IoU — average only over classes that actually appeared.
    num_valid: int = valid_classes.sum().item()
    miou: float = (iou_per_class.sum() / max(num_valid, 1)).item()

    # ---- Pixel accuracy -------------------------------------------
    pixel_accuracy: float = (
        self.total_correct.float() / (self.total_valid.float() + eps)
    ).item()

    # ---- Dice per class: 2 * intersection / (pred_sum + target_sum)
    p_sum: Tensor = self.pred_sum.float()
    t_sum: Tensor = self.target_sum.float()

    dice_per_class: Tensor = torch.zeros(
        self.num_classes, dtype=torch.float32, device=self.device
    )
    dice_per_class[valid_classes] = (
        2.0 * inter[valid_classes]
        / (p_sum[valid_classes] + t_sum[valid_classes] + eps)
    )

    mean_dice: float = (dice_per_class.sum() / max(num_valid, 1)).item()

    return {
        "miou": miou,
        "pixel_accuracy": pixel_accuracy,
        "mean_dice": mean_dice,
        "per_class_iou": iou_per_class,
        "per_class_dice": dice_per_class,
    }

reset()

Reset all accumulator buffers to zero.

Source code in src/corecv/metrics/segmentation.py

def reset(self) -> None:
    """Reset all accumulator buffers to zero."""
    self.intersection.zero_()
    self.union.zero_()
    self.pred_sum.zero_()
    self.target_sum.zero_()
    self.total_correct.zero_()
    self.total_valid.zero_()

update(preds, targets)

Accumulate a batch of predictions and ground-truth masks.

Parameters:

Name	Type	Description	Default
preds	Tensor	Logits of shape (B, C, H, W) where C is the number of classes.	required
targets	Tensor	Ground-truth class indices of shape (B, H, W) with values in [0, C) and optionally self.ignore_index for ignored pixels.	required

Raises:

Type	Description
ValueError	If tensor shapes or value ranges are invalid.

Source code in src/corecv/metrics/segmentation.py

def update(self, preds: Tensor, targets: Tensor) -> None:
    """Accumulate a batch of predictions and ground-truth masks.

    Args:
        preds: Logits of shape ``(B, C, H, W)`` where ``C`` is the
            number of classes.
        targets: Ground-truth class indices of shape ``(B, H, W)``
            with values in ``[0, C)`` and optionally
            ``self.ignore_index`` for ignored pixels.

    Raises:
        ValueError: If tensor shapes or value ranges are invalid.
    """
    self._validate_inputs(preds, targets)

    # Move to accumulator device (no-op if already there).
    preds = preds.to(device=self.device)
    targets = targets.to(device=self.device)

    # Derive predicted class indices: argmax over channel dim.
    pred_classes: Tensor = preds.argmax(dim=1)  # (B, H, W)

    # Build a validity mask — exclude ignore_index pixels.
    valid_mask: Tensor = targets != self.ignore_index  # (B, H, W)

    # Replace ignored targets with a safe dummy (0) so that
    # F.one_hot does not produce out-of-range indices.
    safe_targets: Tensor = targets.clone()
    safe_targets[~valid_mask] = 0

    # ---- One-hot encode preds and targets --------------------------
    # (B, H, W) -> (B, H, W, C)
    oh_pred: Tensor = F.one_hot(pred_classes, self.num_classes)
    oh_target: Tensor = F.one_hot(safe_targets, self.num_classes)

    # Apply validity mask: (B, H, W) -> (B, H, W, 1) for broadcasting.
    mask_4d: Tensor = valid_mask.unsqueeze(-1)  # (B, H, W, 1)
    oh_pred = oh_pred.masked_fill(~mask_4d, 0)
    oh_target = oh_target.masked_fill(~mask_4d, 0)

    # Flatten spatial and batch dims: (B, H, W, C) -> (N, C)
    N: int = oh_pred.shape[0] * oh_pred.shape[1] * oh_pred.shape[2]
    oh_pred_flat: Tensor = oh_pred.reshape(N, self.num_classes)
    oh_target_flat: Tensor = oh_target.reshape(N, self.num_classes)

    # ---- Accumulate per-class counts (pure tensor ops) -------------
    # intersection[c] = sum of (oh_pred[:, c] * oh_target[:, c])
    # union[c]        = sum of clamp(oh_pred[:, c] + oh_target[:, c])
    inter_batch: Tensor = (oh_pred_flat * oh_target_flat).sum(dim=0)  # (C,)
    union_batch: Tensor = oh_pred_flat.sum(dim=0) + oh_target_flat.sum(
        dim=0
    ) - inter_batch  # (C,)

    self.intersection += inter_batch.to(dtype=torch.int64)
    self.union += union_batch.to(dtype=torch.int64)
    self.pred_sum += oh_pred_flat.sum(dim=0).to(dtype=torch.int64)
    self.target_sum += oh_target_flat.sum(dim=0).to(dtype=torch.int64)

    # ---- Pixel accuracy (correct / valid) -------------------------
    correct_batch: Tensor = (
        (pred_classes == targets) & valid_mask
    ).sum()
    valid_batch: Tensor = valid_mask.sum()

    self.total_correct += correct_batch.to(dtype=torch.int64)
    self.total_valid += valid_batch.to(dtype=torch.int64)

---

Datasets & Caching

Classification Datasets

corecv.data.datasets.classification

Image classification dataset with Albumentations transform support.

Provides a standard ImageFolder-style dataset that loads images from class-named subdirectories, applies optional Albumentations transforms (including those from :mod:corecv.data.transforms), and returns (image, label) tuples where image is a [C, H, W] tensor and label is an integer class index.

The dataset is designed to integrate seamlessly with :class:~corecv.data.transforms.CoordinatedTransform but also accepts any callable that receives an image keyword argument and returns a result with an image attribute (e.g. :class:~albumentations.Compose which returns a dict with an "image" key).

Example

from corecv.data.datasets import ClassificationDataset dataset = ClassificationDataset("path/to/data") image, label = dataset[0] image.shape torch.Size([3, 224, 224]) label 0

ClassificationDataset(root, transform=None, transforms=None, image_size=None)

Image classification dataset following the ImageFolder convention.

Organises images by class subdirectories under a root directory. Supports optional Albumentations transforms (e.g. :class:~corecv.data.transforms.CoordinatedTransform) and returns (image, label) tuples ready for model training.

Attributes:

Name	Type	Description
root	Path	Root directory containing class-named subdirectories.
transform	Callable[..., object] | None	Optional callable that accepts image as a keyword argument and returns a transformed result with an image attribute or "image" key.
classes	list[str]	Sorted list of class names derived from folder names.
class_to_idx	dict[str, int]	Mapping from class name to integer index.
samples	list[tuple[str, int]]	List of (filepath, class_index) tuples.

Initialise the dataset by scanning root for class folders.

Parameters:

Name	Type	Description	Default
root	str | Path	Path to dataset root containing class subdirectories.	required
transform	Callable[..., object] | bool | None	Transform pipeline or boolean flag. Pass True to enable standard default augmentations (random flip, rotation, resize, ImageNet normalization).	None
transforms	Callable[..., object] | bool | None	Alias for transform.	None
image_size	tuple[int, int] | None	Target (height, width) tuple. Defaults to (224, 224) when transform is True.	None

Raises:

Type	Description
FileNotFoundError	If root does not exist.

Source code in src/corecv/data/datasets/classification.py

def __init__(
    self,
    root: str | Path,
    transform: Callable[..., object] | bool | None = None,
    transforms: Callable[..., object] | bool | None = None,
    image_size: tuple[int, int] | None = None,
) -> None:
    """Initialise the dataset by scanning *root* for class folders.

    Args:
        root: Path to dataset root containing class subdirectories.
        transform: Transform pipeline or boolean flag. Pass ``True``
            to enable standard default augmentations (random flip,
            rotation, resize, ImageNet normalization).
        transforms: Alias for *transform*.
        image_size: Target ``(height, width)`` tuple. Defaults to
            ``(224, 224)`` when *transform* is ``True``.

    Raises:
        FileNotFoundError: If *root* does not exist.
    """
    self.root: Path = Path(root)
    self.image_size: tuple[int, int] | None = image_size

    tf: Callable[..., object] | bool | None = (
        transform if transform is not None else transforms
    )
    if tf is True:
        target_size: tuple[int, int] = image_size or (224, 224)
        self.transform: Callable[..., object] | None = build_transforms(
            ClassificationTransformConfig(
                image_size=target_size,
                horizontal_flip_p=0.5,
                rotate_limit=15,
            )
        )
    elif callable(tf):
        self.transform = tf
    else:
        self.transform = None

    # Discover classes from sorted subdirectory names
    classes: list[str] = sorted([d.name for d in self.root.iterdir() if d.is_dir()])
    self.classes: list[str] = classes
    self.class_to_idx: dict[str, int] = {cls: idx for idx, cls in enumerate(classes)}

    # Build the flat sample list
    self.samples: list[tuple[str, int]] = []
    for cls_name, cls_idx in self.class_to_idx.items():
        cls_dir: Path = self.root / cls_name
        for fpath in sorted(cls_dir.iterdir()):
            if fpath.suffix.lower() in _IMAGE_EXTENSIONS:
                self.samples.append((str(fpath), cls_idx))

__getitem__(index)

Return the image-label pair at the given index.

Source code in src/corecv/data/datasets/classification.py

def __getitem__(self, index: int) -> tuple[torch.Tensor, int]:
    """Return the image-label pair at the given index."""
    path: str
    label: int
    path, label = self.samples[index]

    raw_img: Image.Image = Image.open(path).convert("RGB")

    # Apply transforms if configured
    if self.transform is not None:
        image_np: np.ndarray = np.array(raw_img)
        result: object = self.transform(image=image_np)
        image_arr: np.ndarray = self._extract_image(result)
        image_tensor: torch.Tensor = torch.from_numpy(image_arr).permute(2, 0, 1)
    else:
        if self.image_size is not None and raw_img.size != (
            self.image_size[1],
            self.image_size[0],
        ):
            raw_img = raw_img.resize(
                (self.image_size[1], self.image_size[0]),
                Image.Resampling.BILINEAR,
            )
        image_arr = np.array(raw_img)
        image_tensor = torch.from_numpy(image_arr).permute(2, 0, 1)

    return image_tensor, label

__len__()

Return the total number of samples in the dataset.

Source code in src/corecv/data/datasets/classification.py

def __len__(self) -> int:
    """Return the total number of samples in the dataset."""
    return len(self.samples)

Detection Datasets (COCO)

corecv.data.datasets.detection

Detection dataset for object detection tasks.

Provides a :class:DetectionDataset that supports COCO JSON and YOLO txt annotation formats with a hybrid cache strategy. On first initialisation an O(N) pre-flight check validates all annotations and writes a persistent cache file (.npz) containing compact NumPy arrays for bounding boxes and labels. Subsequent runs load directly from the cache, bypassing annotation-file I/O each epoch. Cache validity is determined by comparing file modification timestamps.

DetectionDataset(root, annotation_path=None, format='coco', transform=None, transforms=None, image_size=(640, 640), cache_dir=None, use_cache=True, bbox_format='norm_xyxy', class_names=None)

Bases: Dataset[tuple[Tensor, Tensor, Tensor]]

Dataset for object detection with COCO and YOLO annotation formats.

Implements a hybrid cache strategy:

1. On first use, an O(N) pre-flight check scans every annotation file, validates bounding boxes, and writes a persistent cache file (.npz) containing compact NumPy arrays for bboxes and labels.
2. On subsequent runs the cache is loaded directly into RAM, avoiding per-epoch I/O on annotation files.
3. Cache invalidation compares stored file modification times against the current filesystem mtimes. If any source file has changed the cache is rebuilt.

Each sample is a (image, bboxes, labels) tuple where:

• image — torch.Tensor, shape (C, H, W), dtype torch.float32, values in [0, 1] (if untransformed) or normalised per the transform pipeline.
• bboxes — torch.Tensor, shape (N, 4) in XYXY format, either absolute pixels or normalised [0, 1].
• labels — torch.Tensor, shape (N,) with class indices.

Parameters:

Name	Type	Description	Default
root	str | Path	Root directory containing images and (for YOLO) label files.	required
annotation_path	str | Path | None	Path to annotation file (COCO JSON) or label directory (YOLO). If None, inferred from root: root / "annotations.json" for COCO, root / "labels" for YOLO.	None
format	str	Annotation format — "coco" or "yolo".	'coco'
transform	CoordinatedTransform | bool | None	Optional :class:CoordinatedTransform pipeline for synchronised image + bbox + label augmentation. The pipeline's bbox_format should be "pascal_voc" (recommended); other formats are converted automatically.	None
image_size	tuple[int, int]	Target (height, width) for image loading. Ignored when a transform is provided (the transform config controls resizing).	(640, 640)
cache_dir	str | Path | None	Directory for the cache file. If None, uses root / ".cache".	None
use_cache	bool	Whether to enable the on-disk cache. Pass False to force re-validation on every initialisation.	True
bbox_format	Literal['xyxy', 'norm_xyxy']	Output bounding box format. "xyxy" for absolute pixel coordinates, "norm_xyxy" for normalised [0, 1] values.	'norm_xyxy'
class_names	Sequence[str] | None	Optional list of class names for COCO. If None, extracted from the COCO JSON categories.	None

Raises:

Type	Description
FileNotFoundError	If the root directory, annotation path, or an image referenced in the annotations does not exist.
ValueError	If an unsupported format is given, or if an annotation contains invalid bounding boxes.

Example

from corecv.data.datasets import DetectionDataset dataset = DetectionDataset( ... root="data/coco", ... format="coco", ... bbox_format="norm_xyxy", ... ) image, bboxes, labels = dataset[0] image.shape torch.Size([3, 640, 640]) bboxes.shape torch.Size([3, 4]) labels.shape torch.Size([3])

Initialise the detection dataset.

Parameters:

Name	Type	Description	Default
root	str | Path	Root directory containing images.	required
annotation_path	str | Path | None	Path to annotation file or label dir.	None
format	str	Annotation format ("coco" or "yolo").	'coco'
transform	CoordinatedTransform | bool | None	Transform pipeline or boolean flag. Pass True to enable standard default detection augmentations (random flip, rotation, resize, ImageNet normalization). If False or None, applies safe baseline transforms (resize to image_size, normalize to [0.0, 1.0], and convert to [C, H, W] float32 Tensor).	None
transforms	CoordinatedTransform | bool | None	Alias for transform.	None
image_size	tuple[int, int]	Target (height, width) for images.	(640, 640)
cache_dir	str | Path | None	Directory for cache file.	None
use_cache	bool	Whether to use on-disk caching.	True
bbox_format	Literal['xyxy', 'norm_xyxy']	Output bbox format ("xyxy" or "norm_xyxy").	'norm_xyxy'
class_names	Sequence[str] | None	Optional list of class names.	None

Source code in src/corecv/data/datasets/detection.py

def __init__(  # noqa: PLR0913
    self,
    root: str | Path,
    annotation_path: str | Path | None = None,
    format: str = "coco",
    transform: CoordinatedTransform | bool | None = None,
    transforms: CoordinatedTransform | bool | None = None,
    image_size: tuple[int, int] = (640, 640),
    cache_dir: str | Path | None = None,
    use_cache: bool = True,
    bbox_format: Literal["xyxy", "norm_xyxy"] = "norm_xyxy",
    class_names: Sequence[str] | None = None,
) -> None:
    """Initialise the detection dataset.

    Args:
        root: Root directory containing images.
        annotation_path: Path to annotation file or label dir.
        format: Annotation format (``"coco"`` or ``"yolo"``).
        transform: Transform pipeline or boolean flag. Pass ``True``
            to enable standard default detection augmentations
            (random flip, rotation, resize, ImageNet normalization).
            If ``False`` or ``None``, applies safe baseline transforms
            (resize to *image_size*, normalize to ``[0.0, 1.0]``,
            and convert to ``[C, H, W]`` float32 Tensor).
        transforms: Alias for *transform*.
        image_size: Target ``(height, width)`` for images.
        cache_dir: Directory for cache file.
        use_cache: Whether to use on-disk caching.
        bbox_format: Output bbox format (``"xyxy"`` or ``"norm_xyxy"``).
        class_names: Optional list of class names.
    """
    super().__init__()

    self._root: Path = Path(root)
    if not self._root.is_dir():
        msg = f"Root directory does not exist: {self._root}"
        raise FileNotFoundError(msg)

    if format not in self.SUPPORTED_FORMATS:
        msg = (
            f"Unsupported annotation format: {format!r}. "
            f"Supported: {sorted(self.SUPPORTED_FORMATS)}"
        )
        raise ValueError(msg)
    self._format: str = format

    self._image_size: tuple[int, int] = image_size
    self._bbox_format: str = bbox_format
    self._use_cache: bool = use_cache

    tf: CoordinatedTransform | bool | None = (
        transform if transform is not None else transforms
    )
    if tf is True:
        self._transform: CoordinatedTransform | None = build_transforms(
            DetectionTransformConfig(
                image_size=self._image_size,
                horizontal_flip_p=0.5,
                rotate_limit=10,
                bbox_format="pascal_voc",
            )
        )
    elif callable(tf):
        self._transform = tf
    else:
        self._transform = None

    # Resolve annotation path
    if annotation_path is not None:
        self._annotation_path: Path = Path(annotation_path)
    elif format == "coco":
        self._annotation_path = self._root / "annotations.json"
    else:
        self._annotation_path = self._root / "labels"

    # Resolve cache directory
    if cache_dir is not None:
        self._cache_dir: Path = Path(cache_dir)
    else:
        self._cache_dir = self._root / ".cache"

    # Derive cache file name from annotation path hash
    self._cache_path: Path = self._cache_dir / self._cache_filename()

    # User-supplied class names (COCO override)
    self._user_class_names: Sequence[str] | None = class_names

    # In-memory state (populated by _initialize)
    self._image_paths: list[Path] = []
    self._bboxes: list[np.ndarray] = []
    self._labels: list[np.ndarray] = []
    self._class_names: list[str] = []

    # Original image dimensions (needed for normalisation)
    self._image_dims: list[tuple[int, int]] = []

    self._initialize()

CACHE_VERSION = 3 class-attribute

Integer version identifier bumped when cache format changes.

SUPPORTED_FORMATS = frozenset({'coco', 'yolo'}) class-attribute

Annotation formats recognised by this dataset.

class_names property

Return the list of class name strings.

Returns:

Type	Description
list[str]	Ordered list of class names (index = class id).

image_paths property

Return the list of image file paths.

Returns:

Type	Description
list[Path]	Ordered list of paths, one per sample.

num_classes property

Return the number of unique classes.

Returns:

Type	Description
int	Class count.

__getitem__(index)

Return the (image, bboxes, labels) tuple at index.

Loads the image from disk and retrieves cached bounding boxes and labels from the in-memory cache.

Parameters:

Name	Type	Description	Default
index	int	Sample index.	required

Returns:

Type	Description
Tensor	A tuple (image, bboxes, labels) where
Tensor	image has shape (C, H, W).
Tensor	bboxes has shape (N, 4) in XYXY format.
tuple[Tensor, Tensor, Tensor]	labels has shape (N,) with integer class indices.

Source code in src/corecv/data/datasets/detection.py

def __getitem__(self, index: int) -> tuple[Tensor, Tensor, Tensor]:
    """Return the ``(image, bboxes, labels)`` tuple at *index*.

    Loads the image from disk and retrieves cached bounding boxes
    and labels from the in-memory cache.

    Args:
        index: Sample index.

    Returns:
        A tuple ``(image, bboxes, labels)`` where

        - ``image`` has shape ``(C, H, W)``.
        - ``bboxes`` has shape ``(N, 4)`` in XYXY format.
        - ``labels`` has shape ``(N,)`` with integer class indices.
    """
    img_path: Path = self._image_paths[index]
    image: np.ndarray = cv2.imread(str(img_path))
    if image is None:
        msg = f"Failed to read image at index {index}: {img_path}"
        raise ValueError(msg)
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

    # Retrieve cached bboxes (internal format: absolute XYXY)
    bboxes: np.ndarray = self._bboxes[index].copy()
    labels: np.ndarray = self._labels[index].copy()

    img_h: int
    img_w: int
    img_h, img_w = image.shape[:2]

    # ------------------------------------------------------------------
    # Apply transforms if available
    # ------------------------------------------------------------------
    if self._transform is not None:
        # Determine expected bbox format from the transform config
        expected_fmt: str = "pascal_voc"  # default / recommended
        if hasattr(self._transform, "config"):
            cfg: Any = self._transform.config
            if hasattr(cfg, "bbox_format"):
                expected_fmt = cfg.bbox_format  # type: ignore[union-attr]

        # Convert internal (absolute XYXY) to the pipeline's format
        tf_bboxes: np.ndarray = _convert_bbox_format(
            bboxes, "pascal_voc", expected_fmt, float(img_w), float(img_h),
        )

        transformed = self._transform(
            image=image,
            bboxes=tf_bboxes.tolist() if len(tf_bboxes) > 0 else None,
            labels=labels.tolist() if len(labels) > 0 else None,
        )

        image = transformed.image  # (H, W, C), values depend on pipeline
        out_bboxes: np.ndarray = np.array(
            transformed.bboxes, dtype=np.float32
        )
        out_labels: np.ndarray = np.array(
            transformed.labels, dtype=np.int64
        )

        # Convert back to absolute XYXY if needed
        if expected_fmt != "pascal_voc":
            out_bboxes = _convert_bbox_format(
                out_bboxes,
                expected_fmt,
                "pascal_voc",
                float(image.shape[1]),
                float(image.shape[0]),
            )

        # Determine output image dimensions for normalisation
        out_h: int
        out_w: int
        out_h, out_w = image.shape[:2]
    else:
        # Manual normalisation (no transform pipeline)
        image = image.astype(np.float32) / 255.0
        out_bboxes = bboxes.astype(np.float32)
        out_labels = labels.astype(np.int64)
        out_h, out_w = img_h, img_w

    # ------------------------------------------------------------------
    # Convert to output bbox format
    # ------------------------------------------------------------------
    if self._bbox_format == "norm_xyxy":
        out_bboxes = (
            out_bboxes / [float(out_w), float(out_h), float(out_w), float(out_h)]
        ).astype(np.float32)

    # Convert image to tensor (C, H, W)
    image_tensor: Tensor = (
        torch.from_numpy(image).permute(2, 0, 1).float()
    )
    bboxes_tensor: Tensor = torch.from_numpy(out_bboxes)
    labels_tensor: Tensor = torch.from_numpy(out_labels)

    return image_tensor, bboxes_tensor, labels_tensor

__len__()

Return the number of samples in the dataset.

Returns:

Type	Description
int	Total sample count.

Source code in src/corecv/data/datasets/detection.py

def __len__(self) -> int:
    """Return the number of samples in the dataset.

    Returns:
        Total sample count.
    """
    return len(self._image_paths)

Segmentation Datasets

corecv.data.datasets.segmentation

Semantic segmentation dataset with on-disk mask loading and cache manifest.

Provides :class:SegmentationDataset for reading paired images and uint8-indexed masks from images/ and masks/ subdirectories with matched 1:1 filenames. The dataset implements an O(N) pre-flight check that validates all annotations and generates a compressed .npz cache file containing only a manifest of validated pairs with file modification timestamps — no mask pixel arrays are stored in the cache.

Masks are read from disk dynamically at __getitem__ time, avoiding memory explosion on large datasets.

Example

from corecv.data.datasets import SegmentationDataset dataset = SegmentationDataset("path/to/data", num_classes=21) image, mask = dataset[0] image.shape torch.Size([3, 480, 640]) mask.shape torch.Size([480, 640])

SegmentationDataset(root, num_classes, transforms=None, transform=None, image_size=None, cache_dir=None, ignore_index=None)

Bases: Dataset[tuple[Tensor, Tensor]]

PyTorch Dataset for semantic segmentation with on-disk mask loading.

Expects the following directory structure under root::

root/
    images/
        image_001.jpg
        image_002.png
        ...
    masks/
        image_001.png    # uint8, class indices 0..num_classes-1
        image_002.png
        ...

Features:

• 1:1 matched filename pairing between images/ and masks/.
• Masks are validated as uint8 with values in [0, num_classes) (plus an optional ignore_index).
• O(N) pre-flight check generates a .npz cache on first run.
• The cache stores only a manifest of validated pairs with file modification timestamps — no mask pixel arrays are persisted, avoiding memory explosion on large datasets.
• Mask pixel data is read from disk dynamically at __getitem__ time via PIL in uint8 format.
• Cache invalidation via filesystem mtime comparison — any change to a source file triggers a rebuild.
• Optional :class:~corecv.data.transforms.CoordinatedTransform integration for synchronised image + mask augmentations.

Attributes:

Name	Type	Description
root	Path	Resolved path to the dataset root directory.
num_classes	int	Number of semantic classes (used for validation).
transforms	CoordinatedTransform | None	Optional transform pipeline for synchronised image + mask augmentations.
ignore_index	int | None	Optional label value that is excluded from range validation (e.g. 255 for ignore/crop border).
image_paths	list[Path]	List of resolved image file paths.
mask_paths	list[Path]	List of resolved mask file paths.
cache_path	Path	Path to the on-disk .npz cache file.

Initialise the dataset, validating pairs and loading / building cache.

Parameters:

Name	Type	Description	Default
root	str | Path	Root directory containing images/ and masks/ subdirectories.	required
num_classes	int	Number of semantic classes.	required
transforms	CoordinatedTransform | bool | None	Transform pipeline or boolean flag. Pass True to enable standard default segmentation augmentations (random flip, rotation, resize, ImageNet normalization).	None
transform	CoordinatedTransform | bool | None	Alias for transforms.	None
image_size	tuple[int, int] | None	Target (height, width) tuple.	None
cache_dir	str | Path | None	Directory for cache file. Defaults to root.	None
ignore_index	int | None	Optional class index to exclude from range validation.	None

Source code in src/corecv/data/datasets/segmentation.py

def __init__(  # noqa: PLR0913
    self,
    root: str | Path,
    num_classes: int,
    transforms: CoordinatedTransform | bool | None = None,
    transform: CoordinatedTransform | bool | None = None,
    image_size: tuple[int, int] | None = None,
    cache_dir: str | Path | None = None,
    ignore_index: int | None = None,
) -> None:
    """Initialise the dataset, validating pairs and loading / building cache.

    Args:
        root: Root directory containing ``images/`` and ``masks/``
            subdirectories.
        num_classes: Number of semantic classes.
        transforms: Transform pipeline or boolean flag. Pass ``True``
            to enable standard default segmentation augmentations
            (random flip, rotation, resize, ImageNet normalization).
        transform: Alias for *transforms*.
        image_size: Target ``(height, width)`` tuple.
        cache_dir: Directory for cache file. Defaults to *root*.
        ignore_index: Optional class index to exclude from range validation.
    """
    self._root: Path = Path(root).resolve().absolute()
    self._num_classes: int = num_classes
    self._ignore_index: int | None = ignore_index
    self._image_size: tuple[int, int] | None = image_size

    tf: CoordinatedTransform | bool | None = (
        transforms if transforms is not None else transform
    )
    if tf is True:
        target_size: tuple[int, int] = image_size or (512, 512)
        self._transforms: CoordinatedTransform | None = build_transforms(
            SegmentationTransformConfig(
                image_size=target_size,
                ignore_index=ignore_index if ignore_index is not None else 255,
                horizontal_flip_p=0.5,
                rotate_limit=15,
            )
        )
    elif callable(tf):
        self._transforms = tf
    else:
        self._transforms = None

    self._images_dir: Path = self._root / "images"
    self._masks_dir: Path = self._root / "masks"

    # Validate directories exist
    if not self._images_dir.is_dir():
        raise NotADirectoryError(
            _ERROR_IMAGES_DIR.format(path=self._images_dir)
        )
    if not self._masks_dir.is_dir():
        raise NotADirectoryError(
            _ERROR_MASKS_DIR.format(path=self._masks_dir)
        )

    # Cache path
    cache_dir_resolved: Path = (
        Path(cache_dir).resolve().absolute()
        if cache_dir is not None
        else self._root
    )
    self._cache_path: Path = cache_dir_resolved / _CACHE_FILENAME

    # Internal state populated by cache load / build
    self._image_paths: list[Path] = []
    self._mask_paths: list[Path] = []

    # Run pre-flight check: validate pairs and load / build cache
    self._load_or_build_cache()

cache_path property

Return the path to the on-disk .npz cache file.

ignore_index property

Return the optional ignore index.

image_paths property

Return the list of resolved image file paths.

mask_paths property

Return the list of resolved mask file paths.

num_classes property

Return the number of semantic classes.

root property

Return the dataset root directory.

transforms property

Return the optional transform pipeline.

__getitem__(index)

Return a single (image, mask) pair at the given index.

Both the image and mask are read from disk dynamically. The image is returned as a [C, H, W] float32 tensor and the mask as a [H, W] torch.long tensor.

Parameters:

Name	Type	Description	Default
index	int	Sample index (0-based).	required

Returns:

Type	Description
tuple[Tensor, Tensor]	A tuple (image_tensor, mask_tensor).

Source code in src/corecv/data/datasets/segmentation.py

def __getitem__(self, index: int) -> tuple[Tensor, Tensor]:
    """Return a single (image, mask) pair at the given index.

    Both the image and mask are read from disk dynamically.  The
    image is returned as a ``[C, H, W]`` float32 tensor and the
    mask as a ``[H, W]`` ``torch.long`` tensor.

    Args:
        index: Sample index (0-based).

    Returns:
        A tuple ``(image_tensor, mask_tensor)``.
    """
    image_path: Path = self._image_paths[index]
    mask_path: Path = self._mask_paths[index]

    # Load image from disk (uint8 HWC)
    image: np.ndarray = self._load_image(image_path)

    # Load mask from disk (uint8 HW)
    mask: np.ndarray = self._load_mask(mask_path)

    # Apply synchronised transforms if configured
    if self._transforms is not None:
        result = self._transforms(image=image, mask=mask)
        image = result.image
        mask = result.mask  # type: ignore[assignment]
        image_tensor: Tensor = torch.from_numpy(image).permute(2, 0, 1).float()
    else:
        # Safe default fallback when transforms=False / None:
        if self._image_size is not None:
            h, w = image.shape[:2]
            if (h, w) != self._image_size:
                img_pil = Image.fromarray(image).resize(
                    (self._image_size[1], self._image_size[0]),
                    Image.Resampling.BILINEAR,
                )
                mask_pil = Image.fromarray(mask).resize(
                    (self._image_size[1], self._image_size[0]),
                    Image.Resampling.NEAREST,
                )
                image = np.array(img_pil, dtype=np.uint8)
                mask = np.array(mask_pil, dtype=np.uint8)
        image_tensor = torch.from_numpy(image).permute(2, 0, 1).float() / 255.0

    mask_tensor: Tensor = torch.from_numpy(mask).long()

    return image_tensor, mask_tensor

__len__()

Return the number of samples in the dataset.

Returns:

Type	Description
int	Total number of valid image-mask pairs.

Source code in src/corecv/data/datasets/segmentation.py

def __len__(self) -> int:
    """Return the number of samples in the dataset.

    Returns:
        Total number of valid image-mask pairs.
    """
    return len(self._image_paths)