532 lines
18 KiB
TypeScript
532 lines
18 KiB
TypeScript
/**
|
||
* 通用算法插件子进程入口
|
||
* ------------------------------------------------------------
|
||
* fork 自 algorithmJob.ts，负责单次插件运行的全过程：
|
||
* 1. 校验 manifest 是否注册；
|
||
* 2. 准备 workDir = /data/tasks/<taskId>/analysis/<resultId>，写 input.json；
|
||
* 3. 按 runtime.kind 走 process / mock 两条路径：
|
||
* - process：spawn Python，按行解析 stdout 进度 JSON，等子进程退出后读 result.json；
|
||
* - mock：读 fixtures/<file>.json，假装跑 simulateMs ms 后写到 outDir/result.json；
|
||
* 4. 把 result.json 合并到 task_analysis_results.resultData，更新 status。
|
||
*
|
||
* 进程参数：
|
||
* node algorithmJobRunner.js <resultId> <taskId> <pluginId> <paramsJSON>
|
||
*/
|
||
import 'dotenv/config';
|
||
|
||
import * as fs from 'node:fs';
|
||
import * as path from 'node:path';
|
||
import { spawn } from 'node:child_process';
|
||
import { setTimeout as sleep } from 'node:timers/promises';
|
||
|
||
import * as db from './db.js';
|
||
import { getPlugin, adaptersDir, fixturesDir } from './plugins/registry.js';
|
||
import {
|
||
inspectionObjectContextFromName,
|
||
logCompatibilityWarnings,
|
||
normalizeMeasurements,
|
||
type InspectionObjectContext,
|
||
} from './measurementCompatibility.js';
|
||
import type {
|
||
PluginManifest,
|
||
PluginProgressEvent,
|
||
PluginResult,
|
||
PluginResultData,
|
||
} from './plugins/types.js';
|
||
|
||
const DATA_DIR = process.env.DATA_DIR || '/data';
|
||
const TASKS_DIR = path.join(DATA_DIR, 'tasks');
|
||
const PYTHON_BIN = process.env.PYTHON3_PATH || 'python3';
|
||
/** 算法源码根。生产镜像/开发宿主机上把 alg/ 挂到此处。 */
|
||
const ALG_ROOT = process.env.ALG_ROOT || '/home/lipan/massive-cloud/alg';
|
||
const DEFAULT_TIMEOUT_MS = 20 * 60 * 1000; // 20 分钟（验收要求）
|
||
|
||
function workDirFor(taskId: number, resultId: number): string {
|
||
return path.join(TASKS_DIR, String(taskId), 'analysis', String(resultId));
|
||
}
|
||
|
||
function ensureDir(dir: string) {
|
||
fs.mkdirSync(dir, { recursive: true });
|
||
}
|
||
|
||
function formatRegistrationTimestamp(date: Date): string {
|
||
const pad = (n: number) => String(n).padStart(2, '0');
|
||
return [
|
||
date.getFullYear(),
|
||
pad(date.getMonth() + 1),
|
||
pad(date.getDate()),
|
||
'-',
|
||
pad(date.getHours()),
|
||
pad(date.getMinutes()),
|
||
pad(date.getSeconds()),
|
||
].join('');
|
||
}
|
||
|
||
function registeredPointcloudDisplayName(sourceName: string | undefined, derivedFile: string, resultId: number): string {
|
||
const sourceParsed = path.parse(sourceName || 'pointcloud');
|
||
const stem = sourceParsed.name || 'pointcloud';
|
||
const ext = path.extname(derivedFile) || sourceParsed.ext || '.ply';
|
||
return `${stem}_已配准_${formatRegistrationTimestamp(new Date())}_R${resultId}${ext}`;
|
||
}
|
||
|
||
/** 把 process / mock cmd 模板里的占位符替换成实际路径。 */
|
||
function resolvePlaceholders(value: string, ctx: Record<string, string>): string {
|
||
return value.replace(/\{(\w+)\}/g, (_, key) => ctx[key] ?? `{${key}}`);
|
||
}
|
||
|
||
/** 收集任务相关上下文：点云路径、CAD 路径等。 */
|
||
async function gatherInputs(
|
||
taskId: number,
|
||
manifest: PluginManifest,
|
||
pointCloudFileIds?: number[],
|
||
cadFileIds?: number[],
|
||
) {
|
||
const task = await db.getInspectionTaskById(taskId);
|
||
if (!task) throw new Error(`task ${taskId} not found`);
|
||
|
||
let pointCloudRecords = await db.listTaskPointcloudFiles(taskId);
|
||
if (Array.isArray(pointCloudFileIds) && pointCloudFileIds.length > 0) {
|
||
const idSet = new Set(pointCloudFileIds);
|
||
pointCloudRecords = pointCloudRecords.filter((r) => idSet.has(r.id));
|
||
}
|
||
const pointCloudPaths = pointCloudRecords.map((r) => r.filePath);
|
||
if (manifest.requiresPointCloud !== false && pointCloudPaths.length === 0) {
|
||
throw new Error('当前任务下未上传点云，请先上传。');
|
||
}
|
||
|
||
let cadPath: string | null = null;
|
||
if (manifest.requiresCad || (Array.isArray(cadFileIds) && cadFileIds.length > 0)) {
|
||
// 优先使用任务级 CAD（前端选中的）
|
||
if (Array.isArray(cadFileIds) && cadFileIds.length > 0) {
|
||
const taskCads = await db.listTaskCadFiles(taskId);
|
||
const idSet = new Set(cadFileIds);
|
||
const selected = taskCads.filter((r) => idSet.has(r.id));
|
||
if (selected.length > 0) {
|
||
cadPath = selected[0].filePath;
|
||
}
|
||
}
|
||
// 回退到车型级 CAD
|
||
if (!cadPath) {
|
||
const modelCads = await db.listModelCadFiles(task.modelId);
|
||
if (modelCads.length > 0) {
|
||
cadPath = modelCads[0].filePath;
|
||
}
|
||
}
|
||
if (!cadPath && manifest.requiresCad) {
|
||
throw new Error('该插件需要 CAD 文件，但当前任务和车型下均无 CAD。');
|
||
}
|
||
}
|
||
|
||
const inspectionObject = task.inspectionObjectId
|
||
? await db.getInspectionObjectById(task.inspectionObjectId)
|
||
: undefined;
|
||
const inspectionContext = inspectionObjectContextFromName(inspectionObject?.name);
|
||
|
||
return { task, pointCloudRecords, pointCloudPaths, cadPath, inspectionContext };
|
||
}
|
||
|
||
async function writeInputJson(
|
||
workDir: string,
|
||
taskId: number,
|
||
resultId: number,
|
||
manifest: PluginManifest,
|
||
params: Record<string, unknown>,
|
||
pointCloudPaths: string[],
|
||
cadPath: string | null,
|
||
inputPointCloudFileIds: number[],
|
||
inputCadFileIds: number[],
|
||
inspectionContext: InspectionObjectContext,
|
||
) {
|
||
const inputJson = {
|
||
schemaVersion: 1,
|
||
pluginId: manifest.id,
|
||
pluginVersion: manifest.version,
|
||
resultId,
|
||
taskId,
|
||
workDir,
|
||
outDir: workDir,
|
||
context: inspectionContext,
|
||
inputs: {
|
||
pointCloudPaths,
|
||
pointCloudFileIds: inputPointCloudFileIds,
|
||
cadPath,
|
||
cadFileIds: inputCadFileIds,
|
||
config: params,
|
||
},
|
||
};
|
||
const target = path.join(workDir, 'input.json');
|
||
fs.writeFileSync(target, JSON.stringify(inputJson, null, 2), 'utf-8');
|
||
return target;
|
||
}
|
||
|
||
async function patchProgress(
|
||
resultId: number,
|
||
pluginId: string,
|
||
progress: number | undefined,
|
||
stage: string | undefined,
|
||
) {
|
||
const data: Record<string, unknown> = {
|
||
kind: 'plugin',
|
||
pluginId,
|
||
progress: progress != null ? Math.max(0, Math.min(1, progress)) : undefined,
|
||
stage,
|
||
};
|
||
await db.patchTaskAnalysisResultData(resultId, data);
|
||
}
|
||
|
||
function normalizeProgress(percent: number): number {
|
||
// 适配器允许 0~1 也允许 0~100；> 1 时按百分比处理
|
||
if (percent <= 1.0 && percent >= 0) return percent;
|
||
return Math.max(0, Math.min(100, percent)) / 100;
|
||
}
|
||
|
||
function readResultJson(workDir: string): PluginResult | null {
|
||
const p = path.join(workDir, 'result.json');
|
||
if (!fs.existsSync(p)) return null;
|
||
try {
|
||
// JSON 标准不支持 NaN/Infinity；Python json 模块有时会输出 NaN，用正则替换掉
|
||
const raw = fs.readFileSync(p, 'utf-8').replace(/:\s*NaN\b/g, ': null').replace(/:\s*Infinity\b/g, ': null').replace(/:\s*-Infinity\b/g, ': null');
|
||
return JSON.parse(raw) as PluginResult;
|
||
} catch (e) {
|
||
console.error('[AlgorithmJobRunner] result.json 解析失败:', e);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
const DEMO_FLOW_RESULT_DATA_KEYS = [
|
||
'demoFlowRunId',
|
||
'demoFlowRole',
|
||
'demoFlowIndex',
|
||
'demoFlowTotal',
|
||
'demoFlowSourcePointCloudFileId',
|
||
'demoFlowCadFileId',
|
||
'demoFlowCancelledAt',
|
||
] as const;
|
||
|
||
async function readDemoFlowResultData(resultId: number): Promise<Record<string, unknown>> {
|
||
const row = await db.getTaskAnalysisResultById(resultId);
|
||
const existing = (row?.resultData as Record<string, unknown> | null) ?? {};
|
||
const picked: Record<string, unknown> = {};
|
||
for (const key of DEMO_FLOW_RESULT_DATA_KEYS) {
|
||
if (existing[key] !== undefined) {
|
||
picked[key] = existing[key];
|
||
}
|
||
}
|
||
return picked;
|
||
}
|
||
|
||
async function runProcessRuntime(
|
||
manifest: PluginManifest,
|
||
workDir: string,
|
||
inputJsonPath: string,
|
||
resultId: number,
|
||
): Promise<{ exitCode: number | null; lastStage?: string }> {
|
||
if (manifest.runtime.kind !== 'process') {
|
||
throw new Error('runtime.kind !== process');
|
||
}
|
||
const rt = manifest.runtime;
|
||
const ctx: Record<string, string> = {
|
||
pythonBin: PYTHON_BIN,
|
||
inputJson: inputJsonPath,
|
||
workDir,
|
||
adapterDir: adaptersDir(),
|
||
algRoot: ALG_ROOT,
|
||
};
|
||
const [cmd0, ...rawArgs] = rt.cmd;
|
||
if (!cmd0) throw new Error('runtime.cmd 为空');
|
||
const cmd = resolvePlaceholders(cmd0, ctx);
|
||
const args = rawArgs.map((a) => resolvePlaceholders(a, ctx));
|
||
|
||
const envExtra: Record<string, string> = {};
|
||
for (const [k, v] of Object.entries(rt.envExtra ?? {})) {
|
||
envExtra[k] = resolvePlaceholders(String(v), ctx);
|
||
}
|
||
|
||
console.log(`[AlgorithmJobRunner] spawn ${cmd} ${args.join(' ')}`);
|
||
const proc = spawn(cmd, args, {
|
||
cwd: workDir,
|
||
env: { ...process.env, ...envExtra },
|
||
stdio: ['ignore', 'pipe', 'inherit'],
|
||
});
|
||
|
||
const timeoutMs = rt.timeoutMs ?? DEFAULT_TIMEOUT_MS;
|
||
const timer = setTimeout(() => {
|
||
console.error(`[AlgorithmJobRunner] 超时 ${timeoutMs}ms，杀掉子进程`);
|
||
proc.kill('SIGKILL');
|
||
}, timeoutMs);
|
||
|
||
let buf = '';
|
||
let lastStage: string | undefined;
|
||
proc.stdout.setEncoding('utf-8');
|
||
proc.stdout.on('data', (chunk: string) => {
|
||
buf += chunk;
|
||
let nl: number;
|
||
while ((nl = buf.indexOf('\n')) >= 0) {
|
||
const line = buf.slice(0, nl).trim();
|
||
buf = buf.slice(nl + 1);
|
||
if (!line) continue;
|
||
// 行可能是进度 JSON，也可能是普通日志；普通日志直接落控制台
|
||
if (line.startsWith('{') && line.includes('"progress"')) {
|
||
try {
|
||
const evt = JSON.parse(line) as PluginProgressEvent & { progress?: number; stage?: string };
|
||
// 兼容两种字段名：percent / progress
|
||
const rawPercent = (evt as any).percent ?? (evt as any).progress;
|
||
const percent = typeof rawPercent === 'number' ? normalizeProgress(rawPercent) : undefined;
|
||
const stage = (evt as any).message ?? (evt as any).stage;
|
||
lastStage = stage ?? lastStage;
|
||
void patchProgress(resultId, manifest.id, percent, stage);
|
||
} catch {
|
||
process.stdout.write(`[plugin:${manifest.id}] ${line}\n`);
|
||
}
|
||
} else {
|
||
process.stdout.write(`[plugin:${manifest.id}] ${line}\n`);
|
||
}
|
||
}
|
||
});
|
||
|
||
const exitCode: number | null = await new Promise((resolve) => {
|
||
proc.on('error', (err) => {
|
||
console.error('[AlgorithmJobRunner] spawn error:', err);
|
||
resolve(-1);
|
||
});
|
||
proc.on('exit', (code) => resolve(code));
|
||
});
|
||
clearTimeout(timer);
|
||
return { exitCode, lastStage };
|
||
}
|
||
|
||
async function runMockRuntime(
|
||
manifest: PluginManifest,
|
||
workDir: string,
|
||
resultId: number,
|
||
): Promise<void> {
|
||
if (manifest.runtime.kind !== 'mock') return;
|
||
const fixturePath = path.join(fixturesDir(), manifest.runtime.fixture);
|
||
if (!fs.existsSync(fixturePath)) {
|
||
throw new Error(`mock fixture 不存在: ${fixturePath}`);
|
||
}
|
||
const fixture = JSON.parse(fs.readFileSync(fixturePath, 'utf-8')) as PluginResult;
|
||
const simulateMs = Math.max(0, manifest.runtime.simulateMs ?? 1200);
|
||
const steps = 8;
|
||
for (let i = 1; i <= steps; i++) {
|
||
await patchProgress(resultId, manifest.id, i / steps, `Mock 阶段 ${i}/${steps}`);
|
||
await sleep(simulateMs / steps);
|
||
}
|
||
fs.writeFileSync(path.join(workDir, 'result.json'), JSON.stringify(fixture, null, 2), 'utf-8');
|
||
}
|
||
|
||
export async function runAlgorithmJob(
|
||
resultId: number,
|
||
taskId: number,
|
||
pluginId: string,
|
||
params: Record<string, unknown>,
|
||
pointCloudFileIds?: number[],
|
||
cadFileIds?: number[],
|
||
): Promise<void> {
|
||
const manifest = getPlugin(pluginId);
|
||
if (!manifest) {
|
||
const demoFlowResultData = await readDemoFlowResultData(resultId);
|
||
await db.updateTaskAnalysisResult(resultId, {
|
||
status: 'error',
|
||
resultData: {
|
||
kind: 'plugin',
|
||
pluginId,
|
||
error: `未注册的插件：${pluginId}`,
|
||
...demoFlowResultData,
|
||
} as PluginResultData,
|
||
});
|
||
return;
|
||
}
|
||
|
||
const workDir = workDirFor(taskId, resultId);
|
||
ensureDir(workDir);
|
||
|
||
try {
|
||
await patchProgress(resultId, pluginId, 0.02, '准备输入');
|
||
const { pointCloudRecords, pointCloudPaths, cadPath, inspectionContext } = await gatherInputs(
|
||
taskId,
|
||
manifest,
|
||
pointCloudFileIds,
|
||
cadFileIds,
|
||
);
|
||
const inputPointCloudFileIds = pointCloudRecords.map((r) => r.id);
|
||
const inputCadFileIds = Array.isArray(cadFileIds) ? cadFileIds.filter((id) => Number.isInteger(id)) : [];
|
||
|
||
const inputJsonPath = await writeInputJson(
|
||
workDir,
|
||
taskId,
|
||
resultId,
|
||
manifest,
|
||
params,
|
||
pointCloudPaths,
|
||
cadPath,
|
||
inputPointCloudFileIds,
|
||
inputCadFileIds,
|
||
inspectionContext,
|
||
);
|
||
|
||
let exitCode: number | null = 0;
|
||
if (manifest.runtime.kind === 'process') {
|
||
const r = await runProcessRuntime(manifest, workDir, inputJsonPath, resultId);
|
||
exitCode = r.exitCode;
|
||
} else {
|
||
await runMockRuntime(manifest, workDir, resultId);
|
||
}
|
||
|
||
const result = readResultJson(workDir);
|
||
const finalStatus: 'completed' | 'error' = result?.status === 'error'
|
||
? 'error'
|
||
: exitCode === 0 || manifest.runtime.kind === 'mock'
|
||
? 'completed'
|
||
: 'error';
|
||
|
||
const demoFlowResultData = await readDemoFlowResultData(resultId);
|
||
const normalized = normalizeMeasurements({
|
||
pluginId,
|
||
measurements: result?.measurements,
|
||
extra: result?.extra,
|
||
context: inspectionContext,
|
||
algorithmProtocolVersion: result?.schemaVersion ?? manifest.schemaVersion,
|
||
});
|
||
logCompatibilityWarnings(resultId, normalized.warnings);
|
||
const normalizedMeasurements = Array.isArray(result?.measurements)
|
||
? normalized.measurements
|
||
: undefined;
|
||
const resultData: PluginResultData = {
|
||
kind: 'plugin',
|
||
pluginId,
|
||
progress: finalStatus === 'completed' ? 1 : undefined,
|
||
stage: finalStatus === 'completed' ? '完成' : '失败',
|
||
exitCode,
|
||
inputPointCloudFileIds,
|
||
inputCadFileIds,
|
||
// 把 result.json 内容平展到 resultData 里，方便前端直接使用
|
||
schemaVersion: result?.schemaVersion ?? manifest.schemaVersion,
|
||
message: result?.message,
|
||
measurements: normalizedMeasurements,
|
||
errorStats: result?.errorStats,
|
||
overlay3d: result?.overlay3d,
|
||
images: result?.images,
|
||
files: result?.files,
|
||
extra: normalized.extra,
|
||
error: finalStatus === 'error'
|
||
? result?.message ?? `子进程退出码 ${exitCode}`
|
||
: undefined,
|
||
...demoFlowResultData,
|
||
};
|
||
|
||
await db.updateTaskAnalysisResult(resultId, {
|
||
status: finalStatus,
|
||
errorStats: result?.errorStats ?? null,
|
||
measurements: normalizedMeasurements ?? null,
|
||
resultData,
|
||
});
|
||
|
||
// ── 回写配准产出点云 ─────────────────────────────────────────────────────
|
||
// 若 result.json 包含 derivedPointClouds，把对应 PLY 注册为任务点云，
|
||
// 供后续测量算法直接选用，无需用户手动上传。
|
||
if (finalStatus === 'completed') {
|
||
const derived = (result as Record<string, unknown> | null)?.derivedPointClouds;
|
||
if (Array.isArray(derived) && derived.length > 0) {
|
||
for (const item of derived) {
|
||
const d = item as { file?: string; name?: string };
|
||
if (!d.file) continue;
|
||
const absPath = path.join(workDir, d.file);
|
||
if (!fs.existsSync(absPath)) {
|
||
console.warn(`[AlgorithmJobRunner] derivedPointCloud not found: ${absPath}`);
|
||
continue;
|
||
}
|
||
try {
|
||
const sourcePointcloud = pointCloudRecords[0];
|
||
const displayName = registeredPointcloudDisplayName(sourcePointcloud?.name, d.file, resultId);
|
||
await db.createRegisteredPointcloudFile(
|
||
taskId,
|
||
displayName,
|
||
absPath,
|
||
resultId,
|
||
sourcePointcloud?.id ?? null,
|
||
);
|
||
console.log(`[AlgorithmJobRunner] registered derived pointcloud: ${displayName}`);
|
||
} catch (err) {
|
||
console.error('[AlgorithmJobRunner] failed to register derived pointcloud:', err);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if (finalStatus === 'completed') {
|
||
console.log(
|
||
`[AlgorithmJobRunner] ok pluginId=${pluginId} resultId=${resultId} taskId=${taskId}`,
|
||
);
|
||
} else {
|
||
console.error(
|
||
`[AlgorithmJobRunner] failed pluginId=${pluginId} resultId=${resultId} exit=${exitCode}`,
|
||
);
|
||
}
|
||
} catch (err) {
|
||
console.error('[AlgorithmJobRunner]', err);
|
||
const message = err instanceof Error ? err.message : String(err);
|
||
const demoFlowResultData = await readDemoFlowResultData(resultId);
|
||
await db.updateTaskAnalysisResult(resultId, {
|
||
status: 'error',
|
||
resultData: {
|
||
kind: 'plugin',
|
||
pluginId,
|
||
error: message,
|
||
...demoFlowResultData,
|
||
} as PluginResultData,
|
||
});
|
||
}
|
||
}
|
||
|
||
// ─── CLI 入口 ────────────────────────────────────────────────────────────────
|
||
//
|
||
// node algorithmJobRunner.js <resultId> <taskId> <pluginId> <paramsJSON> [pcFileIdsJSON] [cadFileIdsJSON]
|
||
|
||
const [, , a, b, c, d, e, f] = process.argv;
|
||
if (a && b && c) {
|
||
const resultId = Number(a);
|
||
const taskId = Number(b);
|
||
const pluginId = String(c);
|
||
let params: Record<string, unknown> = {};
|
||
if (d) {
|
||
try {
|
||
params = JSON.parse(d) as Record<string, unknown>;
|
||
} catch (err) {
|
||
console.error('[AlgorithmJobRunner] params JSON 解析失败:', err);
|
||
}
|
||
}
|
||
let pointCloudFileIds: number[] | undefined;
|
||
if (e) {
|
||
try {
|
||
const parsed = JSON.parse(e) as unknown;
|
||
if (Array.isArray(parsed) && parsed.length > 0) {
|
||
pointCloudFileIds = parsed.map(Number).filter(Number.isFinite);
|
||
}
|
||
} catch (err) {
|
||
console.error('[AlgorithmJobRunner] pcFileIds JSON 解析失败:', err);
|
||
}
|
||
}
|
||
let cadFileIds: number[] | undefined;
|
||
if (f) {
|
||
try {
|
||
const parsed = JSON.parse(f) as unknown;
|
||
if (Array.isArray(parsed) && parsed.length > 0) {
|
||
cadFileIds = parsed.map(Number).filter(Number.isFinite);
|
||
}
|
||
} catch (err) {
|
||
console.error('[AlgorithmJobRunner] cadFileIds JSON 解析失败:', err);
|
||
}
|
||
}
|
||
if (!Number.isFinite(resultId) || !Number.isFinite(taskId) || resultId <= 0 || taskId <= 0) {
|
||
console.error('[AlgorithmJobRunner] 用法: <resultId> <taskId> <pluginId> [paramsJSON] [pcFileIdsJSON] [cadFileIdsJSON]');
|
||
process.exit(1);
|
||
}
|
||
runAlgorithmJob(resultId, taskId, pluginId, params, pointCloudFileIds, cadFileIds)
|
||
.then(() => process.exit(0))
|
||
.catch((err) => {
|
||
console.error('[AlgorithmJobRunner]', err);
|
||
process.exit(1);
|
||
});
|
||
}
|