Consensus simple merging

.vscode/launch.json

@@ -125,6 +125,26 @@
             ],
             "justMyCode": false,
         },
+        {
+            "name": "REST API tests: Attach to RQ consensus worker",
+            "type": "debugpy",
+            "request": "attach",
+            "connect": {
+                "host": "127.0.0.1",
+                "port": 9096
+            },
+            "pathMappings": [
+                {
+                    "localRoot": "${workspaceFolder}",
+                    "remoteRoot": "/home/django/"
+                },
+                {
+                    "localRoot": "${workspaceFolder}/.env",
+                    "remoteRoot": "/opt/venv",
+                }
+            ],
+            "justMyCode": false,
+        },
         {
             "type": "pwa-chrome",
             "request": "launch",
@@ -383,6 +403,28 @@
             },
             "console": "internalConsole"
         },
+        {
+            "name": "server: RQ - consensus",
+            "type": "debugpy",
+            "request": "launch",
+            "stopOnEntry": false,
+            "justMyCode": false,
+            "python": "${command:python.interpreterPath}",
+            "program": "${workspaceRoot}/manage.py",
+            "args": [
+                "rqworker",
+                "consensus",
+                "--worker-class",
+                "cvat.rqworker.SimpleWorker"
+            ],
+            "django": true,
+            "cwd": "${workspaceFolder}",
+            "env": {
+                "DJANGO_LOG_SERVER_HOST": "localhost",
+                "DJANGO_LOG_SERVER_PORT": "8282"
+            },
+            "console": "internalConsole"
+        },
         {
             "name": "server: migrate",
             "type": "debugpy",
@@ -566,6 +608,7 @@
                 "server: RQ - analytics reports",
                 "server: RQ - cleaning",
                 "server: RQ - chunks",
+                "server: RQ - consensus",
             ]
         }
     ]

cvat-core/src/api-implementation.ts

@@ -33,6 +33,7 @@ import { ArgumentError } from './exceptions';
 import {
     AnalyticsReportFilter, QualityConflictsFilter, QualityReportsFilter,
     QualitySettingsFilter, SerializedAsset,
+    ConsensusSettingsFilter,
 } from './server-response-types';
 import QualityReport from './quality-report';
 import QualityConflict, { ConflictSeverity } from './quality-conflict';
@@ -46,6 +47,7 @@ import { convertDescriptions, getServerAPISchema } from './server-schema';
 import { JobType } from './enums';
 import { PaginatedResource } from './core-types';
 import CVATCore from '.';
+import ConsensusSettings from './consensus-settings';
 
 function implementationMixin(func: Function, implementation: Function): void {
     Object.assign(func, { implementation });
@@ -414,6 +416,20 @@ export default function implementAPI(cvat: CVATCore): CVATCore {
         return webhooks;
     });
 
+    implementationMixin(cvat.consensus.settings.get, async (filter: ConsensusSettingsFilter) => {
+        checkFilter(filter, {
+            taskID: isInteger,
+        });
+
+        const params = fieldsToSnakeCase(filter);
+
+        const settings = await serverProxy.consensus.settings.get(params);
+        const schema = await getServerAPISchema();
+        const descriptions = convertDescriptions(schema.components.schemas.ConsensusSettings.properties);
+
+        return new ConsensusSettings({ ...settings, descriptions });
+    });
+
     implementationMixin(cvat.analytics.quality.reports, async (filter: QualityReportsFilter) => {
         checkFilter(filter, {
             page: isInteger,

cvat-core/src/api.ts

@@ -378,6 +378,14 @@ function build(): CVATCore {
                 return result;
             },
         },
+        consensus: {
+            settings: {
+                async get(filter = {}) {
+                    const result = await PluginRegistry.apiWrapper(cvat.consensus.settings.get, filter);
+                    return result;
+                },
+            },
+        },
         analytics: {
             performance: {
                 async reports(filter = {}) {
@@ -482,6 +490,7 @@ function build(): CVATCore {
     cvat.cloudStorages = Object.freeze(cvat.cloudStorages);
     cvat.organizations = Object.freeze(cvat.organizations);
     cvat.webhooks = Object.freeze(cvat.webhooks);
+    cvat.consensus = Object.freeze(cvat.consensus);
     cvat.analytics = Object.freeze(cvat.analytics);
     cvat.classes = Object.freeze(cvat.classes);
     cvat.utils = Object.freeze(cvat.utils);

cvat-core/src/consensus-settings.ts

@@ -0,0 +1,87 @@
+// Copyright (C) 2024 CVAT.ai Corporation
+//
+// SPDX-License-Identifier: MIT
+
+import { SerializedConsensusSettingsData } from './server-response-types';
+import PluginRegistry from './plugins';
+import serverProxy from './server-proxy';
+import { convertDescriptions, getServerAPISchema } from './server-schema';
+
+export default class ConsensusSettings {
+    #id: number;
+    #task: number;
+    #iouThreshold: number;
+    #quorum: number;
+    #descriptions: Record<string, string>;
+
+    constructor(initialData: SerializedConsensusSettingsData) {
+        this.#id = initialData.id;
+        this.#task = initialData.task;
+        this.#iouThreshold = initialData.iou_threshold;
+        this.#quorum = initialData.quorum;
+        this.#descriptions = initialData.descriptions;
+    }
+
+    get id(): number {
+        return this.#id;
+    }
+
+    get task(): number {
+        return this.#task;
+    }
+
+    get iouThreshold(): number {
+        return this.#iouThreshold;
+    }
+
+    set iouThreshold(newVal: number) {
+        this.#iouThreshold = newVal;
+    }
+
+    get quorum(): number {
+        return this.#quorum;
+    }
+
+    set quorum(newVal: number) {
+        this.#quorum = newVal;
+    }
+
+    get descriptions(): Record<string, string> {
+        const descriptions: Record<string, string> = Object.keys(this.#descriptions).reduce((acc, key) => {
+            const camelCaseKey = _.camelCase(key);
+            acc[camelCaseKey] = this.#descriptions[key];
+            return acc;
+        }, {});
+
+        return descriptions;
+    }
+
+    public toJSON(): SerializedConsensusSettingsData {
+        const result: SerializedConsensusSettingsData = {
+            iou_threshold: this.#iouThreshold,
+            quorum: this.#quorum,
+        };
+
+        return result;
+    }
+
+    public async save(): Promise<ConsensusSettings> {
+        const result = await PluginRegistry.apiWrapper.call(this, ConsensusSettings.prototype.save);
+        return result;
+    }
+}
+
+Object.defineProperties(ConsensusSettings.prototype.save, {
+    implementation: {
+        writable: false,
+        enumerable: false,
+        value: async function implementation(): Promise<ConsensusSettings> {
+            const result = await serverProxy.consensus.settings.update(
+                this.id, this.toJSON(),
+            );
+            const schema = await getServerAPISchema();
+            const descriptions = convertDescriptions(schema.components.schemas.ConsensusSettings.properties);
+            return new ConsensusSettings({ ...result, descriptions });
+        },
+    },
+});

cvat-core/src/index.ts

@@ -4,6 +4,7 @@
 
 import {
     AnalyticsReportFilter, QualityConflictsFilter, QualityReportsFilter, QualitySettingsFilter,
+    ConsensusSettingsFilter,
 } from './server-response-types';
 import PluginRegistry from './plugins';
 import serverProxy from './server-proxy';
@@ -30,6 +31,7 @@ import Webhook from './webhook';
 import QualityReport from './quality-report';
 import QualityConflict from './quality-conflict';
 import QualitySettings from './quality-settings';
+import ConsensusSettings from './consensus-settings';
 import AnalyticsReport from './analytics-report';
 import AnnotationGuide from './guide';
 import { JobValidationLayout, TaskValidationLayout } from './validation-layout';
@@ -139,6 +141,11 @@ export default interface CVATCore {
     webhooks: {
         get: any;
     };
+    consensus: {
+        settings: {
+            get: (filter: ConsensusSettingsFilter) => Promise<ConsensusSettings>;
+        };
+    }
     analytics: {
         quality: {
             reports: (filter: QualityReportsFilter) => Promise<PaginatedResource<QualityReport>>;

cvat-core/src/server-proxy.ts

@@ -19,6 +19,7 @@
     SerializedInvitationData, SerializedCloudStorage, SerializedFramesMetaData, SerializedCollection,
     SerializedQualitySettingsData, APIQualitySettingsFilter, SerializedQualityConflictData, APIQualityConflictsFilter,
     SerializedQualityReportData, APIQualityReportsFilter, SerializedAnalyticsReport, APIAnalyticsReportFilter,
+    SerializedConsensusSettingsData, APIConsensusSettingsFilter,
     SerializedRequest, SerializedJobValidationLayout, SerializedTaskValidationLayout,
 } from './server-response-types';
 import { PaginatedResource, UpdateStatusData } from './core-types';
@@ -244,7 +245,7 @@
     return new ServerError(message, 0);
 }
 
 function prepareData(details) {
     const data = new FormData();
     for (const [key, value] of Object.entries(details)) {
         if (Array.isArray(value)) {
@@ -286,7 +287,7 @@
             return requestId++;
         }
 
         async function get(url: string, requestConfig) {
             return new Promise((resolve, reject) => {
                 const newRequestId = getRequestId();
                 requests[newRequestId] = { resolve, reject };
@@ -767,6 +768,41 @@
     }
 }
 
+async function mergeConsensusJobs(id: number, instanceType: string): Promise<void> {
+    const { backendAPI } = config;
+    const url = `${backendAPI}/consensus/merges`;
+    const params = {
+        rq_id: null,
+    };
+    const requestBody = {
+        task_id: undefined,
+        job_id: undefined,
+    };
+
+    if (instanceType === 'task') requestBody.task_id = id;
+    else requestBody.job_id = id;
+
+    return new Promise<void>((resolve, reject) => {
+        async function request() {
+            try {
+                const response = await Axios.post(url, requestBody, { params });
+                params.rq_id = response.data.rq_id;
+                const { status } = response;
+                if (status === 202) {
+                    setTimeout(request, 3000);
+                } else if (status === 201) {
+                    resolve();
+                } else {
+                    reject(generateError(response));
+                }
+            } catch (errorData) {
+                reject(generateError(errorData));
+            }
+        }
+        setTimeout(request);
+    });
+}
+
 async function getLabels(filter: {
     job_id?: number,
     task_id?: number,
@@ -2182,6 +2218,42 @@
     }
 }
 
+async function getConsensusSettings(
+    filter: APIConsensusSettingsFilter,
+): Promise<SerializedConsensusSettingsData> {
+    const { backendAPI } = config;
+
+    try {
+        const response = await Axios.get(`${backendAPI}/consensus/settings`, {
+            params: {
+                ...filter,
+            },
+        });
+
+        return response.data.results[0];
+    } catch (errorData) {
+        throw generateError(errorData);
+    }
+}
+
+async function updateConsensusSettings(
+    settingsID: number,
+    settingsData: SerializedConsensusSettingsData,
+): Promise<SerializedConsensusSettingsData> {
+    const params = enableOrganization();
+    const { backendAPI } = config;
+
+    try {
+        const response = await Axios.patch(`${backendAPI}/consensus/settings/${settingsID}`, settingsData, {
+            params,
+        });
+
+        return response.data;
+    } catch (errorData) {
+        throw generateError(errorData);
+    }
+}
+
 async function getQualityConflicts(
     filter: APIQualityConflictsFilter,
 ): Promise<SerializedQualityConflictData[]> {
@@ -2411,6 +2483,7 @@
         backup: backupTask,
         restore: restoreTask,
         validationLayout: validationLayout('tasks'),
+        mergeConsensusJobs,
     }),
 
     labels: Object.freeze({
@@ -2427,6 +2500,7 @@
         delete: deleteJob,
         exportDataset: exportDataset('jobs'),
         validationLayout: validationLayout('jobs'),
+        mergeConsensusJobs,
     }),
 
     users: Object.freeze({
@@ -2531,6 +2605,13 @@
         }),
     }),
 
+    consensus: Object.freeze({
+        settings: Object.freeze({
+            get: getConsensusSettings,
+            update: updateConsensusSettings,
+        }),
+    }),
+
     requests: Object.freeze({
         list: getRequestsList,
         status: getRequestStatus,

cvat-core/src/server-response-types.ts

@@ -240,8 +240,15 @@ export interface SerializedOrganization {
 export interface APIQualitySettingsFilter extends APICommonFilterParams {
     task_id?: number;
 }
+
 export type QualitySettingsFilter = Camelized<APIQualitySettingsFilter>;
 
+export interface APIConsensusSettingsFilter extends APICommonFilterParams {
+    task_id?: number;
+}
+
+export type ConsensusSettingsFilter = Camelized<APIConsensusSettingsFilter>;
+
 export interface SerializedQualitySettingsData {
     id?: number;
     task?: number;
@@ -330,6 +337,14 @@ export interface SerializedQualityReportData {
     };
 }
 
+export interface SerializedConsensusSettingsData {
+    id?: number;
+    task?: number;
+    quorum?: number;
+    iou_threshold?: number;
+    descriptions?: Record<string, string>;
+}
+
 export interface SerializedDataEntry {
     date?: string;
     value?: number | Record<string, number>