diff --git a/changelog.d/20251229_140724_grigorio714_annotations_download_error.md b/changelog.d/20251229_140724_grigorio714_annotations_download_error.md new file mode 100644 index 000000000000..4fa0c8d1cfb7 --- /dev/null +++ b/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. + () diff --git a/cvat/apps/dataset_manager/annotation.py b/cvat/apps/dataset_manager/annotation.py index 7d48daeecce9..ab9aa2ddee20 100644 --- a/cvat/apps/dataset_manager/annotation.py +++ b/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) diff --git a/cvat/apps/dataset_manager/tests/test_annotation_3D.py b/cvat/apps/dataset_manager/tests/test_annotation_3D.py new file mode 100644 index 000000000000..b1acd996796e --- /dev/null +++ b/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) diff --git a/tests/python/rest_api/test_tasks.py b/tests/python/rest_api/test_tasks.py index ee5584259f62..a709c764e143 100644 --- a/tests/python/rest_api/test_tasks.py +++ b/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: