[GSoC'24] Merge consensus jobs

.vscode/launch.json

@@ -360,6 +360,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",
@@ -537,7 +559,8 @@
                 "server: RQ - scheduler",
                 "server: RQ - quality reports",
                 "server: RQ - analytics reports",
-                "server: RQ - cleaning"
+                "server: RQ - cleaning",
+                "server: RQ - consensus",
             ]
         }
     ]

cvat-core/src/api-implementation.ts

@@ -31,7 +31,7 @@ import Webhook from './webhook';
 import { ArgumentError } from './exceptions';
 import {
     AnalyticsReportFilter, QualityConflictsFilter, QualityReportsFilter,
-    QualitySettingsFilter, SerializedAsset,
+    SettingsFilter, SerializedAsset,
 } from './server-response-types';
 import QualityReport from './quality-report';
 import QualityConflict, { ConflictSeverity } from './quality-conflict';
@@ -42,6 +42,7 @@ import { listActions, registerAction, runActions } from './annotations-actions';
 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 });
@@ -505,7 +506,7 @@ export default function implementAPI(cvat: CVATCore): CVATCore {
 
         return mergedConflicts;
     });
-    implementationMixin(cvat.analytics.quality.settings.get, async (filter: QualitySettingsFilter) => {
+    implementationMixin(cvat.analytics.quality.settings.get, async (filter: SettingsFilter) => {
         checkFilter(filter, {
             taskID: isInteger,
         });
@@ -515,6 +516,16 @@ export default function implementAPI(cvat: CVATCore): CVATCore {
         const settings = await serverProxy.analytics.quality.settings.get(params);
         return new QualitySettings({ ...settings });
     });
+    implementationMixin(cvat.consensus.settings.get, async (filter: SettingsFilter) => {
+        checkFilter(filter, {
+            taskID: isInteger,
+        });
+
+        const params = fieldsToSnakeCase(filter);
+
+        const settings = await serverProxy.consensus.settings.get(params);
+        return new ConsensusSettings({ ...settings });
+    });
     implementationMixin(cvat.analytics.performance.reports, async (filter: AnalyticsReportFilter) => {
         checkFilter(filter, {
             jobID: isInteger,

cvat-core/src/api.ts

@@ -374,6 +374,14 @@ function build(): CVATCore {
                 },
             },
         },
+        consensus: {
+            settings: {
+                async get(filter = {}) {
+                    const result = await PluginRegistry.apiWrapper(cvat.consensus.settings.get, filter);
+                    return result;
+                },
+            },
+        },
         classes: {
             User,
             Project: implementProject(Project),
@@ -418,6 +426,7 @@ function build(): CVATCore {
     cvat.organizations = Object.freeze(cvat.organizations);
     cvat.webhooks = Object.freeze(cvat.webhooks);
     cvat.analytics = Object.freeze(cvat.analytics);
+    cvat.consensus = Object.freeze(cvat.consensus);
     cvat.classes = Object.freeze(cvat.classes);
     cvat.utils = Object.freeze(cvat.utils);
 

cvat-core/src/consensus-settings.ts

@@ -0,0 +1,81 @@
+// 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';
+
+export default class ConsensusSettings {
+    #id: number;
+    #task: number;
+    #iouThreshold: number;
+    #quorum: number;
+    #agreementScoreThreshold: number;
+
+    constructor(initialData: SerializedConsensusSettingsData) {
+        this.#id = initialData.id;
+        this.#task = initialData.task;
+        this.#iouThreshold = initialData.iou_threshold;
+        this.#agreementScoreThreshold = initialData.agreement_score_threshold;
+        this.#quorum = initialData.quorum;
+    }
+
+    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 agreementScoreThreshold(): number {
+        return this.#agreementScoreThreshold;
+    }
+
+    set agreementScoreThreshold(newVal: number) {
+        this.#agreementScoreThreshold = newVal;
+    }
+
+    public toJSON(): SerializedConsensusSettingsData {
+        const result: SerializedConsensusSettingsData = {
+            iou_threshold: this.#iouThreshold,
+            quorum: this.#quorum,
+            agreement_score_threshold: this.#agreementScoreThreshold,
+        };
+
+        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() {
+            const result = await serverProxy.consensus.settings.update(this.id, this.toJSON());
+            return new ConsensusSettings(result);
+        },
+    },
+});

cvat-core/src/index.ts

@@ -2,8 +2,9 @@
 //
 // SPDX-License-Identifier: MIT
 
+import ConsensusSettings from 'consensus-settings';
 import {
-    AnalyticsReportFilter, QualityConflictsFilter, QualityReportsFilter, QualitySettingsFilter,
+    AnalyticsReportFilter, QualityConflictsFilter, QualityReportsFilter, SettingsFilter,
 } from './server-response-types';
 import PluginRegistry from './plugins';
 import serverProxy from './server-proxy';
@@ -131,12 +132,17 @@ export default interface CVATCore {
     webhooks: {
         get: any;
     };
+    consensus: {
+        settings: {
+            get: (filter: SettingsFilter) => Promise<ConsensusSettings>;
+        };
+    }
     analytics: {
         quality: {
             reports: (filter: QualityReportsFilter) => Promise<PaginatedResource<QualityReport>>;
             conflicts: (filter: QualityConflictsFilter) => Promise<QualityConflict[]>;
             settings: {
-                get: (filter: QualitySettingsFilter) => Promise<QualitySettings>;
+                get: (filter: SettingsFilter) => Promise<QualitySettings>;
             };
         };
         performance: {

cvat-core/src/server-proxy.ts

@@ -16,8 +16,9 @@ import {
     SerializedAbout, SerializedRemoteFile, SerializedUserAgreement,
     SerializedRegister, JobsFilter, SerializedJob, SerializedGuide, SerializedAsset, SerializedAPISchema,
     SerializedInvitationData, SerializedCloudStorage, SerializedFramesMetaData, SerializedCollection,
-    SerializedQualitySettingsData, APIQualitySettingsFilter, SerializedQualityConflictData, APIQualityConflictsFilter,
+    SerializedQualitySettingsData, APISettingsFilter, SerializedQualityConflictData, APIQualityConflictsFilter,
     SerializedQualityReportData, APIQualityReportsFilter, SerializedAnalyticsReport, APIAnalyticsReportFilter,
+    SerializedConsensusSettingsData,
 } from './server-response-types';
 import { PaginatedResource } from './core-types';
 import { Storage } from './storage';
@@ -796,6 +797,30 @@ async function deleteTask(id: number, organizationID: string | null = null): Pro
     }
 }
 
+async function mergeConsensusJobs(id: number): Promise<void> {
+    const { backendAPI } = config;
+    const url = `${backendAPI}/tasks/${id}/aggregate`;
+
+    return new Promise<void>((resolve, reject) => {
+        async function request() {
+            try {
+                const response = await Axios.put(url);
+                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,
@@ -2334,7 +2359,7 @@ async function createAsset(file: File, guideId: number): Promise<SerializedAsset
 }
 
 async function getQualitySettings(
-    filter: APIQualitySettingsFilter,
+    filter: APISettingsFilter,
 ): Promise<SerializedQualitySettingsData> {
     const { backendAPI } = config;
 
@@ -2369,6 +2394,42 @@ async function updateQualitySettings(
     }
 }
 
+async function getConsensusSettings(
+    filter: APISettingsFilter,
+): 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[]> {
@@ -2548,6 +2609,7 @@ export default Object.freeze({
         getPreview: getPreview('tasks'),
         backup: backupTask,
         restore: restoreTask,
+        mergeConsensusJobs,
     }),
 
     labels: Object.freeze({
@@ -2666,4 +2728,11 @@ export default Object.freeze({
             }),
         }),
     }),
+
+    consensus: Object.freeze({
+        settings: Object.freeze({
+            get: getConsensusSettings,
+            update: updateConsensusSettings,
+        }),
+    }),
 });

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

@@ -235,10 +235,10 @@ export interface SerializedOrganization {
     contact?: SerializedOrganizationContact,
 }
 
-export interface APIQualitySettingsFilter extends APICommonFilterParams {
+export interface APISettingsFilter extends APICommonFilterParams {
     task_id?: number;
 }
-export type QualitySettingsFilter = Camelized<APIQualitySettingsFilter>;
+export type SettingsFilter = Camelized<APISettingsFilter>;
 
 export interface SerializedQualitySettingsData {
     id?: number;
@@ -257,6 +257,14 @@ export interface SerializedQualitySettingsData {
     compare_attributes?: boolean;
 }
 
+export interface SerializedConsensusSettingsData {
+    id?: number;
+    task?: number;
+    agreement_score_threshold?: number;
+    quorum?: number;
+    iou_threshold?: number;
+}
+
 export interface APIQualityConflictsFilter extends APICommonFilterParams {
     report_id?: number;
 }

cvat-core/src/session-implementation.ts

@@ -740,6 +740,14 @@ export function implementTask(Task: typeof TaskClass): typeof TaskClass {
         },
     });
 
+    Object.defineProperty(Task.prototype.mergeConsensusJobs, 'implementation', {
+        value: function mergeConsensusJobsImplementation(
+            this: TaskClass,
+        ): ReturnType<typeof TaskClass.prototype.mergeConsensusJobs> {
+            return serverProxy.tasks.mergeConsensusJobs(this.id);
+        },
+    });
+
     Object.defineProperty(Task.prototype.issues, 'implementation', {
         value: function issuesImplementation(
             this: TaskClass,

cvat-core/src/session.ts

@@ -1139,6 +1139,11 @@ export class Task extends Session {
         return result;
     }
 
+    async mergeConsensusJobs(): Promise<void> {
+        const result = await PluginRegistry.apiWrapper.call(this, Task.prototype.mergeConsensusJobs);
+        return result;
+    }
+
     async backup(targetStorage: Storage, useDefaultSettings: boolean, fileName?: string): Promise<string | void> {
         const result = await PluginRegistry.apiWrapper.call(
             this,