954 lines
17 KiB
JavaScript
954 lines
17 KiB
JavaScript
const path = require('path');
|
||
const { spawn } = require('child_process');
|
||
const { randomUUID } = require('crypto');
|
||
|
||
const db = require('../db/knex');
|
||
|
||
const PROJECT_ROOT = path.resolve(
|
||
__dirname,
|
||
'..',
|
||
);
|
||
|
||
const DEFAULT_DATABASE_PATH = path.join(
|
||
PROJECT_ROOT,
|
||
'api_v1',
|
||
'database.sqlite',
|
||
);
|
||
|
||
const DEFAULT_PYTHON_PATH = path.join(
|
||
PROJECT_ROOT,
|
||
'.venv-clustering',
|
||
'bin',
|
||
'python',
|
||
);
|
||
|
||
const HDBSCAN_SCRIPT_PATH = path.join(
|
||
PROJECT_ROOT,
|
||
'tools',
|
||
'clustering',
|
||
'face_hdbscan.py',
|
||
);
|
||
|
||
|
||
function parsePositiveInteger(
|
||
value,
|
||
optionName,
|
||
) {
|
||
const number = Number(value);
|
||
|
||
if (
|
||
!Number.isInteger(number)
|
||
|| number <= 0
|
||
) {
|
||
throw new Error(
|
||
`${optionName} deve essere un intero positivo`,
|
||
);
|
||
}
|
||
|
||
return number;
|
||
}
|
||
|
||
|
||
function parseArguments(argv) {
|
||
const options = {
|
||
user: '',
|
||
databasePath: DEFAULT_DATABASE_PATH,
|
||
pythonPath: DEFAULT_PYTHON_PATH,
|
||
minClusterSize: 2,
|
||
minSamples: 2,
|
||
metric: 'euclidean',
|
||
selectionMethod: 'leaf',
|
||
scope: 'all',
|
||
persist: false,
|
||
};
|
||
|
||
for (
|
||
let index = 0;
|
||
index < argv.length;
|
||
index += 1
|
||
) {
|
||
const argument = argv[index];
|
||
const value = argv[index + 1];
|
||
|
||
if (argument === '--user') {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo --user',
|
||
);
|
||
}
|
||
|
||
options.user = String(value).trim();
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--database') {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo --database',
|
||
);
|
||
}
|
||
|
||
options.databasePath = path.resolve(value);
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--python') {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo --python',
|
||
);
|
||
}
|
||
|
||
options.pythonPath = path.resolve(value);
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (
|
||
argument === '--min-cluster-size'
|
||
) {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo '
|
||
+ '--min-cluster-size',
|
||
);
|
||
}
|
||
|
||
options.minClusterSize =
|
||
parsePositiveInteger(
|
||
value,
|
||
'--min-cluster-size',
|
||
);
|
||
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--min-samples') {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo '
|
||
+ '--min-samples',
|
||
);
|
||
}
|
||
|
||
options.minSamples =
|
||
parsePositiveInteger(
|
||
value,
|
||
'--min-samples',
|
||
);
|
||
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--metric') {
|
||
if (!value) {
|
||
throw new Error(
|
||
'Valore mancante dopo --metric',
|
||
);
|
||
}
|
||
|
||
options.metric = String(value).trim();
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (
|
||
argument === '--selection-method'
|
||
) {
|
||
if (
|
||
value !== 'eom'
|
||
&& value !== 'leaf'
|
||
) {
|
||
throw new Error(
|
||
'--selection-method deve essere '
|
||
+ 'eom oppure leaf',
|
||
);
|
||
}
|
||
|
||
options.selectionMethod = value;
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--scope') {
|
||
if (
|
||
value !== 'all'
|
||
&& value !== 'unclustered'
|
||
) {
|
||
throw new Error(
|
||
'--scope deve essere '
|
||
+ 'all oppure unclustered',
|
||
);
|
||
}
|
||
|
||
options.scope = value;
|
||
index += 1;
|
||
continue;
|
||
}
|
||
|
||
if (argument === '--persist') {
|
||
options.persist = true;
|
||
continue;
|
||
}
|
||
|
||
if (
|
||
argument === '--help'
|
||
|| argument === '-h'
|
||
) {
|
||
options.help = true;
|
||
continue;
|
||
}
|
||
|
||
throw new Error(
|
||
`Argomento sconosciuto: ${argument}`,
|
||
);
|
||
}
|
||
|
||
if (!options.help && !options.user) {
|
||
throw new Error(
|
||
'--user è obbligatorio',
|
||
);
|
||
}
|
||
|
||
if (options.minClusterSize < 2) {
|
||
throw new Error(
|
||
'--min-cluster-size deve essere '
|
||
+ 'almeno 2',
|
||
);
|
||
}
|
||
|
||
if (
|
||
options.persist
|
||
&& options.scope !== 'all'
|
||
) {
|
||
throw new Error(
|
||
'La persistenza iniziale supporta '
|
||
+ 'soltanto --scope all',
|
||
);
|
||
}
|
||
|
||
return options;
|
||
}
|
||
|
||
|
||
function printHelp() {
|
||
console.log(`
|
||
Uso:
|
||
|
||
node tools/runFaceClustering.js \\
|
||
--user Fabio \\
|
||
[--min-cluster-size 2] \\
|
||
[--min-samples 2] \\
|
||
[--selection-method leaf] \\
|
||
[--metric euclidean] \\
|
||
[--scope all] \\
|
||
[--persist]
|
||
|
||
Comportamento:
|
||
|
||
Senza --persist:
|
||
esegue HDBSCAN e mostra il riepilogo,
|
||
senza modificare il database.
|
||
|
||
Con --persist:
|
||
salva run, cluster e membership
|
||
in una singola transazione SQLite.
|
||
`.trim());
|
||
}
|
||
|
||
|
||
function runPythonClustering(options) {
|
||
return new Promise((resolve, reject) => {
|
||
const argumentsList = [
|
||
HDBSCAN_SCRIPT_PATH,
|
||
'--database',
|
||
options.databasePath,
|
||
'--user',
|
||
options.user,
|
||
'--min-cluster-size',
|
||
String(options.minClusterSize),
|
||
'--min-samples',
|
||
String(options.minSamples),
|
||
'--metric',
|
||
options.metric,
|
||
'--selection-method',
|
||
options.selectionMethod,
|
||
'--scope',
|
||
options.scope,
|
||
];
|
||
|
||
const child = spawn(
|
||
options.pythonPath,
|
||
argumentsList,
|
||
{
|
||
cwd: PROJECT_ROOT,
|
||
stdio: [
|
||
'ignore',
|
||
'pipe',
|
||
'pipe',
|
||
],
|
||
},
|
||
);
|
||
|
||
let stdout = '';
|
||
let stderr = '';
|
||
|
||
child.stdout.setEncoding('utf8');
|
||
child.stderr.setEncoding('utf8');
|
||
|
||
child.stdout.on('data', (chunk) => {
|
||
stdout += chunk;
|
||
});
|
||
|
||
child.stderr.on('data', (chunk) => {
|
||
stderr += chunk;
|
||
});
|
||
|
||
child.on('error', (error) => {
|
||
reject(
|
||
new Error(
|
||
'Impossibile avviare Python: '
|
||
+ error.message,
|
||
),
|
||
);
|
||
});
|
||
|
||
child.on('close', (code, signal) => {
|
||
if (code !== 0) {
|
||
reject(
|
||
new Error(
|
||
'Clustering Python fallito'
|
||
+ ` (code=${code}, signal=${signal}):\n`
|
||
+ stderr.trim(),
|
||
),
|
||
);
|
||
|
||
return;
|
||
}
|
||
|
||
let result;
|
||
|
||
try {
|
||
result = JSON.parse(stdout);
|
||
} catch (error) {
|
||
reject(
|
||
new Error(
|
||
'Output JSON Python non valido: '
|
||
+ error.message
|
||
+ '\nOutput:\n'
|
||
+ stdout.slice(0, 2000),
|
||
),
|
||
);
|
||
|
||
return;
|
||
}
|
||
|
||
resolve(result);
|
||
});
|
||
});
|
||
}
|
||
|
||
|
||
function collectAssignments(result) {
|
||
const assignments = [];
|
||
|
||
for (
|
||
const cluster
|
||
of result.clusters || []
|
||
) {
|
||
for (
|
||
const member
|
||
of cluster.members || []
|
||
) {
|
||
assignments.push({
|
||
faceId: member.id,
|
||
label: Number(cluster.label),
|
||
probability:
|
||
Number(member.probability),
|
||
noiseFlag: 0,
|
||
});
|
||
}
|
||
}
|
||
|
||
for (
|
||
const member
|
||
of result.noise || []
|
||
) {
|
||
assignments.push({
|
||
faceId: member.id,
|
||
label: -1,
|
||
probability:
|
||
Number(member.probability || 0),
|
||
noiseFlag: 1,
|
||
});
|
||
}
|
||
|
||
return assignments;
|
||
}
|
||
|
||
|
||
function validateResult(
|
||
result,
|
||
options,
|
||
) {
|
||
if (
|
||
!result
|
||
|| typeof result !== 'object'
|
||
) {
|
||
throw new Error(
|
||
'Risultato clustering mancante',
|
||
);
|
||
}
|
||
|
||
const parameters =
|
||
result.parameters || {};
|
||
|
||
const summary =
|
||
result.summary || {};
|
||
|
||
if (parameters.user !== options.user) {
|
||
throw new Error(
|
||
'Utente risultato non coerente: '
|
||
+ `${parameters.user}`,
|
||
);
|
||
}
|
||
|
||
if (
|
||
Number(parameters.min_cluster_size)
|
||
!== options.minClusterSize
|
||
) {
|
||
throw new Error(
|
||
'min_cluster_size non coerente',
|
||
);
|
||
}
|
||
|
||
if (
|
||
Number(parameters.min_samples)
|
||
!== options.minSamples
|
||
) {
|
||
throw new Error(
|
||
'min_samples non coerente',
|
||
);
|
||
}
|
||
|
||
if (
|
||
parameters.cluster_selection_method
|
||
!== options.selectionMethod
|
||
) {
|
||
throw new Error(
|
||
'selection_method non coerente',
|
||
);
|
||
}
|
||
|
||
const totalFaces =
|
||
Number(summary.total_faces);
|
||
|
||
const clusteredFaces =
|
||
Number(summary.clustered_faces);
|
||
|
||
const noiseFaces =
|
||
Number(summary.noise_faces);
|
||
|
||
const clusterCount =
|
||
Number(summary.cluster_count);
|
||
|
||
if (
|
||
!Number.isInteger(totalFaces)
|
||
|| totalFaces < 2
|
||
) {
|
||
throw new Error(
|
||
'Numero totale volti non valido',
|
||
);
|
||
}
|
||
|
||
if (
|
||
clusteredFaces + noiseFaces
|
||
!== totalFaces
|
||
) {
|
||
throw new Error(
|
||
'Riepilogo incoerente: '
|
||
+ 'clustered + noise != total',
|
||
);
|
||
}
|
||
|
||
if (
|
||
!Array.isArray(result.clusters)
|
||
|| result.clusters.length
|
||
!== clusterCount
|
||
) {
|
||
throw new Error(
|
||
'Numero cluster non coerente',
|
||
);
|
||
}
|
||
|
||
const assignments =
|
||
collectAssignments(result);
|
||
|
||
if (
|
||
assignments.length !== totalFaces
|
||
) {
|
||
throw new Error(
|
||
'Numero membership non coerente',
|
||
);
|
||
}
|
||
|
||
const faceIds = new Set();
|
||
|
||
for (const assignment of assignments) {
|
||
if (!assignment.faceId) {
|
||
throw new Error(
|
||
'Membership senza face_id',
|
||
);
|
||
}
|
||
|
||
if (faceIds.has(assignment.faceId)) {
|
||
throw new Error(
|
||
'Volto duplicato nel risultato: '
|
||
+ assignment.faceId,
|
||
);
|
||
}
|
||
|
||
faceIds.add(assignment.faceId);
|
||
|
||
if (
|
||
!Number.isFinite(
|
||
assignment.probability,
|
||
)
|
||
|| assignment.probability < 0
|
||
|| assignment.probability > 1
|
||
) {
|
||
throw new Error(
|
||
'Probabilità non valida per '
|
||
+ assignment.faceId,
|
||
);
|
||
}
|
||
|
||
if (
|
||
assignment.noiseFlag === 1
|
||
&& assignment.label !== -1
|
||
) {
|
||
throw new Error(
|
||
'Noise con label differente da -1',
|
||
);
|
||
}
|
||
|
||
if (
|
||
assignment.noiseFlag === 0
|
||
&& assignment.label < 0
|
||
) {
|
||
throw new Error(
|
||
'Cluster con label negativa',
|
||
);
|
||
}
|
||
}
|
||
|
||
for (const cluster of result.clusters) {
|
||
const members =
|
||
cluster.members || [];
|
||
|
||
if (
|
||
Number(cluster.size)
|
||
!== members.length
|
||
) {
|
||
throw new Error(
|
||
'Dimensione cluster incoerente: '
|
||
+ cluster.label,
|
||
);
|
||
}
|
||
}
|
||
|
||
return {
|
||
assignments,
|
||
totalFaces,
|
||
clusteredFaces,
|
||
noiseFaces,
|
||
clusterCount,
|
||
};
|
||
}
|
||
|
||
|
||
async function validateDatabaseFaces(
|
||
trx,
|
||
options,
|
||
assignments,
|
||
) {
|
||
const faceIds = assignments.map(
|
||
(assignment) => assignment.faceId,
|
||
);
|
||
|
||
const rows = await trx('faces as f')
|
||
.join(
|
||
'photos as p',
|
||
'p.id',
|
||
'f.photo_id',
|
||
)
|
||
.where(
|
||
'p.user',
|
||
options.user,
|
||
)
|
||
.whereIn(
|
||
'f.id',
|
||
faceIds,
|
||
)
|
||
.select(
|
||
'f.id',
|
||
'f.embedding',
|
||
);
|
||
|
||
if (rows.length !== faceIds.length) {
|
||
const foundIds = new Set(
|
||
rows.map((row) => row.id),
|
||
);
|
||
|
||
const missingIds = faceIds.filter(
|
||
(faceId) => !foundIds.has(faceId),
|
||
);
|
||
|
||
throw new Error(
|
||
'Volti mancanti o appartenenti '
|
||
+ 'a un altro utente: '
|
||
+ missingIds.join(', '),
|
||
);
|
||
}
|
||
|
||
for (const row of rows) {
|
||
if (
|
||
!Buffer.isBuffer(row.embedding)
|
||
|| row.embedding.length !== 512
|
||
) {
|
||
throw new Error(
|
||
'Embedding non valido per '
|
||
+ row.id,
|
||
);
|
||
}
|
||
}
|
||
|
||
const allUserFaces = await trx(
|
||
'faces as f',
|
||
)
|
||
.join(
|
||
'photos as p',
|
||
'p.id',
|
||
'f.photo_id',
|
||
)
|
||
.where(
|
||
'p.user',
|
||
options.user,
|
||
)
|
||
.count({
|
||
count: 'f.id',
|
||
})
|
||
.first();
|
||
|
||
const totalUserFaces = Number(
|
||
allUserFaces?.count || 0,
|
||
);
|
||
|
||
if (
|
||
options.scope === 'all'
|
||
&& totalUserFaces !== faceIds.length
|
||
) {
|
||
throw new Error(
|
||
'La run scope=all non contiene '
|
||
+ 'tutti i volti dell’utente: '
|
||
+ `DB=${totalUserFaces}, `
|
||
+ `risultato=${faceIds.length}`,
|
||
);
|
||
}
|
||
}
|
||
|
||
|
||
async function persistResult(
|
||
result,
|
||
validation,
|
||
options,
|
||
) {
|
||
const runId = randomUUID();
|
||
const now = new Date().toISOString();
|
||
|
||
const clusterIdByLabel = new Map();
|
||
|
||
for (const cluster of result.clusters) {
|
||
clusterIdByLabel.set(
|
||
Number(cluster.label),
|
||
randomUUID(),
|
||
);
|
||
}
|
||
|
||
await db.transaction(async (trx) => {
|
||
await validateDatabaseFaces(
|
||
trx,
|
||
options,
|
||
validation.assignments,
|
||
);
|
||
|
||
await trx('face_cluster_runs')
|
||
.insert({
|
||
id: runId,
|
||
user: options.user,
|
||
status: 'processing',
|
||
scope: options.scope,
|
||
embedding_count:
|
||
validation.totalFaces,
|
||
min_cluster_size:
|
||
options.minClusterSize,
|
||
min_samples:
|
||
options.minSamples,
|
||
metric: options.metric,
|
||
selection_method:
|
||
options.selectionMethod,
|
||
engine: 'sklearn',
|
||
engine_version: null,
|
||
cluster_count: null,
|
||
clustered_faces: null,
|
||
noise_faces: null,
|
||
noise_percent: null,
|
||
last_error: null,
|
||
started_at: now,
|
||
completed_at: null,
|
||
created_at: now,
|
||
updated_at: now,
|
||
});
|
||
|
||
const clusterRows =
|
||
result.clusters.map(
|
||
(cluster) => ({
|
||
id: clusterIdByLabel.get(
|
||
Number(cluster.label),
|
||
),
|
||
run_id: runId,
|
||
label: Number(cluster.label),
|
||
size: Number(cluster.size),
|
||
min_samples:
|
||
options.minSamples,
|
||
min_cluster_size:
|
||
options.minClusterSize,
|
||
person_id: null,
|
||
created_at: now,
|
||
updated_at: now,
|
||
}),
|
||
);
|
||
|
||
if (clusterRows.length > 0) {
|
||
await trx('face_clusters')
|
||
.insert(clusterRows);
|
||
}
|
||
|
||
const membershipRows =
|
||
validation.assignments.map(
|
||
(assignment) => ({
|
||
id: randomUUID(),
|
||
run_id: runId,
|
||
cluster_id:
|
||
assignment.noiseFlag === 1
|
||
? null
|
||
: clusterIdByLabel.get(
|
||
assignment.label,
|
||
),
|
||
face_id: assignment.faceId,
|
||
label: assignment.label,
|
||
probability:
|
||
assignment.probability,
|
||
noise_flag:
|
||
assignment.noiseFlag,
|
||
review_status: 'suggested',
|
||
reviewed_by: null,
|
||
reviewed_at: null,
|
||
created_at: now,
|
||
updated_at: now,
|
||
}),
|
||
);
|
||
|
||
await trx(
|
||
'face_cluster_memberships',
|
||
).insert(membershipRows);
|
||
|
||
if (options.scope === 'all') {
|
||
await trx('faces as f')
|
||
.whereIn(
|
||
'f.id',
|
||
function selectUserFaceIds() {
|
||
this.select('f2.id')
|
||
.from('faces as f2')
|
||
.join(
|
||
'photos as p2',
|
||
'p2.id',
|
||
'f2.photo_id',
|
||
)
|
||
.where(
|
||
'p2.user',
|
||
options.user,
|
||
);
|
||
},
|
||
)
|
||
.update({
|
||
cluster_id: null,
|
||
status: 'detected',
|
||
updated_at: now,
|
||
});
|
||
}
|
||
|
||
for (
|
||
const assignment
|
||
of validation.assignments
|
||
) {
|
||
if (assignment.noiseFlag === 1) {
|
||
continue;
|
||
}
|
||
|
||
const clusterId =
|
||
clusterIdByLabel.get(
|
||
assignment.label,
|
||
);
|
||
|
||
if (!clusterId) {
|
||
throw new Error(
|
||
'Cluster ID mancante per label '
|
||
+ assignment.label,
|
||
);
|
||
}
|
||
|
||
const updated = await trx('faces')
|
||
.where({
|
||
id: assignment.faceId,
|
||
})
|
||
.update({
|
||
cluster_id: clusterId,
|
||
status: 'clustered',
|
||
updated_at: now,
|
||
});
|
||
|
||
if (updated !== 1) {
|
||
throw new Error(
|
||
'Impossibile aggiornare il volto '
|
||
+ assignment.faceId,
|
||
);
|
||
}
|
||
}
|
||
|
||
await trx('face_cluster_runs')
|
||
.where({
|
||
id: runId,
|
||
})
|
||
.update({
|
||
status: 'done',
|
||
cluster_count:
|
||
validation.clusterCount,
|
||
clustered_faces:
|
||
validation.clusteredFaces,
|
||
noise_faces:
|
||
validation.noiseFaces,
|
||
noise_percent:
|
||
Number(
|
||
result.summary.noise_percent,
|
||
),
|
||
completed_at: now,
|
||
updated_at: now,
|
||
});
|
||
});
|
||
|
||
return {
|
||
run_id: runId,
|
||
cluster_count:
|
||
validation.clusterCount,
|
||
clustered_faces:
|
||
validation.clusteredFaces,
|
||
noise_faces:
|
||
validation.noiseFaces,
|
||
};
|
||
}
|
||
|
||
|
||
function printSummary(
|
||
result,
|
||
validation,
|
||
options,
|
||
) {
|
||
console.log({
|
||
mode: options.persist
|
||
? 'persist'
|
||
: 'dry-run',
|
||
user: options.user,
|
||
parameters: result.parameters,
|
||
summary: result.summary,
|
||
validated: {
|
||
total_faces:
|
||
validation.totalFaces,
|
||
cluster_count:
|
||
validation.clusterCount,
|
||
clustered_faces:
|
||
validation.clusteredFaces,
|
||
noise_faces:
|
||
validation.noiseFaces,
|
||
unique_memberships:
|
||
validation.assignments.length,
|
||
},
|
||
});
|
||
}
|
||
|
||
|
||
async function main() {
|
||
const options = parseArguments(
|
||
process.argv.slice(2),
|
||
);
|
||
|
||
if (options.help) {
|
||
printHelp();
|
||
return;
|
||
}
|
||
|
||
const result =
|
||
await runPythonClustering(options);
|
||
|
||
const validation =
|
||
validateResult(
|
||
result,
|
||
options,
|
||
);
|
||
|
||
printSummary(
|
||
result,
|
||
validation,
|
||
options,
|
||
);
|
||
|
||
if (!options.persist) {
|
||
console.log(
|
||
'[faceClustering] dry-run completato. '
|
||
+ 'Database non modificato.',
|
||
);
|
||
|
||
return;
|
||
}
|
||
|
||
const persisted = await persistResult(
|
||
result,
|
||
validation,
|
||
options,
|
||
);
|
||
|
||
console.log(
|
||
'[faceClustering] run persistita:',
|
||
persisted,
|
||
);
|
||
}
|
||
|
||
|
||
main()
|
||
.catch((error) => {
|
||
console.error(
|
||
'[faceClustering] errore:',
|
||
error.stack
|
||
|| error.message
|
||
|| error,
|
||
);
|
||
|
||
process.exitCode = 1;
|
||
})
|
||
.finally(async () => {
|
||
await db.destroy();
|
||
});
|