diff --git a/client/dive-common/apispec.ts b/client/dive-common/apispec.ts index 57ba46287..2b6fd3bd6 100644 --- a/client/dive-common/apispec.ts +++ b/client/dive-common/apispec.ts @@ -11,6 +11,7 @@ import { ImageEnhancements } from 'vue-media-annotator/use/useImageEnhancements' import type { CameraHomographies, CameraCorrespondences, CameraTransformTypes, RegistrationSource, } from 'vue-media-annotator/alignedView/CameraRegistrationStore'; +import type { CameraRole } from 'dive-common/pipelineCameraOrder'; import type { PercentileStretch } from 'vue-media-annotator/use/useImageEnhancements'; type DatasetType = 'image-sequence' | 'video' | 'multi' | 'large-image'; @@ -78,6 +79,22 @@ interface PipeMetadata { * the two conventional keys are used. */ calibrationKeys?: string[]; + /** + * Camera role per pipeline input for 2-cam/3-cam pipes (e.g. ["EO", "UV", "IR"]: + * input1 is optical, input2 ultraviolet, input3 thermal), parsed from a + * `# Camera Order: [cam...]` header. Labels the slots of the pre-run + * camera-assignment step and drives its prefill (dive-common/pipelineCameraOrder.ts); + * pipes without it show bare input1..N slots. + */ + cameraOrder?: string[]; + /** + * Input positions whose detections/images the pipe warps onto camera 1 + * (`process warpN :: warp_detections | warp_image` in the pipe body), e.g. + * [2, 3]. Each such camera needs a fitted registration (Camera Registration tab) onto + * camera 1; DIVE checks that before the run instead of letting the pipe + * fail at configure time on a missing file. + */ + registrationWarps?: number[]; } interface PipelineRuntimeParams { @@ -87,6 +104,12 @@ interface PipelineRuntimeParams { interface PipelineParams { kwiverParams?: Record; runtimeParams?: PipelineRuntimeParams; + /** + * 2-cam/3-cam pipes: the dataset camera to feed each inputN, in order, as + * confirmed by the user before the run. When omitted (API callers) the + * dataset's stored camera order is used. + */ + cameraOrder?: string[]; /** Filter / transcode / disparity pipelines: name for the newly created dataset. */ outputDatasetName?: string; /** @@ -270,9 +293,16 @@ interface DatasetConfigMutable { cameraTransformTypes?: CameraTransformTypes; /** Producer provenance of the camera registration (see RegistrationSource). */ cameraRegistrationSource?: RegistrationSource | null; + /** + * Sensor role per multicam camera name (eo / ir / uv), inferred at import + * from the camera and image names and editable afterwards; used to place + * cameras onto a pipeline's declared camera slots. Cameras with no known + * role are absent. + */ + cameraRoles?: Record; error?: string; } -const DatasetConfigMutableKeys = ['attributes', 'confidenceFilters', 'timeFilters', 'imageEnhancements', 'customTypeStyling', 'customGroupStyling', 'attributeTrackFilters', 'datasetInfo', 'cameraHomographies', 'cameraCorrespondences', 'cameraTransformTypes', 'cameraRegistrationSource', 'typeHierarchy']; +const DatasetConfigMutableKeys = ['attributes', 'confidenceFilters', 'timeFilters', 'imageEnhancements', 'customTypeStyling', 'customGroupStyling', 'attributeTrackFilters', 'datasetInfo', 'cameraHomographies', 'cameraCorrespondences', 'cameraTransformTypes', 'cameraRegistrationSource', 'typeHierarchy', 'cameraRoles']; /** * Cross-dataset color/style overrides, reused across every dataset when the * "shared" color scope is enabled (see clientSettings.typeSettings.colorScope). diff --git a/client/dive-common/components/PipelineCameraAssignDialog.vue b/client/dive-common/components/PipelineCameraAssignDialog.vue new file mode 100644 index 000000000..42f7ee02a --- /dev/null +++ b/client/dive-common/components/PipelineCameraAssignDialog.vue @@ -0,0 +1,245 @@ + + + diff --git a/client/dive-common/components/RunPipelineMenu.vue b/client/dive-common/components/RunPipelineMenu.vue index f280e5f97..efda23a32 100644 --- a/client/dive-common/components/RunPipelineMenu.vue +++ b/client/dive-common/components/RunPipelineMenu.vue @@ -37,6 +37,13 @@ import pipelineTypeDisplay from 'dive-common/pipelineTypeDisplay'; import { useRequest } from 'dive-common/use'; import { usePrompt } from 'dive-common/vue-utilities/prompt-service'; import PipelineParamsDialog from 'dive-common/components/PipelineParamsDialog.vue'; +import PipelineCameraAssignDialog, { + PipelineCameraAssignRequest, PipelineCameraAssignResult, +} from 'dive-common/components/PipelineCameraAssignDialog.vue'; +import { orderedMultiCamCameraNames } from 'dive-common/multicamDisplay'; +import { + CameraRole, pipelineCameraSlots, prefillPipelineCameraOrder, +} from 'dive-common/pipelineCameraOrder'; import PipelineCalibrationWarningIcon from 'dive-common/components/PipelineCalibrationWarningIcon.vue'; type MenuState = 'idle' | 'configuring'; @@ -48,6 +55,7 @@ export default defineComponent({ JobLaunchDialog, JobConfigFilterTranscodeDialog, PipelineParamsDialog, + PipelineCameraAssignDialog, RunPipelineToast, PipelineCalibrationWarningIcon, }, @@ -111,7 +119,9 @@ export default defineComponent({ setup(props) { const { prompt } = usePrompt(); - const { runPipeline, getPipelineList, hasCalibrationFile } = useApi(); + const { + runPipeline, getPipelineList, hasCalibrationFile, loadConfig, saveConfig, + } = useApi(); const unsortedPipelines = ref({} as Pipelines); const { request: _runPipelineRequest, @@ -257,6 +267,76 @@ export default defineComponent({ return false; }); + // --- Multicam camera assignment ------------------------------------- + // Before a 2-cam/3-cam run the user sees which dataset camera DIVE + // proposes for each pipeline input (by camera role, else by name) and + // confirms or corrects it. Nothing about the placement is inferred at + // run time behind their back. + const cameraAssignRequest = ref(null); + let cameraAssignResolve: ((result: PipelineCameraAssignResult | null) => void) | null = null; + + function askCameraAssignment(request: PipelineCameraAssignRequest) { + return new Promise((resolve) => { + cameraAssignResolve = resolve; + cameraAssignRequest.value = request; + }); + } + function settleCameraAssignment(result: PipelineCameraAssignResult | null) { + cameraAssignRequest.value = null; + const resolve = cameraAssignResolve; + cameraAssignResolve = null; + resolve?.(result); + } + + /** + * Confirmed input1..N camera order per dataset, or null when the user + * cancelled. Persists confirmed roles onto the dataset when asked so the + * next run (and other pipelines) prefill correctly. + */ + async function confirmCameraOrders( + pipeline: Pipe, + datasetIds: string[], + ): Promise | null> { + const orders: Record = {}; + // eslint-disable-next-line no-restricted-syntax + for (const id of datasetIds) { + // eslint-disable-next-line no-await-in-loop + const config = await loadConfig(id); + const cameras = orderedMultiCamCameraNames(config.multiCamMedia); + if (!cameras.length) { + throw new Error(`${config.name} is not a multi-camera dataset`); + } + const slots = pipelineCameraSlots(pipeline.metadata?.cameraOrder, cameras.length); + if (slots.length !== cameras.length) { + throw new Error(`${pipeline.name} expects ${slots.length} cameras but ${config.name} has ${cameras.length} (${cameras.join(', ')})`); + } + const roles: Record = config.cameraRoles ?? {}; + // eslint-disable-next-line no-await-in-loop + const result = await askCameraAssignment({ + datasetName: config.name, + pipelineName: pipeline.name, + slots, + cameras, + proposed: prefillPipelineCameraOrder(slots, cameras, roles), + roles, + registrationWarps: pipeline.metadata?.registrationWarps ?? [], + fittedPairs: Object.keys(config.cameraHomographies ?? {}), + }); + if (!result) { + return null; + } + orders[id] = result.order; + if (result.roles) { + const merged = { ...roles, ...result.roles }; + if (JSON.stringify(merged) !== JSON.stringify(roles)) { + // eslint-disable-next-line no-await-in-loop + await saveConfig(id, { cameraRoles: merged }); + } + } + } + return orders; + } + async function _runPipelineOnSelectedItemInner( pipeline: Pipe, outputDatasetNameById?: Record, @@ -286,6 +366,14 @@ export default defineComponent({ || stereoPipelineMarker === pipeline.type) { datasetIds = props.selectedDatasetIds.map((item) => parentDatasetId(item)); } + let cameraOrderById: Record = {}; + if (multiCamPipelineMarkers.includes(pipeline.type)) { + const confirmed = await confirmCameraOrders(pipeline, datasetIds); + if (!confirmed) { + return; + } + cameraOrderById = confirmed; + } selectedPipeline.value = pipeline; const frameRange = props.timeFilter; await _runPipelineRequest(() => Promise.all( @@ -294,6 +382,7 @@ export default defineComponent({ outputDatasetName: outputDatasetNameById?.[id], outputParentFolderId, kwiverParams: kwiverParamsById?.[id], + cameraOrder: cameraOrderById[id], })), )); } @@ -373,6 +462,8 @@ export default defineComponent({ pipelineHasParams, categoryHasParams, categoryHasCalibrationWarning, + cameraAssignRequest, + settleCameraAssignment, }; }, }); @@ -598,6 +689,12 @@ export default defineComponent({ :params="pipelineParams" @confirm="confirmPipelineExecution" /> + diff --git a/client/dive-common/pipelineCameraOrder.spec.ts b/client/dive-common/pipelineCameraOrder.spec.ts new file mode 100644 index 000000000..b24dd5c51 --- /dev/null +++ b/client/dive-common/pipelineCameraOrder.spec.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'vitest'; +import { + camerasForSlot, inferCameraRole, inferCameraRoles, missingRegistrations, + parseCameraOrderHeader, pipelineCameraSlots, prefillPipelineCameraOrder, +} from './pipelineCameraOrder'; + +describe('pipelineCameraOrder', () => { + it('parses the header value into slot tokens', () => { + expect(parseCameraOrderHeader(' EO, UV, IR ')).toStrictEqual(['EO', 'UV', 'IR']); + expect(parseCameraOrderHeader('left right')).toStrictEqual(['left', 'right']); + expect(parseCameraOrderHeader('')).toStrictEqual([]); + }); + + it('matches slots by exact name, name segment, or role alias', () => { + const cameras = ['rgb', 'CENT_IR', 'uv_cam']; + expect(camerasForSlot('EO', cameras)).toStrictEqual(['rgb']); + expect(camerasForSlot('IR', cameras)).toStrictEqual(['CENT_IR']); + expect(camerasForSlot('ultraviolet', cameras)).toStrictEqual(['uv_cam']); + expect(camerasForSlot('rgb', cameras)).toStrictEqual(['rgb']); + // Exact name wins over role matching elsewhere. + expect(camerasForSlot('ir', ['ir', 'thermal'])).toStrictEqual(['ir']); + // Literal segments for non-role tokens. + expect(camerasForSlot('left', ['left_cam', 'right_cam'])).toStrictEqual(['left_cam']); + }); + + it('infers roles from camera names, then image names, only when unanimous', () => { + expect(inferCameraRole('rgb')).toBe('eo'); + expect(inferCameraRole('CENT_IR')).toBe('ir'); + expect(inferCameraRole('uv_cam')).toBe('uv'); + expect(inferCameraRole('cam1', ['flight_0001_rgb.jpg', 'flight_0002_rgb.jpg'])).toBe('eo'); + expect(inferCameraRole('cam1', ['a_rgb.jpg', 'b_ir.tif'])).toBeNull(); + expect(inferCameraRole('eo_ir')).toBeNull(); + expect(inferCameraRole('center', ['0001.png'])).toBeNull(); + expect(inferCameraRoles({ rgb: [], center: ['x_ir.tif'], other: ['a.png'] })) + .toStrictEqual({ rgb: 'eo', center: 'ir' }); + }); + + it('assigned roles beat name matching, and prefill leaves ambiguity open', () => { + // "thermal" is named like IR but the user marked it optical. + expect(prefillPipelineCameraOrder(['EO', 'IR'], ['thermal', 'other'], { thermal: 'eo', other: 'ir' })) + .toStrictEqual(['thermal', 'other']); + expect(prefillPipelineCameraOrder(['EO', 'UV', 'IR'], ['rgb', 'ir', 'uv'])) + .toStrictEqual(['rgb', 'uv', 'ir']); + // Two EO-ish names, no roles: the EO slot stays open, IR still fills. + expect(prefillPipelineCameraOrder(['EO', 'IR'], ['rgb', 'color', 'ir'])) + .toStrictEqual([null, 'ir']); + // Bare positional slots: nothing to match on, all open. + expect(prefillPipelineCameraOrder(['input1', 'input2'], ['rgb', 'ir'])) + .toStrictEqual([null, null]); + expect(pipelineCameraSlots(['EO', 'IR'], 2)).toStrictEqual(['EO', 'IR']); + expect(pipelineCameraSlots(undefined, 3)).toStrictEqual(['input1', 'input2', 'input3']); + }); + + it('reports warped cameras with no fitted registration onto camera 1', () => { + const order = ['rgb', 'uv', 'ir']; + // Stored in the reverse orientation still counts. + expect(missingRegistrations(order, [2, 3], ['rgb::ir'])) + .toStrictEqual([{ input: 2, camera: 'uv', target: 'rgb' }]); + expect(missingRegistrations(order, [2, 3], ['rgb::ir', 'uv::rgb'])).toStrictEqual([]); + expect(missingRegistrations(order, undefined, [])).toStrictEqual([]); + expect(missingRegistrations([null, 'uv'], [2], [])).toStrictEqual([]); + // An unfilled slot is reported by the assignment problems, not here. + expect(missingRegistrations(['rgb', null], [2], [])).toStrictEqual([]); + }); +}); diff --git a/client/dive-common/pipelineCameraOrder.ts b/client/dive-common/pipelineCameraOrder.ts new file mode 100644 index 000000000..0cf9c6794 --- /dev/null +++ b/client/dive-common/pipelineCameraOrder.ts @@ -0,0 +1,197 @@ +/** + * Camera roles, and how a multicam pipeline's input slots are filled from a + * dataset's cameras. + * + * A 2-cam/3-cam pipe wires each `inputN` to a specific role (the arctic seal + * 3-cam pipe runs the thermal detector on input3 and projects optical boxes + * onto input2), so which dataset camera lands on which input is the pipe's + * contract, not something DIVE can infer from display order. Pipes state it + * with a `# Camera Order: EO, UV, IR` header (one token per input, in order). + * + * Datasets carry a role per camera (`cameraRoles`, inferred once at import + * from the camera / image names and editable afterwards). Before a run the + * user is shown each slot with the camera DIVE proposes for it -- by role + * when both sides have one, else by name -- and confirms or corrects it; the + * confirmed order is what the job runs with. + * + * Role inference is mirrored in server/dive_tasks/multicam_pipeline.py. + */ + +/** The sensor modalities DIVE knows how to match. */ +export type CameraRole = 'eo' | 'ir' | 'uv'; +export const CAMERA_ROLES: readonly CameraRole[] = ['eo', 'ir', 'uv']; + +export const CAMERA_ROLE_LABELS: Record = { + eo: 'Optical (EO)', + ir: 'Thermal (IR)', + uv: 'Ultraviolet (UV)', +}; + +/** + * Role aliases: a slot token and a camera name match when they share a role, + * or when the token itself appears as a segment of the camera name (so pipes + * can name cameras literally, e.g. `# Camera Order: left, right`). + */ +export const CAMERA_ROLE_ALIASES: Record = { + eo: ['eo', 'rgb', 'optical', 'color', 'colour', 'vis', 'visible'], + ir: ['ir', 'thermal', 'lwir', 'mwir', 'flir'], + uv: ['uv', 'ultraviolet'], +}; + +function segments(name: string): string[] { + return name.toLowerCase().split(/[^a-z0-9]+/).filter((s) => s); +} + +/** The role a slot token or camera-name segment denotes, if any. */ +export function roleOfToken(token: string): CameraRole | null { + const lower = token.toLowerCase(); + const found = CAMERA_ROLES.find((role) => CAMERA_ROLE_ALIASES[role].includes(lower)); + return found ?? null; +} + +/** + * Infer a camera's role from its name, falling back to tokens in its image + * file names (KAMERA style `..._rgb.jpg` / `_ir.tif` / `_uv.jpg`). Only a + * unanimous answer counts: a name or file set naming two roles yields null. + */ +export function inferCameraRole(cameraName: string, imageNames: string[] = []): CameraRole | null { + const fromName = new Set(segments(cameraName).map(roleOfToken).filter((r): r is CameraRole => !!r)); + if (fromName.size === 1) { + return [...fromName][0]; + } + if (fromName.size > 1) { + return null; + } + const fromImages = new Set(); + imageNames.slice(0, 50).forEach((image) => { + const base = image.split(/[\\/]/).pop() ?? image; + const stem = base.replace(/\.[^.]+$/, ''); + segments(stem).map(roleOfToken).forEach((r) => { if (r) fromImages.add(r); }); + }); + return fromImages.size === 1 ? [...fromImages][0] : null; +} + +/** Infer roles for a whole rig; cameras DIVE cannot classify are omitted. */ +export function inferCameraRoles( + cameras: Record, +): Record { + const roles: Record = {}; + Object.entries(cameras).forEach(([name, images]) => { + const role = inferCameraRole(name, images ?? []); + if (role) { + roles[name] = role; + } + }); + return roles; +} + +/** + * Cameras that could fill a slot. A camera whose assigned role equals the + * slot's role wins over name matching, so a corrected role beats a + * misleading name; otherwise the slot token must equal the camera name, or + * share a role (or, for non-role tokens like `left`, appear literally) with + * one of the name's segments. + */ +export function camerasForSlot( + slot: string, + cameras: string[], + roles: Record = {}, +): string[] { + const role = roleOfToken(slot); + if (role) { + const byRole = cameras.filter((camera) => roles[camera] === role); + if (byRole.length) { + return byRole; + } + } + const lower = slot.toLowerCase(); + const exact = cameras.filter((camera) => camera.toLowerCase() === lower); + if (exact.length) { + return exact; + } + const aliases = new Set(role ? CAMERA_ROLE_ALIASES[role] : [lower]); + return cameras.filter((camera) => segments(camera).some((seg) => aliases.has(seg))); +} + +/** + * Propose a camera for every slot for the confirmation step. Never fails: a + * slot with no unique candidate is proposed as null and left for the user. + */ +export function prefillPipelineCameraOrder( + slots: string[], + cameras: string[], + roles: Record = {}, +): (string | null)[] { + const taken = new Set(); + const proposed: (string | null)[] = slots.map(() => null); + // Unique matches first, so an ambiguous slot cannot steal a camera another + // slot needs unambiguously. + slots.forEach((slot, index) => { + const matches = camerasForSlot(slot, cameras, roles).filter((c) => !taken.has(c)); + if (matches.length === 1) { + [proposed[index]] = matches; + taken.add(matches[0]); + } + }); + return proposed; +} + +/** + * Slot labels for a pipe: its declared `# Camera Order:` tokens, else the + * bare `input1..N` positions. + */ +export function pipelineCameraSlots(declaredOrder: string[] | null | undefined, count: number): string[] { + if (declaredOrder?.length) { + return declaredOrder; + } + return Array.from({ length: count }, (_, i) => `input${i + 1}`); +} + +export interface MissingRegistration { + /** 1-based pipeline input position. */ + input: number; + camera: string; + /** Camera 1, the frame the warp maps onto. */ + target: string; +} + +/** + * Cameras a pipe will warp (its `warpN` processes) that have no fitted + * registration onto camera 1 of the given order. `fittedPairs` are the + * dataset's homography keys (`a::b`, either orientation counts). Checked + * before a run so the failure is "register camera X onto Y first" rather + * than the pipe dying at configure time on a missing file. + */ +export function missingRegistrations( + order: (string | null)[], + registrationWarps: number[] | null | undefined, + fittedPairs: string[], +): MissingRegistration[] { + const target = order[0]; + if (!target || !registrationWarps?.length) { + return []; + } + const fitted = new Set(fittedPairs); + const missing: MissingRegistration[] = []; + registrationWarps.forEach((input) => { + const camera = order[input - 1]; + if (!camera || camera === target) { + return; + } + if (!fitted.has(`${camera}::${target}`) && !fitted.has(`${target}::${camera}`)) { + missing.push({ input, camera, target }); + } + }); + return missing; +} + +export function describeMissingRegistration(missing: MissingRegistration, pipelineName?: string): string { + const where = pipelineName ? ` before running ${pipelineName}` : ''; + return `Camera "${missing.camera}" (input${missing.input}) has no registration onto camera 1 ` + + `("${missing.target}"). Register ${missing.camera} → ${missing.target} in the Camera Registration tab${where}.`; +} + +/** Parse the value of a `# Camera Order:` header into slot tokens. */ +export function parseCameraOrderHeader(value: string): string[] { + return value.trim().split(/[\s,]+/).filter((token) => token); +} diff --git a/client/platform/desktop/backend/native/cameraRegistration.spec.ts b/client/platform/desktop/backend/native/cameraRegistration.spec.ts new file mode 100644 index 000000000..3133f5a2c --- /dev/null +++ b/client/platform/desktop/backend/native/cameraRegistration.spec.ts @@ -0,0 +1,203 @@ +import mockfs from 'mock-fs'; +import npath from 'path'; +import fs from 'fs-extra'; +import { + it, expect, describe, afterAll, vi, +} from 'vitest'; + +import { Settings, JsonConfig } from 'platform/desktop/constants'; +import { buildRegistrationPipelineArgs } from './cameraRegistration'; + +// mock-fs no longer intercepts fs-extra's exists checks on newer Node; +// route them through statSync like common.spec.ts does. +vi.mock('fs-extra', async () => { + const actual = await vi.importActual('fs-extra'); + const fsNode = await import('node:fs'); + const existsByStat = (targetPath: Parameters[0]) => { + try { + fsNode.statSync(targetPath); + return true; + } catch { + return false; + } + }; + + const patchedDefault = { + ...actual.default, + existsSync: existsByStat, + pathExistsSync: existsByStat, + }; + + return { + ...actual, + default: patchedDefault, + existsSync: existsByStat, + pathExistsSync: existsByStat, + }; +}); + +const settings: Settings = { + version: 1, + dataPath: '/home/user/viamedata', + viamePath: '/opt/viame', + readonlyMode: false, + overrides: {}, +}; + +const identity = [[1, 0, 0], [0, 1, 0], [0, 0, 1]]; +const irToRgb = [[1, 0, 5], [0, 1, -3], [0, 0, 1]]; +const uvToRgb = [[1, 0, -8], [0, 1, 2], [0, 0, 1]]; + +function registrationFile(pairs: unknown[]) { + return JSON.stringify({ type: 'dive-camera-registration', version: 1, pairs }); +} + +const projectFiles = { 'meta.json': '{}', 'result_1.json': '{}' }; + +mockfs({ + [npath.join(settings.dataPath, 'DIVE_Projects')]: { + withreg: { + ...projectFiles, + 'ir_to_rgb_registration.json': registrationFile([ + { + left: 'ir', right: 'rgb', points: [], leftToRight: irToRgb, rightToLeft: identity, + }, + ]), + 'uv_to_rgb_registration.json': registrationFile([ + { + left: 'uv', right: 'rgb', points: [], leftToRight: uvToRgb, rightToLeft: identity, + }, + ]), + }, + partial: { + ...projectFiles, + 'ir_to_rgb_registration.json': registrationFile([ + { + left: 'ir', right: 'rgb', points: [], leftToRight: irToRgb, rightToLeft: identity, + }, + ]), + // Non-star pair (two non-reference cameras): explicitly unsupported + // for pipelines even though fitted. + 'uv_to_ir_registration.json': registrationFile([ + { + left: 'uv', right: 'ir', points: [], leftToRight: uvToRgb, rightToLeft: identity, + }, + ]), + }, + noreg: { ...projectFiles }, + seeded: { ...projectFiles }, + }, + '/home/user/job': {}, +}); + +afterAll(() => mockfs.restore()); + +function multiCamMeta(id: string, cameras: string[], defaultDisplay: string): JsonConfig { + return { + version: 1, + type: 'multi', + id, + fps: 1, + originalBasePath: '/home/user/data', + originalImageFiles: [], + originalVideoFile: '', + transcodedVideoFile: '', + transcodedImageFiles: [], + name: id, + createdAt: 'now', + subType: 'multicam', + multiCam: { + cameras: Object.fromEntries(cameras.map((name) => [name, { + type: 'image-sequence', + originalBasePath: `/home/user/data/${name}`, + originalImageFiles: [], + originalVideoFile: '', + transcodedVideoFile: '', + transcodedImageFiles: [], + }])), + defaultDisplay, + }, + } as unknown as JsonConfig; +} + +describe('buildRegistrationPipelineArgs', () => { + it('writes one file per camera pair and pins each warp pair/direction', async () => { + // Display order is irrelevant: the given pipeline order (rgb, uv, ir) + // puts uv on input2 / warp2 and ir on input3 / warp3. + const meta = multiCamMeta('withreg', ['ir', 'rgb', 'uv'], 'rgb'); + const jobWorkDir = '/home/user/job/full'; + await fs.ensureDir(jobWorkDir); + const args = await buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'uv', 'ir']); + + const irPath = npath.join(jobWorkDir, 'ir_to_rgb_registration.json'); + const uvPath = npath.join(jobWorkDir, 'uv_to_rgb_registration.json'); + expect(args).toStrictEqual({ + 'warp2:transformation_file': uvPath, + 'warp2:transform_reader:type': 'dive', + 'warp2:transform_reader:dive:from_camera': 'uv', + 'warp2:transform_reader:dive:to_camera': 'rgb', + 'warp3:transformation_file': irPath, + 'warp3:transform_reader:type': 'dive', + 'warp3:transform_reader:dive:from_camera': 'ir', + 'warp3:transform_reader:dive:to_camera': 'rgb', + }); + const written = await fs.readJSON(irPath); + expect(written.type).toBe('dive-camera-registration'); + expect(written.pairs).toHaveLength(1); + expect(written.pairs[0].left).toBe('ir'); + expect(written.pairs[0].right).toBe('rgb'); + expect(written.pairs[0].leftToRight).toStrictEqual(irToRgb); + expect((await fs.readJSON(uvPath)).pairs[0].left).toBe('uv'); + }); + + it('skips cameras without a fitted reference pair; non-star pairs never reach the job', async () => { + const meta = multiCamMeta('partial', ['rgb', 'ir', 'uv'], 'rgb'); + const jobWorkDir = '/home/user/job/partial'; + await fs.ensureDir(jobWorkDir); + const args = await buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'uv', 'ir']); + // uv (warp2) has only the unsupported uv-to-ir pair, so it gets nothing + // (no warps declared, so that is not an error); ir (warp3) has a + // reference pair. + expect(Object.keys(args).some((key) => key.startsWith('warp2'))).toBe(false); + expect(args['warp3:transform_reader:dive:from_camera']).toBe('ir'); + const irPath = npath.join(jobWorkDir, 'ir_to_rgb_registration.json'); + expect(args['warp3:transformation_file']).toBe(irPath); + // The uv-to-ir pair is dropped from ir's job file too. + const written = await fs.readJSON(irPath); + expect(written.pairs).toHaveLength(1); + expect(written.pairs[0].right).toBe('rgb'); + expect(await fs.readdir(jobWorkDir)).toStrictEqual(['ir_to_rgb_registration.json']); + }); + + it('refuses a warped input whose camera has no registration onto camera 1', async () => { + const meta = multiCamMeta('partial', ['rgb', 'ir', 'uv'], 'rgb'); + const jobWorkDir = '/home/user/job/partial-warps'; + await fs.ensureDir(jobWorkDir); + // warp3 (ir) is fitted; warp2 (uv) is not. + await expect(buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'uv', 'ir'], [3], 'pipe')) + .resolves.toHaveProperty('warp3:transformation_file'); + await expect(buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'uv', 'ir'], [2, 3], 'pipe')) + .rejects.toThrow(/Camera "uv" \(input2\) has no registration onto camera 1 \("rgb"\)/); + }); + + it('returns no args when the dataset has no registration', async () => { + const meta = multiCamMeta('noreg', ['rgb', 'ir'], 'rgb'); + const jobWorkDir = '/home/user/job/noreg'; + await fs.ensureDir(jobWorkDir); + const args = await buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'ir']); + expect(args).toStrictEqual({}); + expect(await fs.readdir(jobWorkDir)).toStrictEqual([]); + }); + + it('falls back to the import-time seed in dataset meta when no files exist', async () => { + const meta = multiCamMeta('seeded', ['rgb', 'ir'], 'rgb'); + meta.cameraHomographies = { + 'ir::rgb': { AtoB: irToRgb, BtoA: identity }, + }; + const jobWorkDir = '/home/user/job/seeded'; + await fs.ensureDir(jobWorkDir); + const args = await buildRegistrationPipelineArgs(settings, meta, jobWorkDir, ['rgb', 'ir']); + expect(args['warp2:transform_reader:dive:from_camera']).toBe('ir'); + expect(args['warp2:transform_reader:dive:to_camera']).toBe('rgb'); + }); +}); diff --git a/client/platform/desktop/backend/native/cameraRegistration.ts b/client/platform/desktop/backend/native/cameraRegistration.ts index 157e247e4..e89381751 100644 --- a/client/platform/desktop/backend/native/cameraRegistration.ts +++ b/client/platform/desktop/backend/native/cameraRegistration.ts @@ -12,12 +12,14 @@ import fs from 'fs-extra'; import { TransformType, DEFAULT_TRANSFORM_TYPE } from 'vue-media-annotator/alignedView/transform'; import { buildPerCameraRegistrationFiles, registrationValuesSummary, filterRegistrationValues, - mergeRegistrationValues, mergeRegistrationSources, CameraRegistrationValues, + mergeRegistrationValues, mergeRegistrationSources, registrationFileName, + CameraRegistrationValues, } from 'vue-media-annotator/alignedView/cameraRegistrationFiles'; import { readTransformMatrix } from 'vue-media-annotator/alignedView/alignedView'; import { invert3, Matrix3 } from 'vue-media-annotator/alignedView/homography'; import { DatasetConfigMutable } from 'dive-common/apispec'; import { referenceCameraName as multicamReferenceCameraName } from 'dive-common/multicamDisplay'; +import { describeMissingRegistration } from 'dive-common/pipelineCameraOrder'; import { RegistrationFileNamePattern, compareRegistrationCandidates, @@ -191,6 +193,59 @@ export async function loadEffectiveRegistration( }; } +/** + * Build the kwiver -s settings that hand a dataset's camera registration to + * a 2-cam/3-cam pipeline. `cameraOrder` is the pipeline's input1..N cameras; + * camera 1 is the frame the pipe warps everything onto, so each other camera + * with a fitted pair onto camera 1 gets a single-pair + * _to__registration.json in the job work dir bound to its + * warpN, with the direction pinned through the reader's from/to config since + * a pair may be stored in either orientation. Pairs between two other + * cameras are unsupported (no transform composition) and never reach the + * job. A warped input (`registrationWarps`) whose camera has no fitted pair + * onto camera 1 is an error before the run, not a missing file at pipe + * configure time. + */ +export async function buildRegistrationPipelineArgs( + settings: Settings, + meta: JsonConfig, + jobWorkDir: string, + cameraOrder: string[], + registrationWarps: number[] = [], + pipelineName = '', +): Promise> { + const args: Record = {}; + const [reference] = cameraOrder; + if (!meta.multiCam || !reference) { + return args; + } + const projectDirInfo = await getValidatedProjectDir(settings, meta.id); + const values = await loadEffectiveRegistration(projectDirInfo.basePath, meta); + const files = buildPerCameraRegistrationFiles(values, reference); + const writes: Promise[] = []; + cameraOrder.slice(1).forEach((camera, offset) => { + const input = offset + 2; + const file = files.find((candidate) => candidate.camera === camera); + const pair = file?.body.pairs.find((candidate) => ( + (candidate.left === reference || candidate.right === reference) + && (candidate.leftToRight || candidate.rightToLeft))); + if (!file || !pair) { + if (registrationWarps.includes(input)) { + throw new Error(describeMissingRegistration({ input, camera, target: reference }, pipelineName)); + } + return; + } + const registrationPath = npath.join(jobWorkDir, registrationFileName(camera, reference)); + writes.push(writeJsonFile(registrationPath, { ...file.body, pairs: [pair] })); + args[`warp${input}:transformation_file`] = registrationPath; + args[`warp${input}:transform_reader:type`] = 'dive'; + args[`warp${input}:transform_reader:dive:from_camera`] = camera; + args[`warp${input}:transform_reader:dive:to_camera`] = reference; + }); + await Promise.all(writes); + return args; +} + /** * Persist camera registration to standalone per-camera files in a dataset * directory, merging partial updates with whatever is already on disk. diff --git a/client/platform/desktop/backend/native/common.ts b/client/platform/desktop/backend/native/common.ts index e1c8ae6e4..6cc912675 100644 --- a/client/platform/desktop/backend/native/common.ts +++ b/client/platform/desktop/backend/native/common.ts @@ -31,6 +31,7 @@ import { FrameMetadataSourcesResponse, } from 'dive-common/apispec'; import { orderedMultiCamCameraNames } from 'dive-common/multicamDisplay'; +import { parseCameraOrderHeader } from 'dive-common/pipelineCameraOrder'; import isFrameMetadataSourceName from 'dive-common/frameMetadata/naming'; import { METADATA_ATTACHMENT_UNAVAILABLE, isFrameMetadataReadableName, @@ -284,11 +285,25 @@ async function extractPipeMetadata(filePath: string): Promise { const lines = await readLines(filePath); let inDescription = false; let fullDescription = ''; + // `process warpN` followed by `:: warp_detections|warp_image` marks an + // input whose camera must be registered onto camera 1. + let lastProcessName: string | null = null; + const registrationWarps: number[] = []; lines.forEach((line) => { const trimmed = line.trim(); if (!trimmed) return; + const processMatch = trimmed.match(/^process\s+(\S+)/); + if (processMatch) { + [, lastProcessName] = processMatch; + } else if (/^::\s*(warp_detections|warp_image)\b/.test(trimmed) && lastProcessName) { + const warpMatch = lastProcessName.match(/^warp(\d+)$/); + if (warpMatch) { + registrationWarps.push(Number.parseInt(warpMatch[1], 10)); + } + } + // --- Description extraction (Multiline) --- if (/^#\s*Description:\s*/i.test(line)) { inDescription = true; @@ -297,7 +312,7 @@ async function extractPipeMetadata(filePath: string): Promise { } if (inDescription) { - if (/^#\s*$/.test(line) || /^#\s*=/.test(line) || /^#\s*(Input|Output|Requires\s+Calibration|Metadata\s+File|Image\s+List\s+Keys?|Calibration\s+Keys?):/i.test(line) || !line.startsWith('#')) { + if (/^#\s*$/.test(line) || /^#\s*=/.test(line) || /^#\s*(Input|Output|Requires\s+Calibration|Metadata\s+File|Image\s+List\s+Keys?|Calibration\s+Keys?|Camera\s+Order):/i.test(line) || !line.startsWith('#')) { inDescription = false; } else { fullDescription += ` ${line.replace(/^#\s*/, '').trim()}`; @@ -353,7 +368,20 @@ async function extractPipeMetadata(filePath: string): Promise { metadata.calibrationKeys = keys; } } + + // `# Camera Order: EO, UV, IR` names the camera role fed to each inputN of + // a 2-cam/3-cam pipe; DIVE matches dataset cameras onto it by name. + const cameraOrderMatch = line.match(/^#\s*Camera\s+Order:\s*(.+)/i); + if (cameraOrderMatch) { + const slots = parseCameraOrderHeader(cameraOrderMatch[1]); + if (slots.length) { + metadata.cameraOrder = slots; + } + } }); + if (registrationWarps.length) { + metadata.registrationWarps = [...new Set(registrationWarps)].sort((a, b) => a - b); + } metadata.description = fullDescription.trim() || undefined; } catch (error) { console.error(`Error while reading ${filePath} metadata`, error); diff --git a/client/platform/desktop/backend/native/multiCamImport.ts b/client/platform/desktop/backend/native/multiCamImport.ts index 9119f7517..48a425e7b 100644 --- a/client/platform/desktop/backend/native/multiCamImport.ts +++ b/client/platform/desktop/backend/native/multiCamImport.ts @@ -12,6 +12,7 @@ import { MultiType, } from 'dive-common/constants'; import { preferEoIrSubfolderOrder } from 'dive-common/components/ImportMultiCamDialog/multicamSubfolderLayout'; +import { inferCameraRoles } from 'dive-common/pipelineCameraOrder'; import { JsonConfig, JsonConfigCurrentVersion, DesktopMediaImportResponse, @@ -371,6 +372,19 @@ async function beginMultiCamImport(args: MultiCamImportArgs): Promise [ + name, + camera.originalImageFiles.length ? camera.originalImageFiles : [camera.originalVideoFile], + ]), + )); + if (Object.keys(cameraRoles).length) { + jsonConfig.cameraRoles = cameraRoles; + } + return { jsonConfig, globPattern: '', diff --git a/client/platform/desktop/backend/native/multiCamUtils.ts b/client/platform/desktop/backend/native/multiCamUtils.ts index 1ddcf1166..fcf0ceb6a 100644 --- a/client/platform/desktop/backend/native/multiCamUtils.ts +++ b/client/platform/desktop/backend/native/multiCamUtils.ts @@ -73,11 +73,24 @@ function getTranscodedMultiCamType(imageListFile: string, jsonConfig: JsonConfig throw new Error(`No associate type for ${imageListFile} in multiCam data`); } -async function writeMultiCamStereoPipelineArgs(jobWorkDir: string, meta: JsonConfig, settings: Settings, utility = false, forceTranscoded = false) { +async function writeMultiCamStereoPipelineArgs( + jobWorkDir: string, + meta: JsonConfig, + settings: Settings, + utility = false, + forceTranscoded = false, + // Explicit input1..N camera order for 2-cam/3-cam pipes; stereo + // measurement keeps the stored left/right order when omitted. + cameraOrder: string[] | undefined = undefined, +) { const argFilePair: Record = {}; const outFiles: Record = {}; if (meta.multiCam && meta.multiCam.cameras) { - const cameraList = Object.entries(meta.multiCam.cameras); + const { cameras } = meta.multiCam; + const cameraNames = cameraOrder + ? cameraOrder.filter((name) => name in cameras) + : Object.keys(cameras); + const cameraList = cameraNames.map((name) => [name, cameras[name]] as const); for (let i = 0; i < cameraList.length; i += 1) { const [key, list] = cameraList[i]; const { originalBasePath } = list; @@ -149,6 +162,7 @@ function getMultiCamUrls( } const multiCamMedia: MultiCamMedia = { cameras: {}, + cameraOrder: projectMetaData.multiCam.cameraOrder, defaultDisplay: projectMetaData.multiCam.defaultDisplay, }; diff --git a/client/platform/desktop/backend/native/viame.ts b/client/platform/desktop/backend/native/viame.ts index f9668b925..faf2524f7 100644 --- a/client/platform/desktop/backend/native/viame.ts +++ b/client/platform/desktop/backend/native/viame.ts @@ -32,6 +32,7 @@ import { jobFileEchoMiddleware, createWorkingDirectory, createCustomWorkingDirectory, splitExt, buildTrainingExitManifest, } from './utils'; +import { buildRegistrationPipelineArgs } from './cameraRegistration'; import { getMultiCamImageFiles, getMultiCamVideoPath, writeMultiCamStereoPipelineArgs, @@ -162,6 +163,22 @@ async function importNewMedia( /** * a node.js implementation of dive_tasks.tasks.run_pipeline */ +/** + * The input1..N camera order for a 2-cam/3-cam run: the order the user + * confirmed in the camera-assignment step (validated against the dataset's + * cameras), else the dataset's camera order as stored. + */ +function multiCamOrderFor(meta: JsonConfig, confirmed?: string[]): string[] { + const cameras = Object.keys(meta.multiCam?.cameras ?? {}); + if (!confirmed?.length) { + return cameras; + } + if ([...confirmed].sort().join('\n') !== [...cameras].sort().join('\n')) { + throw new Error(`Camera assignment [${confirmed.join(', ')}] does not match the dataset cameras [${cameras.join(', ')}]`); + } + return confirmed; +} + async function runPipeline( settings: Settings, runPipelineArgs: RunPipeline, @@ -366,7 +383,13 @@ async function runPipeline( let multiOutFiles: Record; if (meta.multiCam && stereoOrMultiCam) { - const { argFilePair, outFiles } = await writeMultiCamStereoPipelineArgs(jobWorkDir, meta, settings, requiresInput); + const isMultiCamPipeline = multiCamPipelineMarkers.includes(pipeline.type); + // 2-cam/3-cam pipes: which camera feeds which inputN is the order the + // user confirmed before the run. + const multiCamOrder = isMultiCamPipeline + ? multiCamOrderFor(meta, runPipelineArgs.pipelineParams?.cameraOrder) + : undefined; + const { argFilePair, outFiles } = await writeMultiCamStereoPipelineArgs(jobWorkDir, meta, settings, requiresInput, false, multiCamOrder); Object.entries(argFilePair).forEach(([arg, file]) => { command.push(`-s ${arg}="${file}"`); }); @@ -394,6 +417,22 @@ async function runPipeline( command.push(`-s ${key}="${meta.multiCam?.calibration}"`); }); } + if (multiCamOrder) { + // Hand the camera registration to the pipeline's warp processes; a + // warped camera with no registration onto camera 1 fails here, before + // the job exists. + const registrationArgs = await buildRegistrationPipelineArgs( + settings, + meta, + jobWorkDir, + multiCamOrder, + pipeline.metadata?.registrationWarps, + pipeline.name, + ); + Object.entries(registrationArgs).forEach(([arg, value]) => { + command.push(`-s ${arg}="${value}"`); + }); + } } else if (pipeline.type === stereoPipelineMarker) { throw new Error('Attempting to run a multicam pipeline on non multicam data'); } diff --git a/docs/Pipeline-Import-Export.md b/docs/Pipeline-Import-Export.md index 17970d0f0..d4250cdb4 100644 --- a/docs/Pipeline-Import-Export.md +++ b/docs/Pipeline-Import-Export.md @@ -75,6 +75,16 @@ Example: | `# Calibration Keys: [k…]` | Opt-in: binds the dataset's stereo calibration file to each listed KWIVER config key at run time (one `-s =` per key). Keys may be space- or comma-separated. Use this when the pipe's calibration consumer is not the conventional `measurer:calibration_file` / `calibration_reader:file` pair (those are used when the header is unset). Needed because `$CONFIG{global:…}` indirection cannot receive `-s` overrides (macros expand at parse time; `-s` blocks are appended last). | | `# Metadata File: :` | Opt-in: when the dataset has an attached **Metadata File**, DIVE appends a KWIVER override `-s :=` at run time. The same CSV/TXT attachment is also considered for [Frame Metadata](Frame-Metadata.md). Without this header, no metadata file is injected. | | `# Image List Keys: [k…]` | Opt-in: binds the run's per-camera input image list(s) to each listed KWIVER key. Keys may be space- or comma-separated. A key containing `{cam}` is expanded once per camera (1-based), e.g. `stabilizer:image_list{cam}` → `image_list1`, `image_list2`, …. A key without `{cam}` receives camera 1's list only. | +| `# Camera Order: [cam…]` | 2-cam/3-cam pipes only: names the camera role fed to each `inputN`, in order (e.g. `# Camera Order: EO, UV, IR` → `input1` optical, `input2` ultraviolet, `input3` thermal). Role tokens are `EO`, `IR`, `UV` (aliases: eo/rgb/optical/color/vis, ir/thermal/lwir/flir, uv/ultraviolet); any other token (`left`, `right`) matches a camera name literally. See [Multicam camera assignment](#multicam-camera-assignment) for how DIVE places a dataset's cameras onto these slots. Camera 1 is the frame the pipe's warp processes map onto: each other camera's registration (Camera Registration tab) onto camera 1 is written to the job as `_to__registration.json` and bound to `warpN`. | + +### Multicam camera assignment + +Which dataset camera feeds which input of a 2-cam/3-cam pipe is decided in the open, not inferred from display order: + +1. **Camera roles.** Each camera of a multicam dataset carries a sensor role (`eo`, `ir`, `uv`) in `cameraRoles`, inferred once at import from the camera (subfolder) name and, failing that, from tokens in its image file names (KAMERA style `…_rgb.jpg`, `…_ir.tif`, `…_uv.jpg`). Only a unanimous answer is recorded; ambiguous cameras get no role. +2. **Assignment step.** When you run a 2-cam/3-cam pipe, DIVE shows one row per pipeline camera (from the `# Camera Order:` header, or plain `input1..N` when the pipe has none) with the dataset camera it proposes — matched by role when both sides have one, else by name — and you confirm or change it before anything runs. Unfilled or duplicated slots block the run. Confirming a role-labelled slot saves the roles back onto the dataset (uncheck *Save these as the dataset's camera roles* to skip), so a corrected role wins over a misleading name next time and for every other pipeline. +3. **Registration check.** DIVE reads which inputs the pipe warps onto camera 1 (`process warpN :: warp_detections` / `warp_image` in the pipe body) and, in the same dialog, shows for each such row whether the chosen camera has a fitted registration onto camera 1. A missing one blocks the run with "Register X → Y in the Camera Registration tab first" — the pipe never gets to fail at configure time on a missing `registration_cameraN_to_camera1.json`. The same check runs server-side (and on desktop) before the job is created, so CLI runs get the same message. +4. **Job.** The confirmed order is what the job runs with (`cameraOrder` in the pipeline params; visible in the desktop job manifest). API callers that omit it get the dataset's stored camera order, as before. ### Metadata File vs Configuration File diff --git a/server/dive_server/crud_dataset.py b/server/dive_server/crud_dataset.py index ef1dda5fb..694bb52f5 100644 --- a/server/dive_server/crud_dataset.py +++ b/server/dive_server/crud_dataset.py @@ -16,6 +16,7 @@ from dive_server import crud, crud_annotation from dive_tasks import tasks +from dive_tasks.multicam_pipeline import infer_camera_roles from dive_utils import ( TRUTHY_META_VALUES, asbool, @@ -1701,6 +1702,9 @@ def create_multicam( 'folderId': str(child['_id']), 'type': camera_types_by_name[name], } + # Sensor role per camera from its name; the pipeline camera-assignment + # step prefills from it and the user can correct it there. + camera_roles = infer_camera_roles(camera_order) calibration_source_item_id = None json_calibration_item_id = None @@ -1780,6 +1784,7 @@ def create_multicam( else {} ), }, + **({'cameraRoles': camera_roles} if camera_roles else {}), } parent_folder_doc['meta'].setdefault( constants.ConfidenceFiltersMarker, diff --git a/server/dive_server/crud_rpc.py b/server/dive_server/crud_rpc.py index 70f18dfe2..9ae6acebd 100644 --- a/server/dive_server/crud_rpc.py +++ b/server/dive_server/crud_rpc.py @@ -18,7 +18,13 @@ from dive_server import crud, crud_annotation, crud_dataset from dive_tasks import tasks -from dive_tasks.multicam_pipeline import is_stereo_or_multicam_pipeline, pipeline_requires_input +from dive_tasks.multicam_pipeline import ( + build_registration_pairs, + describe_missing_registration, + is_stereo_or_multicam_pipeline, + missing_registrations, + pipeline_requires_input, +) from dive_tasks.utils import choose_annotation_fps from dive_utils import ( TRUTHY_META_VALUES, @@ -320,12 +326,29 @@ def run_pipeline( multicam_default_display = '' calibration_item_id: Optional[str] = None default_camera_folder: Optional[types.GirderModel] = None + is_warp_pipeline = False + reference_camera = '' if dataset_type == constants.MultiType: multi_cam = fromMeta(folder, constants.MultiCamMarker, required=True) multicam_default_display = multi_cam['defaultDisplay'] camera_order = crud_dataset._multicam_camera_order(multi_cam) cameras_meta = multi_cam.get('cameras') or {} + is_warp_pipeline = pipeline['type'] in constants.MultiCamPipelineMarkers + if is_warp_pipeline and camera_order: + # 2-cam/3-cam pipes warp everything onto camera 1; the order the + # user confirmed in the camera-assignment step is which camera + # feeds which inputN. Without one, the dataset's stored order. + confirmed_order = (pipeline_params or {}).get('cameraOrder') + if confirmed_order: + if sorted(confirmed_order) != sorted(camera_order): + raise RestException( + f'Camera assignment [{", ".join(confirmed_order)}] does not match ' + f'the dataset cameras [{", ".join(camera_order)}]', + code=400, + ) + camera_order = list(confirmed_order) + reference_camera = camera_order[0] for name in camera_order: cam_info = cameras_meta[name] folder_id = cam_info.get('folderId') @@ -407,6 +430,36 @@ def run_pipeline( params['multicam_requires_input'] = multicam_requires_input if calibration_item_id: params['calibration_item_id'] = calibration_item_id + if is_warp_pipeline and reference_camera: + # Refuse up front when a warped camera has no registration onto + # camera 1, rather than letting the pipe die at configure time. + fitted_pairs = [ + key + for key, value in ( + (folder.get('meta') or {}).get('cameraHomographies') or {} + ).items() + if value and (value.get('AtoB') or value.get('BtoA')) + ] + missing = missing_registrations( + camera_order, + (pipeline.get('metadata') or {}).get('registrationWarps'), + fitted_pairs, + ) + if missing: + raise RestException( + ' '.join( + describe_missing_registration(*entry, pipeline['name']) for entry in missing + ), + code=400, + ) + registration_pairs = build_registration_pairs(folder.get('meta') or {}) + if any( + pair.get('leftToRight') or pair.get('rightToLeft') for pair in registration_pairs + ): + params['multicam_registration'] = { + 'reference': reference_camera, + 'pairs': registration_pairs, + } if metadata_file_key and metadata_file_item_id: params['metadata_file_key'] = metadata_file_key params['metadata_file_item_id'] = metadata_file_item_id diff --git a/server/dive_tasks/multicam_pipeline.py b/server/dive_tasks/multicam_pipeline.py index 2094520e9..bb2cbd284 100644 --- a/server/dive_tasks/multicam_pipeline.py +++ b/server/dive_tasks/multicam_pipeline.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from pathlib import Path import re import shlex @@ -9,7 +10,7 @@ from dive_tasks.pipeline_creates_dataset import is_disparity_image_pipeline from dive_utils import constants -from dive_utils.types import MulticamCameraJob, PipelineDescription +from dive_utils.types import MulticamCameraJob, MulticamRegistrationJob, PipelineDescription _PIPELINE_INPUT_PATTERN = re.compile(r'utility_|filter_|transcode_|measurement_') @@ -89,6 +90,183 @@ def append_metadata_file_kwiver_settings( command.append(f'-s {shlex.quote(kwiver_key)}={shlex.quote(str(metadata_path))}') +# Sensor-role aliases for camera names; mirrors CAMERA_ROLE_ALIASES in +# client/dive-common/pipelineCameraOrder.ts. +CAMERA_ROLE_ALIASES: Dict[str, Tuple[str, ...]] = { + 'eo': ('eo', 'rgb', 'optical', 'color', 'colour', 'vis', 'visible'), + 'ir': ('ir', 'thermal', 'lwir', 'mwir', 'flir'), + 'uv': ('uv', 'ultraviolet'), +} + + +def infer_camera_role(camera_name: str) -> Optional[str]: + """ + The sensor role (eo / ir / uv) a camera name denotes, or None when it names + none or more than one. Set once at multicam import; the pipeline + camera-assignment step prefills from it and lets the user correct it. + """ + segments = [seg for seg in re.split(r'[^a-z0-9]+', camera_name.lower()) if seg] + roles = { + role for role, aliases in CAMERA_ROLE_ALIASES.items() if any(s in aliases for s in segments) + } + return next(iter(roles)) if len(roles) == 1 else None + + +def infer_camera_roles(camera_names: List[str]) -> Dict[str, str]: + """Roles for a whole rig; cameras that cannot be classified are omitted.""" + return {name: role for name in camera_names for role in [infer_camera_role(name)] if role} + + +def build_registration_pairs(folder_meta: dict) -> List[dict]: + """ + Convert a dataset folder's camera registration meta (cameraHomographies / + cameraCorrespondences / cameraTransformTypes, keyed by directional + "left::right") into dive-camera-registration file pairs (format v2). + + Meta stores each pair's points as observations -- one entry per image + pair, carrying its own points -- so this is the inverse of + registration_output._from_registration_pairs: the store's imageA/imageB + become the file's imageLeft/imageRight, and each point's a/b pair becomes + one `leftX leftY rightX rightY` row. + + VIAME's dive transform reader only consumes the matrices; the + observations travel for provenance and so a file round-trips back into + DIVE without losing which frame contributed what. + """ + homographies = folder_meta.get('cameraHomographies') or {} + correspondences = folder_meta.get('cameraCorrespondences') or {} + transform_types = folder_meta.get('cameraTransformTypes') or {} + keys = set(homographies) | set(correspondences) | set(transform_types) + pairs: List[dict] = [] + for key in sorted(keys): + left, _, right = key.partition('::') + homography = homographies.get(key) + observations = [] + for obs in correspondences.get(key) or []: + observations.append( + { + 'imageLeft': obs.get('imageA'), + 'imageRight': obs.get('imageB'), + 'frame': obs.get('frame'), + 'enabled': obs.get('enabled', True), + 'source': obs.get('source') or 'manual', + **({'stats': obs['stats']} if obs.get('stats') is not None else {}), + 'points': [ + [p['a'][0], p['a'][1], p['b'][0], p['b'][1]] + for p in obs.get('points') or [] + ], + } + ) + pairs.append( + { + 'left': left, + 'right': right, + 'observations': observations, + 'leftToRight': homography.get('AtoB') if homography else None, + 'rightToLeft': homography.get('BtoA') if homography else None, + 'transformType': transform_types.get(key, 'similarity'), + } + ) + return pairs + + +def missing_registrations( + order: List[str], registration_warps: Optional[List[int]], fitted_pairs: List[str] +) -> List[Tuple[int, str, str]]: + """ + (input, camera, camera1) for each warped input whose camera has no fitted + registration onto camera 1. `fitted_pairs` are the dataset's homography + keys (`a::b`, either orientation counts). Checked before a run so the + failure is "register camera X onto Y first" rather than the pipe dying at + configure time on a missing file. + """ + if not order or not registration_warps: + return [] + target = order[0] + fitted = set(fitted_pairs) + missing = [] + for position in registration_warps: + if position < 2 or position > len(order): + continue + camera = order[position - 1] + if camera == target: + continue + if f'{camera}::{target}' not in fitted and f'{target}::{camera}' not in fitted: + missing.append((position, camera, target)) + return missing + + +def describe_missing_registration( + position: int, camera: str, target: str, pipeline_name: str +) -> str: + return ( + f'Camera "{camera}" (input{position}) has no registration onto camera 1 ("{target}"). ' + f'Register {camera} -> {target} in the Camera Registration tab ' + f'before running {pipeline_name}.' + ) + + +def build_registration_kwiver_settings( + work_dir: Path, + cameras: List[MulticamCameraJob], + registration: MulticamRegistrationJob, +) -> Dict[str, str]: + """ + Build the -s settings handing the camera registration to a 2-cam/3-cam + pipeline. One standard _to__registration.json per + non-reference camera is written into the work dir; each camera's warp + process (warp2, warp3, ... matching the job camera order) gets its own + single-pair file. The pair and direction are still pinned via the + reader's from_camera/to_camera config, since a pair may be stored in + either orientation. + + Only pairs registering a camera directly onto the reference are + supported: pairs between two non-reference cameras are explicitly + unsupported here (there is no transform composition) and never reach the + pipeline. Cameras without a fitted reference pair get no settings. + """ + reference = registration.get('reference') + if not reference: + return {} + reference_pairs = [ + pair + for pair in registration.get('pairs') or [] + if reference in (pair['left'], pair['right']) and pair['left'] != pair['right'] + ] + pairs_by_camera: Dict[str, List[dict]] = {} + for pair in reference_pairs: + camera = pair['left'] if pair['right'] == reference else pair['right'] + pairs_by_camera.setdefault(camera, []).append(pair) + fitted = { + (pair['left'], pair['right']) + for pair in reference_pairs + if pair.get('leftToRight') or pair.get('rightToLeft') + } + settings: Dict[str, str] = {} + for index, camera in enumerate(cameras): + name = camera['name'] + if index == 0 or name == reference: + continue + if (name, reference) not in fitted and (reference, name) not in fitted: + continue + camera_pairs = pairs_by_camera.get(name) + if not camera_pairs: + continue + registration_path = work_dir / f'{name}_to_{reference}_registration.json' + with open(registration_path, 'w', encoding='utf-8') as registration_file: + json.dump( + {'type': 'dive-camera-registration', 'version': 2, 'pairs': camera_pairs}, + registration_file, + indent=2, + ) + warp = f'warp{index + 1}' + settings[f'{warp}:transformation_file'] = str(registration_path) + settings[f'{warp}:transform_reader:type'] = 'dive' + settings[f'{warp}:transform_reader:dive:from_camera'] = name + settings[f'{warp}:transform_reader:dive:to_camera'] = reference + return settings + + def build_multicam_kwiver_settings( work_dir: Path, cameras: List[MulticamCameraJob], diff --git a/server/dive_tasks/pipeline_discovery.py b/server/dive_tasks/pipeline_discovery.py index 590ee844c..9975e73a4 100644 --- a/server/dive_tasks/pipeline_discovery.py +++ b/server/dive_tasks/pipeline_discovery.py @@ -181,6 +181,10 @@ def extract_pipe_metadata(file_path: Path) -> PipeMetadata: in_description = False full_description_parts: List[str] = [] + # `process warpN` followed by `:: warp_detections|warp_image` marks an + # input whose camera must be registered onto camera 1. + last_process_name: Optional[str] = None + registration_warps: List[int] = [] try: with open(file_path, 'r', encoding='utf-8') as f: @@ -190,6 +194,16 @@ def extract_pipe_metadata(file_path: Path) -> PipeMetadata: if not trimmed: continue + process_match = re.match(r'^process\s+(\S+)', trimmed) + if process_match: + last_process_name = process_match.group(1) + elif ( + re.match(r'^::\s*(warp_detections|warp_image)\b', trimmed) and last_process_name + ): + warp_match = re.match(r'^warp(\d+)$', last_process_name) + if warp_match: + registration_warps.append(int(warp_match.group(1))) + # --- Description extraction (Multiline) --- desc_start_match = re.match(r'^#\s*Description:\s*(.*)', line_raw, re.IGNORECASE) if desc_start_match: @@ -205,7 +219,7 @@ def extract_pipe_metadata(file_path: Path) -> PipeMetadata: or re.match(r'^#\s*=', line_raw) or re.match( r'^#\s*(Input|Output|Requires\s+Calibration|Metadata\s+File' - r'|Image\s+List\s+Keys?|Calibration\s+Keys?):', + r'|Image\s+List\s+Keys?|Calibration\s+Keys?|Camera\s+Order):', line_raw, re.IGNORECASE, ) @@ -277,6 +291,22 @@ def extract_pipe_metadata(file_path: Path) -> PipeMetadata: if keys: metadata["calibrationKeys"] = keys + # `# Camera Order: EO, UV, IR` names the camera role fed to each + # inputN of a 2-cam/3-cam pipe; DIVE matches dataset cameras onto + # it by name at run time (multicam_pipeline.resolve_pipeline_camera_order). + camera_order_match = re.match( + r'^#\s*Camera\s+Order:\s*(.+)', line_raw, re.IGNORECASE + ) + if camera_order_match: + slots = [ + s for s in re.split(r'[\s,]+', camera_order_match.group(1).strip()) if s + ] + if slots: + metadata["cameraOrder"] = slots + + if registration_warps: + metadata["registrationWarps"] = sorted(set(registration_warps)) + if full_description_parts: metadata["description"] = " ".join(full_description_parts) else: diff --git a/server/dive_tasks/run_pipeline.py b/server/dive_tasks/run_pipeline.py index ab3a4aab2..f7819c263 100644 --- a/server/dive_tasks/run_pipeline.py +++ b/server/dive_tasks/run_pipeline.py @@ -17,6 +17,7 @@ append_metadata_file_kwiver_settings, append_stereo_calibration_kwiver_settings, build_multicam_kwiver_settings, + build_registration_kwiver_settings, find_downloaded_calibration_file, is_stereo_measurement_pipeline, ) @@ -327,6 +328,14 @@ def run_pipeline(self: Task, params: PipelineJob): for arg, file_name in arg_file_pair.items(): command.append(f"-s {shlex.quote(arg)}={shlex.quote(file_name)}") + multicam_registration = multicam_params.get('multicam_registration') + if multicam_registration: + registration_settings = build_registration_kwiver_settings( + _working_directory_path, multicam_cameras, multicam_registration + ) + for arg, value in registration_settings.items(): + command.append(f'-s {shlex.quote(arg)}={shlex.quote(value)}') + transcoded_video: Optional[str] = None if creates_new_dataset: video_name = None diff --git a/server/dive_utils/models.py b/server/dive_utils/models.py index db5039e00..7a4492e98 100644 --- a/server/dive_utils/models.py +++ b/server/dive_utils/models.py @@ -242,6 +242,9 @@ class PairHomography(BaseModel): CameraTransformType = Literal['translation', 'rigid', 'similarity', 'affine', 'homography'] TypeHierarchy = Dict[StrictStr, StrictStr] +# Sensor modality of a multicam camera; see dive_tasks.multicam_pipeline.CAMERA_ROLE_ALIASES. +CameraRole = Literal['eo', 'ir', 'uv'] + class MetadataMutable(BaseModel): version = ( @@ -271,6 +274,10 @@ class MetadataMutable(BaseModel): # DIVE; preserved verbatim so refined calibrations can be traced back to the # model version they were made against. cameraRegistrationSource: Optional[Dict[str, Any]] + # Sensor role per multicam camera name, inferred at import from the camera + # and image names and editable afterwards; used to place cameras onto a + # pipeline's declared camera slots. Cameras with no known role are absent. + cameraRoles: Optional[Dict[str, CameraRole]] fps: Optional[float] @staticmethod diff --git a/server/dive_utils/types.py b/server/dive_utils/types.py index 162187bcf..e5c5489a1 100644 --- a/server/dive_utils/types.py +++ b/server/dive_utils/types.py @@ -92,6 +92,14 @@ class PipeMetadata(TypedDict): # not the conventional `measurer`/`calibration_reader` declare their own keys # here; when unset the two conventional keys are used. calibrationKeys: NotRequired[Optional[list[str]]] + # Camera role per pipeline input for 2-cam/3-cam pipes (e.g. ["EO", "UV", "IR"]), + # parsed from `# Camera Order: [cam...]`. Labels the slots of the client's + # pre-run camera-assignment step; pipes without it show bare input1..N slots. + cameraOrder: NotRequired[Optional[list[str]]] + # Input positions the pipe warps onto camera 1 (`process warpN :: warp_detections | + # warp_image`), e.g. [2, 3]; each such camera needs a fitted registration onto + # camera 1, checked before the run. + registrationWarps: NotRequired[Optional[list[int]]] class PipelineDescription(TypedDict): @@ -114,6 +122,10 @@ class PipelineRuntimeParams(TypedDict, total=False): class PipelineParams(TypedDict, total=False): kwiverParams: Dict[str, str] runtimeParams: PipelineRuntimeParams + # 2-cam/3-cam pipes: the dataset camera to feed each inputN, in order, as + # confirmed by the user before the run. When omitted (API callers) the + # dataset's stored camera order is used. + cameraOrder: List[str] # Name for the newly created dataset (filter / transcode / disparity). outputDatasetName: str # Optional Girder folder that should own the new dataset (else sibling of input). @@ -155,6 +167,17 @@ class PipelineJob(TypedDict): output_parent_folder_id: NotRequired[Optional[str]] +class MulticamRegistrationJob(TypedDict): + """Camera registration handed to a 2-cam/3-cam pipeline's warp processes. + + Pairs use the dive-camera-registration file layout: left/right camera + names, correspondence points, and leftToRight/rightToLeft 3x3 matrices. + """ + + reference: str + pairs: List[dict] + + class MulticamPipelineJob(PipelineJob, total=False): """Pipeline job fields set when running stereo/multicam pipelines on a multi dataset.""" @@ -162,6 +185,7 @@ class MulticamPipelineJob(PipelineJob, total=False): multicam_default_display: str calibration_item_id: Optional[str] multicam_requires_input: bool + multicam_registration: Optional[MulticamRegistrationJob] class TrainingJob(TypedDict): diff --git a/server/tests/test_multicam_pipeline.py b/server/tests/test_multicam_pipeline.py index 0595f2785..19769bf95 100644 --- a/server/tests/test_multicam_pipeline.py +++ b/server/tests/test_multicam_pipeline.py @@ -1,12 +1,18 @@ +import json from pathlib import Path from dive_tasks.multicam_pipeline import ( DEFAULT_CALIBRATION_KEYS, append_stereo_calibration_kwiver_settings, build_multicam_kwiver_settings, + build_registration_kwiver_settings, + build_registration_pairs, find_downloaded_calibration_file, + infer_camera_role, + infer_camera_roles, is_stereo_measurement_pipeline, is_stereo_or_multicam_pipeline, + missing_registrations, pipeline_requires_input, stereo_calibration_keys, ) @@ -120,6 +126,174 @@ def test_build_multicam_kwiver_settings_image_sequence(tmp_path: Path): ) +def test_infer_camera_role(): + assert infer_camera_role('rgb') == 'eo' + assert infer_camera_role('CENT_IR') == 'ir' + assert infer_camera_role('uv_cam') == 'uv' + assert infer_camera_role('eo_ir') is None + assert infer_camera_role('center') is None + assert infer_camera_roles(['rgb', 'CENT_IR', 'center']) == {'rgb': 'eo', 'CENT_IR': 'ir'} + + +def test_missing_registrations(): + order = ['rgb', 'uv', 'ir'] + fitted = ['rgb::ir'] # stored in the reverse orientation still counts + assert missing_registrations(order, [2, 3], fitted) == [(2, 'uv', 'rgb')] + assert missing_registrations(order, [2, 3], fitted + ['uv::rgb']) == [] + # No warps declared (or no order): nothing to check. + assert missing_registrations(order, None, []) == [] + assert missing_registrations([], [2], []) == [] + # Out-of-range warp positions are ignored rather than crashing. + assert missing_registrations(['rgb', 'ir'], [3], []) == [] + + +IR_TO_RGB = [[1, 0, 5], [0, 1, -3], [0, 0, 1]] +RGB_TO_IR = [[1, 0, -5], [0, 1, 3], [0, 0, 1]] + + +def test_build_registration_pairs(): + folder_meta = { + 'cameraHomographies': {'ir::rgb': {'AtoB': IR_TO_RGB, 'BtoA': RGB_TO_IR}}, + 'cameraCorrespondences': { + 'ir::rgb': [ + { + 'imageA': 'ir_0001.png', + 'imageB': 'rgb_0001.jpg', + 'frame': 1, + 'enabled': True, + 'source': 'manual', + 'points': [{'id': 1, 'a': [1, 2], 'b': [3, 4]}], + } + ], + 'uv::rgb': [ + { + 'imageA': 'uv_0002.jpg', + 'imageB': 'rgb_0002.jpg', + 'frame': 2, + 'enabled': True, + 'source': 'minima_loftr', + 'points': [{'id': 1, 'a': [5, 6], 'b': [7, 8]}], + } + ], + }, + 'cameraTransformTypes': {'ir::rgb': 'affine'}, + } + pairs = build_registration_pairs(folder_meta) + assert pairs == [ + { + 'left': 'ir', + 'right': 'rgb', + 'observations': [ + { + 'imageLeft': 'ir_0001.png', + 'imageRight': 'rgb_0001.jpg', + 'frame': 1, + 'enabled': True, + 'source': 'manual', + 'points': [[1, 2, 3, 4]], + } + ], + 'leftToRight': IR_TO_RGB, + 'rightToLeft': RGB_TO_IR, + 'transformType': 'affine', + }, + { + 'left': 'uv', + 'right': 'rgb', + 'observations': [ + { + 'imageLeft': 'uv_0002.jpg', + 'imageRight': 'rgb_0002.jpg', + 'frame': 2, + 'enabled': True, + 'source': 'minima_loftr', + 'points': [[5, 6, 7, 8]], + } + ], + 'leftToRight': None, + 'rightToLeft': None, + 'transformType': 'similarity', + }, + ] + assert build_registration_pairs({}) == [] + + +def test_build_registration_kwiver_settings(tmp_path: Path): + cameras = [ + {'name': 'rgb', 'folder_id': '1', 'media_type': constants.ImageSequenceType}, + {'name': 'ir', 'folder_id': '2', 'media_type': constants.ImageSequenceType}, + {'name': 'uv', 'folder_id': '3', 'media_type': constants.ImageSequenceType}, + ] + registration = { + 'reference': 'rgb', + 'pairs': [ + { + 'left': 'ir', + 'right': 'rgb', + 'observations': [], + 'leftToRight': IR_TO_RGB, + 'rightToLeft': RGB_TO_IR, + 'transformType': 'similarity', + }, + # Points-only pair: uv has nothing fitted, so no warp3 settings. + { + 'left': 'uv', + 'right': 'rgb', + 'observations': [ + { + 'imageLeft': 'uv_0001.jpg', + 'imageRight': 'rgb_0001.jpg', + 'frame': 1, + 'enabled': True, + 'source': 'manual', + 'points': [[1, 2, 3, 4]], + } + ], + 'leftToRight': None, + 'rightToLeft': None, + 'transformType': 'similarity', + }, + # Non-star pair (two non-reference cameras): explicitly + # unsupported, never reaches the pipeline even though fitted. + { + 'left': 'uv', + 'right': 'ir', + 'observations': [], + 'leftToRight': IR_TO_RGB, + 'rightToLeft': RGB_TO_IR, + 'transformType': 'similarity', + }, + ], + } + settings = build_registration_kwiver_settings(tmp_path, cameras, registration) + # One file per camera pair; uv is points-only so it gets no file or settings. + registration_path = str(tmp_path / 'ir_to_rgb_registration.json') + assert settings == { + 'warp2:transformation_file': registration_path, + 'warp2:transform_reader:type': 'dive', + 'warp2:transform_reader:dive:from_camera': 'ir', + 'warp2:transform_reader:dive:to_camera': 'rgb', + } + written = json.loads((tmp_path / 'ir_to_rgb_registration.json').read_text(encoding='utf-8')) + assert written['type'] == 'dive-camera-registration' + # DIVE's format-v2 loader rejects any other version rather than reading + # a matrix-only pair with its points silently dropped; VIAME's dive + # transform reader only consumes the matrices and ignores the version. + assert written['version'] == 2 + assert len(written['pairs']) == 1 + assert written['pairs'][0]['left'] == 'ir' + # uv produced no file: its only fitted pair skips the reference. + assert list(tmp_path.iterdir()) == [tmp_path / 'ir_to_rgb_registration.json'] + + +def test_build_registration_kwiver_settings_empty(tmp_path: Path): + cameras = [{'name': 'rgb', 'folder_id': '1', 'media_type': constants.ImageSequenceType}] + assert ( + build_registration_kwiver_settings(tmp_path, cameras, {'reference': '', 'pairs': []}) == {} + ) + assert list(tmp_path.iterdir()) == [] + + def test_build_multicam_kwiver_settings_video(tmp_path: Path): cameras = [ {'name': 'left', 'folder_id': 'l', 'media_type': constants.VideoType}, diff --git a/server/tests/test_pipeline_discovery.py b/server/tests/test_pipeline_discovery.py index f9feab1df..a0546116c 100644 --- a/server/tests/test_pipeline_discovery.py +++ b/server/tests/test_pipeline_discovery.py @@ -120,6 +120,56 @@ def test_extract_pipe_metadata_calibration_keys(tmp_path: Path): assert metadata['description'] == 'rectified disparity' +def test_extract_pipe_metadata_camera_order(tmp_path: Path): + pipe = tmp_path / 'detector_seal_3-cam.pipe' + pipe.write_text( + '\n'.join( + [ + '# Description: three camera detector', + '# Camera Order: EO, UV, IR', + '# Input: IMAGE (per camera)', + ] + ) + ) + + metadata = extract_pipe_metadata(pipe) + + assert metadata['cameraOrder'] == ['EO', 'UV', 'IR'] + assert 'registrationWarps' not in metadata + # The header must not bleed into the multi-line description. + assert metadata['description'] == 'three camera detector' + assert 'cameraOrder' not in extract_pipe_metadata( + _write(tmp_path, 'detector_plain.pipe', ['# Description: none']) + ) + + +def test_extract_pipe_metadata_registration_warps(tmp_path: Path): + pipe = _write( + tmp_path, + 'detector_seal_3-cam.pipe', + [ + '# Description: three camera detector', + 'process input1', + ' :: video_input', + 'process warp3', + ' :: warp_detections', + ' :transformation_file registration_camera3_to_camera1.json', + 'process warp2', + ' :: warp_detections', + ' :inverse true', + 'process not_a_warp', + ' :: warp_image', + ], + ) + assert extract_pipe_metadata(pipe)['registrationWarps'] == [2, 3] + + +def _write(tmp_path: Path, name: str, lines: list) -> Path: + pipe = tmp_path / name + pipe.write_text('\n'.join(lines)) + return pipe + + def test_extract_pipe_metadata_parses_metadata_file_key(tmp_path: Path): pipe = tmp_path / 'detector_stabilize.pipe' pipe.write_text(