[Recording oracle] Apply comments && small fixes && remove unused code

Author: Marishka17

packages/examples/cvat/recording-oracle/src/cvat/api_calls.py

@@ -40,7 +40,7 @@ def get_last_task_quality_report(task_id: int) -> models.QualityReport | None:
 def compute_task_quality_report(
     task_id: int,
     *,
-    max_waiting_time: int = 10 * 60,
+    max_waiting_time: int = 60 * 60,
     sleep_interval: float = 0.5,
 ) -> models.QualityReport:
     logger = logging.getLogger("app")
@@ -86,7 +86,10 @@ def get_task_quality_report(
     report = get_last_task_quality_report(task_id)
     if report and report.created_date > report.target_last_updated:
         if logger.isEnabledFor(logging.DEBUG):
-            logger.debug(f"The latest task({task_id}) quality report({report.id}) is actual")
+            logger.debug(
+                f"The latest task({task_id}) quality report({report.id}) is actual. "
+                "Do not recreate it."
+            )
         return report
 
     return compute_task_quality_report(
@@ -103,20 +106,8 @@ def get_quality_report_data(report_id: int) -> QualityReportData:
             )
             return QualityReportData(**response.json())
 
-        except exceptions.ApiException as e:
-            logger.exception(f"Exception when calling QualityApi.retrieve_report_data: {e}\n")
-            raise
-
-
-def get_job_validation_layout(job_id: int) -> models.JobValidationLayoutRead:
-    logger = logging.getLogger("app")
-    with get_api_client() as api_client:
-        try:
-            layout, _ = api_client.jobs_api.retrieve_validation_layout(job_id)
-            return layout
-
-        except exceptions.ApiException as e:
-            logger.exception(f"Exception when calling JobApi.retrieve_validation_layout: {e}\n")
+        except exceptions.ApiException as ex:
+            logger.exception(f"Exception when calling QualityApi.retrieve_report_data: {ex}\n")
             raise
 
 
@@ -125,18 +116,21 @@ def get_task_validation_layout(task_id: int) -> models.TaskValidationLayoutRead:
     with get_api_client() as api_client:
         try:
             layout, _ = api_client.tasks_api.retrieve_validation_layout(task_id)
+
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(f"Retrieved validation layout: {layout}")
+
             return layout
 
-        except exceptions.ApiException as e:
-            logger.exception(f"Exception when calling TaskApi.retrieve_validation_layout: {e}\n")
+        except exceptions.ApiException as ex:
+            logger.exception(f"Exception when calling TaskApi.retrieve_validation_layout: {ex}\n")
             raise
 
 
 def get_jobs_quality_reports(parent_id: int) -> dict[int, models.QualityReport]:
     logger = logging.getLogger("app")
     with get_api_client() as api_client:
         try:
-            # TODO: optimize
             reports: list[models.QualityReport] = get_paginated_collection(
                 api_client.quality_api.list_reports_endpoint, parent_id=parent_id, target="job"
             )
@@ -148,16 +142,33 @@ def get_jobs_quality_reports(parent_id: int) -> dict[int, models.QualityReport]:
 
 
 def update_task_validation_layout(
-    task_id: int, *, disabled_frames: list[int], honeypot_real_frames: list[int]
+    task_id: int,
+    *,
+    disabled_frames: list[int],
+    shuffle_honeypots: bool = True,
 ) -> None:
     logger = logging.getLogger("app")
+    params = {
+        "disabled_frames": disabled_frames,
+    }
+    if shuffle_honeypots:
+        params["frame_selection_method"] = models.FrameSelectionMethod("random_uniform")
+
     with get_api_client() as api_client:
-        api_client.tasks_api.partial_update_validation_layout(
-            task_id,
-            patched_task_validation_layout_write_request=models.PatchedTaskValidationLayoutWriteRequest(
-                frame_selection_method="manual",
-                disabled_frames=disabled_frames,
-                honeypot_real_frames=honeypot_real_frames,
-            ),
-        )
+        try:
+            validation_layout, _ = api_client.tasks_api.partial_update_validation_layout(
+                task_id,
+                patched_task_validation_layout_write_request=models.PatchedTaskValidationLayoutWriteRequest(
+                    **params
+                ),
+            )
+        except exceptions.ApiException as ex:
+            logger.exception(
+                f"Exception when calling TasksApi.partial_update_validation_layout: {ex}\n"
+            )
+            raise
+
         logger.info(f"Validation layout for the task {task_id} has been updated.")
+
+        if logger.isEnabledFor(logging.DEBUG):
+            logger.debug(f"Validation layout: {validation_layout}")

packages/examples/cvat/recording-oracle/src/cvat/interface.py

@@ -4,20 +4,11 @@
 class AnnotationInfo(BaseModel):
     accuracy: float | int
 
-    class Config:
-        extra = "allow"
-
 
 class FrameResult(BaseModel):
     conflicts: list[dict]
     annotations: AnnotationInfo
 
-    class Config:
-        extra = "allow"
-
 
 class QualityReportData(BaseModel):
     frame_results: dict[str, FrameResult]
-
-    class Config:
-        extra = "allow"

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

@@ -94,7 +94,6 @@ def __init__(
         chain_id: int,
         manifest: TaskManifest,
         *,
-        job_annotations: dict[int, io.IOBase],
         merged_annotations: io.IOBase,
         meta: AnnotationMeta,
         gt_stats: _GtStats | None = None,
@@ -104,7 +103,6 @@ def __init__(
         self.manifest = manifest
 
         self._initial_gt_stats: _GtStats = gt_stats or {}
-        self._job_annotations: dict[int, io.IOBase] = job_annotations
         self._merged_annotations: io.IOBase = merged_annotations
 
         self._updated_merged_dataset_archive: io.IOBase | None = None
@@ -147,16 +145,16 @@ def _load_job_dataset(self, job_id: int, job_dataset_path: Path) -> dm.Dataset:
         )
 
     def _validate_jobs(self):
-        tempdir = self._require_field(self._temp_dir)
         manifest = self._require_field(self.manifest)
-        job_annotations = self._require_field(self._job_annotations)
+        meta = self._require_field(self._meta)
 
         job_results: _JobResults = {}
         rejected_jobs: _RejectedJobs = {}
         self._updated_gt_stats = {}
 
-        cvat_task_ids = {job_meta.task_id for job_meta in self._meta.jobs}
-        job_id_to_task_id = {job_meta.job_id: job_meta.task_id for job_meta in self._meta.jobs}
+        cvat_task_ids = {job_meta.task_id for job_meta in meta.jobs}
+        cvat_job_ids = {job_meta.job_id for job_meta in meta.jobs}
+        job_id_to_task_id = {job_meta.job_id: job_meta.task_id for job_meta in meta.jobs}
 
         task_id_to_quality_report: dict[int, dict] = {}
         task_id_to_quality_report_data: dict[int, dict] = {}
@@ -192,12 +190,9 @@ def _validate_jobs(self):
 
         job_id_to_quality_report = cvat_api.get_jobs_quality_reports(task_quality_report.id)
 
-        for cvat_job_id, job_annotations_file in job_annotations.items():
+        for cvat_job_id in cvat_job_ids:
             cvat_task_id = job_id_to_task_id[cvat_job_id]
 
-            job_dataset_path = tempdir / str(cvat_job_id)
-            extract_zip_archive(job_annotations_file, job_dataset_path)
-
             job_quality_report = job_id_to_quality_report[cvat_job_id]
 
             accuracy = job_quality_report.summary.accuracy
@@ -340,7 +335,6 @@ def process_intermediate_results(  # noqa: PLR0912
     escrow_address: str,
     chain_id: int,
     meta: AnnotationMeta,
-    job_annotations: dict[int, io.RawIOBase],
     merged_annotations: io.RawIOBase,
     manifest: TaskManifest,
     logger: logging.Logger,
@@ -353,6 +347,7 @@ def process_intermediate_results(  # noqa: PLR0912
         ),  # should not happen, but waiting should not block processing
     )
     if not task:
+        # 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)
         task = db_service.get_task_by_id(session, task_id, for_update=True)
 
@@ -369,7 +364,6 @@ def process_intermediate_results(  # noqa: PLR0912
         escrow_address=escrow_address,
         chain_id=chain_id,
         manifest=manifest,
-        job_annotations=job_annotations,
         merged_annotations=merged_annotations,
         meta=meta,
         gt_stats=initial_gt_stats,
@@ -398,26 +392,21 @@ def process_intermediate_results(  # noqa: PLR0912
         for cvat_task_id, val_frame_ids in cvat_task_id_to_failed_val_frames.items():
             task_validation_layout = validation_result.task_id_to_val_layout[cvat_task_id]
             intersection = set(val_frame_ids) & set(task_validation_layout.disabled_frames)
+
             if intersection:
                 logger.error(f"Unexpected case: frames {intersection} were disabled earlier")
 
-            updated_disable_frames = task_validation_layout.disabled_frames + val_frame_ids
-            not_disabled_frames = list(
-                set(task_validation_layout.validation_frames) - set(updated_disable_frames)
-            )
+            upd_disabled_frames = task_validation_layout.disabled_frames + val_frame_ids
 
-            rng = np.random.default_rng()
-            upd_honeypot_real_frames = [
-                frame
-                if frame not in updated_disable_frames
-                else int(rng.choice(not_disabled_frames))
-                for frame in task_validation_layout.honeypot_real_frames
-            ]
+            shuffle_honeypots = True
+            if set(upd_disabled_frames) == set(task_validation_layout.validation_frames):
+                logger.error("All validation frames were banned. Honeypots will not be shuffled")
+                shuffle_honeypots = False
 
             cvat_api.update_task_validation_layout(
                 cvat_task_id,
-                disabled_frames=updated_disable_frames,
-                honeypot_real_frames=upd_honeypot_real_frames,
+                disabled_frames=upd_disabled_frames,
+                shuffle_honeypots=shuffle_honeypots,
             )
 
         if logger.isEnabledFor(logging.DEBUG):

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

@@ -48,7 +48,6 @@ def __init__(
         self.data_bucket = BucketAccessInfo.parse_obj(Config.exchange_oracle_storage_config)
 
         self.annotation_meta: annotation.AnnotationMeta | None = None
-        self.job_annotations: dict[int, bytes] | None = None
         self.merged_annotations: bytes | None = None
 
     def set_logger(self, logger: Logger):
@@ -70,23 +69,15 @@ def _download_annotations(self):
 
         data_bucket_client = make_cloud_client(self.data_bucket)
 
-        job_annotations = {}
-        for job_meta in self.annotation_meta.jobs:
-            job_filename = compose_annotation_results_bucket_filename(
-                self.escrow_address,
-                self.chain_id,
-                job_meta.annotation_filename,
-            )
-            job_annotations[job_meta.job_id] = data_bucket_client.download_file(job_filename)
-
-        excor_merged_annotation_path = compose_annotation_results_bucket_filename(
+        exchange_oracle_merged_annotation_path = compose_annotation_results_bucket_filename(
             self.escrow_address,
             self.chain_id,
             annotation.RESULTING_ANNOTATIONS_FILE,
         )
-        merged_annotations = data_bucket_client.download_file(excor_merged_annotation_path)
+        merged_annotations = data_bucket_client.download_file(
+            exchange_oracle_merged_annotation_path
+        )
 
-        self.job_annotations = job_annotations
         self.merged_annotations = merged_annotations
 
     def _download_results(self):
@@ -97,7 +88,6 @@ def _download_results(self):
 
     def _process_annotation_results(self) -> ValidationResult:
         assert self.annotation_meta is not None
-        assert self.job_annotations is not None
         assert self.merged_annotations is not None
 
         # TODO: refactor further
@@ -106,7 +96,6 @@ def _process_annotation_results(self) -> ValidationResult:
             escrow_address=self.escrow_address,
             chain_id=self.chain_id,
             meta=self.annotation_meta,
-            job_annotations={k: io.BytesIO(v) for k, v in self.job_annotations.items()},
             merged_annotations=io.BytesIO(self.merged_annotations),
             manifest=self.manifest,
             logger=self.logger,
@@ -201,7 +190,6 @@ def _handle_validation_result(self, validation_result: ValidationResult):
                 OracleWebhookTypes.exchange_oracle,
                 event=RecordingOracleEvent_SubmissionRejected(
                     # TODO: send all assignments, handle rejection reason in Exchange Oracle
-                    # change validation frames in these jobs once possible
                     assignments=[
                         RecordingOracleEvent_SubmissionRejected.RejectedAssignmentInfo(
                             assignment_id=job_id_to_assignment_id[rejected_job_id],

packages/examples/cvat/recording-oracle/src/services/validation.py

@@ -132,7 +132,7 @@ def get_task_gt_stats(
     )
 
 
-def update_gt_stats(session: Session, task_id: int, values: dict[tuple[int, int], int]):
+def update_gt_stats(session: Session, task_id: str, values: dict[tuple[int, int], int]):
     # Read more about upsert:
     # https://docs.sqlalchemy.org/en/20/orm/queryguide/dml.html#orm-upsert-statements