Allow manifest changes in recording oracle

Author: zhiltsov-max

packages/examples/cvat/recording-oracle/src/core/annotation_meta.py

@@ -1,8 +1,13 @@
-from collections.abc import Iterator
-from pathlib import Path
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
+from pathlib import Path
 from pydantic import BaseModel
 
+if TYPE_CHECKING:
+    from collections.abc import Collection, Iterator
+
 ANNOTATION_RESULTS_METAFILE_NAME = "annotation_meta.json"
 RESULTING_ANNOTATIONS_FILE = "resulting_annotations.zip"
 
@@ -24,5 +29,7 @@ def job_frame_range(self) -> Iterator[int]:
 class AnnotationMeta(BaseModel):
     jobs: list[JobMeta]
 
-    def skip_jobs(self, job_ids: list[int]):
-        return AnnotationMeta(jobs=[job for job in self.jobs if job.job_id not in job_ids])
+    def skip_assignments(self, assignment_ids: Collection[int]) -> AnnotationMeta:
+        return AnnotationMeta(
+            jobs=[job for job in self.jobs if job.assignment_id not in assignment_ids]
+        )

packages/examples/cvat/recording-oracle/src/handlers/process_intermediate_results.py

@@ -375,13 +375,12 @@ def process_intermediate_results(  # noqa: PLR0912
         ),  # should not happen, but waiting should not block processing
     )
     if task:
-        # skip jobs that have already been accepted on the previous epochs
-        accepted_cvat_job_ids = [
-            validation_result.job.cvat_id
+        # Skip assignments that were validated earlier
+        validated_assignment_ids = {
+            validation_result.assignment_id
             for validation_result in db_service.get_task_validation_results(session, task.id)
-            if validation_result.annotation_quality >= manifest.validation.min_quality
-        ]
-        unchecked_jobs_meta = meta.skip_jobs(accepted_cvat_job_ids)
+        }
+        unchecked_jobs_meta = meta.skip_assignments(validated_assignment_ids)
     else:
         # Recording Oracle task represents all CVAT tasks related with the escrow
         task_id = db_service.create_task(session, escrow_address=escrow_address, chain_id=chain_id)
@@ -566,6 +565,16 @@ def process_intermediate_results(  # noqa: PLR0912
         else:
             assignment_validation_result_id = assignment_validation_result.id
 
+        # We consider only the last assignment as final even if there were assignments with higher
+        # quality score. The reason for this is that during escrow annotation there are various
+        # task changes possible, for instance:
+        # - GT can be changed in the middle of the task annotation
+        # - manifest can be updated with different quality parameters
+        # etc. It can be considered more of a development or testing conditions so far,
+        # according to the current system requirements, but it's likely to be
+        # a normal requirement in the future.
+        # Therefore, we use the logic: only the last job assignment can be considered
+        # a final annotation result, regardless of the assignment quality.
         job_final_result_ids[job.id] = assignment_validation_result_id
 
     task_jobs = task.jobs

packages/examples/cvat/recording-oracle/tests/conftest.py

@@ -2,9 +2,10 @@
 
 import pytest
 from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
 
 from src import app
-from src.db import Base, engine
+from src.db import Base, SessionLocal, engine
 
 
 @pytest.fixture(autouse=True)
@@ -17,3 +18,14 @@ def db():
 def client() -> Generator:
     with TestClient(app) as c:
         yield c
+
+
+@pytest.fixture
+def session() -> Generator[Session, None, None]:
+    session = SessionLocal()
+
+    try:
+        yield session
+    finally:
+        session.rollback()
+        session.close()

packages/examples/cvat/recording-oracle/tests/integration/services/test_validation_service.py

@@ -1,9 +1,21 @@
+import io
 import random
 import unittest
 import uuid
+from contextlib import ExitStack
+from logging import Logger
+from types import SimpleNamespace
+from unittest import mock
 
+from sqlalchemy.orm import Session
+
+from src.core.annotation_meta import AnnotationMeta, JobMeta
+from src.core.manifest import TaskManifest, parse_manifest
 from src.core.types import Networks
+from src.core.validation_results import ValidationFailure, ValidationSuccess
+from src.cvat import api_calls as cvat_api
 from src.db import SessionLocal
+from src.handlers.process_intermediate_results import process_intermediate_results
 from src.services.validation import (
     create_job,
     create_task,
@@ -16,19 +28,19 @@
     get_validation_result_by_assignment_id,
 )
 
+from tests.utils.constants import ESCROW_ADDRESS, WALLET_ADDRESS1
+
 
 class ServiceIntegrationTest(unittest.TestCase):
     def setUp(self):
         random.seed(42)
         self.session = SessionLocal()
-        self.escrow_address = "0x" + "".join([str(random.randint(0, 9)) for _ in range(40)])
+        self.escrow_address = ESCROW_ADDRESS
         self.chain_id = Networks.localhost
         self.cvat_id = 0
-        self.annotator_wallet_address = "0x" + "".join(
-            [str(random.randint(0, 9)) for _ in range(40)]
-        )
+        self.annotator_wallet_address = WALLET_ADDRESS1
         self.annotation_quality = 0.9
-        self.assigment_id = str(uuid.uuid4())
+        self.assignment_id = str(uuid.uuid4())
 
     def tearDown(self):
         self.session.close()
@@ -67,12 +79,214 @@ def test_create_and_get_validation_result(self):
             job_id,
             self.annotator_wallet_address,
             self.annotation_quality,
-            self.assigment_id,
+            self.assignment_id,
         )
 
-        vr = get_validation_result_by_assignment_id(self.session, self.assigment_id)
+        vr = get_validation_result_by_assignment_id(self.session, self.assignment_id)
         assert vr.id == vr_id
 
         vrs = get_task_validation_results(self.session, task_id)
         assert len(vrs) == 1
         assert vrs[0] == vr
+
+
+class TestManifestChange:
+    @staticmethod
+    def _generate_manifest(*, min_quality: float = 0.8) -> TaskManifest:
+        data = {
+            "data": {"data_url": "http://localhost:9010/datasets/sample"},
+            "annotation": {
+                "labels": [{"name": "person"}],
+                "description": "",
+                "user_guide": "",
+                "type": "image_points",
+                "job_size": 10,
+            },
+            "validation": {
+                "min_quality": min_quality,
+                "val_size": 2,
+                "gt_url": "http://localhost:9010/datasets/sample/annotations/sample_gt.json",
+            },
+            "job_bounty": "0.0001",
+        }
+
+        return parse_manifest(data)
+
+    def test_can_handle_lowered_quality_requirements_in_manifest(self, session: Session):
+        escrow_address = ESCROW_ADDRESS
+        chain_id = Networks.localhost
+
+        min_quality1 = 0.8
+        min_quality2 = 0.5
+        frame_count = 10
+
+        manifest = self._generate_manifest(min_quality=min_quality1)
+
+        cvat_task_id = 1
+        cvat_job_id = 1
+        annotator1 = WALLET_ADDRESS1
+
+        assignment1_id = f"0x{0:040d}"
+        assignment1_quality = 0.7
+
+        assignment2_id = f"0x{1:040d}"
+        assignment2_quality = 0.6
+
+        # create a validation input
+        with ExitStack() as common_lock_es:
+            logger = mock.Mock(Logger)
+
+            mock_make_cloud_client = common_lock_es.enter_context(
+                mock.patch("src.handlers.process_intermediate_results.make_cloud_client")
+            )
+            mock_make_cloud_client.return_value.download_file = mock.Mock(return_value=b"")
+
+            mock_get_task_validation_layout = common_lock_es.enter_context(
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_task_validation_layout"
+                )
+            )
+            mock_get_task_validation_layout.return_value = mock.Mock(
+                cvat_api.models.ITaskValidationLayoutRead,
+                honeypot_frames=[0, 1],
+                honeypot_real_frames=[0, 1],
+            )
+
+            mock_get_task_data_meta = common_lock_es.enter_context(
+                mock.patch("src.handlers.process_intermediate_results.cvat_api.get_task_data_meta")
+            )
+            mock_get_task_data_meta.return_value = mock.Mock(
+                cvat_api.models.IDataMetaRead,
+                frames=[SimpleNamespace(name=f"frame_{i}.jpg") for i in range(frame_count)],
+            )
+
+            common_lock_es.enter_context(
+                mock.patch("src.handlers.process_intermediate_results.dm.Dataset.import_from")
+            )
+            common_lock_es.enter_context(
+                mock.patch("src.handlers.process_intermediate_results.extract_zip_archive")
+            )
+            common_lock_es.enter_context(
+                mock.patch("src.handlers.process_intermediate_results.write_dir_to_zip_archive")
+            )
+
+            def patched_prepare_merged_dataset(self):
+                self._updated_merged_dataset_archive = io.BytesIO()
+
+            common_lock_es.enter_context(
+                mock.patch(
+                    "src.handlers.process_intermediate_results._TaskValidator._prepare_merged_dataset",
+                    patched_prepare_merged_dataset,
+                )
+            )
+
+            annotation_meta = AnnotationMeta(
+                jobs=[
+                    JobMeta(
+                        job_id=cvat_job_id,
+                        task_id=cvat_task_id,
+                        annotation_filename="",
+                        annotator_wallet_address=annotator1,
+                        assignment_id=assignment1_id,
+                        start_frame=0,
+                        stop_frame=manifest.annotation.job_size + manifest.validation.val_size,
+                    )
+                ]
+            )
+
+            with (
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_task_quality_report"
+                ) as mock_get_task_quality_report,
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_quality_report_data"
+                ) as mock_get_quality_report_data,
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_jobs_quality_reports"
+                ) as mock_get_jobs_quality_reports,
+            ):
+                mock_get_task_quality_report.return_value = mock.Mock(
+                    cvat_api.models.IQualityReport, id=1
+                )
+                mock_get_quality_report_data.return_value = mock.Mock(
+                    cvat_api.QualityReportData,
+                    frame_results={
+                        "0": mock.Mock(annotations=mock.Mock(accuracy=assignment1_quality)),
+                        "1": mock.Mock(annotations=mock.Mock(accuracy=assignment1_quality)),
+                    },
+                )
+                mock_get_jobs_quality_reports.return_value = [
+                    mock.Mock(
+                        cvat_api.models.IQualityReport,
+                        job_id=1,
+                        summary=mock.Mock(accuracy=assignment1_quality),
+                    ),
+                ]
+
+                vr1 = process_intermediate_results(
+                    session,
+                    escrow_address=escrow_address,
+                    chain_id=chain_id,
+                    meta=annotation_meta,
+                    merged_annotations=io.BytesIO(),
+                    manifest=manifest,
+                    logger=logger,
+                )
+
+            assert isinstance(vr1, ValidationFailure)
+            assert len(vr1.rejected_jobs) == 1
+
+            manifest.validation.min_quality = min_quality2
+
+            annotation_meta.jobs[0].assignment_id = assignment2_id
+
+            with (
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_task_quality_report"
+                ) as mock_get_task_quality_report,
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_quality_report_data"
+                ) as mock_get_quality_report_data,
+                mock.patch(
+                    "src.handlers.process_intermediate_results.cvat_api.get_jobs_quality_reports"
+                ) as mock_get_jobs_quality_reports,
+            ):
+                mock_get_task_quality_report.return_value = mock.Mock(
+                    cvat_api.models.IQualityReport, id=2
+                )
+                mock_get_quality_report_data.return_value = mock.Mock(
+                    cvat_api.QualityReportData,
+                    frame_results={
+                        "0": mock.Mock(annotations=mock.Mock(accuracy=assignment2_quality)),
+                        "1": mock.Mock(annotations=mock.Mock(accuracy=assignment2_quality)),
+                    },
+                )
+                mock_get_jobs_quality_reports.return_value = [
+                    mock.Mock(
+                        cvat_api.models.IQualityReport,
+                        job_id=1,
+                        summary=mock.Mock(accuracy=assignment2_quality),
+                    ),
+                ]
+
+                vr2 = process_intermediate_results(
+                    session,
+                    escrow_address=escrow_address,
+                    chain_id=chain_id,
+                    meta=annotation_meta,
+                    merged_annotations=io.BytesIO(),
+                    manifest=manifest,
+                    logger=logger,
+                )
+
+        assert isinstance(vr2, ValidationSuccess)
+        assert vr2.job_results[cvat_job_id] == assignment2_quality
+
+        assert len(vr2.validation_meta.jobs) == 1
+        assert len(vr2.validation_meta.results) == 2
+        assert (
+            vr2.validation_meta.results[
+                vr2.validation_meta.jobs[0].final_result_id
+            ].annotation_quality
+            == assignment2_quality
+        )

packages/examples/cvat/recording-oracle/tests/utils/constants.py

@@ -98,3 +98,5 @@
     -----END PGP PUBLIC KEY BLOCK-----
 """
 )
+
+WALLET_ADDRESS1 = "0x5aAeb6053F3E94C9b9A09f33669435E7Ef1BeAed"