[CVAT] Refactor oracle configs (#2780)

* Remove extra files

* Remove outdated validation settings

* Remove unused aws storage region

* Remove cvat_ prefix for cvat settings

* Refactor env var access

* Update exchange oracle cron settings

* Add comment

* Add missing template vars

* Format code

* Merge

* Add missing type conversions

* Fix invalid webhook handling

* Fix merge

* Remove extra config parameter for polygons iou

* Update code formatting

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/pyproject.toml

@@ -114,6 +114,7 @@ ignore = [
     "ANN002",  # Missing type annotation for `*args`
     "TRY300",  # Consider moving this statement to an `else` block
     "C901",    # Function is too complex
+    "PLW1508", # invalid-envvar-default. Alerts only for os.getenv(), but not for os.environ.get()
     "PLW2901", # Variable overwritten by assignment target
     "PTH118",  # Prefer pathlib instead of os.path
     "PTH119",  # `os.path.basename()` should be replaced by `Path.name`

packages/examples/cvat/exchange-oracle/src/.env.template

@@ -44,9 +44,7 @@ PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE=
 PROCESS_REPUTATION_ORACLE_WEBHOOKS_CHUNK_SIZE=
 PROCESS_REPUTATION_ORACLE_WEBHOOKS_INT=
 TRACK_COMPLETED_PROJECTS_INT=
-TRACK_COMPLETED_PROJECTS_CHUNK_SIZE=
 TRACK_COMPLETED_TASKS_INT=
-TRACK_COMPLETED_TASKS_CHUNK_SIZE=
 TRACK_COMPLETED_ESCROWS_INT=
 TRACK_COMPLETED_ESCROWS_CHUNK_SIZE=
 TRACK_ESCROW_VALIDATIONS_INT=
@@ -79,13 +77,11 @@ CVAT_IOU_THRESHOLD=
 CVAT_OKS_SIGMA=
 CVAT_EXPORT_TIMEOUT=
 CVAT_IMPORT_TIMEOUT=
-CVAT_POLYGONS_IOU_THRESHOLD=
 
 # Storage Config (S3/GCS)
 
 STORAGE_PROVIDER=
 STORAGE_ENDPOINT_URL=
-STORAGE_REGION=
 STORAGE_ACCESS_KEY=
 STORAGE_SECRET_KEY=
 STORAGE_KEY_FILE_PATH=
@@ -96,6 +92,7 @@ STORAGE_USE_SSL=
 
 ENABLE_CUSTOM_CLOUD_HOST=
 REQUEST_LOGGING_ENABLED=
+PROFILING_ENABLED=
 MANIFEST_CACHE_TTL=
 
 # Core
@@ -112,16 +109,22 @@ HUMAN_APP_JWT_KEY=
 # API config
 
 DEFAULT_API_PAGE_SIZE=
+MIN_API_PAGE_SIZE=
+MAX_API_PAGE_SIZE=
+STATS_RPS_LIMIT=
 
 # Localhost
 
+LOCALHOST_RPC_API_URL=
+LOCALHOST_AMOY_ADDR=
 LOCALHOST_RECORDING_ORACLE_ADDRESS=
 LOCALHOST_RECORDING_ORACLE_URL=
 LOCALHOST_JOB_LAUNCHER_URL=
 LOCALHOST_REPUTATION_ORACLE_ADDRESS=
 LOCALHOST_REPUTATION_ORACLE_URL=
 
 # Encryption
+
 PGP_PRIVATE_KEY=
 PGP_PASSPHRASE=
 PGP_PUBLIC_KEY_URL=

packages/examples/cvat/exchange-oracle/src/core/config.py

@@ -60,7 +60,6 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
                 )
                 return
 
-            chunk_size = CronConfig.accepted_projects_chunk_size
             project_ids = cvat_db_service.get_project_cvat_ids_by_escrow_address(
                 db_session, webhook.escrow_address
             )
@@ -71,6 +70,7 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
                 )
                 return
 
+            chunk_size = CronConfig.process_accepted_projects_chunk_size
             for ids_chunk in take_by(project_ids, chunk_size):
                 projects_chunk = cvat_db_service.get_projects_by_cvat_ids(
                     db_session, ids_chunk, for_update=True, limit=chunk_size
@@ -138,7 +138,7 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
             )
             rejected_project_cvat_ids = set(j.cvat_project_id for j in rejected_jobs)
 
-            chunk_size = CronConfig.rejected_projects_chunk_size
+            chunk_size = CronConfig.process_rejected_projects_chunk_size
             for chunk_ids in take_by(rejected_project_cvat_ids, chunk_size):
                 projects_chunk = cvat_db_service.get_projects_by_cvat_ids(
                     db_session, chunk_ids, for_update=True, limit=chunk_size

packages/examples/cvat/exchange-oracle/src/crons/webhooks/recording_oracle.py

@@ -77,7 +77,7 @@ def _get_annotations(
     time_begin = utcnow()
 
     if timeout is _NOTSET:
-        timeout = Config.cvat_config.cvat_export_timeout
+        timeout = Config.cvat_config.export_timeout
 
     while True:
         (_, response) = endpoint.call_with_http_info(
@@ -118,23 +118,23 @@ def get_api_client() -> ApiClient:
         return current_api_client
 
     configuration = Configuration(
-        host=Config.cvat_config.cvat_url,
-        username=Config.cvat_config.cvat_admin,
-        password=Config.cvat_config.cvat_admin_pass,
+        host=Config.cvat_config.host_url,
+        username=Config.cvat_config.admin_login,
+        password=Config.cvat_config.admin_pass,
     )
 
     api_client = ApiClient(configuration=configuration)
-    api_client.set_default_header("X-organization", Config.cvat_config.cvat_org_slug)
+    api_client.set_default_header("X-organization", Config.cvat_config.org_slug)
 
     return api_client
 
 
 def get_sdk_client() -> Client:
     client = make_client(
-        host=Config.cvat_config.cvat_url,
-        credentials=(Config.cvat_config.cvat_admin, Config.cvat_config.cvat_admin_pass),
+        host=Config.cvat_config.host_url,
+        credentials=(Config.cvat_config.admin_login, Config.cvat_config.admin_pass),
     )
-    client.organization_slug = Config.cvat_config.cvat_org_slug
+    client.organization_slug = Config.cvat_config.org_slug
 
     return client
 
@@ -291,11 +291,11 @@ def create_cvat_webhook(project_id: int) -> models.WebhookRead:
     logger = logging.getLogger("app")
     with get_api_client() as api_client:
         webhook_write_request = models.WebhookWriteRequest(
-            target_url=Config.cvat_config.cvat_incoming_webhooks_url,
+            target_url=Config.cvat_config.incoming_webhooks_url,
             description="Exchange Oracle notification",
             type=models.WebhookType("project"),
             content_type=models.WebhookContentType("application/json"),
-            secret=Config.cvat_config.cvat_webhook_secret,
+            secret=Config.cvat_config.webhook_secret,
             is_active=True,
             # enable_ssl=True,
             project_id=project_id,
@@ -403,7 +403,7 @@ def put_task_data(
         data_request = models.DataRequest(
             chunk_size=chunk_size,
             cloud_storage_id=cloudstorage_id,
-            image_quality=Config.cvat_config.cvat_default_image_quality,
+            image_quality=Config.cvat_config.image_quality,
             use_cache=True,
             use_zip_chunks=True,
             sorting_method=sorting_method,
@@ -641,7 +641,7 @@ def upload_gt_annotations(
     *,
     format_name: str,
     sleep_interval: int = 5,
-    timeout: int | None = Config.cvat_config.cvat_import_timeout,
+    timeout: int | None = Config.cvat_config.import_timeout,
 ) -> None:
     # FUTURE-TODO: use job.import_annotations when CVAT supports a waiting timeout
     start_time = datetime.now(timezone.utc)
@@ -728,8 +728,8 @@ def update_quality_control_settings(
     *,
     target_metric_threshold: float,
     target_metric: str = "accuracy",
-    max_validations_per_job: int = Config.cvat_config.cvat_max_validation_checks,
-    iou_threshold: float = Config.cvat_config.cvat_iou_threshold,
+    max_validations_per_job: int = Config.cvat_config.max_validation_checks,
+    iou_threshold: float = Config.cvat_config.iou_threshold,
     oks_sigma: float | None = None,
     point_size_base: str | None = None,
     match_empty_frames: bool | None = None,
@@ -823,7 +823,7 @@ def get_user_id(user_email: str) -> int:
         try:
             (invitation, _) = api_client.invitations_api.create(
                 models.InvitationWriteRequest(role="worker", email=user_email),
-                org=Config.cvat_config.cvat_org_slug,
+                org=Config.cvat_config.org_slug,
             )
         except exceptions.ApiException as e:
             logger.exception(f"Exception when calling get_user_id(): {e}\n")
@@ -839,7 +839,7 @@ def remove_user_from_org(user_id: int):
         try:
             (page, _) = api_client.users_api.list(
                 filter='{"==":[{"var":"id"},"%s"]}' % user_id,  # noqa: UP031
-                org=Config.cvat_config.cvat_org_slug,
+                org=Config.cvat_config.org_slug,
             )
             if not page.results:
                 return
@@ -849,7 +849,7 @@ def remove_user_from_org(user_id: int):
 
             (page, _) = api_client.memberships_api.list(
                 user=user.username,
-                org=Config.cvat_config.cvat_org_slug,
+                org=Config.cvat_config.org_slug,
             )
             if page.results:
                 api_client.memberships_api.destroy(page.results[0].id)

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

@@ -203,7 +203,7 @@ def _wait_task_creation(self, task_id: int) -> cvat_api.UploadStatus:
             if task_status not in [cvat_api.UploadStatus.STARTED, cvat_api.UploadStatus.QUEUED]:
                 return task_status
 
-            sleep(Config.cvat_config.cvat_task_creation_check_interval)
+            sleep(Config.cvat_config.task_creation_check_interval)
 
     def _setup_gt_job_for_cvat_task(
         self, task_id: int, gt_dataset: dm.Dataset, *, dm_export_format: str = "coco"
@@ -409,7 +409,7 @@ def build(self):
 
             for data_subset in self._split_dataset_per_task(
                 data_to_be_annotated,
-                subset_size=Config.cvat_config.cvat_max_jobs_per_task * segment_size,
+                subset_size=Config.cvat_config.max_jobs_per_task * segment_size,
             ):
                 cvat_task = cvat_api.create_task(
                     cvat_project.id, escrow_address, segment_size=segment_size
@@ -552,7 +552,7 @@ def build(self):
 
 class PolygonTaskBuilder(SimpleTaskBuilder):
     def _setup_quality_settings(self, task_id: int, **overrides) -> None:
-        values = {"iou_threshold": Config.cvat_config.cvat_polygons_iou_threshold, **overrides}
+        values = {"iou_threshold": Config.cvat_config.iou_threshold, **overrides}
         super()._setup_quality_settings(task_id, **values)
 
 
@@ -1554,7 +1554,7 @@ def _create_on_cvat(self):
 
             for data_subset in self._split_dataset_per_task(
                 self._roi_filenames_to_be_annotated,
-                subset_size=Config.cvat_config.cvat_max_jobs_per_task * segment_size,
+                subset_size=Config.cvat_config.max_jobs_per_task * segment_size,
             ):
                 cvat_task = cvat_api.create_task(
                     cvat_project.id, self.escrow_address, segment_size=segment_size
@@ -2354,7 +2354,7 @@ def _prepare_task_params(self):
                         label_id=label_id, roi_ids=task_data_roi_ids, roi_gt_ids=label_gt_roi_ids
                     )
                     for task_data_roi_ids in take_by(
-                        label_data_roi_ids, Config.cvat_config.cvat_max_jobs_per_task * segment_size
+                        label_data_roi_ids, Config.cvat_config.max_jobs_per_task * segment_size
                     )
                 ]
             )
@@ -2603,7 +2603,7 @@ def _save_cvat_gt_dataset_to_oracle_bucket(
 
     def _setup_quality_settings(self, task_id: int, **overrides) -> None:
         values = {
-            "oks_sigma": Config.cvat_config.cvat_oks_sigma,
+            "oks_sigma": Config.cvat_config.oks_sigma,
             "point_size_base": "image_size",  # we don't expect any boxes or groups, so ignore them
         }
         values.update(overrides)
@@ -2767,7 +2767,7 @@ def _task_params_label_key(ts):
                             cvat_task.id, gt_point_dataset, dm_export_format="cvat"
                         )
                         self._setup_quality_settings(
-                            cvat_task.id, oks_sigma=Config.cvat_config.cvat_oks_sigma
+                            cvat_task.id, oks_sigma=Config.cvat_config.oks_sigma
                         )
 
                         db_service.create_data_upload(session, cvat_task.id)

packages/examples/cvat/exchange-oracle/src/handlers/job_creation.py

@@ -24,7 +24,7 @@ def compose_assignment_url(task_id: int, job_id: int, *, project: Project) -> st
     if project.job_type == TaskTypes.image_skeletons_from_boxes:
         query_params += "&defaultPointsCount=1"
 
-    return urljoin(Config.cvat_config.cvat_url, f"/tasks/{task_id}/jobs/{job_id}{query_params}")
+    return urljoin(Config.cvat_config.host_url, f"/tasks/{task_id}/jobs/{job_id}{query_params}")
 
 
 def get_default_assignment_timeout(task_type: TaskTypes) -> int:

packages/examples/cvat/exchange-oracle/src/utils/assignments.py

@@ -32,7 +32,7 @@ async def validate_cvat_signature(request: Request, x_signature_256: str):
     signature = (
         "sha256="
         + hmac.new(
-            Config.cvat_config.cvat_webhook_secret.encode("utf-8"),
+            Config.cvat_config.webhook_secret.encode("utf-8"),
             data,
             digestmod=sha256,
         ).hexdigest()

packages/examples/cvat/exchange-oracle/src/validators/signature.py

@@ -43,9 +43,9 @@ def tearDown(self):
         self.session.close()
 
     def test_process_incoming_job_launcher_webhooks_escrow_created_type(self):
-        webhok_id = str(uuid.uuid4())
+        webhook_id = str(uuid.uuid4())
         webhook = Webhook(
-            id=webhok_id,
+            id=webhook_id,
             signature="signature",
             escrow_address=escrow_address,
             chain_id=chain_id,
@@ -93,7 +93,7 @@ def test_process_incoming_job_launcher_webhooks_escrow_created_type(self):
             process_incoming_job_launcher_webhooks()
 
         updated_webhook = (
-            self.session.execute(select(Webhook).where(Webhook.id == webhok_id)).scalars().first()
+            self.session.execute(select(Webhook).where(Webhook.id == webhook_id)).scalars().first()
         )
 
         assert updated_webhook.status == OracleWebhookStatuses.completed.value

packages/examples/cvat/exchange-oracle/tests/integration/cron/test_process_job_launcher_webhooks.py

@@ -18,7 +18,7 @@ def generate_cvat_signature(data: dict):
     return (
         "sha256="
         + hmac.new(
-            CvatConfig.cvat_webhook_secret.encode("utf-8"),
+            CvatConfig.webhook_secret.encode("utf-8"),
             b_data,
             digestmod=sha256,
         ).hexdigest()