1.10: Add position information to PointsCategories (#1702)

- This PR adds optional position information as meta-info for (key)point annotations.
- The position information can be used to define one or multiple default relative positions of a set of keypoints.
- This type of meta-information about points data is useful in annotation tools.

Author: djdameln

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Q4 2024 Release 1.10.0
 
 ### New features
+- Add default position information to PointsCategories class
+  (<https://github.com/openvinotoolkit/datumaro/pull/1702>)
 - Support KITTI 3D format
   (<https://github.com/openvinotoolkit/datumaro/pull/1619>)
   (<https://github.com/openvinotoolkit/datumaro/pull/1621>)

src/datumaro/components/annotation.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2021-2024 Intel Corporation
+# Copyright (C) 2021-2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -16,6 +16,7 @@
     Iterator,
     List,
     Optional,
+    Sequence,
     Set,
     Tuple,
     Type,
@@ -31,6 +32,7 @@
 
 from datumaro.components.media import Image
 from datumaro.util.attrs_util import default_if_none, not_empty
+from datumaro.util.points_util import normalize_points
 
 
 class AnnotationType(IntEnum):
@@ -1414,14 +1416,57 @@ class Category:
         # Pairs of connected point indices
         joints: Set[Tuple[int, int]] = field(factory=set, validator=default_if_none(set))
 
+        # Set of default x, y coordinates of the points
+        positions: List[float] = field(default=[])
+
+        @positions.validator
+        def positions_validator(
+            self, attribute: attr.Attribute[list], positions: list[float] | None
+        ) -> None:
+            """
+            Validate a list of point positions in the format [x1, y1, x2, y2, ..., xn, yn].
+
+            To be used as an attrs validator for the positions field of PointsCategories.Category.
+
+            Args:
+                attribute (attr.Attribute[list]): The attribute being validated.
+                positions (list[float]): A list of point positions.
+
+            Raises:
+                ValueError: If the provided value cannot be converted to a list of floats,
+                    or if the number of positions is invalid.
+            """
+            if positions is None:
+                positions = []
+            else:
+                # convert to a list of floats
+                try:
+                    positions = list(map(float, positions))
+                except (TypeError, ValueError):
+                    msg = "Cannot convert positions to list of floats. Check your input data."
+                    raise ValueError(msg)
+                # check if number of coordinates is even
+                if len(positions) % 2 != 0:
+                    msg = "positions must have an even number of elements"
+                    raise ValueError(msg)
+                if len(positions) > 0:
+                    # check if the number of positions is equal to the number of labels
+                    if len(self.labels) > 0 and len(positions) != len(self.labels) * 2:
+                        msg = "The number of positions should be equal to the number of labels"
+                        raise ValueError(msg)
+                    # normalize the positions
+                    positions = normalize_points(positions)
+            setattr(self, attribute.name, positions)
+
     items: Dict[int, Category] = field(factory=dict, validator=default_if_none(dict))
 
     @classmethod
     def from_iterable(
         cls,
         iterable: Union[
-            Tuple[int, List[str]],
-            Tuple[int, List[str], Set[Tuple[int, int]]],
+            Iterable[Sequence[int, List[str]]],
+            Iterable[Sequence[int, List[str], Set[Tuple[int, int]]]],
+            Iterable[Sequence[int, List[str], Set[Tuple[int, int]], List[float]]],
         ],
     ) -> PointsCategories:
         """
@@ -1447,11 +1492,12 @@ def add(
         label_id: int,
         labels: Optional[Iterable[str]] = None,
         joints: Iterable[Tuple[int, int]] = None,
+        positions: Iterable[float] = None,
     ):
         if joints is None:
             joints = []
         joints = set(map(tuple, joints))
-        self.items[label_id] = self.Category(labels, joints)
+        self.items[label_id] = self.Category(labels, joints, positions)
 
     def __contains__(self, idx: int) -> bool:
         return idx in self.items

src/datumaro/plugins/data_formats/datumaro/base.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2024 Intel Corporation
+# Copyright (C) 2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -115,7 +115,12 @@ def _load_categories(parsed) -> Dict:
         if parsed_points_cat:
             point_categories = PointsCategories()
             for item in parsed_points_cat["items"]:
-                point_categories.add(int(item["label_id"]), item["labels"], joints=item["joints"])
+                point_categories.add(
+                    int(item["label_id"]),
+                    item["labels"],
+                    joints=item["joints"],
+                    positions=item.get("positions"),
+                )
 
             categories[AnnotationType.points] = point_categories
 

src/datumaro/plugins/data_formats/datumaro/exporter.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2024 Intel Corporation
+# Copyright (C) 2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -107,6 +107,7 @@ def _convert_points_categories(cls, obj):
                     "label_id": int(label_id),
                     "labels": [cast(label, str) for label in item.labels],
                     "joints": [list(map(int, j)) for j in item.joints],
+                    "positions": list(map(float, item.positions)),
                 }
             )
         return converted

src/datumaro/util/points_util.py

@@ -0,0 +1,38 @@
+# Copyright (C) 2025 Intel Corporation
+#
+# SPDX-License-Identifier: MIT
+
+import numpy as np
+
+
+def normalize_points(positions: list[float]) -> list[float]:
+    """
+    Normalize a set of keypoints to fit within a unit bounding box [0, 1] x [0, 1],
+    maintaining the aspect ratio.
+
+    Parameters:
+        keypoints (list[float]): A list of point positions in the format [x1, y1, x2, y2, ..., xn, yn].
+
+    Returns:
+        list[float]: Normalized keypoints within the unit bounding box, preserving aspect ratio.
+    """
+    # Convert keypoints to a NumPy array for easier manipulation
+    positions = np.array(positions, dtype=float).reshape(-1, 2)
+
+    # Find the minimum and maximum values for x and y
+    min = positions.min(axis=0)
+    max = positions.max(axis=0)
+
+    # Compute the width and height of the bounding box
+    size = max - min
+
+    # Handle edge case where all keypoints are the same (zero width or height)
+    size[np.where(size == 0)] = 1e-6
+
+    # Determine the scaling factor to maintain aspect ratio
+    scale = size.max()
+
+    # Normalize the keypoints to fit within [0, 1] x [0, 1], preserving aspect ratio
+    normalized_positions = (positions - min) / scale
+
+    return list(map(float, normalized_positions.flatten()))

tests/unit/data_formats/datumaro/conftest.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2024 Intel Corporation
+# Copyright (C) 2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -680,3 +680,54 @@ def fxt_legacy_dataset_pair(test_dir):
     )
 
     yield source_dataset, target_dataset
+
+
+@pytest.fixture
+def fxt_test_pointscategories_without_positions():
+    points_categories = PointsCategories()
+    for index in range(5):
+        points_categories.add(index, labels=["label1", "label2", "label3"], joints=[[0, 1], [1, 2]])
+
+    return Dataset.from_iterable(
+        [
+            DatasetItem(
+                id=100,
+                subset="train",
+                media=Image.from_numpy(data=np.ones((10, 6, 3))),
+                annotations=[
+                    Points([1, 2, 0, 0, 1, 1]),
+                ],
+            ),
+        ],
+        categories={
+            AnnotationType.points: points_categories,
+        },
+    )
+
+
+@pytest.fixture
+def fxt_test_pointscategories_with_positions():
+    points_categories = PointsCategories()
+    for index in range(5):
+        points_categories.add(
+            index,
+            labels=["label1", "label2", "label3"],
+            joints=[[0, 1], [1, 2]],
+            positions=[0, 1, 1, 2, 2, 3],
+        )
+
+    return Dataset.from_iterable(
+        [
+            DatasetItem(
+                id=100,
+                subset="train",
+                media=Image.from_numpy(data=np.ones((10, 6, 3))),
+                annotations=[
+                    Points([1, 2, 0, 0, 1, 1]),
+                ],
+            ),
+        ],
+        categories={
+            AnnotationType.points: points_categories,
+        },
+    )

tests/unit/data_formats/datumaro/test_datumaro_format.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2024 Intel Corporation
+# Copyright (C) 2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -122,6 +122,18 @@ def _test_save_and_load(
                 True,
                 id="test_can_save_and_load_infos",
             ),
+            pytest.param(
+                "fxt_test_pointscategories_without_positions",
+                compare_datasets,
+                False,
+                id="test_pointscategories_without_positions",
+            ),
+            pytest.param(
+                "fxt_test_pointscategories_with_positions",
+                compare_datasets,
+                False,
+                id="test_pointscategories_with_positions",
+            ),
         ],
     )
     @pytest.mark.parametrize("stream", [True, False])

tests/unit/test_annotation.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2020-2024 Intel Corporation
+# Copyright (C) 2020-2025 Intel Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -17,9 +17,11 @@
     ExtractedMask,
     HashKey,
     Mask,
+    PointsCategories,
     RotatedBbox,
 )
 from datumaro.util.image import lazy_image
+from datumaro.util.points_util import normalize_points
 
 
 class EllipseTest:
@@ -142,3 +144,117 @@ def test_get_semantic_seg_mask_binary_mask(self, fxt_index_mask, dtype):
         semantic_seg_mask = annotations.get_semantic_seg_mask(ignore_index=255, dtype=dtype)
 
         assert np.allclose(semantic_seg_mask, fxt_index_mask)
+
+
+class PointsCategoriesTest:
+    @pytest.mark.parametrize(
+        "positions, expected",
+        [
+            (
+                [2, 3, 4, 6, 3, 5],
+                [0.0, 0.0, 0.666667, 1.0, 0.333333, 0.666667],
+            ),  # basic functionality
+            ([1, 1, 1, 1, 1, 1], [0.0, 0.0, 0.0, 0.0, 0.0, 0.0]),  # all points are the same
+            ([1, 1, 3, 1, 5, 1], [0.0, 0.0, 0.5, 0.0, 1.0, 0.0]),  # points form horizontal line
+            ([1, 1, 1, 3, 1, 5], [0.0, 0.0, 0.0, 0.5, 0.0, 1.0]),  # points form vertical line
+            (
+                [-2, -3, -4, -6, -3, -5],
+                [0.666667, 1.0, 0.0, 0.0, 0.333333, 0.333333],
+            ),  # negative coords
+            (
+                [1000, 2000, 4000, 6000, 3000, 5000],
+                [0.0, 0.0, 0.75, 1.0, 0.50, 0.75],
+            ),  # large range
+            (
+                [0.001, 0.002, 0.004, 0.006, 0.003, 0.005],
+                [0.0, 0.0, 0.75, 1.0, 0.50, 0.75],
+            ),  # small range
+            ([2, 3], [0.0, 0.0]),  # single point
+        ],
+    )
+    def test_normalize_positions(self, positions, expected):
+        result = normalize_points(positions)
+        assert np.allclose(result, expected), f"Expected {expected}, got {result}"
+
+    class PointsPositionsValidatorTest:
+        """Tests for the validator of the `positions` field in PointsCategories.Category."""
+
+        @staticmethod
+        def test_empty_positions_list():
+            """Test that an empty list of positions is allowed."""
+            obj = PointsCategories.Category(positions=[])
+            assert obj.positions == []  # Should allow empty list
+
+        @staticmethod
+        def test_empty_positions_tuple():
+            """Test that an empty list of positions is allowed."""
+            obj = PointsCategories.Category(positions=())
+            assert obj.positions == []  # Should allow empty list
+
+        @staticmethod
+        def test_none_positions():
+            """Test that None is allowed and converted to an empty list."""
+            obj = PointsCategories.Category(positions=None)
+            assert obj.positions == []  # Should allow None and convert to empty list
+
+        @staticmethod
+        def test_valid_positions():
+            """Test that valid positions are allowed."""
+            labels = ["p1", "p2"]
+            positions = [1.0, 2.0, 3.0, 4.0]
+            obj = PointsCategories.Category(labels=labels, positions=positions)
+            assert obj
+
+        @staticmethod
+        def test_type_not_list():
+            """Test that a non-list type for positions raises an error."""
+            with pytest.raises(ValueError, match="Cannot convert positions to list of floats"):
+                PointsCategories.Category(positions=56)
+
+        @staticmethod
+        def test_coordinates_as_string():
+            """Test that the coordinates may be represented as a string."""
+            labels = ["p1", "p2"]
+            positions = ["1", "2", "3", "4"]
+            obj = PointsCategories.Category(labels=labels, positions=positions)
+            assert obj
+
+        @staticmethod
+        def test_non_numeric_elements():
+            """Test that passing non-numeric elements raises an error."""
+            labels = ["p1", "p2"]
+            positions = [1.0, 2.0, 3.0, "not_a_number"]
+            with pytest.raises(ValueError, match="Cannot convert positions to list of floats"):
+                PointsCategories.Category(labels=labels, positions=positions)
+
+        @staticmethod
+        def test_uneven_number_of_elements():
+            """Test that an uneven number of elements raises an error."""
+            with pytest.raises(ValueError, match="positions must have an even number of elements"):
+                PointsCategories.Category(positions=[1.0, 2.0, 3.0])
+
+        @staticmethod
+        def test_num_positions_not_same_as_num_labels():
+            """Test that the number of positions must match the number of labels."""
+            labels = ["p1", "p2", "p3"]  # 3 labels
+            positions = [1.0, 2.0, 3.0, 4.0]  # 2 positions
+            with pytest.raises(
+                ValueError, match="number of positions should be equal to the number of labels"
+            ):
+                PointsCategories.Category(labels=labels, positions=positions)
+
+        @staticmethod
+        def test_positions_allowed_with_empty_labels():
+            """Test that the the number of coordinates check is skipped when labels is empty."""
+            labels = []  # no labels
+            positions = [1.0, 2.0, 3.0, 4.0]  # 2 positions
+            obj = PointsCategories.Category(labels=labels, positions=positions)
+            assert obj
+
+        @staticmethod
+        def test_labels_allowed_with_empty_positions():
+            """Test that the the number of coordinates check is skipped when positions is empty."""
+            labels = ["p1", "p2", "p3"]  # 3 labels
+            positions = []  # no positions
+            obj = PointsCategories.Category(labels=labels, positions=positions)
+            assert obj