Migrate anomaly task to use the new Datumaro dataset

library/pyproject.toml

@@ -27,7 +27,7 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
-    "datumaro==1.10.0",
+    "datumaro[experimental] @ git+https://github.com/open-edge-platform/datumaro.git@gppayend/anomaly",
     "omegaconf==2.3.0",
     "rich==14.0.0",
     "jsonargparse==4.35.0",
@@ -37,7 +37,6 @@ dependencies = [
     "docstring_parser==0.16", # CLI help-formatter
     "rich_argparse==1.7.0", # CLI help-formatter
     "einops==0.8.1",
-    "decord==0.6.0",
     "typeguard>=4.3,<4.5",
     # TODO(ashwinvaidya17): https://github.com/openvinotoolkit/anomalib/issues/2126
     "setuptools<70",
@@ -51,6 +50,8 @@ dependencies = [
     "onnxconverter-common==1.14.0",
     "nncf==2.17.0",
     "anomalib[core]==1.1.3",
+    "numpy<2.0.0",
+    "tensorboardX>=1.8",
 ]
 
 [project.optional-dependencies]

library/src/otx/backend/native/callbacks/gpu_mem_monitor.py

@@ -29,7 +29,7 @@ def _get_and_log_device_stats(
             batch_size (int): batch size.
         """
         device = trainer.strategy.root_device
-        if device.type in ["cpu", "xpu"]:
+        if device.type in ["cpu", "xpu", "mps"]:
             return
 
         device_stats = trainer.accelerator.get_device_stats(device)

library/src/otx/backend/native/models/__init__.py

@@ -3,6 +3,11 @@
 
 """Module for OTX custom models."""
 
+import multiprocessing
+
+if multiprocessing.get_start_method(allow_none=True) is None:
+    multiprocessing.set_start_method("forkserver")
+
 from .anomaly import Padim, Stfpm, Uflow
 from .classification import (
     EfficientNet,

library/src/otx/backend/native/models/detection/ssd.py

@@ -15,7 +15,7 @@
 from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 import numpy as np
-from datumaro.components.annotation import Bbox
+from datumaro.experimental.dataset import Dataset as DmDataset
 
 from otx.backend.native.exporter.base import OTXModelExporter
 from otx.backend.native.exporter.native import OTXNativeModelExporter
@@ -30,6 +30,7 @@
 from otx.backend.native.models.utils.support_otx_v1 import OTXv1Helper
 from otx.backend.native.models.utils.utils import load_checkpoint
 from otx.config.data import TileConfig
+from otx.data.entity.sample import DetectionSample
 from otx.metrics.fmeasure import MeanAveragePrecisionFMeasureCallable
 
 if TYPE_CHECKING:
@@ -231,7 +232,7 @@ def _get_new_anchors(self, dataset: OTXDataset, anchor_generator: SSDAnchorGener
         return self._get_anchor_boxes(wh_stats, group_as)
 
     @staticmethod
-    def _get_sizes_from_dataset_entity(dataset: OTXDataset, target_wh: list[int]) -> list[tuple[int, int]]:
+    def _get_sizes_from_dataset_entity(dataset: OTXDataset, target_wh: list[int]) -> np.ndarray:
         """Function to get width and height size of items in OTXDataset.
 
         Args:
@@ -240,20 +241,34 @@ def _get_sizes_from_dataset_entity(dataset: OTXDataset, target_wh: list[int]) ->
         Return
             list[tuple[int, int]]: tuples with width and height of each instance
         """
-        wh_stats: list[tuple[int, int]] = []
+        wh_stats = np.empty((0, 2), dtype=np.float32)
+        if not isinstance(dataset.dm_subset, DmDataset):
+            exc_str = "The variable dataset.dm_subset must be an instance of DmDataset"
+            raise TypeError(exc_str)
+
         for item in dataset.dm_subset:
-            for ann in item.annotations:
-                if isinstance(ann, Bbox):
-                    x1, y1, x2, y2 = ann.points
-                    x1 = x1 / item.media.size[1] * target_wh[0]
-                    y1 = y1 / item.media.size[0] * target_wh[1]
-                    x2 = x2 / item.media.size[1] * target_wh[0]
-                    y2 = y2 / item.media.size[0] * target_wh[1]
-                    wh_stats.append((x2 - x1, y2 - y1))
+            if not isinstance(item, DetectionSample):
+                exc_str = "The variable item must be an instance of DetectionSample"
+                raise TypeError(exc_str)
+
+            if item.img_info is None:
+                exc_str = "The image info must not be None"
+                raise RuntimeError(exc_str)
+
+            height, width = item.img_info.img_shape
+            x1 = item.bboxes[:, 0]
+            y1 = item.bboxes[:, 1]
+            x2 = item.bboxes[:, 2]
+            y2 = item.bboxes[:, 3]
+
+            w = (x2 - x1) / width * target_wh[0]
+            h = (y2 - y1) / height * target_wh[1]
+
+            wh_stats = np.concatenate((wh_stats, np.stack((w, h), axis=1)), axis=0)
         return wh_stats
 
     @staticmethod
-    def _get_anchor_boxes(wh_stats: list[tuple[int, int]], group_as: list[int]) -> tuple:
+    def _get_anchor_boxes(wh_stats: np.ndarray, group_as: list[int]) -> tuple:
         """Get new anchor box widths & heights using KMeans."""
         from sklearn.cluster import KMeans
 

library/src/otx/backend/native/models/instance_segmentation/base.py

@@ -442,7 +442,7 @@ def _convert_pred_entity_to_compute_metric(
 
             rles = (
                 [encode_rle(mask) for mask in masks.data]
-                if len(masks)
+                if masks is not None
                 else polygon_to_rle(polygons, *imgs_info.ori_shape)  # type: ignore[union-attr,arg-type]
             )
             target_info.append(

library/src/otx/backend/native/models/instance_segmentation/heads/roi_head_tv.py

@@ -15,13 +15,13 @@
 from otx.data.utils.structures.mask import mask_target
 
 if TYPE_CHECKING:
-    from datumaro import Polygon
+    import numpy as np
 
 
 def maskrcnn_loss(
     mask_logits: Tensor,
     proposals: list[Tensor],
-    gt_masks: list[list[Tensor]] | list[list[Polygon]],
+    gt_masks: list[list[Tensor]] | list[np.ndarray],
     gt_labels: list[Tensor],
     mask_matched_idxs: list[Tensor],
     image_shapes: list[tuple[int, int]],
@@ -31,7 +31,7 @@ def maskrcnn_loss(
     Args:
         mask_logits (Tensor): the mask predictions.
         proposals (list[Tensor]): the region proposals.
-        gt_masks (list[list[Tensor]] | list[list[Polygon]]): the ground truth masks.
+        gt_masks (list[list[Tensor]] | list[np.ndarray]): the ground truth masks as ragged arrays.
         gt_labels (list[Tensor]): the ground truth labels.
         mask_matched_idxs (list[Tensor]): the matched indices.
         image_shapes (list[tuple[int, int]]): the image shapes.
@@ -142,7 +142,9 @@ def forward(
                     raise ValueError(msg)
 
                 gt_masks = (
-                    [t["masks"] for t in targets] if len(targets[0]["masks"]) else [t["polygons"] for t in targets]
+                    [t["masks"] for t in targets]
+                    if targets[0]["masks"] is not None
+                    else [t["polygons"] for t in targets]
                 )
                 gt_labels = [t["labels"] for t in targets]
                 rcnn_loss_mask = maskrcnn_loss(

library/src/otx/backend/native/models/instance_segmentation/heads/rtmdet_inst_head.py

@@ -18,7 +18,6 @@
 import numpy as np
 import torch
 import torch.nn.functional
-from datumaro import Polygon
 from torch import Tensor, nn
 
 from otx.backend.native.models.common.utils.nms import batched_nms, multiclass_nms
@@ -644,7 +643,7 @@ def prepare_loss_inputs(self, x: tuple[Tensor], entity: OTXDataBatch) -> dict:
         )
 
         # Convert polygon masks to bitmap masks
-        if isinstance(batch_gt_instances[0].masks[0], Polygon):
+        if isinstance(batch_gt_instances[0].masks, np.ndarray):
             for gt_instances, img_meta in zip(batch_gt_instances, batch_img_metas):
                 ndarray_masks = polygon_to_bitmap(gt_instances.masks, *img_meta["img_shape"])
                 if len(ndarray_masks) == 0:

library/src/otx/backend/native/models/instance_segmentation/rotated_det.py

@@ -4,13 +4,27 @@
 """Rotated Detection Prediction Mixin."""
 
 import cv2
+import numpy as np
 import torch
-from datumaro import Polygon
 from torchvision import tv_tensors
 
 from otx.data.entity.torch.torch import OTXPredBatch
 
 
+def get_polygon_area(points: np.ndarray) -> float:
+    """Calculate polygon area using the shoelace formula.
+
+    Args:
+        points: Array of polygon vertices with shape (N, 2)
+
+    Returns:
+        float: Area of the polygon
+    """
+    x = points[:, 0]
+    y = points[:, 1]
+    return 0.5 * np.abs(np.dot(x, np.roll(y, 1)) - np.dot(y, np.roll(x, 1)))
+
+
 def convert_masks_to_rotated_predictions(preds: OTXPredBatch) -> OTXPredBatch:
     """Convert masks to rotated bounding boxes and polygons.
 
@@ -58,8 +72,10 @@ def convert_masks_to_rotated_predictions(preds: OTXPredBatch) -> OTXPredBatch:
             for contour, hierarchy in zip(contours, hierarchies[0]):
                 if hierarchy[3] != -1 or len(contour) <= 2:
                     continue
-                rbox_points = Polygon(cv2.boxPoints(cv2.minAreaRect(contour)).reshape(-1))
-                rbox_polygons.append((rbox_points, rbox_points.get_area()))
+                # Get rotated bounding box points and convert to ragged array format
+                box_points = cv2.boxPoints(cv2.minAreaRect(contour)).astype(np.float32)
+                area = get_polygon_area(box_points)
+                rbox_polygons.append((box_points, area))
 
             if rbox_polygons:
                 rbox_polygons.sort(key=lambda x: x[1], reverse=True)

library/src/otx/backend/native/models/instance_segmentation/utils/utils.py

@@ -53,7 +53,7 @@ def unpack_inst_seg_entity(entity: OTXDataBatch) -> tuple:
         }
         batch_img_metas.append(metainfo)
 
-        gt_masks = mask if len(mask) else polygon
+        gt_masks = mask if mask is not None else polygon
 
         batch_gt_instances.append(
             InstanceData(

library/src/otx/backend/native/utils/utils.py

@@ -80,8 +80,8 @@ def mock_modules_for_chkpt() -> Iterator[None]:
         setattr(sys.modules["otx.types.task"], "OTXTrainType", OTXTrainType)  # noqa: B010
 
         sys.modules["otx.core"] = types.ModuleType("otx.core")
-        sys.modules["otx.core.config"] = otx.config
-        sys.modules["otx.core.config.data"] = otx.config.data
+        sys.modules["otx.core.config"] = otx.config  # type: ignore[attr-defined]
+        sys.modules["otx.core.config.data"] = otx.config.data  # type: ignore[attr-defined]
         sys.modules["otx.core.types"] = otx.types
         sys.modules["otx.core.types.task"] = otx.types.task
         sys.modules["otx.core.types.label"] = otx.types.label

library/src/otx/data/dataset/anomaly_new.py

@@ -0,0 +1,36 @@
+# Copyright (C) 2023-2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Module for OTXSegmentationDataset."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from datumaro.experimental.categories import LabelCategories, LabelSemantic
+
+from otx.data.dataset.base_new import OTXDataset
+from otx.data.entity.sample import AnomalySample
+from otx.types.label import AnomalyLabelInfo
+from otx.types.task import OTXTaskType
+
+if TYPE_CHECKING:
+    from datumaro.experimental import Dataset
+
+
+class OTXAnomalyDataset(OTXDataset):
+    """OTXDataset class for anomaly task."""
+
+    def __init__(self, task_type: OTXTaskType, dm_subset: Dataset, **kwargs) -> None:
+        self.task_type = task_type
+        sample_type = AnomalySample
+        categories = {
+            "label": LabelCategories(
+                labels=["normal", "anomalous"],
+                label_semantics={LabelSemantic.NORMAL: "normal", LabelSemantic.ANOMALOUS: "anomalous"},
+            )
+        }
+        dm_subset = dm_subset.convert_to_schema(sample_type, target_categories=categories)
+        super().__init__(dm_subset=dm_subset, sample_type=sample_type, **kwargs)
+
+        self.label_info = AnomalyLabelInfo()

library/src/otx/data/dataset/base.py

@@ -6,13 +6,14 @@
 from __future__ import annotations
 
 from abc import abstractmethod
+from collections import defaultdict
 from collections.abc import Iterable
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, Callable, Iterator, List, Union
 
 import cv2
 import numpy as np
-from datumaro.components.annotation import AnnotationType
+from datumaro.components.annotation import AnnotationType, LabelCategories
 from datumaro.util.image import IMAGE_BACKEND, IMAGE_COLOR_CHANNEL, ImageBackend
 from datumaro.util.image import ImageColorChannel as DatumaroImageColorChannel
 from torch.utils.data import Dataset
@@ -196,3 +197,23 @@ def _get_item_impl(self, idx: int) -> OTXDataItem | None:
     def collate_fn(self) -> Callable:
         """Collection function to collect KeypointDetDataEntity into KeypointDetBatchDataEntity in data loader."""
         return OTXDataItem.collate_fn
+
+    def get_idx_list_per_classes(self, use_string_label: bool = False) -> dict[int | str, list[int]]:
+        """Get a dictionary mapping class labels (string or int) to lists of samples.
+
+        Args:
+            use_string_label (bool): If True, use string class labels as keys.
+                If False, use integer indices as keys.
+        """
+        stats: dict[int | str, list[int]] = defaultdict(list)
+        for item_idx, item in enumerate(self.dm_subset):
+            for ann in item.annotations:
+                if use_string_label:
+                    labels = self.dm_subset.categories().get(AnnotationType.label, LabelCategories())
+                    stats[labels.items[ann.label].name].append(item_idx)
+                else:
+                    stats[ann.label].append(item_idx)
+        # Remove duplicates in label stats idx: O(n)
+        for k in stats:
+            stats[k] = list(dict.fromkeys(stats[k]))
+        return stats