[CVAT] Adapt exchange/recording oracles for honeypots

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/pyproject.toml

@@ -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/run.py

@@ -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/src/.env.template

@@ -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/__init__.py

@@ -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/chain/escrow.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/core/annotation_meta.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/config.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/manifest.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
 

packages/examples/cvat/exchange-oracle/src/core/tasks/points.py

@@ -0,0 +1,31 @@
+import os
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import datumaro as dm
+
+# These details are relevant for image_points tasks
+
+
+class TaskMetaLayout:
+    GT_FILENAME = "gt.json"
+
+
+class TaskMetaSerializer:
+    GT_DATASET_FORMAT = "datumaro"
+
+    def serialize_gt_annotations(self, gt_dataset: dm.Dataset) -> bytes:
+        with TemporaryDirectory() as temp_dir:
+            gt_dataset_dir = os.path.join(temp_dir, "gt_dataset")
+            gt_dataset.export(gt_dataset_dir, self.GT_DATASET_FORMAT)
+            return (Path(gt_dataset_dir) / "annotations" / "default.json").read_bytes()
+
+    def parse_gt_annotations(self, gt_dataset_data: bytes) -> dm.Dataset:
+        with TemporaryDirectory() as temp_dir:
+            annotations_filename = os.path.join(temp_dir, "default.json")
+            with open(annotations_filename, "wb") as f:
+                f.write(gt_dataset_data)
+
+            dataset = dm.Dataset.import_from(temp_dir, format=self.GT_DATASET_FORMAT)
+            dataset.init_cache()
+            return dataset

packages/examples/cvat/exchange-oracle/src/core/tasks/simple.py

@@ -4,7 +4,7 @@
 
 import datumaro as dm
 
-# These details are relevant for image_points and image_boxes tasks
+# These details are relevant for image_boxes and image_polygons tasks
 
 
 class TaskMetaLayout:

packages/examples/cvat/exchange-oracle/src/crons/cvat/state_trackers.py

@@ -186,6 +186,8 @@ def track_task_creation(logger: logging.Logger, session: Session) -> None:
                         upload.task_id,
                         upload.task.cvat_project_id,
                         status=JobStatuses(cvat_job.state),
+                        start_frame=cvat_job.start_frame,
+                        stop_frame=cvat_job.stop_frame,
                     )
 
                 completed.append(upload)