[CVAT] Adapt exchange/recording oracles for honeypots (#2720)

* Update exchange oracle

* Add wait timeout when importing GT annotations

* Extract a base class for task creation

* Move gt setup into the base class

* Add draft implementation for points task creation

* [draft] Update recording oracle

* Upgrade cvat-sdk dep

* [Exchnage oracle] apply some comments && small fixes

* Refactor some code, fix errors

* Add quality settings setup

* Fix update quality settings

* Use inbound bbox circle radius for point validation

* Fix linter errors

* Fix quality settings update call

* Move common function to the base class

* Expect point group annotations in points annotation task

* Fix response check

* Improve formatting in the log message

* Use single shape mode for image_points task

* Refactor recording oracle updates

* [Recording oracle] update deps

* Update assignment urls for skeleton tasks

* Rename quality parameter

* Fix linter error

* [Ex oracle] Move gt dataset preparation into separate method for skeletons task

* [Ex oracle] update deps

* [Ex oracle] Improve handling oracle mode(dev/prod/test)

* [Ex oracle] Fix test

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

* [Exchnage oracle] Pass job start/stop frame from Ex oracle to Rec oracle

* Update recording oracle

* t

* [Exchange oracle] Fix tests

* [Exchange oracle] Add migration

* Fix test

* [Exchange oracle] mark job start/stop frame as not nullable

* Fix tests

* [Recording oracle] Clean up the code

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

Co-authored-by: Maxim Zhiltsov <[email protected]>

* [Recording oracle] Apply comments

* [Exchnage oracle] use_bbox_size_for_points -> point_size_base

* [Rec oracle] Use MT19937 generator

* [Exchange oracle] Move BoxesFromPointsTaskBuilder::_prepare_gt_roi_dataset before _extract_and_upload_rois

* [Exchange oracle] Fix checking which files should be uploaded to the ex oracle bucket after escrow is annotated

* [Exchange oracle] Update down_revision

* fix typo

* [Exchange oracle] Include val_size into chunk_size

* Fix some errors

* Enable empty frame matching

* Use the added parameter

* [Recording orcale] Fix get_task_quality_report

* [Ex oracle] Fix missing segment_size

* [Rec oracle] Small fixes

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

* [Exchange oracle] Bump cvat-sdk version

* Fix roi GT dataset in boxes_from_points tasks

- fixed invalid image ids
- removed unnecessary image reading during dataset preparation

* Add more clever default for sort_images

* Fix linter error, remove gt image data callback

* Fix type annotation

* Fix quality settings for skeletons_from_boxes

* Fix GT datasets for points in skeletons_from_boxes

- removed invalid mappings in the class
- fixed incorrect gt point preparations
- refactored some code

* Simplify roi info id

* Remove unused field from skeleton roi info

* Allow optional joints in skeleton task manifest

* Add points task meta

* Basic fix for merged dataset annotations

Fixes how the merged datasets get annotations for all the task types. Probably, need to change it further to use the original GT, as excluded GT annotations will be missing from the result now.

* Basic fix for premature escrow validation requests

* Fix incorrect GT preparation in boxes_from_points

Some images might have incorrect ids in the generated COCO annotation file. Reindexing is not desirable, so clear the original image ids from roi images instead

* Refactor some code

* Fix linter problem

* Use the original GT for final annotation merging

* Fix dataset merging for the points task

* Update comment

* [Exchange Oracle] Move cvat timeout settings to cvat config, update .env template

* [Recording Oracle] Update .env template, add some variables

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

---------

Co-authored-by: Maxim Zhiltsov <[email protected]>

Author: Marishka17

packages/examples/cvat/exchange-oracle/alembic/versions/1730963700_job_start_stop_frame_0a91b6a5f7b6.py

@@ -0,0 +1,31 @@
+"""job_start_stop_frame
+
+Revision ID: 0a91b6a5f7b6
+Revises: 4fc740e8c6ff
+Create Date: 2024-11-07 08:15:00.780982
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0a91b6a5f7b6"
+down_revision = "4fc740e8c6ff"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("jobs", sa.Column("start_frame", sa.Integer(), nullable=False))
+    op.add_column("jobs", sa.Column("stop_frame", sa.Integer(), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("jobs", "stop_frame")
+    op.drop_column("jobs", "start_frame")
+    # ### end Alembic commands ###

packages/examples/cvat/exchange-oracle/poetry.lock

@@ -16,7 +16,7 @@ sqlalchemy-utils = "^0.41.1"
 alembic = "^1.11.1"
 httpx = "^0.24.1"
 pytest = "^7.2.2"
-cvat-sdk = "2.6.0"
+cvat-sdk = "^2.22.0"
 sqlalchemy = "^2.0.16"
 apscheduler = "^3.10.1"
 xmltodict = "^0.13.0"

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

@@ -4,7 +4,7 @@
 from src.core.config import Config
 
 if __name__ == "__main__":
-    is_dev = Config.environment == "development"
+    is_dev = Config.is_development_mode()
     Config.validate()
     register_in_kvstore()
 

packages/examples/cvat/exchange-oracle/run.py

@@ -69,6 +69,14 @@ CVAT_ADMIN_USER_ID=
 CVAT_INCOMING_WEBHOOKS_URL=
 CVAT_WEBHOOK_SECRET=
 CVAT_ORG_SLUG=
+CVAT_TASK_SEGMENT_SIZE=
+CVAT_MAX_JOBS_PER_TASK=
+CVAT_TASK_CREATION_CHECK_INTERVAL=
+CVAT_MAX_VALIDATION_CHECKS=
+CVAT_IOU_THRESHOLD=
+CVAT_OKS_SIGMA=
+CVAT_EXPORT_TIMEOUT=
+CVAT_IMPORT_TIMEOUT=
 
 # Storage Config (S3/GCS)
 
@@ -84,6 +92,7 @@ STORAGE_USE_SSL=
 # Features
 
 ENABLE_CUSTOM_CLOUD_HOST=
+REQUEST_LOGGING_ENABLED=
 
 # Core
 
@@ -105,6 +114,7 @@ DEFAULT_API_PAGE_SIZE=
 LOCALHOST_RECORDING_ORACLE_ADDRESS=
 LOCALHOST_RECORDING_ORACLE_URL=
 LOCALHOST_JOB_LAUNCHER_URL=
+LOCALHOST_REPUTATION_ORACLE_ADDRESS=
 LOCALHOST_REPUTATION_ORACLE_URL=
 
 # Encryption

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

@@ -32,6 +32,5 @@ async def startup_event():
     logger.info("Exchange Oracle is up and running!")
 
 
-is_test = Config.environment == "test"
-if not is_test:
+if not Config.is_test_mode():
     setup_cron_jobs(app)

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

@@ -67,6 +67,6 @@ def get_available_webhook_types(
             Config.localhost.recording_oracle_address or escrow.recording_oracle
         ).lower(): OracleWebhookTypes.recording_oracle,
         (
-            Config.localhost.reputation_oracle_url or escrow.reputation_oracle
+            Config.localhost.reputation_oracle_address or escrow.reputation_oracle
         ).lower(): OracleWebhookTypes.reputation_oracle,
     }

packages/examples/cvat/exchange-oracle/src/chain/escrow.py

@@ -8,9 +8,12 @@
 
 class JobMeta(BaseModel):
     job_id: int
+    task_id: int
     annotation_filename: Path
     annotator_wallet_address: str
     assignment_id: str
+    start_frame: int
+    stop_frame: int
 
 
 class AnnotationMeta(BaseModel):

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

@@ -4,7 +4,8 @@
 import inspect
 import os
 from collections.abc import Iterable
-from typing import ClassVar
+from enum import Enum
+from typing import ClassVar, Optional
 
 from attrs.converters import to_bool
 from dotenv import load_dotenv
@@ -104,6 +105,8 @@ class LocalhostConfig(_NetworkConfig):
 
     recording_oracle_address = os.environ.get("LOCALHOST_RECORDING_ORACLE_ADDRESS")
     recording_oracle_url = os.environ.get("LOCALHOST_RECORDING_ORACLE_URL")
+
+    reputation_oracle_address = os.environ.get("LOCALHOST_REPUTATION_ORACLE_ADDRESS")
     reputation_oracle_url = os.environ.get("LOCALHOST_REPUTATION_ORACLE_URL")
 
 
@@ -167,14 +170,30 @@ class CronConfig:
 
 
 class CvatConfig:
+    # TODO: remove cvat_ prefix
     cvat_url = os.environ.get("CVAT_URL", "http://localhost:8080")
     cvat_admin = os.environ.get("CVAT_ADMIN", "admin")
     cvat_admin_pass = os.environ.get("CVAT_ADMIN_PASS", "admin")
     cvat_org_slug = os.environ.get("CVAT_ORG_SLUG", "")
 
     cvat_job_overlap = int(os.environ.get("CVAT_JOB_OVERLAP", 0))
-    cvat_job_segment_size = int(os.environ.get("CVAT_JOB_SEGMENT_SIZE", 150))
+    cvat_task_segment_size = int(os.environ.get("CVAT_TASK_SEGMENT_SIZE", 150))
     cvat_default_image_quality = int(os.environ.get("CVAT_DEFAULT_IMAGE_QUALITY", 70))
+    cvat_max_jobs_per_task = int(os.environ.get("CVAT_MAX_JOBS_PER_TASK", 1000))
+    cvat_task_creation_check_interval = int(os.environ.get("CVAT_TASK_CREATION_CHECK_INTERVAL", 5))
+
+    cvat_export_timeout = int(os.environ.get("CVAT_EXPORT_TIMEOUT", 5 * 60))
+    "Timeout, in seconds, for annotations or dataset export waiting"
+
+    cvat_import_timeout = int(os.environ.get("CVAT_IMPORT_TIMEOUT", 60 * 60))
+    "Timeout, in seconds, for waiting on GT annotations import"
+
+    # quality control settings
+    cvat_max_validation_checks = int(os.environ.get("CVAT_MAX_VALIDATION_CHECKS", 3))
+    "Maximum number of attempts to run a validation check on a job after completing annotation"
+
+    cvat_iou_threshold = float(os.environ.get("CVAT_IOU_THRESHOLD", 0.8))
+    cvat_oks_sigma = float(os.environ.get("CVAT_OKS_SIGMA", 0.1))
 
     cvat_incoming_webhooks_url = os.environ.get("CVAT_INCOMING_WEBHOOKS_URL")
     cvat_webhook_secret = os.environ.get("CVAT_WEBHOOK_SECRET", "thisisasamplesecret")
@@ -220,9 +239,6 @@ class FeaturesConfig:
     enable_custom_cloud_host = to_bool(os.environ.get("ENABLE_CUSTOM_CLOUD_HOST", "no"))
     "Allows using a custom host in manifest bucket urls"
 
-    default_export_timeout = int(os.environ.get("DEFAULT_EXPORT_TIMEOUT", 60))
-    "Timeout, in seconds, for annotations or dataset export waiting"
-
     request_logging_enabled = to_bool(os.getenv("REQUEST_LOGGING_ENABLED", "0"))
     "Allow to log request details for each request"
 
@@ -282,10 +298,25 @@ def validate(cls) -> None:
                 raise Exception(" ".join([ex_prefix, str(ex)]))
 
 
+class Environment(str, Enum):
+    PRODUCTION = "production"
+    DEVELOPMENT = "development"
+    TEST = "test"
+
+    @classmethod
+    def _missing_(cls, value: str) -> Optional["Environment"]:
+        value = value.lower()
+        for member in cls:
+            if member.value == value:
+                return member
+
+        return None
+
+
 class Config:
     debug = to_bool(os.environ.get("DEBUG", "false"))
     port = int(os.environ.get("PORT", 8000))
-    environment = os.environ.get("ENVIRONMENT", "development")
+    environment = Environment(os.environ.get("ENVIRONMENT", Environment.DEVELOPMENT.value))
     workers_amount = int(os.environ.get("WORKERS_AMOUNT", 1))
     webhook_max_retries = int(os.environ.get("WEBHOOK_MAX_RETRIES", 5))
     webhook_delay_if_failed = int(os.environ.get("WEBHOOK_DELAY_IF_FAILED", 60))
@@ -307,6 +338,21 @@ class Config:
     core_config = CoreConfig
     encryption_config = EncryptionConfig
 
+    @classmethod
+    def is_development_mode(cls) -> bool:
+        """Returns whether the oracle is running in development mode or not"""
+        return cls.environment == Environment.DEVELOPMENT
+
+    @classmethod
+    def is_test_mode(cls) -> bool:
+        """Returns whether the oracle is running in testing mode or not"""
+        return cls.environment == Environment.TEST
+
+    @classmethod
+    def is_production_mode(cls) -> bool:
+        """Returns whether the oracle is running in production mode or not"""
+        return cls.environment == Environment.PRODUCTION
+
     @classmethod
     def validate(cls) -> None:
         for attr_or_method in cls.__dict__:

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

@@ -105,14 +105,15 @@ def validate_type(cls, values: dict[str, Any]) -> dict[str, Any]:
             existing_names.add(node_name.lower())
 
         nodes_count = len(values["nodes"])
-        joints = values["joints"]
-        for joint_idx, joint in enumerate(joints):
-            for v in joint:
-                if not (0 <= v < nodes_count):
-                    raise ValueError(
-                        f"Skeleton '{skeleton_name}' joint #{joint_idx}: invalid value. "
-                        f"Expected a number in the range [0; {nodes_count - 1}]"
-                    )
+        joints = values.get("joints")
+        if joints is not None:
+            for joint_idx, joint in enumerate(joints):
+                for v in joint:
+                    if not (0 <= v < nodes_count):
+                        raise ValueError(
+                            f"Skeleton '{skeleton_name}' joint #{joint_idx}: invalid value. "
+                            f"Expected a number in the range [0; {nodes_count - 1}]"
+                        )
 
         return values