Fix interpolation selection and add 3D annotation tests

changelog.d/20251229_140724_grigorio714_annotations_download_error.md

@@ -0,0 +1,4 @@
+### Fixed
+
+- Fixed a bug that occurred during the download of 3D annotations.
+  (<https://github.com/cvat-ai/cvat/pull/10165>)

cvat/apps/dataset_manager/annotation.py

@@ -1080,7 +1080,7 @@ def interpolate(shape0, shape1):
 
             if dimension == DimensionType.DIM_3D:
                 yield from simple_3d_interpolation(shape0, shape1)
-            if is_rectangle or is_cuboid or is_ellipse or is_skeleton:
+            elif is_rectangle or is_cuboid or is_ellipse or is_skeleton:
                 yield from simple_interpolation(shape0, shape1)
             elif is_points:
                 yield from points_interpolation(shape0, shape1)

cvat/apps/dataset_manager/tests/test_annotation_3D.py

@@ -0,0 +1,263 @@
+# Copyright (C) 2020-2022 Intel Corporation
+#
+# SPDX-License-Identifier: MIT
+from unittest import mock
+
+from django.test import TestCase
+
+from cvat.apps.dataset_manager import task as task_module
+from cvat.apps.dataset_manager.annotation import AnnotationIR, TrackManager
+from cvat.apps.engine import models
+
+
+class TrackManager3DTest(TestCase):
+    def _check_interpolation(self, track):
+        interpolated = TrackManager.get_interpolated_shapes(
+            track, 0, 7, models.DimensionType.DIM_3D
+        )
+
+        self.assertEqual(
+            [
+                {"frame": 0, "keyframe": True, "outside": False},
+                {"frame": 1, "keyframe": False, "outside": False},
+                {"frame": 2, "keyframe": True, "outside": True},
+                # frame = 3 should be skipped as it is outside and interpolated
+                {"frame": 4, "keyframe": True, "outside": False},
+                {"frame": 5, "keyframe": False, "outside": False},
+                {"frame": 6, "keyframe": False, "outside": False},
+            ],
+            [
+                {k: v for k, v in shape.items() if k in ["frame", "keyframe", "outside"]}
+                for shape in interpolated
+            ],
+        )
+
+    def _cuboid_points(self, base=0.0):
+        # Return a 16-element list similar to other tests in the repo.
+        # Structure: center x,y,z, angles (3), some zeros, lengths (3), rest zeros
+        return [
+            1.0 + base,
+            2.0 + base,
+            3.0 + base,
+            0.0,
+            0.0,
+            0.0,
+            4.0 + base,
+            4.0 + base,
+            4.0 + base,
+            0.0,
+            0.0,
+            0.0,
+            0.0,
+            0.0,
+            0.0,
+            0.0,
+        ]
+
+    def _make_cuboid_shape(self, frame, base=0.0, outside=False, rotation=0, attributes=None):
+        return {
+            "frame": frame,
+            "points": self._cuboid_points(base),
+            "rotation": rotation,
+            "type": "cuboid",
+            "occluded": False,
+            "outside": outside,
+            "attributes": attributes or [],
+        }
+
+    def _make_track(self, shapes, frame=0, label_id=0, source="manual", attributes=None):
+        return {
+            "frame": frame,
+            "label_id": label_id,
+            "group": None,
+            "source": source,
+            "attributes": attributes or [],
+            "shapes": shapes,
+        }
+
+    def test_cuboid_interpolation(self):
+        shapes = [
+            self._make_cuboid_shape(0, base=0.0, outside=False),
+            self._make_cuboid_shape(2, base=2.0, outside=True),
+            self._make_cuboid_shape(4, base=4.0, outside=False),
+        ]
+
+        track = self._make_track(shapes)
+
+        self._check_interpolation(track)
+
+    def test_outside_cuboid_interpolation(self):
+        shapes = [
+            self._make_cuboid_shape(0, base=0.0, outside=False),
+            self._make_cuboid_shape(2, base=2.0, outside=True),
+            self._make_cuboid_shape(4, base=4.0, outside=True),
+        ]
+
+        track = self._make_track(shapes)
+
+        interpolated = TrackManager.get_interpolated_shapes(
+            track, 0, 5, models.DimensionType.DIM_3D
+        )
+
+        # ensure frames/keyframe/outside sequence is reasonable for 3D cuboids
+        expected = [0, 1, 2, 4]
+        got = [shape["frame"] for shape in interpolated]
+        self.assertEqual(expected, got)
+
+    def test_deleted_frames_with_keyframes_are_ignored_3d(self):
+        deleted_frames = [2]
+        end_frame = 5
+
+        shapes = [
+            self._make_cuboid_shape(0, base=0.0, outside=False),
+            self._make_cuboid_shape(2, base=2.0, outside=False),
+            self._make_cuboid_shape(4, base=4.0, outside=False),
+        ]
+
+        track = self._make_track(shapes)
+
+        interpolated_shapes = TrackManager.get_interpolated_shapes(
+            track, 0, end_frame, models.DimensionType.DIM_3D, deleted_frames=deleted_frames
+        )
+
+        expected_frames = [0, 1, 3, 4]
+        self.assertEqual(expected_frames, [s["frame"] for s in interpolated_shapes])
+
+    def test_duplicated_shape_interpolation_3d(self):
+        # Ensure duplicated shapes in a track are deduplicated
+        shape0 = {
+            "type": "cuboid",
+            "occluded": False,
+            "outside": False,
+            "points": self._cuboid_points(0.0),
+            "frame": 0,
+            "attributes": [],
+            "rotation": 0,
+        }
+        shape1 = {
+            "type": "cuboid",
+            "occluded": False,
+            "outside": True,
+            "points": self._cuboid_points(0.0),
+            "frame": 1,
+            "attributes": [],
+            "rotation": 0,
+        }
+        shapes = [shape0, shape1, shape1]
+
+        track = {
+            "id": 777,
+            "frame": 0,
+            "group": None,
+            "source": "manual",
+            "attributes": [],
+            "elements": [],
+            "label": "car",
+            "shapes": shapes,
+        }
+
+        interpolated_shapes = TrackManager.get_interpolated_shapes(
+            track, 0, 2, models.DimensionType.DIM_3D
+        )
+        # Expect only two shapes (no duplicated last one)
+        self.assertEqual(2, len(interpolated_shapes))
+
+
+class AnnotationIR3DTest(TestCase):
+    def test_slice_track_does_not_duplicate_outside_frame_on_the_end_3d(self):
+        track_shapes = [
+            {
+                "type": "cuboid",
+                "occluded": False,
+                "outside": False,
+                "points": [100, 100, 100] + [0] * 13,
+                "frame": 0,
+                "attributes": [],
+                "rotation": 0,
+            },
+            {
+                "type": "cuboid",
+                "occluded": False,
+                "outside": True,
+                "points": [100, 100, 100] + [0] * 13,
+                "frame": 1,
+                "attributes": [],
+                "rotation": 0,
+            },
+            {
+                "type": "cuboid",
+                "occluded": False,
+                "outside": False,
+                "points": [111, 111, 111] + [0] * 13,
+                "frame": 10,
+                "attributes": [],
+                "rotation": 0,
+            },
+        ]
+        data = {
+            "tags": [],
+            "shapes": [],
+            "tracks": [
+                {
+                    "id": 666,
+                    "frame": 0,
+                    "group": None,
+                    "source": "manual",
+                    "attributes": [],
+                    "elements": [],
+                    "label": "car",
+                    "shapes": track_shapes,
+                }
+            ],
+        }
+        annotation = AnnotationIR(dimension=models.DimensionType.DIM_3D, data=data)
+        sliced_annotation = annotation.slice(0, 1)
+        self.assertEqual(sliced_annotation.data["tracks"][0]["shapes"], track_shapes[0:2])
+
+
+class TestTaskAnnotation3D(TestCase):
+    def test_reads_ordered_jobs_3d(self):
+        # replicate the job ordering test but ensure AnnotationIR used is 3D
+        user = models.User.objects.create_superuser(username="admin", email="", password="admin")
+
+        db_data = models.Data.objects.create(size=31, stop_frame=30, image_quality=50)
+
+        data = {"name": "my task", "owner": user, "overlap": 1, "segment_size": 11}
+        db_task = models.Task.objects.create(data=db_data, **data)
+
+        models.Job.objects.create(
+            segment=models.Segment.objects.create(task=db_task, start_frame=0, stop_frame=10),
+            type=models.JobType.ANNOTATION,
+            id=456789,
+        )
+        models.Job.objects.create(
+            segment=models.Segment.objects.create(task=db_task, start_frame=10, stop_frame=20),
+            type=models.JobType.ANNOTATION,
+            id=123456,
+        )
+        models.Job.objects.create(
+            segment=models.Segment.objects.create(task=db_task, start_frame=20, stop_frame=30),
+            type=models.JobType.ANNOTATION,
+            id=345678,
+        )
+
+        unordered_ids = list(
+            models.Job.objects.filter(segment__task_id=db_task.id).values_list("id", flat=True)
+        )
+        assert sorted(unordered_ids) != unordered_ids
+
+        class DummyJobAnnotation(task_module.JobAnnotation):
+            called_ids = []
+
+            def __init__(self, job_id, db_job=None):
+                self.called_ids.append(job_id)
+                self.ir_data = AnnotationIR(models.DimensionType.DIM_3D)
+
+            def init_from_db(self, *, streaming: bool = False):
+                pass
+
+        with mock.patch.object(task_module, "JobAnnotation", DummyJobAnnotation):
+            ta = task_module.TaskAnnotation(db_task.id)
+            ta.init_from_db()
+
+        assert DummyJobAnnotation.called_ids == sorted(unordered_ids)

tests/python/rest_api/test_tasks.py

@@ -989,6 +989,59 @@ def test_datumaro_export_without_annotations_includes_image_info(
                     if "size" in related_image:
                         assert tuple(related_image["size"]) > (0, 0)
 
+    @pytest.mark.usefixtures("restore_db_per_function")
+    def test_can_export_3d_annotations_via_rest_api(self, admin_user, tasks):
+        """Export 3D annotations via REST API v2 and verify returned archive."""
+        # find a task with 3d dimension and at least one frame
+        try:
+            task = next(t for t in tasks if t.get("dimension") == "3d" and t.get("size"))
+        except StopIteration:
+            pytest.skip("No 3D task found in fixtures")
+
+        with make_api_client(admin_user) as api_client:
+            dataset_bytes = export_dataset(
+                api_client.tasks_api,
+                id=task["id"],
+                save_images=False,
+                format="Datumaro 3D 1.0",
+            )
+
+        assert dataset_bytes is not None
+        assert zipfile.is_zipfile(io.BytesIO(dataset_bytes))
+
+        with zipfile.ZipFile(io.BytesIO(dataset_bytes)) as zf:
+            # basic sanity: annotations folder should be present in the archive
+            names = zf.namelist()
+            assert any(
+                name.startswith("annotations/") for name in names
+            ), f"No annotations folder in export archive: {names}"
+
+    @pytest.mark.usefixtures("restore_db_per_function")
+    def test_can_export_3d_annotations_from_job_via_rest_api(self, admin_user, tasks, jobs):
+        """Export 3D annotations from a Job via REST API v2 and verify returned archive."""
+        # find a 3D job (job belongs to a task with dimension == '3d')
+        try:
+            job = next(j for j in jobs if tasks[j["task_id"]]["dimension"] == "3d")
+        except StopIteration:
+            pytest.skip("No 3D job found in fixtures")
+
+        with make_api_client(admin_user) as api_client:
+            dataset_bytes = export_dataset(
+                api_client.jobs_api,
+                id=job["id"],
+                save_images=False,
+                format="Datumaro 3D 1.0",
+            )
+
+        assert dataset_bytes is not None
+        assert zipfile.is_zipfile(io.BytesIO(dataset_bytes))
+
+        with zipfile.ZipFile(io.BytesIO(dataset_bytes)) as zf:
+            names = zf.namelist()
+            assert any(
+                name.startswith("annotations/") for name in names
+            ), f"No annotations folder in job export archive: {names}"
+
 
 @pytest.mark.usefixtures("restore_db_per_function")
 class TestPatchTaskLabel: