diff --git a/.github/workflows/critical-fixes-ci.yml b/.github/workflows/critical-fixes-ci.yml new file mode 100644 index 0000000..8120cf6 --- /dev/null +++ b/.github/workflows/critical-fixes-ci.yml @@ -0,0 +1,86 @@ +name: Critical fixes CI + +on: + push: + branches: + - fix/critical-job-races + pull_request: + paths: + - 'source app folder/dashboard/**' + - 'source app folder/tracker/**' + - '.github/workflows/critical-fixes-ci.yml' + +permissions: + contents: read + +jobs: + validate-linux: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + cache-dependency-path: source app folder/dashboard/package-lock.json + + - name: Install dashboard dependencies + working-directory: source app folder/dashboard + run: npm ci + + - name: Lint backend and regression tests + working-directory: source app folder/dashboard + run: npm run lint:backend + + - name: Run utility, HTTP race, lifecycle, and metric tests + working-directory: source app folder/dashboard + run: npm run test:server + + - name: Build dashboard + working-directory: source app folder/dashboard + run: npm run build + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + cache: pip + cache-dependency-path: source app folder/tracker/requirements.txt + + - name: Install tracker dependencies + run: python -m pip install -r 'source app folder/tracker/requirements.txt' + + - name: Check Python syntax + run: python -m py_compile 'source app folder/tracker/tracker.py' + + - name: Run tracker scientific regression tests + run: python -m unittest discover -s 'source app folder/tracker/tests' -p 'test_*.py' -v + + validate-windows-node: + runs-on: windows-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + cache-dependency-path: source app folder/dashboard/package-lock.json + + - name: Install dashboard dependencies + working-directory: source app folder/dashboard + run: npm ci + + - name: Lint backend and regression tests + working-directory: source app folder/dashboard + run: npm run lint:backend + + - name: Run Node utility, HTTP, lifecycle, and metric tests + working-directory: source app folder/dashboard + run: npm run test:server + + - name: Build dashboard + working-directory: source app folder/dashboard + run: npm run build diff --git a/source app folder/dashboard/eslint.config.js b/source app folder/dashboard/eslint.config.js index 4fa125d..3922469 100644 --- a/source app folder/dashboard/eslint.config.js +++ b/source app folder/dashboard/eslint.config.js @@ -26,4 +26,10 @@ export default defineConfig([ 'no-unused-vars': ['error', { varsIgnorePattern: '^[A-Z_]' }], }, }, + { + files: ['server*.js', 'tests/**/*.js'], + languageOptions: { + globals: globals.node, + }, + }, ]) diff --git a/source app folder/dashboard/package.json b/source app folder/dashboard/package.json index cecd59c..e9dd113 100644 --- a/source app folder/dashboard/package.json +++ b/source app folder/dashboard/package.json @@ -9,6 +9,10 @@ "dev:server": "node server.js", "build": "vite build", "lint": "eslint .", + "lint:backend": "eslint server.js server-utils.js tests", + "test": "npm run test:server", + "test:server": "node --test tests/*.test.js", + "check": "npm run lint:backend && npm run test:server && npm run build", "preview": "vite preview" }, "dependencies": { diff --git a/source app folder/dashboard/server-utils.js b/source app folder/dashboard/server-utils.js new file mode 100644 index 0000000..01ce2ba --- /dev/null +++ b/source app folder/dashboard/server-utils.js @@ -0,0 +1,665 @@ +import { randomUUID } from 'crypto'; +import { spawn as nodeSpawn } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +export class AbortRunError extends Error { + constructor(message = 'Run aborted') { + super(message); + this.name = 'AbortError'; + } +} + +export function ensureDir(dirPath) { + fs.mkdirSync(dirPath, { recursive: true }); +} + +export function safeUnlink(filePath) { + if (!filePath) return; + try { fs.rmSync(filePath, { force: true }); } catch (error) { + console.error(`[Server] Failed to remove ${filePath}: ${error.message}`); + } +} + +export function safeRemoveDir(dirPath) { + if (!dirPath) return; + try { fs.rmSync(dirPath, { recursive: true, force: true }); } catch (error) { + console.error(`[Server] Failed to remove ${dirPath}: ${error.message}`); + } +} + +export function removeTransactionalFile(filePath) { + if (!filePath) return; + fs.rmSync(filePath, { force: true }); + if (fs.existsSync(filePath)) { + throw new Error(`Could not remove transactional path: ${filePath}`); + } +} + +export function recoverRuntimeArtifacts(uploadsDir) { + ensureDir(uploadsDir); + fs.readdirSync(uploadsDir, { withFileTypes: true }).forEach((entry) => { + const fullPath = path.join(uploadsDir, entry.name); + if (entry.isFile() && entry.name.startsWith('input-')) safeUnlink(fullPath); + else if (entry.isDirectory() && entry.name.startsWith('run-')) safeRemoveDir(fullPath); + }); +} + +export function readJson(filePath, fallback) { + if (!fs.existsSync(filePath)) return fallback; + try { return JSON.parse(fs.readFileSync(filePath, 'utf8')); } catch { return fallback; } +} + +export function writeJson(filePath, data) { + ensureDir(path.dirname(filePath)); + const tempPath = `${filePath}.${process.pid}.${randomUUID()}.tmp`; + fs.writeFileSync(tempPath, JSON.stringify(data, null, 2)); + fs.renameSync(tempPath, filePath); +} + +export function countCsvRows(csvPath) { + const content = fs.readFileSync(csvPath, 'utf8').trim(); + return content ? content.split(/\r?\n/).length - 1 : 0; +} + +export function parseTrackerSync(stdout) { + const matches = [...String(stdout).matchAll( + /TRACKER_SYNC frames_processed=(\d+) csv_rows=(\d+) expected_video_frames=(\d+) sync_ok=(true|false)/g, + )]; + if (matches.length !== 1) return null; + const match = matches[0]; + return { + framesProcessed: Number(match[1]), + csvRows: Number(match[2]), + expectedVideoFrames: Number(match[3]), + syncOk: match[4] === 'true', + }; +} + +function requirePositiveInteger(name, value) { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`Frame integrity evidence missing or invalid: ${name}=${value}`); + } +} + +export function validateFrameIntegrity({ + syncInfo, + inputFrames, + csvRows, + rawVideoFrames, + finalVideoFrames, +}) { + if (!syncInfo) throw new Error('Frame integrity evidence missing: TRACKER_SYNC marker required'); + if (syncInfo.syncOk !== true) throw new Error('Tracker reported frame synchronization failure'); + + const counts = { + inputFrames, + trackerFrames: syncInfo.framesProcessed, + trackerCsvRows: syncInfo.csvRows, + csvRows, + rawVideoFrames, + finalVideoFrames, + }; + Object.entries(counts).forEach(([name, value]) => requirePositiveInteger(name, value)); + + if (!Number.isInteger(syncInfo.expectedVideoFrames) || syncInfo.expectedVideoFrames < 0) { + throw new Error( + `Frame integrity evidence missing or invalid: trackerExpectedFrames=${syncInfo.expectedVideoFrames}`, + ); + } + + const unique = new Set(Object.values(counts)); + if (unique.size !== 1) { + const detail = Object.entries(counts).map(([key, value]) => `${key}=${value}`).join(', '); + throw new Error(`Frame integrity mismatch (${detail})`); + } + + return { + ...counts, + trackerExpectedFrames: syncInfo.expectedVideoFrames, + expectedMetadataMatches: ( + syncInfo.expectedVideoFrames === 0 || syncInfo.expectedVideoFrames === inputFrames + ), + syncOk: true, + }; +} + +const terminationPromises = new WeakMap(); + +function childHasExited(child) { + return Boolean(child) && ( + (child.exitCode !== null && child.exitCode !== undefined) + || Boolean(child.signalCode) + ); +} + +export function terminateChild(child, { graceMs = 1000, hardKillMs = 3000 } = {}) { + if (!child || childHasExited(child)) return Promise.resolve(); + if (terminationPromises.has(child)) return terminationPromises.get(child); + + const promise = new Promise((resolve, reject) => { + let settled = false; + let graceTimer; + let hardTimer; + const cleanup = () => { + clearTimeout(graceTimer); + clearTimeout(hardTimer); + child.removeListener('close', finish); + child.removeListener('error', onError); + }; + const finish = () => { + if (settled) return; + settled = true; + cleanup(); + resolve(); + }; + const fail = (error) => { + if (settled) return; + settled = true; + cleanup(); + reject(error); + }; + const onError = (error) => { + if (childHasExited(child)) finish(); + else fail(new Error(`Child process termination failed before close: ${error.message}`)); + }; + const sendSignal = (signal) => { + try { + const delivered = child.kill(signal); + if (delivered === false && !childHasExited(child)) { + fail(new Error(`Could not deliver ${signal} to child process`)); + return false; + } + } catch (error) { + if (childHasExited(child)) finish(); + else fail(new Error(`Could not deliver ${signal} to child process: ${error.message}`)); + return false; + } + return true; + }; + + child.once('close', finish); + child.once('error', onError); + if (!sendSignal('SIGTERM')) return; + graceTimer = setTimeout(() => { + if (settled || childHasExited(child)) { + if (childHasExited(child)) finish(); + return; + } + if (!sendSignal('SIGKILL')) return; + hardTimer = setTimeout(() => { + if (childHasExited(child)) finish(); + else fail(new Error('Child process did not close after SIGKILL')); + }, hardKillMs); + hardTimer.unref?.(); + }, graceMs); + graceTimer.unref?.(); + }).finally(() => terminationPromises.delete(child)); + + terminationPromises.set(child, promise); + return promise; +} + +export function runCommand(command, args, { + spawnFn = nodeSpawn, + signal, + children, + onStdout, + onStderr, + killOptions, +} = {}) { + if (signal?.aborted) return Promise.reject(new AbortRunError()); + + return new Promise((resolve, reject) => { + let child; + try { + child = spawnFn(command, args); + } catch (error) { + reject(error); + return; + } + children?.add(child); + let stdout = ''; + let stderr = ''; + let settled = false; + let processError = null; + + const cleanup = ({ removeChild = true } = {}) => { + if (removeChild) children?.delete(child); + signal?.removeEventListener('abort', onAbort); + }; + const settle = (handler, value, options) => { + if (settled) return; + settled = true; + cleanup(options); + handler(value); + }; + const onAbort = () => { + terminateChild(child, killOptions) + .then(() => settle(reject, new AbortRunError(), { removeChild: true })) + .catch((error) => settle(reject, error, { removeChild: childHasExited(child) })); + }; + + child.stdout?.on('data', (chunk) => { + const text = chunk.toString(); + stdout += text; + onStdout?.(text); + }); + child.stderr?.on('data', (chunk) => { + const text = chunk.toString(); + stderr += text; + onStderr?.(text); + }); + child.once('error', (error) => { + processError = error; + if (!child.pid || childHasExited(child)) { + settle(reject, error, { removeChild: true }); + } + }); + child.once('close', (code, closeSignal) => { + children?.delete(child); + if (settled) return; + if (signal?.aborted) settle(reject, new AbortRunError(), { removeChild: true }); + else if (processError) settle(reject, processError, { removeChild: true }); + else settle(resolve, { code, signal: closeSignal, stdout, stderr }, { removeChild: true }); + }); + signal?.addEventListener('abort', onAbort, { once: true }); + if (signal?.aborted) onAbort(); + }); +} + +export async function getVideoFrameCount(ffmpegPath, videoPath, options = {}) { + const result = await runCommand(ffmpegPath, [ + '-hide_banner', '-i', videoPath, '-map', '0:v:0', '-f', 'null', '-', + ], options); + if (result.code !== 0) { + throw new Error(`ffmpeg frame count failed (exit ${result.code}): ${result.stderr.slice(-500)}`); + } + const matches = [...result.stderr.matchAll(/frame=\s*(\d+)/g)]; + if (!matches.length) throw new Error(`Could not determine video frame count for ${videoPath}`); + const count = Number(matches.at(-1)[1]); + requirePositiveInteger('decodedVideoFrames', count); + return count; +} + +export async function transcodeVideo(ffmpegPath, rawPath, finalPath, options = {}) { + const result = await runCommand(ffmpegPath, [ + '-hide_banner', '-i', rawPath, '-vcodec', 'libx264', '-preset', 'veryfast', + '-pix_fmt', 'yuv420p', '-movflags', '+faststart', '-an', '-y', finalPath, + ], options); + if (result.code !== 0) { + throw new Error(`ffmpeg transcode failed (exit ${result.code}): ${result.stderr.slice(-500)}`); + } +} + +export class SimulatedProcessCrash extends Error { + constructor(message = 'Simulated process crash') { + super(message); + this.name = 'SimulatedProcessCrash'; + } +} + +function validatePublishToken(token) { + if (!/^[A-Za-z0-9-]+$/.test(String(token))) { + throw new Error(`Invalid publication token: ${token}`); + } +} + +function publishManifestPath(entries, token, manifestDir) { + if (!entries.length) throw new Error('Publication bundle is empty'); + validatePublishToken(token); + const directory = manifestDir || path.dirname(entries[0].destination); + ensureDir(directory); + return path.join(directory, `.flyt-publish-${token}.json`); +} + +function validatePublishManifest(manifest, manifestPath) { + if (!manifest || manifest.version !== 1 || !Array.isArray(manifest.entries)) { + throw new Error(`Invalid publication transaction manifest: ${manifestPath}`); + } + if (!['prepared', 'publishing', 'committed'].includes(manifest.state)) { + throw new Error(`Invalid publication transaction state in ${manifestPath}`); + } + manifest.entries.forEach((entry) => { + for (const key of ['destination', 'stage', 'backup']) { + if (typeof entry[key] !== 'string' || !path.isAbsolute(entry[key])) { + throw new Error(`Invalid ${key} in publication transaction ${manifestPath}`); + } + } + if (typeof entry.hadDestination !== 'boolean') { + throw new Error(`Invalid hadDestination in publication transaction ${manifestPath}`); + } + }); + return manifest; +} + +function rollbackPublishManifest(manifest, { removeFile = removeTransactionalFile } = {}) { + const errors = []; + [...manifest.entries].reverse().forEach((entry) => { + try { + if (entry.hadDestination) { + if (fs.existsSync(entry.backup)) { + removeFile(entry.destination); + ensureDir(path.dirname(entry.destination)); + fs.renameSync(entry.backup, entry.destination); + } else if (!fs.existsSync(entry.destination)) { + throw new Error(`Cannot restore missing prior destination: ${entry.destination}`); + } + } else { + removeFile(entry.destination); + removeFile(entry.backup); + } + removeFile(entry.stage); + } catch (error) { + errors.push(error); + } + }); + if (errors.length) { + throw new globalThis.AggregateError(errors, 'Publication transaction rollback failed'); + } +} + +function finalizeCommittedManifest(manifest, { removeFile = removeTransactionalFile } = {}) { + const missing = manifest.entries.filter((entry) => !fs.existsSync(entry.destination)); + if (missing.length) { + rollbackPublishManifest(manifest, { removeFile }); + return; + } + manifest.entries.forEach((entry) => { + removeFile(entry.stage); + removeFile(entry.backup); + }); +} + +function listFilesRecursive(directory) { + if (!fs.existsSync(directory)) return []; + const files = []; + fs.readdirSync(directory, { withFileTypes: true }).forEach((entry) => { + const fullPath = path.join(directory, entry.name); + if (entry.isDirectory()) files.push(...listFilesRecursive(fullPath)); + else if (entry.isFile()) files.push(fullPath); + }); + return files; +} + +export function publishBundle(entries, token, { + faultInjector, + manifestDir, + removeFile = removeTransactionalFile, +} = {}) { + const manifestPath = publishManifestPath(entries, token, manifestDir); + const manifest = { + version: 1, + token, + state: 'prepared', + entries: entries.map(({ source, destination }) => ({ + source, + destination: path.resolve(destination), + stage: path.resolve(`${destination}.${token}.new`), + backup: path.resolve(`${destination}.${token}.bak`), + hadDestination: fs.existsSync(destination), + })), + }; + + try { + manifest.entries.forEach((entry) => { + if (!fs.existsSync(entry.source)) throw new Error(`Missing publish source: ${entry.source}`); + ensureDir(path.dirname(entry.destination)); + fs.copyFileSync(entry.source, entry.stage); + }); + writeJson(manifestPath, manifest); + faultInjector?.('prepared', -1); + + manifest.entries.forEach((entry, index) => { + if (entry.hadDestination) fs.renameSync(entry.destination, entry.backup); + faultInjector?.('backed-up', index); + }); + manifest.state = 'publishing'; + writeJson(manifestPath, manifest); + + manifest.entries.forEach((entry, index) => { + fs.renameSync(entry.stage, entry.destination); + faultInjector?.('published', index); + }); + manifest.state = 'committed'; + writeJson(manifestPath, manifest); + faultInjector?.('committed', -1); + + finalizeCommittedManifest(manifest, { removeFile }); + removeFile(manifestPath); + } catch (error) { + if (error instanceof SimulatedProcessCrash) throw error; + if (manifest.state === 'committed') throw error; + try { + rollbackPublishManifest(manifest, { removeFile }); + removeFile(manifestPath); + } catch (rollbackError) { + throw new globalThis.AggregateError([error, rollbackError], 'Publication failed and rollback was incomplete'); + } + throw error; + } +} + +export function recoverPublishArtifacts(directory, { removeFile = removeTransactionalFile } = {}) { + if (!fs.existsSync(directory)) return; + const manifests = fs.readdirSync(directory) + .filter((name) => /^\.flyt-publish-[A-Za-z0-9-]+\.json$/.test(name)) + .map((name) => path.join(directory, name)); + + manifests.forEach((manifestPath) => { + const manifest = validatePublishManifest(readJson(manifestPath, null), manifestPath); + if (manifest.state === 'committed') finalizeCommittedManifest(manifest, { removeFile }); + else rollbackPublishManifest(manifest, { removeFile }); + removeFile(manifestPath); + }); + + const artifacts = listFilesRecursive(directory); + const orphanBackups = artifacts.filter((filePath) => filePath.endsWith('.bak')); + if (orphanBackups.length) { + throw new Error( + `Unrecoverable publication backups without a transaction manifest: ${orphanBackups.join(', ')}`, + ); + } + artifacts.filter((filePath) => filePath.endsWith('.new')).forEach((filePath) => removeFile(filePath)); +} + +export function buildTrackerArgs(trackerScript, inputPath, outputs, overrides = {}, defaults) { + const numberOr = (value, fallback) => { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback; + }; + const integerOr = (value, fallback, minimum = 0) => ( + Math.max(minimum, Math.floor(numberOr(value, fallback))) + ); + const config = { + minArea: integerOr(overrides.minArea, defaults.minArea), + maxArea: integerOr(overrides.maxArea, defaults.maxArea), + proximityThreshold: numberOr(overrides.proximityThreshold, defaults.proximityThreshold), + boutMinFrames: integerOr(overrides.boutMinFrames, defaults.boutMinFrames, 1), + }; + if (config.maxArea > 0 && config.maxArea <= config.minArea) config.maxArea = defaults.maxArea; + return [ + trackerScript, '--input', inputPath, + '--output-video', outputs.rawVideo, + '--output-csv', outputs.csv, + '--output-events', outputs.events, + '--min-area', String(config.minArea), + '--max-area', String(config.maxArea), + '--proximity-threshold', String(config.proximityThreshold), + '--bout-min-frames', String(config.boutMinFrames), + ]; +} + +export function readEventsFps(eventsPath) { + const data = readJson(eventsPath, null); + return data && Number.isFinite(data.fps) ? data.fps : null; +} + +export function readCsvAvgProximity(csvPath) { + if (!fs.existsSync(csvPath)) return null; + const lines = fs.readFileSync(csvPath, 'utf8').trim().split(/\r?\n/); + if (lines.length < 2) return null; + const headers = lines[0].split(','); + const proximityIndex = headers.indexOf('proximity_distance'); + const occlusionIndex = headers.indexOf('occlusion_flag'); + const trackingValidIndex = headers.indexOf('tracking_valid'); + const detectionCountIndex = headers.indexOf('detection_count'); + const fly1AreaIndex = headers.indexOf('fly1_area'); + const fly2AreaIndex = headers.indexOf('fly2_area'); + if (proximityIndex < 0) return null; + + let total = 0; + let count = 0; + lines.slice(1).forEach((line) => { + const columns = line.split(','); + const rawProximity = columns[proximityIndex]?.trim(); + if (!rawProximity) return; + const proximity = Number(rawProximity); + const occluded = occlusionIndex >= 0 && Number(columns[occlusionIndex]) !== 0; + let trackingValid = true; + if (trackingValidIndex >= 0) { + trackingValid = Number(columns[trackingValidIndex]) === 1; + } else if (detectionCountIndex >= 0) { + trackingValid = Number(columns[detectionCountIndex]) >= 2; + } else if (fly1AreaIndex >= 0 && fly2AreaIndex >= 0) { + trackingValid = Number(columns[fly1AreaIndex]) > 0 && Number(columns[fly2AreaIndex]) > 0; + } + if (Number.isFinite(proximity) && trackingValid && !occluded) { + total += proximity; + count += 1; + } + }); + return count ? Math.round((total / count) * 100) / 100 : null; +} + +export function countCourtshipBouts(eventsPath) { + const data = readJson(eventsPath, { events: [] }); + return Array.isArray(data.events) + ? data.events.filter((event) => event.type === 'courtship_bout').length + : 0; +} + +const RUN_ID_PATTERN = /^run-[A-Za-z0-9][A-Za-z0-9-]{2,127}$/; + +function requireRunId(runId) { + if (!RUN_ID_PATTERN.test(String(runId))) throw new Error(`Invalid run ID: ${runId}`); + return String(runId); +} + +export function createRunId() { + return `run-${Date.now()}-${randomUUID().slice(0, 8)}`; +} + +export function scopeEventsForRun(eventsPath, runId, destinationPath = eventsPath) { + const safeRunId = requireRunId(runId); + const prefix = `${safeRunId}:`; + const data = readJson(eventsPath, { version: 1, events: [] }); + data.run_id = safeRunId; + data.events = Array.isArray(data.events) ? data.events.map((event) => { + const currentId = String(event.id || ''); + const originalId = event.original_id + || (currentId.startsWith(prefix) ? currentId.slice(prefix.length) : currentId); + if (!originalId) throw new Error('Event ID missing while scoping run events'); + return { + ...event, + original_id: originalId, + id: `${safeRunId}:${originalId}`, + }; + }) : []; + writeJson(destinationPath, data); + return data; +} + +export function scopeEventsForHistory(eventsPath, runId, destinationPath) { + return scopeEventsForRun(eventsPath, runId, destinationPath); +} + +export function prepareRunHistoryBundle({ + runId, + historyDir, + historyMetaPath, + historyIndexOutput, + filename, + durationSec, + fps, + totalFrames, + csvSrc, + eventsSrc, + verificationSrc, +}) { + const safeRunId = requireRunId(runId); + for (const source of [csvSrc, eventsSrc, verificationSrc]) { + if (!fs.existsSync(source)) throw new Error(`Missing history source: ${source}`); + } + const history = readJson(historyMetaPath, { version: 1, runs: [] }); + if (!Array.isArray(history.runs)) history.runs = []; + if (history.runs.some((run) => run.runId === safeRunId)) { + throw new Error(`History run already exists: ${safeRunId}`); + } + const meta = { + runId: safeRunId, + timestamp: new Date().toISOString(), + filename: filename || 'unknown', + durationSec: Math.round((durationSec || 0) * 10) / 10, + fps: fps || null, + totalFrames: totalFrames || null, + avgProximity: readCsvAvgProximity(csvSrc), + detectedBouts: countCourtshipBouts(eventsSrc), + }; + history.runs.unshift(meta); + writeJson(historyIndexOutput, history); + + const runDir = path.join(historyDir, safeRunId); + return { + meta, + entries: [ + { source: csvSrc, destination: path.join(runDir, 'data.csv') }, + { source: eventsSrc, destination: path.join(runDir, 'events.json') }, + { source: verificationSrc, destination: path.join(runDir, 'verification.json') }, + { source: historyIndexOutput, destination: historyMetaPath }, + ], + }; +} + +export function runIdFromEventId(eventId) { + const value = String(eventId); + const separator = value.indexOf(':'); + if (separator <= 0 || separator === value.length - 1) { + throw new Error('Run-scoped event ID required'); + } + return requireRunId(value.slice(0, separator)); +} + +export function verificationPathForRun(runId, historyDir, { mustExist = true } = {}) { + const safeRunId = requireRunId(runId); + const root = path.resolve(historyDir); + const filePath = path.resolve(root, safeRunId, 'verification.json'); + if (!filePath.startsWith(`${root}${path.sep}`)) throw new Error('Verification path escaped history root'); + if (mustExist && !fs.existsSync(filePath)) throw new Error('Run verification file not found'); + return filePath; +} + +export function eventsPathForRun(runId, historyDir, { mustExist = true } = {}) { + const safeRunId = requireRunId(runId); + const root = path.resolve(historyDir); + const filePath = path.resolve(root, safeRunId, 'events.json'); + if (!filePath.startsWith(`${root}${path.sep}`)) throw new Error('Events path escaped history root'); + if (mustExist && !fs.existsSync(filePath)) throw new Error('Run events file not found'); + return filePath; +} + +export function assertEventBelongsToRun(eventId, historyDir) { + const value = String(eventId); + const runId = runIdFromEventId(value); + const events = readJson(eventsPathForRun(runId, historyDir), null); + if (!events || !Array.isArray(events.events)) throw new Error('Run events file is invalid'); + if (events.run_id && events.run_id !== runId) throw new Error('Run events file has a mismatched run ID'); + if (!events.events.some((event) => String(event.id) === value)) { + throw new Error('Event does not belong to the referenced run'); + } + return runId; +} + +export function resolveVerificationPath(eventId, historyDir) { + return verificationPathForRun(runIdFromEventId(eventId), historyDir); +} diff --git a/source app folder/dashboard/server.js b/source app folder/dashboard/server.js index cbff4d2..5c33429 100644 --- a/source app folder/dashboard/server.js +++ b/source app folder/dashboard/server.js @@ -1,94 +1,50 @@ +import { randomUUID } from 'crypto'; +import { spawn as nodeSpawn } from 'child_process'; import express from 'express'; +import defaultFfmpegPath from 'ffmpeg-static'; +import fs from 'fs'; import multer from 'multer'; -import { spawn } from 'child_process'; import path from 'path'; -import fs from 'fs'; import { fileURLToPath } from 'url'; -import ffmpegPath from 'ffmpeg-static'; - -const __filename = fileURLToPath(import.meta.url); -const __dirname = path.dirname(__filename); - -const app = express(); -const PORT = 3001; - -// Configure multer for file uploads -const uploadsDir = path.join(__dirname, 'uploads'); -if (!fs.existsSync(uploadsDir)) { - fs.mkdirSync(uploadsDir, { recursive: true }); -} -// History snapshots directory (created lazily on first snapshot, but ensure now -// so GET /api/history works even before any run completes). -const publicDirTmp = path.join(__dirname, 'public'); -const historyDirInit = path.join(publicDirTmp, 'history'); -if (!fs.existsSync(historyDirInit)) { - fs.mkdirSync(historyDirInit, { recursive: true }); -} - -const storage = multer.diskStorage({ - destination: (req, file, cb) => cb(null, uploadsDir), - filename: (req, file, cb) => cb(null, 'input_video' + path.extname(file.originalname)) -}); -const upload = multer({ storage, limits: { fileSize: 10 * 1024 * 1024 * 1024 } }); // 10GB limit - -// Paths -const trackerDir = path.join(__dirname, '..', 'tracker'); -const trackerScript = path.join(trackerDir, 'tracker.py'); -// Platform-agnostic venv python: Windows → venv/Scripts/python.exe, POSIX → venv/bin/python -const isWindows = process.platform === 'win32'; -const pythonExe = path.join(trackerDir, 'venv', isWindows ? 'Scripts' : 'bin', isWindows ? 'python.exe' : 'python'); -const publicDir = path.join(__dirname, 'public'); - -// Tracker defaults. Overridable per-run via Settings UI → /api/upload body. -// These mirror the pitch baseline; kept here so an empty Settings payload still -// produces pitch-identical output. -const TRACKER_DEFAULTS = { - minArea: 30, - maxArea: 0, - proximityThreshold: 60, - boutMinFrames: 90, -}; - -const EVENTS_PATH = path.join(publicDir, 'events.json'); -const VERIFICATION_PATH = path.join(publicDir, 'verification.json'); - -app.use(express.json()); - -// Enable CORS for dev -app.use((req, res, next) => { - res.header('Access-Control-Allow-Origin', '*'); - res.header('Access-Control-Allow-Methods', 'GET, POST'); - res.header('Access-Control-Allow-Headers', 'Content-Type'); - next(); -}); - -// ============================================================ -// ASYNC JOB SYSTEM — never blocks the HTTP request -// ============================================================ -let runId = 0; - -let job = { - runId: null, - status: 'idle', // idle | uploading | processing | done | error - progress: '', // human-readable progress string - framesProcessed: 0, - totalFrames: null, - frameCount: null, // verified frame count after sync check - error: null, - startTime: null, - endTime: null, -}; - -let activeTracker = null; -let trackerStdout = ''; - -function resetJob() { - if (activeTracker) { - try { activeTracker.kill(); } catch { /* ignore */ } - activeTracker = null; - } - trackerStdout = ''; - job = { +import { + AbortRunError, + assertEventBelongsToRun, + buildTrackerArgs, + createRunId, + countCsvRows, + ensureDir, + getVideoFrameCount, + parseTrackerSync, + prepareRunHistoryBundle, + publishBundle, + readEventsFps, + readJson, + recoverPublishArtifacts, + recoverRuntimeArtifacts, + resolveVerificationPath, + scopeEventsForRun, + runCommand, + safeRemoveDir, + safeUnlink, + verificationPathForRun, + terminateChild, + transcodeVideo, + validateFrameIntegrity, + writeJson, +} from './server-utils.js'; + +const modulePath = fileURLToPath(import.meta.url); +const moduleDir = path.dirname(modulePath); +const DEFAULT_PORT = 3001; +const DEFAULT_ALLOWED_ORIGINS = [ + 'http://localhost:3001', + 'http://127.0.0.1:3001', + 'http://localhost:5173', + 'http://127.0.0.1:5173', +]; + +function blankJob() { + return { runId: null, status: 'idle', progress: '', @@ -101,498 +57,548 @@ function resetJob() { }; } -function countCsvRows(csvPath) { - const content = fs.readFileSync(csvPath, 'utf8').trim(); - if (!content) return 0; - return content.split('\n').length - 1; +function parseAllowedOrigins(value) { + if (!value) return DEFAULT_ALLOWED_ORIGINS; + return value.split(',').map((item) => item.trim()).filter(Boolean); } -function getVideoFrameCount(videoPath) { - return new Promise((resolve, reject) => { - // Decode to null muxer — frame= count appears in stderr (copy mode skips it). - const ffprobe = spawn(ffmpegPath, [ - '-i', videoPath, - '-map', '0:v:0', - '-f', 'null', - '-', - ]); - - let stderr = ''; - ffprobe.stderr.on('data', (data) => { stderr += data.toString(); }); - - ffprobe.on('close', () => { - const matches = [...stderr.matchAll(/frame=\s*(\d+)/g)]; - if (matches.length > 0) { - resolve(parseInt(matches[matches.length - 1][1], 10)); - return; - } - reject(new Error('Could not determine video frame count from ffmpeg')); - }); - - ffprobe.on('error', (err) => reject(err)); +export function createFlytServer(options = {}) { + const rootDir = options.rootDir || moduleDir; + const uploadsDir = options.uploadsDir || path.join(rootDir, 'uploads'); + const publicDir = options.publicDir || path.join(rootDir, 'public'); + const historyDir = options.historyDir || path.join(publicDir, 'history'); + const historyMetaPath = options.historyMetaPath || path.join(publicDir, 'history.json'); + const verificationPath = options.verificationPath || path.join(publicDir, 'verification.json'); + const trackerDir = options.trackerDir || path.join(rootDir, '..', 'tracker'); + const trackerScript = options.trackerScript || path.join(trackerDir, 'tracker.py'); + const isWindows = process.platform === 'win32'; + const pythonExe = options.pythonExe || path.join( + trackerDir, + 'venv', + isWindows ? 'Scripts' : 'bin', + isWindows ? 'python.exe' : 'python', + ); + const ffmpegPath = options.ffmpegPath || defaultFfmpegPath; + const spawnFn = options.spawnFn || nodeSpawn; + const defaults = options.defaults || { + minArea: 30, + maxArea: 0, + proximityThreshold: 60, + boutMinFrames: 90, + }; + const allowedOrigins = new Set(options.allowedOrigins || parseAllowedOrigins(process.env.FLYT_ALLOWED_ORIGINS)); + const killOptions = options.killOptions || { graceMs: 1000, hardKillMs: 3000 }; + + for (const dir of [uploadsDir, publicDir, historyDir]) ensureDir(dir); + recoverRuntimeArtifacts(uploadsDir); + recoverPublishArtifacts(publicDir); + + const app = express(); + const storage = multer.diskStorage({ + destination: (_req, _file, callback) => callback(null, uploadsDir), + filename: (req, file, callback) => { + const filename = `input-${Date.now()}-${randomUUID()}${path.extname(file.originalname).toLowerCase()}`; + const state = pendingUploads.get(req); + if (state) state.filePath = path.join(uploadsDir, filename); + callback(null, filename); + }, + }); + const upload = multer({ storage, limits: { fileSize: 10 * 1024 * 1024 * 1024 } }); + + app.use(express.json()); + app.use((req, res, next) => { + const origin = req.get('Origin'); + if (!origin) return next(); + if (!allowedOrigins.has(origin)) { + return res.status(403).json({ error: 'Cross-origin request denied.' }); + } + res.header('Access-Control-Allow-Origin', origin); + res.header('Vary', 'Origin'); + res.header('Access-Control-Allow-Methods', 'GET, POST, DELETE, OPTIONS'); + res.header('Access-Control-Allow-Headers', 'Content-Type'); + if (req.method === 'OPTIONS') return res.sendStatus(204); + return next(); }); -} -function parseTrackerSync(stdout) { - const match = stdout.match( - /TRACKER_SYNC frames_processed=(\d+) csv_rows=(\d+) expected_video_frames=(\d+) sync_ok=(true|false)/ - ); - if (!match) return null; - return { - framesProcessed: parseInt(match[1], 10), - csvRows: parseInt(match[2], 10), - expectedVideoFrames: parseInt(match[3], 10), - syncOk: match[4] === 'true', + let epoch = 0; + let uploadReserved = false; + let terminating = false; + let activeRun = null; + let stoppingRun = null; + let job = blankJob(); + const pendingUploads = new Map(); + const observedStoppingChildren = new WeakSet(); + + const services = { + countFrames: options.services?.countFrames || ((filePath, context) => getVideoFrameCount( + ffmpegPath, + filePath, + { + spawnFn, + signal: context.controller.signal, + children: context.children, + killOptions, + }, + )), + runTracker: options.services?.runTracker || ((inputPath, outputs, overrides, context) => runCommand( + pythonExe, + buildTrackerArgs(trackerScript, inputPath, outputs, overrides, defaults), + { + spawnFn, + signal: context.controller.signal, + children: context.children, + killOptions, + onStdout: (text) => { + const match = text.match(/Processed (\d+) frames/); + if (match && isCurrent(context)) { + job.framesProcessed = Number(match[1]); + job.progress = `Processed ${job.framesProcessed} frames...`; + } else if (text.includes('Tracking completed') && isCurrent(context)) { + job.progress = 'Validating tracker output...'; + } + }, + onStderr: (text) => console.error(`[Tracker] ${text.trim()}`), + }, + )), + transcode: options.services?.transcode || ((rawPath, finalPath, context) => transcodeVideo( + ffmpegPath, + rawPath, + finalPath, + { + spawnFn, + signal: context.controller.signal, + children: context.children, + killOptions, + }, + )), + publish: options.services?.publish || ((entries, token) => publishBundle( + entries, token, { manifestDir: publicDir }, + )), + prepareHistory: options.services?.prepareHistory || prepareRunHistoryBundle, }; -} -function writeRunMetadata(metadata) { - const metadataPath = path.join(publicDir, 'run_metadata.json'); - fs.writeFileSync(metadataPath, JSON.stringify(metadata, null, 2)); -} + const isBusy = () => ( + uploadReserved + || terminating + || ['uploading', 'processing', 'stopping'].includes(job.status) + ); + const isCurrent = (context) => ( + activeRun === context + && context.epoch === epoch + && !context.controller.signal.aborted + ); -function readEventsFps() { - if (!fs.existsSync(EVENTS_PATH)) return null; - try { - const eventsData = JSON.parse(fs.readFileSync(EVENTS_PATH, 'utf8')); - return typeof eventsData.fps === 'number' ? eventsData.fps : null; - } catch { - return null; - } -} + function reconcileStoppingState() { + const context = stoppingRun; + const contextFinished = !context || context.finished; + const childrenStopped = !context || context.children.size === 0; + if (!terminating || !contextFinished || !childrenStopped || pendingUploads.size > 0) { + return false; + } -function readVerification() { - if (!fs.existsSync(VERIFICATION_PATH)) return { version: 1, reviews: [] }; - try { - const data = JSON.parse(fs.readFileSync(VERIFICATION_PATH, 'utf8')); - if (!Array.isArray(data.reviews)) return { version: 1, reviews: [] }; - return data; - } catch { - return { version: 1, reviews: [] }; + stoppingRun = null; + terminating = false; + if ( + job.status === 'stopping' + || (job.status === 'error' && job.error?.startsWith('Could not terminate active work:')) + ) { + job = blankJob(); + } + return true; } -} - -function writeVerification(data) { - fs.writeFileSync(VERIFICATION_PATH, JSON.stringify(data, null, 2)); -} -function resetVerification() { - writeVerification({ version: 1, reviews: [] }); -} - -// ============================================================ -// RUN HISTORY — per-run snapshots of CSV + events in public/history// -// Metadata lives in public/history.json. Tracked video is NOT snapshotted -// (disk); past-run video falls back to "not available". -// ============================================================ -const HISTORY_DIR = path.join(publicDir, 'history'); -const HISTORY_META_PATH = path.join(publicDir, 'history.json'); - -function readHistoryMeta() { - if (!fs.existsSync(HISTORY_META_PATH)) return { version: 1, runs: [] }; - try { - const data = JSON.parse(fs.readFileSync(HISTORY_META_PATH, 'utf8')); - if (!Array.isArray(data.runs)) return { version: 1, runs: [] }; - return data; - } catch { - return { version: 1, runs: [] }; + function observeStoppingChildren(context) { + if (!context) return; + context.children.forEach((child) => { + if (observedStoppingChildren.has(child)) return; + observedStoppingChildren.add(child); + child.once('close', () => queueMicrotask(reconcileStoppingState)); + }); } -} -function writeHistoryMeta(data) { - fs.writeFileSync(HISTORY_META_PATH, JSON.stringify(data, null, 2)); -} - - -function readCsvAvgProximity(csvPath) { - if (!fs.existsSync(csvPath)) return null; - try { - const lines = fs.readFileSync(csvPath, 'utf8').trim().split('\n'); - if (lines.length < 2) return null; - const header = lines[0].split(','); - const proxIdx = header.indexOf('proximity_distance'); - const occIdx = header.indexOf('occlusion_flag'); - if (proxIdx < 0) return null; - let sum = 0, count = 0; - for (let i = 1; i < lines.length; i++) { - const cols = lines[i].split(','); - const v = parseFloat(cols[proxIdx]); - const isOccluded = occIdx >= 0 ? parseInt(cols[occIdx], 10) === 1 : false; - if (Number.isFinite(v) && !isOccluded) { - sum += v; - count++; - } + function reserveUpload(req, res, next) { + if (isBusy()) { + return res.status(409).json({ error: 'A video is already uploading, processing, or stopping.' }); } - return count > 0 ? Math.round((sum / count) * 100) / 100 : null; - } catch { - return null; + req.uploadEpoch = epoch; + uploadReserved = true; + let released = false; + let resolveClosed; + const state = { + request: req, + filePath: null, + closed: new Promise((resolve) => { resolveClosed = resolve; }), + release: null, + }; + const release = () => { + if (released) return; + released = true; + pendingUploads.delete(req); + uploadReserved = pendingUploads.size > 0; + queueMicrotask(reconcileStoppingState); + }; + state.release = release; + pendingUploads.set(req, state); + req.once('close', () => resolveClosed()); + res.once('finish', release); + res.once('close', release); + return next(); } -} - -function countCourtshipBouts(eventsPath) { - if (!fs.existsSync(eventsPath)) return 0; - try { - const data = JSON.parse(fs.readFileSync(eventsPath, 'utf8')); - if (!Array.isArray(data.events)) return 0; - return data.events.filter((e) => e.type === 'courtship_bout').length; - } catch { - return 0; - } -} -// Snapshot data.csv + events.json for the just-finished run. -// Call AFTER all sync checks pass, BEFORE deleting raw video. -function snapshotRunToHistory({ runId, filename, durationSec, fps, totalFrames }) { - const csvSrc = path.join(publicDir, 'data.csv'); - const eventsSrc = EVENTS_PATH; - if (!fs.existsSync(csvSrc)) return null; - - const stampedId = `run-${String(runId).padStart(4, '0')}`; - const runDir = path.join(HISTORY_DIR, stampedId); - fs.mkdirSync(runDir, { recursive: true }); - fs.copyFileSync(csvSrc, path.join(runDir, 'data.csv')); - if (fs.existsSync(eventsSrc)) { - fs.copyFileSync(eventsSrc, path.join(runDir, 'events.json')); + function readVerification(filePath) { + const data = readJson(filePath, { version: 1, reviews: [] }); + return Array.isArray(data.reviews) ? data : { version: 1, reviews: [] }; } - const meta = { - runId: stampedId, - timestamp: new Date().toISOString(), - filename: filename || 'unknown', - durationSec: Math.round((durationSec || 0) * 10) / 10, - fps: fps || null, - totalFrames: totalFrames || null, - avgProximity: readCsvAvgProximity(csvSrc), - detectedBouts: countCourtshipBouts(eventsSrc), - }; - - const history = readHistoryMeta(); - history.runs.unshift(meta); // newest first - writeHistoryMeta(history); - return meta; -} - -function transcodeVideo(rawPath, finalPath) { - return new Promise((resolve, reject) => { - const args = [ - '-i', rawPath, - '-vcodec', 'libx264', - '-preset', 'veryfast', - '-pix_fmt', 'yuv420p', - '-movflags', '+faststart', - '-an', - '-y', - finalPath, - ]; - - const ffmpeg = spawn(ffmpegPath, args); - let stderr = ''; - - ffmpeg.stderr.on('data', (data) => { - stderr += data.toString(); - }); - - ffmpeg.on('close', (code) => { - if (code === 0) { + function waitForUploadClose(state) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('Multipart upload did not close after reset')), 3000); + timer.unref?.(); + state.closed.then(() => { + clearTimeout(timer); resolve(); - } else { - reject(new Error(`ffmpeg transcode failed (exit ${code}): ${stderr.slice(-500)}`)); - } + }); }); + } - ffmpeg.on('error', (err) => reject(new Error(`Failed to start ffmpeg: ${err.message}`))); - }); -} + async function removePendingUploadFile(filePath) { + if (!filePath) return; + let lastError = null; + for (let attempt = 0; attempt < 5; attempt += 1) { + try { + fs.rmSync(filePath, { force: true }); + if (!fs.existsSync(filePath)) return; + lastError = new Error(`Could not remove cancelled upload: ${filePath}`); + } catch (error) { + lastError = error; + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw lastError || new Error(`Could not remove cancelled upload: ${filePath}`); + } -function buildTrackerArgs(inputPath, outputVideoRaw, outputCsv, outputEvents, overrides = {}) { - // Parse & clamp per-run settings from the Settings UI. Empty/invalid values - // fall back to TRACKER_DEFAULTS so a missing payload still yields pitch-identical output. - const num = (v, fallback) => { - const n = Number(v); - return Number.isFinite(n) && n >= 0 ? n : fallback; - }; - const cfg = { - minArea: num(overrides.minArea, TRACKER_DEFAULTS.minArea), - maxArea: num(overrides.maxArea, TRACKER_DEFAULTS.maxArea), - proximityThreshold: num(overrides.proximityThreshold, TRACKER_DEFAULTS.proximityThreshold), - boutMinFrames: Math.max(1, Math.floor(num(overrides.boutMinFrames, TRACKER_DEFAULTS.boutMinFrames))), - }; - return [ - trackerScript, - '--input', inputPath, - '--output-video', outputVideoRaw, - '--output-csv', outputCsv, - '--output-events', outputEvents, - '--min-area', String(cfg.minArea), - '--max-area', String(cfg.maxArea), - '--proximity-threshold', String(cfg.proximityThreshold), - '--bout-min-frames', String(cfg.boutMinFrames), - ]; -} + async function abortPendingUploads(states) { + states.forEach((state) => { + if (!state.request.destroyed) state.request.destroy(); + }); + await Promise.all(states.map(async (state) => { + await waitForUploadClose(state); + await removePendingUploadFile(state.filePath); + state.release(); + })); + } -function startTracking(inputPath, overrides = {}) { - const outputVideoRaw = path.join(publicDir, 'tracked_raw.mp4'); - const outputVideoFinal = path.join(publicDir, 'tracked.mp4'); - const outputCsv = path.join(publicDir, 'data.csv'); - const outputEvents = EVENTS_PATH; - - job.status = 'processing'; - job.progress = 'Initializing tracker pipeline...'; - job.framesProcessed = 0; - job.startTime = Date.now(); - job.error = null; - trackerStdout = ''; - job.overrides = overrides; // echo back to UI for transparency - - console.log(`[Server] Starting tracker on: ${inputPath}`); - console.log(`[Server] Tracker config: ${JSON.stringify(overrides)}`); - - const tracker = spawn(pythonExe, buildTrackerArgs(inputPath, outputVideoRaw, outputCsv, outputEvents, overrides)); - - activeTracker = tracker; - - tracker.stdout.on('data', (data) => { - const msg = data.toString().trim(); - trackerStdout += data.toString(); - console.log(`[Tracker] ${msg}`); - - const match = msg.match(/Processed (\d+) frames/); - if (match) { - job.framesProcessed = parseInt(match[1]); - job.progress = `Processed ${job.framesProcessed} frames...`; - } else if (msg.includes('Tracking completed')) { - job.progress = 'Finalizing output files...'; - } else { - job.progress = msg; + async function stopActiveRun() { + const uploadStates = [...pendingUploads.values()]; + const context = activeRun || stoppingRun; + if (activeRun || uploadStates.length > 0) epoch += 1; + if (activeRun) stoppingRun = activeRun; + activeRun = null; + + const hasWork = Boolean(context) || uploadStates.length > 0; + job = { ...blankJob(), status: hasWork ? 'stopping' : 'idle' }; + if (!hasWork) { + reconcileStoppingState(); + return; } - }); - tracker.stderr.on('data', (data) => { - const msg = data.toString().trim(); - trackerStdout += data.toString(); - if (msg.includes('NAL unit') || msg.includes('partial file')) { - console.log(`[Tracker] (codec warning, safe to ignore) ${msg}`); - } else { - console.error(`[Tracker] ${msg}`); + terminating = true; + context?.controller.abort(); + observeStoppingChildren(context); + try { + const children = context ? [...context.children] : []; + await Promise.all([ + abortPendingUploads(uploadStates), + ...children.map((child) => terminateChild(child, killOptions)), + ]); + if (context) { + await context.done.catch((error) => { + if (error?.name !== 'AbortError') throw error; + }); + } + job = blankJob(); + } catch (error) { + job = { + ...blankJob(), + status: 'error', + error: `Could not terminate active work: ${error.message}`, + }; + throw error; + } finally { + reconcileStoppingState(); } - }); + } - tracker.on('close', async (code) => { - activeTracker = null; - job.endTime = Date.now(); - const elapsed = ((job.endTime - job.startTime) / 1000).toFixed(1); + function createRunContext(inputPath, filename, overrides) { + const runId = createRunId(); + const runEpoch = epoch; + const token = `${runId}-${runEpoch}-${randomUUID()}`; + const workDir = path.join(uploadsDir, `run-${token}`); + ensureDir(workDir); + return { + runId, + epoch: runEpoch, + token, + workDir, + inputPath, + filename, + overrides, + startedAt: Date.now(), + controller: new AbortController(), + children: new Set(), + done: null, + finished: false, + }; + } - if (code !== 0) { - job.status = 'error'; - job.error = `Tracker exited with code ${code}`; - job.progress = ''; - console.error(`[Server] ❌ Tracker failed with code ${code}`); - return; - } + async function executeRun(context) { + const outputs = { + rawVideo: path.join(context.workDir, 'tracked_raw.mp4'), + browserVideo: path.join(context.workDir, 'tracked.mp4'), + csv: path.join(context.workDir, 'data.csv'), + events: path.join(context.workDir, 'events.json'), + metadata: path.join(context.workDir, 'run_metadata.json'), + verification: path.join(context.workDir, 'verification.json'), + historyIndex: path.join(context.workDir, 'history.json'), + }; + const ensureCurrent = () => { + if (!isCurrent(context)) throw new AbortRunError(); + }; try { - const syncInfo = parseTrackerSync(trackerStdout); - const csvRows = countCsvRows(outputCsv); - let videoFrames = null; - - if (fs.existsSync(outputVideoRaw)) { - videoFrames = await getVideoFrameCount(outputVideoRaw); + job.status = 'processing'; + job.progress = 'Counting input frames...'; + const inputFrames = await services.countFrames(context.inputPath, context); + ensureCurrent(); + + job.progress = 'Running tracker...'; + const trackerResult = await services.runTracker( + context.inputPath, + outputs, + context.overrides, + context, + ); + ensureCurrent(); + if (trackerResult.code !== 0) { + throw new Error(`Tracker exited with code ${trackerResult.code}: ${trackerResult.stderr.slice(-500)}`); } - const frameCount = syncInfo?.framesProcessed ?? csvRows; - const syncOk = syncInfo?.syncOk !== false - && csvRows === frameCount - && (videoFrames === null || videoFrames === csvRows); + const syncInfo = parseTrackerSync(trackerResult.stdout); + if (!syncInfo) throw new Error('Frame integrity evidence missing: TRACKER_SYNC marker required'); + if (!fs.existsSync(outputs.csv)) throw new Error('Tracker data.csv missing'); + if (!fs.existsSync(outputs.rawVideo)) throw new Error('Tracker raw video missing'); + if (!fs.existsSync(outputs.events)) throw new Error('Tracker events.json missing'); - const eventsFps = readEventsFps(); - const metadata = { - framesProcessed: frameCount, - csvRows, - videoFrames, - expectedVideoFrames: syncInfo?.expectedVideoFrames ?? null, - syncOk, - fps: eventsFps, - timestamp: new Date().toISOString(), - }; - writeRunMetadata(metadata); - job.frameCount = frameCount; + const csvRows = countCsvRows(outputs.csv); + job.progress = 'Counting raw output frames...'; + const rawVideoFrames = await services.countFrames(outputs.rawVideo, context); + ensureCurrent(); - console.log( - `[Server] Frame sync: csv_rows=${csvRows}, video_frames=${videoFrames}, sync_ok=${syncOk}` - ); + job.progress = 'Transcoding browser video...'; + await services.transcode(outputs.rawVideo, outputs.browserVideo, context); + ensureCurrent(); - if (!syncOk) { - throw new Error( - `Frame sync mismatch (csv=${csvRows}, video=${videoFrames}, tracker=${frameCount})` - ); - } + job.progress = 'Counting final output frames...'; + const finalVideoFrames = await services.countFrames(outputs.browserVideo, context); + ensureCurrent(); - // Snapshot this run's CSV + events into history BEFORE touching video files. - try { - snapshotRunToHistory({ - runId: job.runId, - filename: job.uploadedFilename, - durationSec: parseFloat(elapsed), - fps: eventsFps, - totalFrames: frameCount, - }); - console.log(`[Server] 📚 Run snapshotted to history`); - } catch (histErr) { - console.error(`[Server] History snapshot failed (non-fatal): ${histErr.message}`); + const integrity = validateFrameIntegrity({ + syncInfo, + inputFrames, + csvRows, + rawVideoFrames, + finalVideoFrames, + }); + const fps = readEventsFps(outputs.events); + scopeEventsForRun(outputs.events, context.runId, outputs.events); + const metadata = { + ...integrity, + runId: context.runId, + expectedVideoFrames: syncInfo.expectedVideoFrames, + fps, + timestamp: new Date().toISOString(), + }; + writeJson(outputs.metadata, metadata); + writeJson(outputs.verification, { version: 1, run_id: context.runId, reviews: [] }); + const historyBundle = services.prepareHistory({ + runId: context.runId, + historyDir, + historyMetaPath, + historyIndexOutput: outputs.historyIndex, + filename: context.filename, + durationSec: (Date.now() - context.startedAt) / 1000, + fps, + totalFrames: integrity.inputFrames, + csvSrc: outputs.csv, + eventsSrc: outputs.events, + verificationSrc: outputs.verification, + }); + ensureCurrent(); + + job.progress = 'Publishing validated current and historical results...'; + services.publish([ + { source: outputs.csv, destination: path.join(publicDir, 'data.csv') }, + { source: outputs.events, destination: path.join(publicDir, 'events.json') }, + { source: outputs.browserVideo, destination: path.join(publicDir, 'tracked.mp4') }, + { source: outputs.metadata, destination: path.join(publicDir, 'run_metadata.json') }, + ...historyBundle.entries, + ], context.token); + ensureCurrent(); + + job = { + ...job, + frameCount: integrity.inputFrames, + framesProcessed: integrity.inputFrames, + totalFrames: integrity.inputFrames, + status: 'done', + progress: `Tracking complete! Processed ${integrity.inputFrames} frames.`, + endTime: Date.now(), + }; + } catch (error) { + if (isCurrent(context)) { + job = { + ...job, + status: 'error', + progress: '', + error: error.message, + endTime: Date.now(), + }; } + if (error?.name !== 'AbortError') console.error(`[Server] Run failed: ${error.message}`); + throw error; + } finally { + safeUnlink(context.inputPath); + safeRemoveDir(context.workDir); + context.finished = true; + if (activeRun === context && context.children.size === 0) activeRun = null; + reconcileStoppingState(); + } + } - if (!fs.existsSync(outputVideoRaw)) { - throw new Error('tracked_raw.mp4 missing after tracker completed'); - } + app.get('/api/status', (_req, res) => res.json({ ...job })); - job.progress = 'Transcoding video for browser playback...'; - await transcodeVideo(outputVideoRaw, outputVideoFinal); - - fs.unlinkSync(outputVideoRaw); - console.log(`[Server] Transcoded annotated video to ${outputVideoFinal}`); - - job.status = 'done'; - job.progress = `Tracking complete! Processed ${job.framesProcessed} frames in ${elapsed}s`; - console.log(`[Server] ✅ Tracking completed in ${elapsed}s`); - } catch (e) { - job.status = 'error'; - job.error = e.message; - job.progress = ''; - console.error(`[Server] ❌ Post-processing failed: ${e.message}`); - console.error('[Server] CSV may still be valid at public/data.csv'); + app.post('/api/upload', reserveUpload, upload.single('video'), (req, res) => { + if (!req.file) return res.status(400).json({ error: 'No video file provided' }); + if (req.uploadEpoch !== epoch) { + safeUnlink(req.file.path); + return res.status(409).json({ error: 'Upload was cancelled by a reset.' }); } + const overrides = { + minArea: req.body.minArea, + maxArea: req.body.maxArea, + proximityThreshold: req.body.proximityThreshold, + boutMinFrames: req.body.boutMinFrames, + }; + const context = createRunContext(req.file.path, req.file.originalname, overrides); + activeRun = context; + job = { + ...blankJob(), + runId: context.runId, + status: 'uploading', + progress: 'Upload complete. Starting validation pipeline...', + uploadedFilename: context.filename, + startTime: context.startedAt, + }; + context.done = executeRun(context).catch(() => {}); + return res.json({ success: true, runId: context.runId }); }); - tracker.on('error', (err) => { - activeTracker = null; - job.status = 'error'; - job.error = `Failed to start tracker: ${err.message}`; - job.progress = ''; - console.error(`[Server] ❌ ${err.message}`); + app.post('/api/reset', async (_req, res) => { + try { + await stopActiveRun(); + return res.json({ success: true }); + } catch (error) { + return res.status(500).json({ error: error.message }); + } }); -} - -// ============================================================ -// ROUTES -// ============================================================ - -app.get('/api/status', (req, res) => { - res.json({ ...job }); -}); -app.post('/api/upload', upload.single('video'), (req, res) => { - if (!req.file) { - return res.status(400).json({ error: 'No video file provided' }); - } + app.get('/api/events', (_req, res) => { + const data = readJson(path.join(publicDir, 'events.json'), null); + if (!data) return res.status(404).json({ error: 'No events file found.' }); + return res.json(data); + }); - if (job.status === 'processing') { - return res.status(409).json({ error: 'A video is already being processed. Please wait.' }); + function currentRunId() { + return readJson(path.join(publicDir, 'run_metadata.json'), null)?.runId || null; } - resetJob(); - resetVerification(); - runId += 1; - const currentRunId = runId; - job.runId = currentRunId; - job.status = 'uploading'; - job.progress = 'Video uploaded, starting tracker...'; - job.uploadedFilename = req.file.originalname; - - // Per-run Settings overrides (multipart text fields arrive as strings). - // buildTrackerArgs validates/clamps these; invalid → TRACKER_DEFAULTS. - const overrides = { - minArea: req.body.minArea, - maxArea: req.body.maxArea, - proximityThreshold: req.body.proximityThreshold, - boutMinFrames: req.body.boutMinFrames, - }; - - const inputPath = req.file.path; - const uploadedFilename = req.file.originalname; - console.log(`[Server] Received upload: ${uploadedFilename} (${(req.file.size / 1024 / 1024).toFixed(1)} MB)`); - - res.json({ success: true, message: 'Upload received, processing started', runId: currentRunId }); + app.get('/api/verification', (req, res) => { + const runId = req.query.runId || currentRunId(); + if (!runId) return res.status(404).json({ error: 'No current run found' }); + try { + return res.json(readVerification(verificationPathForRun(runId, historyDir))); + } catch (error) { + return res.status(404).json({ error: error.message }); + } + }); - startTracking(inputPath, overrides); -}); + app.post('/api/verification', (req, res) => { + const { eventId, verdict } = req.body || {}; + if (!eventId || !['confirmed', 'rejected'].includes(verdict)) { + return res.status(400).json({ error: 'Body must include eventId and verdict.' }); + } + try { + assertEventBelongsToRun(eventId, historyDir); + const filePath = resolveVerificationPath(eventId, historyDir); + const data = readVerification(filePath); + const review = { event_id: eventId, verdict, reviewed_at: new Date().toISOString() }; + const index = data.reviews.findIndex((item) => item.event_id === eventId); + if (index >= 0) data.reviews[index] = review; + else data.reviews.push(review); + writeJson(filePath, data); + return res.json({ success: true, review }); + } catch (error) { + return res.status(400).json({ error: error.message }); + } + }); -app.post('/api/reset', (req, res) => { - resetJob(); - res.json({ success: true }); -}); + app.post('/api/verification/reset', (req, res) => { + const runId = req.body?.runId || currentRunId(); + if (!runId) return res.status(404).json({ error: 'No current run found' }); + try { + writeJson(verificationPathForRun(runId, historyDir), { version: 1, run_id: runId, reviews: [] }); + return res.json({ success: true }); + } catch (error) { + return res.status(404).json({ error: error.message }); + } + }); -app.get('/api/events', (req, res) => { - if (!fs.existsSync(EVENTS_PATH)) { - return res.status(404).json({ error: 'No events file found. Run tracking first.' }); - } - try { - const eventsData = JSON.parse(fs.readFileSync(EVENTS_PATH, 'utf8')); - res.json(eventsData); - } catch (e) { - res.status(500).json({ error: `Failed to read events: ${e.message}` }); - } -}); + app.get('/api/history', (_req, res) => { + res.json(readJson(historyMetaPath, { version: 1, runs: [] })); + }); -app.get('/api/verification', (req, res) => { - res.json(readVerification()); -}); + app.delete('/api/history', (_req, res) => { + writeJson(historyMetaPath, { version: 1, runs: [] }); + res.json({ success: true }); + }); -app.post('/api/verification', (req, res) => { - const { eventId, verdict } = req.body || {}; - if (!eventId || !['confirmed', 'rejected'].includes(verdict)) { - return res.status(400).json({ error: 'Body must include eventId and verdict (confirmed|rejected)' }); - } + app.get('/api/history/:runId', (req, res) => { + const history = readJson(historyMetaPath, { runs: [] }); + const entry = history.runs?.find((run) => run.runId === req.params.runId); + if (!entry) return res.status(404).json({ error: 'Run not found' }); + return res.json(entry); + }); - const data = readVerification(); - const existing = data.reviews.findIndex((r) => r.event_id === eventId); - const review = { - event_id: eventId, - verdict, - reviewed_at: new Date().toISOString(), + return { + app, + stop: stopActiveRun, + getState: () => ({ + epoch, + uploadReserved, + terminating, + pendingUploads: pendingUploads.size, + activeRun, + stoppingRun, + job: { ...job }, + }), + paths: { uploadsDir, publicDir, historyDir, historyMetaPath, verificationPath }, }; +} - if (existing >= 0) { - data.reviews[existing] = review; - } else { - data.reviews.push(review); - } - - writeVerification(data); - res.json({ success: true, review }); -}); - -app.post('/api/verification/reset', (req, res) => { - resetVerification(); - res.json({ success: true }); -}); - -// ---- Run history (Task E / K-06) ---- -// Per-run CSV + events live under public/history// and are served -// statically by express.static(publicDir) already mounted via Vite proxy. -// These routes expose metadata + convenience loaders. - -app.get('/api/history', (req, res) => { - res.json(readHistoryMeta()); -}); - -app.delete('/api/history', (req, res) => { - // Clear metadata only; leave snapshot files on disk (cheap, recoverable). - writeHistoryMeta({ version: 1, runs: [] }); - res.json({ success: true }); -}); - -app.get('/api/history/:runId', (req, res) => { - const { runId } = req.params; - const meta = readHistoryMeta(); - const entry = meta.runs.find((r) => r.runId === runId); - if (!entry) return res.status(404).json({ error: 'Run not found' }); - res.json(entry); -}); - -app.listen(PORT, () => { - console.log(`\n 🧬 Flyt API Server`); - console.log(` ➜ Running on http://localhost:${PORT}`); - console.log(` ➜ Tracker: ${trackerScript}`); - console.log(` ➜ Python: ${pythonExe}`); - console.log(` ➜ FFmpeg: ${ffmpegPath}\n`); -}); \ No newline at end of file +const isDirectRun = process.argv[1] && path.resolve(process.argv[1]) === path.resolve(modulePath); +if (isDirectRun) { + const { app } = createFlytServer(); + app.listen(DEFAULT_PORT, () => { + console.log(`\n 🧬 Flyt API Server\n ➜ http://localhost:${DEFAULT_PORT}`); + }); +} diff --git a/source app folder/dashboard/src/App.jsx b/source app folder/dashboard/src/App.jsx index cd5c72e..ad01fac 100644 --- a/source app folder/dashboard/src/App.jsx +++ b/source app folder/dashboard/src/App.jsx @@ -1,5 +1,6 @@ import React, { useState, useEffect, useRef } from 'react'; import Papa from 'papaparse'; +import { prismDistance, prismVelocity, proximityValue } from './metrics.js'; import { LineChart, Line, XAxis, YAxis, CartesianGrid, Tooltip, ResponsiveContainer, AreaChart, Area, ScatterChart, Scatter, ZAxis @@ -64,9 +65,9 @@ function exportPrismCsv(rows, fps) { const lines = [header.join(',')]; rows.forEach((r) => { const t = Number(r.frame) / effectiveFps; - const v1 = Number(r.fly1_speed_pxsec ?? r.fly1_speed ?? 0); - const v2 = Number(r.fly2_speed_pxsec ?? r.fly2_speed ?? 0); - const dist = Number(r.proximity_distance ?? 0); + const v1 = prismVelocity(r, 'fly1'); + const v2 = prismVelocity(r, 'fly2'); + const dist = prismDistance(r); lines.push([t, v1, v2, dist].map(csvEscape).join(',')); }); const blob = new Blob([lines.join('\n')], { type: 'text/csv;charset=utf-8' }); @@ -649,17 +650,18 @@ function App() { } } catch { /* events optional until first track */ } - if (!runId) { - try { - const verRes = await fetch(`/api/verification?t=${cacheBust}`, { cache: 'no-store' }); - if (verRes.ok) { - const verData = await verRes.json(); - (verData.reviews || []).forEach((r) => { - reviews[r.event_id] = r; - }); - } - } catch { /* keep empty reviews */ } - } + try { + const verificationUrl = runId + ? `/api/verification?runId=${encodeURIComponent(runId)}&t=${cacheBust}` + : `/api/verification?t=${cacheBust}`; + const verRes = await fetch(verificationUrl, { cache: 'no-store' }); + if (verRes.ok) { + const verData = await verRes.json(); + (verData.reviews || []).forEach((r) => { + reviews[r.event_id] = r; + }); + } + } catch { /* keep empty reviews */ } if (generation !== loadGenerationRef.current) return { fps: 30 }; @@ -668,9 +670,7 @@ function App() { setEvents(eventList); setReviewsByEventId(reviews); setStats((prev) => { - const { detected, verified } = runId - ? { detected: eventList.filter((e) => e.type === 'courtship_bout').length, verified: 0 } - : computeCourtshipStats(eventList, reviews); + const { detected, verified } = computeCourtshipStats(eventList, reviews); return { ...prev, courtshipDetected: detected, courtshipVerified: verified }; }); return { fps, frames }; @@ -721,9 +721,11 @@ function App() { const hmData = []; parsedData.forEach((row, i) => { - // Average proximity only over frames where flies are separate (exclude merged/occluded where proximity=0 by design) - if (row.proximity_distance != null && !row.occlusion_flag) { - totalProx += row.proximity_distance; + // Only measured two-fly observations contribute. Dropouts retain + // display coordinates but carry no scientifically valid proximity. + const observedProximity = proximityValue(row); + if (observedProximity !== null) { + totalProx += observedProximity; proxCount++; } if (row.activity_level > maxAct) maxAct = row.activity_level; @@ -791,11 +793,12 @@ function App() { // Load a past run's snapshot (data.csv + events.json from public/history//) // into the active dashboard. Video for past runs is not snapshotted — clear it // so the player doesn't show a stale frame. - const loadHistoricRun = async (runId) => { + const loadHistoricRun = async (run) => { + const runId = run.runId; const generation = ++loadGenerationRef.current; const cacheBust = Date.now(); setMediaCacheBust(cacheBust); - setRunTimestamp(new Date().toISOString()); + setRunTimestamp(run.timestamp || null); setIsHistoricRun(true); try { // Load events first to get accurate fps (avoids stale state) for sleep + pxsec normalization @@ -1034,7 +1037,7 @@ function App() { history.map((run) => (
loadHistoricRun(run.runId)} + onClick={() => loadHistoricRun(run)} className="grid grid-cols-5 p-4 border-b border-zinc-200 dark:border-zinc-800 text-sm items-center hover:bg-zinc-50 dark:hover:bg-zinc-900/50 cursor-pointer transition-colors group last:border-b-0">
{run.runId}
{formatRunDate(run.timestamp)}
diff --git a/source app folder/dashboard/src/metrics.js b/source app folder/dashboard/src/metrics.js new file mode 100644 index 0000000..76e4efd --- /dev/null +++ b/source app folder/dashboard/src/metrics.js @@ -0,0 +1,42 @@ +export function trackingObservationIsValid(row) { + if (!row || Number(row.occlusion_flag ?? 0) !== 0) return false; + if (row.tracking_valid !== null && row.tracking_valid !== undefined && row.tracking_valid !== '') { + return Number(row.tracking_valid) === 1; + } + if (row.detection_count !== null && row.detection_count !== undefined && row.detection_count !== '') { + return Number(row.detection_count) >= 2; + } + if ( + row.fly1_area !== null && row.fly1_area !== undefined + && row.fly2_area !== null && row.fly2_area !== undefined + ) { + return Number(row.fly1_area) > 0 && Number(row.fly2_area) > 0; + } + return true; +} + +export function proximityValue(row) { + if (!trackingObservationIsValid(row)) return null; + const rawProximity = row.proximity_distance; + if (rawProximity === null || rawProximity === undefined || rawProximity === '') return null; + const proximity = Number(rawProximity); + return Number.isFinite(proximity) ? proximity : null; +} + +export function computeAverageProximity(rows) { + const values = rows.map(proximityValue).filter((value) => value !== null); + if (!values.length) return 0; + return Math.round(values.reduce((sum, value) => sum + value, 0) / values.length); +} + +export function prismDistance(row) { + return proximityValue(row) ?? ''; +} + +export function prismVelocity(row, flyKey) { + if (!trackingObservationIsValid(row)) return ''; + const rawSpeed = row?.[`${flyKey}_speed_pxsec`] ?? row?.[`${flyKey}_speed`]; + if (rawSpeed === null || rawSpeed === undefined || rawSpeed === '') return ''; + const speed = Number(rawSpeed); + return Number.isFinite(speed) ? speed : ''; +} diff --git a/source app folder/dashboard/tests/frontend-contract.test.js b/source app folder/dashboard/tests/frontend-contract.test.js new file mode 100644 index 0000000..d3e531b --- /dev/null +++ b/source app folder/dashboard/tests/frontend-contract.test.js @@ -0,0 +1,30 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import test from 'node:test'; +import { fileURLToPath } from 'node:url'; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const appSource = () => fs.readFileSync(path.join(testDir, '..', 'src', 'App.jsx'), 'utf8'); + +test('historic runs load and count their scoped verification state', () => { + const source = appSource(); + assert.match(source, /api\/verification\?runId=\$\{encodeURIComponent\(runId\)\}/); + assert.doesNotMatch(source, /if \(!runId\) \{\s*try \{\s*const verRes/); + assert.match(source, /const \{ detected, verified \} = computeCourtshipStats\(eventList, reviews\)/); + assert.doesNotMatch(source, /verified:\s*0/); +}); + +test('historic runs display their recorded timestamp rather than load time', () => { + const source = appSource(); + assert.match(source, /setRunTimestamp\(run\.timestamp \|\| null\)/); + assert.match(source, /loadHistoricRun\(run\)/); +}); + + +test('dashboard and Prism export use validity-aware proximity helpers', () => { + const source = appSource(); + assert.match(source, /proximityValue\(row\)/); + assert.match(source, /prismDistance\(r\)/); + assert.doesNotMatch(source, /Number\(r\.proximity_distance \?\? 0\)/); +}); diff --git a/source app folder/dashboard/tests/metrics.test.js b/source app folder/dashboard/tests/metrics.test.js new file mode 100644 index 0000000..b612107 --- /dev/null +++ b/source app folder/dashboard/tests/metrics.test.js @@ -0,0 +1,45 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + computeAverageProximity, + prismDistance, + prismVelocity, + proximityValue, +} from '../src/metrics.js'; + +test('carried coordinates from missing detections do not bias proximity metrics', () => { + const rows = [ + { proximity_distance: 10, tracking_valid: 1, detection_count: 2, occlusion_flag: 0 }, + { proximity_distance: 10, tracking_valid: 0, detection_count: 0, occlusion_flag: 0 }, + { proximity_distance: null, tracking_valid: 0, detection_count: 1, occlusion_flag: 1 }, + { proximity_distance: 30, tracking_valid: 1, detection_count: 2, occlusion_flag: 0 }, + ]; + assert.equal(computeAverageProximity(rows), 20); + assert.equal(proximityValue(rows[1]), null); + assert.equal(prismDistance(rows[1]), ''); +}); + +test('legacy rows remain valid only when they contain a finite non-occluded observation', () => { + assert.equal(proximityValue({ proximity_distance: 12, occlusion_flag: 0 }), 12); + assert.equal(proximityValue({ proximity_distance: '', occlusion_flag: 0 }), null); + assert.equal(proximityValue({ proximity_distance: 0, occlusion_flag: 1 }), null); +}); + + +test('invalid or merged observations export blank individual-fly velocities', () => { + const invalid = { + tracking_valid: 0, + detection_count: 1, + occlusion_flag: 1, + fly1_speed_pxsec: 120, + fly2_speed_pxsec: 95, + }; + assert.equal(prismVelocity(invalid, 'fly1'), ''); + assert.equal(prismVelocity(invalid, 'fly2'), ''); + assert.equal(prismVelocity({ + tracking_valid: 1, + detection_count: 2, + occlusion_flag: 0, + fly1_speed_pxsec: 42.5, + }, 'fly1'), 42.5); +}); diff --git a/source app folder/dashboard/tests/server-integration.test.js b/source app folder/dashboard/tests/server-integration.test.js new file mode 100644 index 0000000..322408f --- /dev/null +++ b/source app folder/dashboard/tests/server-integration.test.js @@ -0,0 +1,326 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import http from 'node:http'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { createFlytServer } from '../server.js'; + +function abortError() { + const error = new Error('aborted'); + error.name = 'AbortError'; + return error; +} + +function createGate({ releaseOnAbort = true } = {}) { + let resolveGate; + let rejectGate; + let aborted = false; + const promise = new Promise((resolve, reject) => { + resolveGate = resolve; + rejectGate = reject; + }); + return { + wait(signal) { + if (signal.aborted) return Promise.reject(abortError()); + signal.addEventListener('abort', () => { + aborted = true; + if (releaseOnAbort) rejectGate(abortError()); + }, { once: true }); + return promise; + }, + resolve(value) { resolveGate(value); }, + reject(error) { rejectGate(error); }, + get aborted() { return aborted; }, + }; +} + +function writeTrackerOutputs(outputs, frames = 4) { + fs.writeFileSync(outputs.rawVideo, 'raw-video'); + fs.writeFileSync(outputs.csv, [ + 'frame,proximity_distance,occlusion_flag,tracking_valid,detection_count,fly1_area,fly2_area', + ...Array.from({ length: frames }, (_, index) => `${index},20,0,1,2,100,100`), + ].join('\n')); + fs.writeFileSync(outputs.events, JSON.stringify({ + fps: 30, + total_frames: frames, + events: [{ id: 'evt-001', type: 'courtship_bout' }], + })); +} + +function standardServices(overrides = {}) { + return { + countFrames: async () => 4, + runTracker: async (_input, outputs) => { + writeTrackerOutputs(outputs); + return { + code: 0, + stderr: '', + stdout: 'TRACKER_SYNC frames_processed=4 csv_rows=4 expected_video_frames=4 sync_ok=true', + }; + }, + transcode: async (raw, final) => fs.copyFileSync(raw, final), + ...overrides, + }; +} + +async function createHarness(services) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'flyt-http-')); + const instance = createFlytServer({ + rootDir: root, + uploadsDir: path.join(root, 'uploads'), + publicDir: path.join(root, 'public'), + historyDir: path.join(root, 'public', 'history'), + historyMetaPath: path.join(root, 'public', 'history.json'), + verificationPath: path.join(root, 'public', 'verification.json'), + trackerDir: path.join(root, 'tracker'), + trackerScript: path.join(root, 'tracker', 'tracker.py'), + pythonExe: 'python', + ffmpegPath: 'ffmpeg', + allowedOrigins: ['http://localhost:5173'], + services, + }); + const listener = await new Promise((resolve) => { + const server = instance.app.listen(0, '127.0.0.1', () => resolve(server)); + }); + const address = listener.address(); + return { + ...instance, + root, + baseUrl: `http://127.0.0.1:${address.port}`, + async close() { + try { await instance.stop(); } catch { /* test cleanup */ } + await new Promise((resolve) => listener.close(resolve)); + fs.rmSync(root, { recursive: true, force: true }); + }, + }; +} + +async function upload(baseUrl, name = 'flies.mp4') { + const form = new FormData(); + form.append('video', new Blob(['video-bytes'], { type: 'video/mp4' }), name); + return fetch(`${baseUrl}/api/upload`, { method: 'POST', body: form }); +} + +async function waitForStatus(baseUrl, expected, timeoutMs = 1500) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const response = await fetch(`${baseUrl}/api/status`); + const status = await response.json(); + if (status.status === expected) return status; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`Timed out waiting for ${expected}`); +} + +test('rejects a simultaneous upload before a second file is accepted', async (t) => { + const gate = createGate(); + const harness = await createHarness(standardServices({ + countFrames: (_path, context) => gate.wait(context.controller.signal), + })); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl, 'first.mp4')).status, 200); + assert.equal((await upload(harness.baseUrl, 'second.mp4')).status, 409); + assert.equal(fs.readdirSync(harness.paths.uploadsDir).filter((name) => name.startsWith('input-')).length, 1); + + const reset = await fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }); + assert.equal(reset.status, 200); + assert.equal(gate.aborted, true); +}); + +test('reset actively aborts a stalled multipart upload and reopens the gate', async (t) => { + const harness = await createHarness(standardServices()); + t.after(() => harness.close()); + const boundary = '----flyt-boundary'; + const prefix = [ + `--${boundary}`, + 'Content-Disposition: form-data; name="video"; filename="slow.mp4"', + 'Content-Type: video/mp4', + '', + 'partial-video-', + ].join('\r\n'); + + const uploadOutcome = new Promise((resolve) => { + const request = http.request(`${harness.baseUrl}/api/upload`, { + method: 'POST', + headers: { 'Content-Type': `multipart/form-data; boundary=${boundary}` }, + }, (response) => { + response.resume(); + response.on('end', () => resolve(response.statusCode)); + }); + request.on('error', () => resolve('aborted')); + request.write(prefix); + }); + + await new Promise((resolve) => setTimeout(resolve, 30)); + const reset = await fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }); + assert.equal(reset.status, 200); + const outcome = await Promise.race([ + uploadOutcome, + new Promise((resolve) => setTimeout(() => resolve('timeout'), 1000)), + ]); + assert.notEqual(outcome, 'timeout'); + assert.ok(outcome === 'aborted' || outcome === 409); + assert.equal( + fs.readdirSync(harness.paths.uploadsDir).filter((name) => name.startsWith('input-')).length, + 0, + ); + const state = harness.getState(); + assert.equal(state.pendingUploads, 0); + assert.equal(state.uploadReserved, false); + assert.equal(state.terminating, false); + + assert.equal((await upload(harness.baseUrl, 'after-reset.mp4')).status, 200); + await waitForStatus(harness.baseUrl, 'done'); +}); + +test('reset remains pending until the active stage has actually stopped', async (t) => { + const gate = createGate({ releaseOnAbort: false }); + const harness = await createHarness(standardServices({ + countFrames: (_path, context) => gate.wait(context.controller.signal), + })); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl)).status, 200); + let resetSettled = false; + const resetPromise = fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }) + .then((response) => { resetSettled = true; return response; }); + await new Promise((resolve) => setTimeout(resolve, 40)); + assert.equal(resetSettled, false); + assert.equal((await upload(harness.baseUrl, 'blocked.mp4')).status, 409); + + gate.resolve(4); + assert.equal((await resetPromise).status, 200); + assert.equal(resetSettled, true); +}); + +test('reset during transcoding prevents stale publication', async (t) => { + const transcodeGate = createGate({ releaseOnAbort: false }); + let publishCalls = 0; + const harness = await createHarness(standardServices({ + transcode: async (raw, final, context) => { + await transcodeGate.wait(context.controller.signal); + fs.copyFileSync(raw, final); + }, + publish: () => { publishCalls += 1; }, + })); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl)).status, 200); + await new Promise((resolve) => setTimeout(resolve, 30)); + const resetPromise = fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(transcodeGate.aborted, true); + assert.equal(publishCalls, 0); + + transcodeGate.resolve(); + assert.equal((await resetPromise).status, 200); + assert.equal(publishCalls, 0); +}); + +test('a complete run publishes only after every frame count agrees', async (t) => { + const harness = await createHarness(standardServices()); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl)).status, 200); + const status = await waitForStatus(harness.baseUrl, 'done'); + assert.equal(status.frameCount, 4); + const metadata = JSON.parse(fs.readFileSync(path.join(harness.paths.publicDir, 'run_metadata.json'))); + assert.equal(metadata.inputFrames, 4); + assert.equal(metadata.rawVideoFrames, 4); + assert.equal(metadata.finalVideoFrames, 4); + assert.equal(metadata.syncOk, true); + assert.match(metadata.runId, /^run-/); + const currentEvents = JSON.parse(fs.readFileSync(path.join(harness.paths.publicDir, 'events.json'))); + assert.equal(currentEvents.events[0].id, `${metadata.runId}:evt-001`); +}); + +test('frame mismatch fails closed and preserves the previous published bundle', async (t) => { + let countCall = 0; + const harness = await createHarness(standardServices({ + countFrames: async () => { + countCall += 1; + return countCall === 3 ? 3 : 4; + }, + })); + t.after(() => harness.close()); + fs.writeFileSync(path.join(harness.paths.publicDir, 'data.csv'), 'old-data'); + + assert.equal((await upload(harness.baseUrl)).status, 200); + const status = await waitForStatus(harness.baseUrl, 'error'); + assert.match(status.error, /Frame integrity mismatch/); + assert.equal(fs.readFileSync(path.join(harness.paths.publicDir, 'data.csv'), 'utf8'), 'old-data'); +}); + +test('destructive cross-origin requests are denied unless explicitly allowed', async (t) => { + const harness = await createHarness(standardServices()); + t.after(() => harness.close()); + + const denied = await fetch(`${harness.baseUrl}/api/reset`, { + method: 'POST', + headers: { Origin: 'https://evil.example' }, + }); + assert.equal(denied.status, 403); + + const allowed = await fetch(`${harness.baseUrl}/api/reset`, { + method: 'POST', + headers: { Origin: 'http://localhost:5173' }, + }); + assert.equal(allowed.status, 200); + assert.equal(allowed.headers.get('access-control-allow-origin'), 'http://localhost:5173'); +}); + + +test('a verdict made on the current run survives subsequent runs and history reload', async (t) => { + const harness = await createHarness(standardServices()); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl, 'first.mp4')).status, 200); + const firstStatus = await waitForStatus(harness.baseUrl, 'done'); + const firstRunId = firstStatus.runId; + const eventsResponse = await fetch(`${harness.baseUrl}/api/events`); + const firstEvents = await eventsResponse.json(); + const eventId = firstEvents.events[0].id; + assert.equal(eventId, `${firstRunId}:evt-001`); + + const verdictResponse = await fetch(`${harness.baseUrl}/api/verification`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ eventId, verdict: 'confirmed' }), + }); + assert.equal(verdictResponse.status, 200); + + const unscopedResponse = await fetch(`${harness.baseUrl}/api/verification`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ eventId: 'evt-001', verdict: 'confirmed' }), + }); + assert.equal(unscopedResponse.status, 400); + + const orphanResponse = await fetch(`${harness.baseUrl}/api/verification`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ eventId: `${firstRunId}:evt-does-not-exist`, verdict: 'confirmed' }), + }); + assert.equal(orphanResponse.status, 400); + + assert.equal((await upload(harness.baseUrl, 'second.mp4')).status, 200); + const secondStatus = await waitForStatus(harness.baseUrl, 'done'); + assert.notEqual(secondStatus.runId, firstRunId); + + const historicResponse = await fetch( + `${harness.baseUrl}/api/verification?runId=${encodeURIComponent(firstRunId)}`, + ); + assert.equal(historicResponse.status, 200); + const historicVerification = await historicResponse.json(); + assert.deepEqual(historicVerification.reviews.map((review) => ({ + event_id: review.event_id, + verdict: review.verdict, + })), [{ event_id: eventId, verdict: 'confirmed' }]); + + const canonical = JSON.parse(fs.readFileSync( + path.join(harness.paths.historyDir, firstRunId, 'verification.json'), + )); + assert.equal(canonical.reviews[0].verdict, 'confirmed'); +}); diff --git a/source app folder/dashboard/tests/server-utils.test.js b/source app folder/dashboard/tests/server-utils.test.js new file mode 100644 index 0000000..f5945bd --- /dev/null +++ b/source app folder/dashboard/tests/server-utils.test.js @@ -0,0 +1,363 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { + assertEventBelongsToRun, + buildTrackerArgs, + parseTrackerSync, + prepareRunHistoryBundle, + publishBundle, + readCsvAvgProximity, + recoverPublishArtifacts, + recoverRuntimeArtifacts, + resolveVerificationPath, + runCommand, + scopeEventsForRun, + SimulatedProcessCrash, + terminateChild, + validateFrameIntegrity, + verificationPathForRun, +} from '../server-utils.js'; + +const defaults = { minArea: 30, maxArea: 0, proximityThreshold: 60, boutMinFrames: 90 }; +const tempDir = (prefix) => fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + +test('normalizes integer tracker settings and rejects invalid max area', () => { + const args = buildTrackerArgs('tracker.py', 'in.mp4', { + rawVideo: 'raw.mp4', csv: 'data.csv', events: 'events.json', + }, { minArea: 30.9, maxArea: 20.2, boutMinFrames: 0.8 }, defaults); + assert.equal(args[args.indexOf('--min-area') + 1], '30'); + assert.equal(args[args.indexOf('--max-area') + 1], '0'); + assert.equal(args[args.indexOf('--bout-min-frames') + 1], '1'); +}); + +test('requires exactly one tracker sync marker', () => { + const marker = 'TRACKER_SYNC frames_processed=12 csv_rows=12 expected_video_frames=12 sync_ok=true'; + assert.deepEqual(parseTrackerSync(marker), { + framesProcessed: 12, csvRows: 12, expectedVideoFrames: 12, syncOk: true, + }); + assert.equal(parseTrackerSync('no marker'), null); + assert.equal(parseTrackerSync(`${marker}\n${marker}`), null); +}); + +test('frame validation fails closed on missing evidence or disagreement', () => { + const evidence = { + syncInfo: { framesProcessed: 12, csvRows: 12, expectedVideoFrames: 12, syncOk: true }, + inputFrames: 12, csvRows: 12, rawVideoFrames: 12, finalVideoFrames: 12, + }; + assert.equal(validateFrameIntegrity(evidence).syncOk, true); + assert.throws(() => validateFrameIntegrity({ ...evidence, syncInfo: null }), /marker required/); + assert.throws(() => validateFrameIntegrity({ ...evidence, finalVideoFrames: 11 }), /mismatch/); + const diagnosticMismatch = validateFrameIntegrity({ + ...evidence, + syncInfo: { ...evidence.syncInfo, expectedVideoFrames: 11 }, + }); + assert.equal(diagnosticMismatch.syncOk, true); + assert.equal(diagnosticMismatch.expectedMetadataMatches, false); +}); + +test('publishes a complete bundle and removes backups', () => { + const dir = tempDir('flyt-publish-'); + const sourceA = path.join(dir, 'source-a'); + const sourceB = path.join(dir, 'source-b'); + const destinationA = path.join(dir, 'a'); + const destinationB = path.join(dir, 'b'); + fs.writeFileSync(sourceA, 'new-a'); + fs.writeFileSync(sourceB, 'new-b'); + fs.writeFileSync(destinationA, 'old-a'); + fs.writeFileSync(destinationB, 'old-b'); + publishBundle([ + { source: sourceA, destination: destinationA }, + { source: sourceB, destination: destinationB }, + ], 'token'); + assert.equal(fs.readFileSync(destinationA, 'utf8'), 'new-a'); + assert.equal(fs.readFileSync(destinationB, 'utf8'), 'new-b'); + assert.equal(fs.readdirSync(dir).some((name) => /\.(bak|new)$/.test(name)), false); +}); + +test('rolls back after one destination has already been published', () => { + const dir = tempDir('flyt-rollback-'); + const sources = ['new-a', 'new-b'].map((content, index) => { + const file = path.join(dir, `source-${index}`); + fs.writeFileSync(file, content); + return file; + }); + const destinations = [path.join(dir, 'a'), path.join(dir, 'b')]; + fs.writeFileSync(destinations[0], 'old-a'); + fs.writeFileSync(destinations[1], 'old-b'); + assert.throws(() => publishBundle([ + { source: sources[0], destination: destinations[0] }, + { source: sources[1], destination: destinations[1] }, + ], 'token', { + faultInjector(stage, index) { + if (stage === 'published' && index === 0) throw new Error('injected failure'); + }, + }), /injected failure/); + assert.equal(fs.readFileSync(destinations[0], 'utf8'), 'old-a'); + assert.equal(fs.readFileSync(destinations[1], 'utf8'), 'old-b'); + assert.equal(fs.readdirSync(dir).some((name) => /\.(bak|new)$/.test(name)), false); +}); + +test('crash recovery rolls back the entire partially published bundle', () => { + const dir = tempDir('flyt-crash-rollback-'); + const sourceA = path.join(dir, 'source-a'); + const sourceB = path.join(dir, 'source-b'); + const destinationA = path.join(dir, 'a'); + const destinationB = path.join(dir, 'b'); + fs.writeFileSync(sourceA, 'new-a'); + fs.writeFileSync(sourceB, 'new-b'); + fs.writeFileSync(destinationA, 'old-a'); + fs.writeFileSync(destinationB, 'old-b'); + + assert.throws(() => publishBundle([ + { source: sourceA, destination: destinationA }, + { source: sourceB, destination: destinationB }, + ], 'crash-token', { + manifestDir: dir, + faultInjector(stage, index) { + if (stage === 'published' && index === 0) throw new SimulatedProcessCrash(); + }, + }), SimulatedProcessCrash); + + assert.equal(fs.readFileSync(destinationA, 'utf8'), 'new-a'); + assert.equal(fs.existsSync(destinationB), false); + recoverPublishArtifacts(dir); + assert.equal(fs.readFileSync(destinationA, 'utf8'), 'old-a'); + assert.equal(fs.readFileSync(destinationB, 'utf8'), 'old-b'); + assert.equal(fs.readdirSync(dir).some((name) => /flyt-publish|\.(bak|new)$/.test(name)), false); +}); + +test('crash recovery finalizes an explicitly committed all-new bundle', () => { + const dir = tempDir('flyt-crash-commit-'); + const source = path.join(dir, 'source'); + const destination = path.join(dir, 'destination'); + fs.writeFileSync(source, 'new'); + fs.writeFileSync(destination, 'old'); + + assert.throws(() => publishBundle([ + { source, destination }, + ], 'commit-token', { + manifestDir: dir, + faultInjector(stage) { + if (stage === 'committed') throw new SimulatedProcessCrash(); + }, + }), SimulatedProcessCrash); + + recoverPublishArtifacts(dir); + assert.equal(fs.readFileSync(destination, 'utf8'), 'new'); + assert.equal(fs.readdirSync(dir).some((name) => /flyt-publish|\.(bak|new)$/.test(name)), false); +}); + +test('legacy orphan backups fail closed instead of reconstructing a mixed bundle', () => { + const dir = tempDir('flyt-orphan-backup-'); + const backup = path.join(dir, 'data.csv.token.bak'); + fs.writeFileSync(backup, 'old-data'); + assert.throws(() => recoverPublishArtifacts(dir), /without a transaction manifest/); + assert.equal(fs.readFileSync(backup, 'utf8'), 'old-data'); +}); + +test('history bundle uses scoped events and excludes invalid proximity observations', () => { + const dir = tempDir('flyt-history-'); + const historyDir = path.join(dir, 'history'); + const historyMetaPath = path.join(dir, 'history.json'); + const historyIndexOutput = path.join(dir, 'next-history.json'); + const csv = path.join(dir, 'data.csv'); + const events = path.join(dir, 'events.json'); + const verification = path.join(dir, 'verification.json'); + const runId = 'run-1234-abcd'; + fs.writeFileSync(csv, [ + 'frame,proximity_distance,occlusion_flag,tracking_valid,detection_count', + '0,10,0,1,2', + '1,10,0,0,0', + '2,,1,0,1', + '3,30,0,1,2', + ].join('\n')); + fs.writeFileSync(events, JSON.stringify({ + events: [{ id: 'evt-001', type: 'courtship_bout' }], + })); + fs.writeFileSync(verification, JSON.stringify({ version: 1, run_id: runId, reviews: [] })); + scopeEventsForRun(events, runId, events); + + const bundle = prepareRunHistoryBundle({ + runId, + historyDir, + historyMetaPath, + historyIndexOutput, + filename: 'flies.mp4', + durationSec: 2.3, + fps: 30, + totalFrames: 4, + csvSrc: csv, + eventsSrc: events, + verificationSrc: verification, + }); + assert.equal(bundle.meta.avgProximity, 20); + assert.equal(bundle.meta.detectedBouts, 1); + assert.equal(readCsvAvgProximity(csv), 20); + publishBundle(bundle.entries, 'history-token', { manifestDir: dir }); + + const scopedEvents = JSON.parse(fs.readFileSync(path.join(historyDir, runId, 'events.json'))); + assert.equal(scopedEvents.events[0].id, `${runId}:evt-001`); + assert.equal( + resolveVerificationPath(`${runId}:evt-001`, historyDir), + path.join(historyDir, runId, 'verification.json'), + ); + assert.equal(verificationPathForRun(runId, historyDir), path.join(historyDir, runId, 'verification.json')); + assert.equal(assertEventBelongsToRun(`${runId}:evt-001`, historyDir), runId); + assert.throws( + () => assertEventBelongsToRun(`${runId}:evt-does-not-exist`, historyDir), + /does not belong/, + ); + assert.throws(() => resolveVerificationPath('evt-001', historyDir), /Run-scoped event ID required/); +}); + +test('rollback remains fail-closed when a new destination cannot be deleted', () => { + const dir = tempDir('flyt-delete-failure-'); + const source = path.join(dir, 'source'); + const destination = path.join(dir, 'new-destination'); + const manifestPath = path.join(dir, '.flyt-publish-delete-failure.json'); + fs.writeFileSync(source, 'new-data'); + let blockDestinationRemoval = true; + const removeFile = (filePath) => { + if (blockDestinationRemoval && filePath === destination) { + throw new Error('injected deletion failure'); + } + fs.rmSync(filePath, { force: true }); + if (fs.existsSync(filePath)) throw new Error(`still exists: ${filePath}`); + }; + + assert.throws(() => publishBundle([ + { source, destination }, + ], 'delete-failure', { + manifestDir: dir, + removeFile, + faultInjector(stage) { + if (stage === 'published') throw new Error('injected publish failure'); + }, + }), /rollback was incomplete/); + assert.equal(fs.readFileSync(destination, 'utf8'), 'new-data'); + assert.equal(fs.existsSync(manifestPath), true); + + blockDestinationRemoval = false; + recoverPublishArtifacts(dir); + assert.equal(fs.existsSync(destination), false); + assert.equal(fs.existsSync(manifestPath), false); +}); + +test('startup runtime recovery removes abandoned inputs and run workspaces only', () => { + const dir = tempDir('flyt-runtime-recovery-'); + fs.writeFileSync(path.join(dir, 'input-stale.mp4'), 'partial'); + fs.mkdirSync(path.join(dir, 'run-stale')); + fs.writeFileSync(path.join(dir, 'run-stale', 'data.tmp'), 'partial'); + fs.writeFileSync(path.join(dir, 'keep.txt'), 'keep'); + + recoverRuntimeArtifacts(dir); + assert.equal(fs.existsSync(path.join(dir, 'input-stale.mp4')), false); + assert.equal(fs.existsSync(path.join(dir, 'run-stale')), false); + assert.equal(fs.readFileSync(path.join(dir, 'keep.txt'), 'utf8'), 'keep'); +}); + +test('termination error without close rejects and never proves process exit', async () => { + class ErrorOnlyChild extends EventEmitter { + constructor() { + super(); + this.exitCode = null; + this.signalCode = null; + } + kill() { + setTimeout(() => this.emit('error', new Error('signal delivery failed')), 1); + return true; + } + } + const child = new ErrorOnlyChild(); + await assert.rejects( + terminateChild(child, { graceMs: 20, hardKillMs: 20 }), + /termination failed before close/, + ); + assert.equal(child.exitCode, null); + assert.equal(child.signalCode, null); +}); + +test('runCommand keeps an unclosed child tracked after termination failure', async () => { + class ErrorOnlyChild extends EventEmitter { + constructor() { + super(); + this.exitCode = null; + this.signalCode = null; + this.pid = 123; + } + kill() { + setTimeout(() => this.emit('error', new Error('kill failed')), 1); + return true; + } + } + const child = new ErrorOnlyChild(); + const controller = new AbortController(); + const children = new Set(); + const command = runCommand('fake', [], { + spawnFn: () => child, + signal: controller.signal, + children, + killOptions: { graceMs: 20, hardKillMs: 20 }, + }); + controller.abort(); + await assert.rejects(command, /termination failed before close/); + assert.equal(children.has(child), true); + child.exitCode = 1; + child.emit('close', 1, null); + assert.equal(children.size, 0); +}); + +test('termination escalates to SIGKILL and resolves only after close', async () => { + class FakeChild extends EventEmitter { + constructor() { + super(); + this.exitCode = null; + this.signalCode = null; + this.signals = []; + } + kill(signal) { + this.signals.push(signal); + if (signal === 'SIGKILL') { + setTimeout(() => { + this.signalCode = signal; + this.emit('close', null, signal); + }, 5); + } + return true; + } + } + const child = new FakeChild(); + let settled = false; + const termination = terminateChild(child, { graceMs: 5, hardKillMs: 100 }) + .then(() => { settled = true; }); + await new Promise((resolve) => setTimeout(resolve, 7)); + assert.equal(settled, false); + await termination; + assert.deepEqual(child.signals, ['SIGTERM', 'SIGKILL']); +}); + +test('abort kills a real child process that ignores SIGTERM', async () => { + const controller = new AbortController(); + const children = new Set(); + const command = runCommand(process.execPath, [ + '-e', + "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000);", + ], { + signal: controller.signal, + children, + killOptions: { graceMs: 20, hardKillMs: 500 }, + }); + + const deadline = Date.now() + 1000; + while (children.size === 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.equal(children.size, 1); + controller.abort(); + await assert.rejects(command, (error) => error?.name === 'AbortError'); + assert.equal(children.size, 0); +}); diff --git a/source app folder/dashboard/tests/termination-lifecycle.test.js b/source app folder/dashboard/tests/termination-lifecycle.test.js new file mode 100644 index 0000000..b923b28 --- /dev/null +++ b/source app folder/dashboard/tests/termination-lifecycle.test.js @@ -0,0 +1,155 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { createFlytServer } from '../server.js'; + +function abortError() { + const error = new Error('aborted'); + error.name = 'AbortError'; + return error; +} + +function writeTrackerOutputs(outputs, frames = 4) { + fs.writeFileSync(outputs.rawVideo, 'raw-video'); + fs.writeFileSync(outputs.csv, [ + 'frame,proximity_distance,occlusion_flag,tracking_valid,detection_count,fly1_area,fly2_area', + ...Array.from({ length: frames }, (_, index) => `${index},20,0,1,2,100,100`), + ].join('\n')); + fs.writeFileSync(outputs.events, JSON.stringify({ + fps: 30, + total_frames: frames, + events: [{ id: 'evt-001', type: 'courtship_bout' }], + })); +} + +async function createHarness(services) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'flyt-stop-lifecycle-')); + const instance = createFlytServer({ + rootDir: root, + uploadsDir: path.join(root, 'uploads'), + publicDir: path.join(root, 'public'), + historyDir: path.join(root, 'public', 'history'), + historyMetaPath: path.join(root, 'public', 'history.json'), + verificationPath: path.join(root, 'public', 'verification.json'), + trackerDir: path.join(root, 'tracker'), + trackerScript: path.join(root, 'tracker', 'tracker.py'), + pythonExe: 'python', + ffmpegPath: 'ffmpeg', + allowedOrigins: ['http://localhost:5173'], + killOptions: { graceMs: 10, hardKillMs: 20 }, + services, + }); + const listener = await new Promise((resolve) => { + const server = instance.app.listen(0, '127.0.0.1', () => resolve(server)); + }); + const address = listener.address(); + return { + ...instance, + baseUrl: `http://127.0.0.1:${address.port}`, + async close() { + try { await instance.stop(); } catch { /* test cleanup */ } + await new Promise((resolve) => listener.close(resolve)); + fs.rmSync(root, { recursive: true, force: true }); + }, + }; +} + +async function upload(baseUrl, name = 'flies.mp4') { + const form = new FormData(); + form.append('video', new Blob(['video-bytes'], { type: 'video/mp4' }), name); + return fetch(`${baseUrl}/api/upload`, { method: 'POST', body: form }); +} + +async function waitFor(check, timeoutMs = 1500) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = await check(); + if (value) return value; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error('Timed out waiting for lifecycle state'); +} + +async function waitForStatus(baseUrl, expected, timeoutMs = 1500) { + return waitFor(async () => { + const response = await fetch(`${baseUrl}/api/status`); + const status = await response.json(); + return status.status === expected ? status : null; + }, timeoutMs); +} + +class InitiallyUnkillableChild extends EventEmitter { + constructor() { + super(); + this.exitCode = null; + this.signalCode = null; + this.pid = 123; + } + + kill() { + setTimeout(() => this.emit('error', new Error('signal delivery failed')), 1); + return true; + } + + closeLater() { + this.exitCode = 1; + this.emit('close', 1, null); + } +} + +test('late child close clears a failed stopping context and reopens uploads', async (t) => { + let child; + let countCalls = 0; + const services = { + countFrames: async (_filePath, context) => { + countCalls += 1; + if (countCalls > 1) return 4; + + child = new InitiallyUnkillableChild(); + context.children.add(child); + return new Promise((_resolve, reject) => { + child.once('close', () => { + context.children.delete(child); + reject(abortError()); + }); + }); + }, + runTracker: async (_input, outputs) => { + writeTrackerOutputs(outputs); + return { + code: 0, + stderr: '', + stdout: 'TRACKER_SYNC frames_processed=4 csv_rows=4 expected_video_frames=4 sync_ok=true', + }; + }, + transcode: async (raw, final) => fs.copyFileSync(raw, final), + }; + const harness = await createHarness(services); + t.after(() => harness.close()); + + assert.equal((await upload(harness.baseUrl, 'first.mp4')).status, 200); + await waitFor(() => child && harness.getState().activeRun?.children.has(child)); + + const firstReset = await fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }); + assert.equal(firstReset.status, 500); + assert.equal(harness.getState().terminating, true); + assert.ok(harness.getState().stoppingRun); + assert.equal((await upload(harness.baseUrl, 'blocked.mp4')).status, 409); + + const secondReset = await fetch(`${harness.baseUrl}/api/reset`, { method: 'POST' }); + assert.equal(secondReset.status, 500); + assert.equal(harness.getState().terminating, true); + assert.ok(harness.getState().stoppingRun?.children.has(child)); + + child.closeLater(); + await waitFor(() => { + const state = harness.getState(); + return !state.terminating && state.stoppingRun === null && state.job.status === 'idle'; + }); + + assert.equal((await upload(harness.baseUrl, 'after-close.mp4')).status, 200); + await waitForStatus(harness.baseUrl, 'done'); +}); diff --git a/source app folder/tracker/tests/test_tracker.py b/source app folder/tracker/tests/test_tracker.py new file mode 100644 index 0000000..db333e7 --- /dev/null +++ b/source app folder/tracker/tests/test_tracker.py @@ -0,0 +1,95 @@ +import importlib.util +import math +import pathlib +import unittest + +TRACKER_PATH = pathlib.Path(__file__).resolve().parents[1] / "tracker.py" +spec = importlib.util.spec_from_file_location("flyt_tracker", TRACKER_PATH) +tracker = importlib.util.module_from_spec(spec) +assert spec.loader is not None +spec.loader.exec_module(tracker) + + +def row(frame, proximity, confidence=1.0, occlusion=0, area=100.0, tracking_valid=None): + if tracking_valid is None: + tracking_valid = int(occlusion == 0 and area > 0) + return { + "frame": frame, + "proximity_distance": proximity, + "identity_confidence": confidence, + "occlusion_flag": occlusion, + "fly1_area": area, + "fly2_area": area, + "tracking_valid": tracking_valid, + } + + +class TrackerTests(unittest.TestCase): + def test_zero_is_not_missing(self): + self.assertEqual(tracker.value_or({"value": 0}, "value", 999), 0) + self.assertEqual(tracker.value_or({"value": 0.0}, "value", 999), 0.0) + + def test_initial_detection_has_zero_velocity(self): + self.assertEqual(tracker.displacement((0, 0), (100, 100), False), 0.0) + self.assertEqual(tracker.displacement((0, 0), (3, 4), True), 5.0) + + def test_invalid_observation_speed_is_missing(self): + self.assertTrue(math.isnan(tracker.observed_speed(25.0, 0))) + self.assertEqual(tracker.observed_speed(25.0, 1), 25.0) + + def test_separate_confident_close_flies_generate_courtship(self): + rows = [row(i, proximity=20, confidence=0.8) for i in range(5)] + events = tracker.detect_events(rows, 30, proximity_threshold=60, bout_min_frames=5) + self.assertEqual([event["type"] for event in events], ["courtship_bout"]) + self.assertEqual(events[0]["occlusion_fraction"], 0) + + def test_occluded_zero_proximity_never_generates_courtship(self): + rows = [row(i, proximity=0, confidence=0.0, occlusion=1) for i in range(30)] + events = tracker.detect_events(rows, 30, proximity_threshold=60, bout_min_frames=5) + self.assertNotIn("courtship_bout", [event["type"] for event in events]) + self.assertEqual([event["type"] for event in events], ["low_confidence_segment"]) + + def test_low_confidence_separate_flies_do_not_generate_courtship(self): + rows = [row(i, proximity=10, confidence=0.1, occlusion=0) for i in range(30)] + events = tracker.detect_events(rows, 30, proximity_threshold=60, bout_min_frames=5) + self.assertNotIn("courtship_bout", [event["type"] for event in events]) + self.assertEqual(events[0]["type"], "low_confidence_segment") + + def test_missing_fly_area_does_not_generate_courtship(self): + rows = [row(i, proximity=10, confidence=1.0, area=0) for i in range(10)] + events = tracker.detect_events(rows, 30, proximity_threshold=60, bout_min_frames=5) + self.assertNotIn("courtship_bout", [event["type"] for event in events]) + + def test_assignment_confidence_can_fall_below_low_confidence_threshold(self): + confidence = tracker.assignment_confidence((5, 0), (-5, 0), (0, 0), (0, 0)) + self.assertLess(confidence, tracker.LOW_CONFIDENCE_THRESHOLD) + + def test_frame_sync_treats_opencv_count_as_diagnostic_only(self): + self.assertTrue(tracker.frame_sync_ok(10, 10, 9, 10)) + self.assertTrue(tracker.frame_sync_ok(8, 8, 7, 10)) + self.assertTrue(tracker.frame_sync_ok(8, 8, 7, 0)) + self.assertFalse(tracker.frame_sync_ok(8, 7, 6, 8)) + + def test_invalid_tracking_observation_cannot_be_courtship(self): + invalid = row(0, proximity=10, confidence=1.0, tracking_valid=0) + self.assertFalse(tracker.is_courtship_frame(invalid, 60)) + self.assertTrue(tracker.is_low_confidence_frame(invalid)) + + def test_event_mean_proximity_ignores_missing_values(self): + segment = [ + row(0, proximity=20, confidence=0.8), + row(1, proximity=float("nan"), confidence=0.0, tracking_valid=0), + ] + event = tracker.build_event_record( + "evt-001", "low_confidence_segment", 0, 1, 30, segment, "test" + ) + self.assertEqual(event["mean_proximity_px"], 20.0) + + def test_roi_validation(self): + self.assertEqual(tracker.parse_roi("1,2,3,4"), (1, 2, 3, 4)) + with self.assertRaises(ValueError): + tracker.parse_roi("1,2") + + +if __name__ == "__main__": + unittest.main() diff --git a/source app folder/tracker/tracker.py b/source app folder/tracker/tracker.py index 9ab9533..f342c2e 100644 --- a/source app folder/tracker/tracker.py +++ b/source app folder/tracker/tracker.py @@ -1,66 +1,51 @@ -import cv2 -import pandas as pd -import numpy as np +import argparse +import json import math -import sys import os -import json -import argparse +import sys +from typing import Any, Callable + +import cv2 +import numpy as np +import pandas as pd -# --- Configuration --- -parser = argparse.ArgumentParser(description='Drosophila Fly Tracker') -parser.add_argument('--input', type=str, default=None, help='Path to input video file') -parser.add_argument('--output-video', type=str, default='output_tracked.mp4', help='Path for output tracked video') -parser.add_argument('--output-csv', type=str, default='fly_tracking_data.csv', help='Path for output CSV data') -parser.add_argument('--no-video', action='store_true', help='Disable video output') -parser.add_argument('--min-area', type=int, default=30, help='Minimum contour area in pixels') -parser.add_argument('--max-area', type=int, default=0, help='Maximum contour area in pixels (0 = no limit)') -parser.add_argument('--roi', type=str, default=None, help='ROI rectangle as x,y,w,h in pixels') -parser.add_argument('--proximity-threshold', type=float, default=60.0, help='Courtship bout proximity threshold (px)') -parser.add_argument('--bout-min-frames', type=int, default=90, help='Minimum consecutive frames for a courtship bout') -parser.add_argument('--output-events', type=str, default=None, help='Path for suspected events JSON output') -args = parser.parse_args() - -if args.input: - VIDEO_PATH = os.path.abspath(args.input) -else: - VIDEO_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "assets", "fly_video.mp4")) - -OUTPUT_VIDEO_PATH = args.output_video -OUTPUT_CSV_PATH = args.output_csv -ENABLE_VIDEO_OUTPUT = not args.no_video -MIN_AREA = args.min_area -MAX_AREA = args.max_area -PROXIMITY_THRESHOLD = args.proximity_threshold -BOUT_MIN_FRAMES = args.bout_min_frames -OUTPUT_EVENTS_PATH = args.output_events LOW_CONFIDENCE_THRESHOLD = 0.2 LOW_CONFIDENCE_MIN_FRAMES = 30 - -ROI_RECT = None -if args.roi: +COURTSHIP_MIN_IDENTITY_CONFIDENCE = 0.2 + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Drosophila Fly Tracker") + parser.add_argument("--input", type=str, default=None) + parser.add_argument("--output-video", type=str, default="output_tracked.mp4") + parser.add_argument("--output-csv", type=str, default="fly_tracking_data.csv") + parser.add_argument("--no-video", action="store_true") + parser.add_argument("--min-area", type=int, default=30) + parser.add_argument("--max-area", type=int, default=0) + parser.add_argument("--roi", type=str, default=None) + parser.add_argument("--proximity-threshold", type=float, default=60.0) + parser.add_argument("--bout-min-frames", type=int, default=90) + parser.add_argument("--output-events", type=str, default=None) + return parser.parse_args(argv) + + +def value_or(row: dict[str, Any], key: str, default: float) -> float: + value = row.get(key) + if value is None or value == "": + return default try: - roi_parts = [int(v) for v in args.roi.split(',')] - if len(roi_parts) != 4: - raise ValueError - ROI_RECT = tuple(roi_parts) - except ValueError: - print(f"Error: --roi must be x,y,w,h (got {args.roi!r})") - sys.exit(1) - - -def contour_area_valid(area: float) -> bool: - if area <= MIN_AREA: - return False - if MAX_AREA > 0 and area >= MAX_AREA: - return False - return True - - -def detect_sustained_segments(rows, condition_fn, min_frames: int): - segments = [] - run_start = None - + return float(value) + except (TypeError, ValueError): + return default + + +def detect_sustained_segments( + rows: list[dict[str, Any]], + condition_fn: Callable[[dict[str, Any]], bool], + min_frames: int, +) -> list[tuple[int, int]]: + segments: list[tuple[int, int]] = [] + run_start: int | None = None for row in rows: frame = int(row["frame"]) if condition_fn(row): @@ -71,34 +56,30 @@ def detect_sustained_segments(rows, condition_fn, min_frames: int): if run_end - run_start + 1 >= min_frames: segments.append((run_start, run_end)) run_start = None - - if run_start is not None: + if run_start is not None and rows: run_end = int(rows[-1]["frame"]) if run_end - run_start + 1 >= min_frames: segments.append((run_start, run_end)) - return segments -def segment_rows(rows, start_frame: int, end_frame: int): - return [r for r in rows if start_frame <= int(r["frame"]) <= end_frame] - - def build_event_record( event_id: str, event_type: str, start_frame: int, end_frame: int, fps: float, - segment, + segment: list[dict[str, Any]], detection_reason: str, -) -> dict: +) -> dict[str, Any]: effective_fps = fps if fps > 0 else 30.0 duration_frames = end_frame - start_frame + 1 - proximities = [float(r.get("proximity_distance") or 0) for r in segment] - confidences = [float(r.get("identity_confidence") or 0) for r in segment] - occlusions = [int(r.get("occlusion_flag") or 0) for r in segment] - + proximities = [ + value_or(row, "proximity_distance", math.nan) for row in segment + if math.isfinite(value_or(row, "proximity_distance", math.nan)) + ] + confidences = [value_or(row, "identity_confidence", 0.0) for row in segment] + occlusions = [value_or(row, "occlusion_flag", 0.0) for row in segment] return { "id": event_id, "type": event_type, @@ -114,298 +95,361 @@ def build_event_record( } -def detect_events(rows, fps: float): - events = [] - event_counter = 1 +def is_courtship_frame(row: dict[str, Any], proximity_threshold: float) -> bool: + proximity = value_or(row, "proximity_distance", math.inf) + confidence = value_or(row, "identity_confidence", 0.0) + occluded = value_or(row, "occlusion_flag", 1.0) != 0.0 + fly1_present = value_or(row, "fly1_area", 0.0) > 0.0 + fly2_present = value_or(row, "fly2_area", 0.0) > 0.0 + tracking_valid = value_or(row, "tracking_valid", 1.0) != 0.0 + return ( + tracking_valid + and math.isfinite(proximity) + and not occluded + and fly1_present + and fly2_present + and confidence >= COURTSHIP_MIN_IDENTITY_CONFIDENCE + and proximity < proximity_threshold + ) - courtship_segments = detect_sustained_segments( - rows, - lambda r: float(r.get("proximity_distance") or 999) < PROXIMITY_THRESHOLD, - BOUT_MIN_FRAMES, + +def is_low_confidence_frame(row: dict[str, Any]) -> bool: + return ( + value_or(row, "tracking_valid", 1.0) == 0.0 + or value_or(row, "occlusion_flag", 0.0) != 0.0 + or value_or(row, "identity_confidence", 1.0) < LOW_CONFIDENCE_THRESHOLD ) - for start_frame, end_frame in courtship_segments: - segment = segment_rows(rows, start_frame, end_frame) - events.append( - build_event_record( - f"evt-{event_counter:03d}", - "courtship_bout", - start_frame, - end_frame, - fps, - segment, - "proximity_sustained", - ) - ) - event_counter += 1 - low_conf_segments = detect_sustained_segments( + +def detect_events( + rows: list[dict[str, Any]], + fps: float, + proximity_threshold: float, + bout_min_frames: int, +) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + counter = 1 + courtship = detect_sustained_segments( rows, - lambda r: float(r.get("identity_confidence") or 1) < LOW_CONFIDENCE_THRESHOLD, + lambda row: is_courtship_frame(row, proximity_threshold), + max(1, bout_min_frames), + ) + low_confidence = detect_sustained_segments( + rows, + is_low_confidence_frame, LOW_CONFIDENCE_MIN_FRAMES, ) - for start_frame, end_frame in low_conf_segments: - segment = segment_rows(rows, start_frame, end_frame) - events.append( - build_event_record( - f"evt-{event_counter:03d}", - "low_confidence_segment", - start_frame, - end_frame, - fps, - segment, - "identity_confidence_low", - ) - ) - event_counter += 1 - - events.sort(key=lambda e: e["start_frame"]) + for event_type, segments, reason in ( + ( + "courtship_bout", + courtship, + "separate_flies_with_sustained_proximity_and_identity_confidence", + ), + ("low_confidence_segment", low_confidence, "identity_or_occlusion_uncertain"), + ): + for start, end in segments: + segment = [row for row in rows if start <= int(row["frame"]) <= end] + events.append(build_event_record( + f"evt-{counter:03d}", event_type, start, end, fps, segment, reason, + )) + counter += 1 + events.sort(key=lambda event: event["start_frame"]) return events -def write_events_json(rows, fps: float, output_path: str) -> None: +def write_events_json( + rows: list[dict[str, Any]], + fps: float, + output_path: str, + proximity_threshold: float, + bout_min_frames: int, +) -> None: effective_fps = fps if fps > 0 else 30.0 - events = detect_events(rows, effective_fps) - envelope = { + events = detect_events(rows, effective_fps, proximity_threshold, bout_min_frames) + payload = { "version": 1, "fps": round(effective_fps, 3), "total_frames": len(rows), "detection_params": { - "proximity_threshold_px": PROXIMITY_THRESHOLD, - "bout_min_frames": BOUT_MIN_FRAMES, + "proximity_threshold_px": proximity_threshold, + "bout_min_frames": max(1, bout_min_frames), + "courtship_min_identity_confidence": COURTSHIP_MIN_IDENTITY_CONFIDENCE, + "courtship_requires_separate_flies": True, "low_confidence_threshold": LOW_CONFIDENCE_THRESHOLD, "low_confidence_min_frames": LOW_CONFIDENCE_MIN_FRAMES, }, "events": events, } - with open(output_path, "w", encoding="utf-8") as f: - json.dump(envelope, f, indent=2) + with open(output_path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2) print(f"Events saved to {output_path} ({len(events)} suspected events)") -def assignment_confidence(cA, cB, prev_f1, prev_f2) -> float: - """Confidence from how clearly nearest-neighbor assignment beats the swap.""" - direct = float(math.dist(cA, prev_f1) + math.dist(cB, prev_f2)) - swapped = float(math.dist(cA, prev_f2) + math.dist(cB, prev_f1)) - margin = abs(swapped - direct) - return float(np.clip(margin / (max(direct, swapped) + 1e-6), 0.2, 1.0)) +def displacement(previous, current, was_initialized: bool) -> float: + return float(math.dist(current, previous)) if was_initialized else 0.0 + +def observed_speed(speed: float, tracking_valid: int) -> float: + return float(speed) if tracking_valid else math.nan -if not os.path.exists(VIDEO_PATH): - print(f"Error: Video file not found at {VIDEO_PATH}") - sys.exit(1) -cap = cv2.VideoCapture(VIDEO_PATH) +def assignment_confidence(ca, cb, prev_f1, prev_f2) -> float: + direct = float(math.dist(ca, prev_f1) + math.dist(cb, prev_f2)) + swapped = float(math.dist(ca, prev_f2) + math.dist(cb, prev_f1)) + margin = abs(swapped - direct) + return float(np.clip(margin / (max(direct, swapped) + 1e-6), 0.0, 1.0)) + + +def frame_sync_ok( + frames_processed: int, + csv_rows: int, + last_frame: int, + expected_frame_count: int, +) -> bool: + internal_sync_ok = ( + csv_rows == frames_processed + and (frames_processed == 0 or last_frame == frames_processed - 1) + ) + # OpenCV CAP_PROP_FRAME_COUNT is diagnostic only. The server performs the + # authoritative full decode with ffmpeg and compares it to tracker/CSV/video output. + _ = expected_frame_count + return internal_sync_ok + -width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) -height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) -fps = cap.get(cv2.CAP_PROP_FPS) -expected_frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) +def parse_roi(value: str | None) -> tuple[int, int, int, int] | None: + if not value: + return None + try: + parts = [int(item) for item in value.split(",")] + except ValueError as error: + raise ValueError(f"--roi must be x,y,w,h (got {value!r})") from error + if len(parts) != 4: + raise ValueError(f"--roi must be x,y,w,h (got {value!r})") + return tuple(parts) # type: ignore[return-value] -# Effective FPS for temporal normalization. Existing fly1_speed/fly2_speed stay -# in px/frame (byte-identical to pitch gold); px/sec is exposed via separate -# *_pxsec columns so pitch parity (Rule #2) is preserved. -effective_fps = fps if fps and fps > 0 else 30.0 -if ENABLE_VIDEO_OUTPUT: - fourcc = cv2.VideoWriter_fourcc(*'mp4v') # type: ignore - out = cv2.VideoWriter(OUTPUT_VIDEO_PATH, fourcc, fps, (width, height)) +def run_tracker(args: argparse.Namespace) -> int: + video_path = os.path.abspath(args.input) if args.input else os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "..", "assets", "fly_video.mp4") + ) + if not os.path.exists(video_path): + print(f"Error: Video file not found at {video_path}") + return 1 + try: + roi_rect = parse_roi(args.roi) + except ValueError as error: + print(f"Error: {error}") + return 1 + + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + print(f"Error: Could not open video at {video_path}") + return 1 + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + fps = float(cap.get(cv2.CAP_PROP_FPS)) + effective_fps = fps if fps > 0 else 30.0 + expected_frame_count = max(0, int(cap.get(cv2.CAP_PROP_FRAME_COUNT))) + + out = None + if not args.no_video: + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + out = cv2.VideoWriter(args.output_video, fourcc, effective_fps, (width, height)) + if not out.isOpened(): + cap.release() + print(f"Error: Could not create output video at {args.output_video}") + return 1 + + fgbg = cv2.createBackgroundSubtractorMOG2( + history=500, varThreshold=50, detectShadows=False, + ) + kernel = np.ones((5, 5), np.uint8) + data: list[dict[str, Any]] = [] + frame_num = 0 + is_initialized = False + prev_f1 = (0, 0) + prev_f2 = (0, 0) + + print(f"Starting tracking on {video_path}...") + if roi_rect: + print( + f"ROI active: x={roi_rect[0]}, y={roi_rect[1]}, " + f"w={roi_rect[2]}, h={roi_rect[3]}" + ) -fgbg = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=50, detectShadows=False) + try: + while cap.isOpened(): + ret, frame = cap.read() + if not ret: + break + fgmask = fgbg.apply(frame) + fgmask = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel) + fgmask = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel) + if roi_rect: + rx, ry, rw, rh = roi_rect + roi_mask = np.zeros_like(fgmask) + roi_mask[ry:ry + rh, rx:rx + rw] = 255 + fgmask = cv2.bitwise_and(fgmask, roi_mask) + + activity_level = int(cv2.countNonZero(fgmask)) + contours, _ = cv2.findContours( + fgmask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE, + ) + valid = [ + contour for contour in contours + if cv2.contourArea(contour) > args.min_area + and (args.max_area <= 0 or cv2.contourArea(contour) < args.max_area) + ] + valid.sort(key=cv2.contourArea, reverse=True) + centroids, bboxes, areas = [], [], [] + for contour in valid[:2]: + x, y, w, h = cv2.boundingRect(contour) + centroids.append((int(x + w / 2), int(y + h / 2))) + bboxes.append((x, y, w, h)) + areas.append(float(cv2.contourArea(contour))) + + f1_coords = f2_coords = (0, 0) + f1_speed = f2_speed = math.nan + proximity = math.nan + occlusion_flag = 0 + identity_confidence = 0.0 + fly1_area = fly2_area = 0.0 + detection_count = len(centroids) + tracking_valid = 0 + + if len(centroids) == 2: + ca, cb = centroids + was_initialized = is_initialized + if not was_initialized: + f1, f2 = ca, cb + is_initialized = True + identity_confidence = 0.5 + else: + direct = math.dist(ca, prev_f1) + math.dist(cb, prev_f2) + swapped = math.dist(ca, prev_f2) + math.dist(cb, prev_f1) + f1, f2 = (ca, cb) if direct < swapped else (cb, ca) + identity_confidence = assignment_confidence(ca, cb, prev_f1, prev_f2) + f1_coords, f2_coords = f1, f2 + if f1 == ca: + fly1_area, fly2_area = areas[0], areas[1] + else: + fly1_area, fly2_area = areas[1], areas[0] + if was_initialized: + f1_speed = displacement(prev_f1, f1_coords, was_initialized) + f2_speed = displacement(prev_f2, f2_coords, was_initialized) + else: + f1_speed = f2_speed = 0.0 + proximity = float(math.dist(f1_coords, f2_coords)) + tracking_valid = 1 + prev_f1, prev_f2 = f1_coords, f2_coords + if out: + for x, y, w, h in bboxes: + cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2) + cv2.circle(frame, f1_coords, 5, (255, 0, 0), -1) + cv2.circle(frame, f2_coords, 5, (0, 0, 255), -1) + cv2.line(frame, f1_coords, f2_coords, (0, 255, 255), 2) + text_pos = ( + min(f1_coords[0], f2_coords[0]), + max(0, min(f1_coords[1], f2_coords[1]) - 10), + ) + cv2.putText( + frame, + f"Dist: {int(proximity)}px", + text_pos, + cv2.FONT_HERSHEY_SIMPLEX, + 0.6, + (0, 255, 255), + 2, + ) + elif len(centroids) == 1: + point = centroids[0] + was_initialized = is_initialized + f1_coords = f2_coords = point + occlusion_flag = 1 + identity_confidence = 0.0 + fly1_area = fly2_area = areas[0] + prev_f1 = prev_f2 = point + is_initialized = True + if out: + x, y, w, h = bboxes[0] + cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 255), 3) + cv2.putText( + frame, + "MERGED", + (x, max(0, y - 10)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.6, + (0, 255, 255), + 2, + ) + elif is_initialized: + # Coordinates are retained only for display continuity. Proximity is + # deliberately missing because no two-fly observation occurred. + f1_coords, f2_coords = prev_f1, prev_f2 + identity_confidence = 0.0 + + measured_f1_speed = observed_speed(f1_speed, tracking_valid) + measured_f2_speed = observed_speed(f2_speed, tracking_valid) + data.append({ + "frame": frame_num, + "fly1_x": f1_coords[0], + "fly1_y": f1_coords[1], + "fly2_x": f2_coords[0], + "fly2_y": f2_coords[1], + "fly1_speed": measured_f1_speed, + "fly2_speed": measured_f2_speed, + "fly1_speed_pxsec": round(measured_f1_speed * effective_fps, 4), + "fly2_speed_pxsec": round(measured_f2_speed * effective_fps, 4), + "activity_level": activity_level, + "proximity_distance": proximity, + "tracking_valid": tracking_valid, + "detection_count": detection_count, + "occlusion_flag": occlusion_flag, + "identity_confidence": round(identity_confidence, 4), + "fly1_area": fly1_area, + "fly2_area": fly2_area, + }) + if out: + out.write(frame) + frame_num += 1 + if frame_num % 100 == 0: + print(f"Processed {frame_num} frames...") + finally: + cap.release() + if out: + out.release() + cv2.destroyAllWindows() + + pd.DataFrame(data).to_csv(args.output_csv, index=False) + if args.output_events: + write_events_json( + data, + effective_fps, + args.output_events, + args.proximity_threshold, + args.bout_min_frames, + ) + last_frame = data[-1]["frame"] if data else -1 + sync_ok = frame_sync_ok( + frame_num, len(data), last_frame, expected_frame_count, + ) + print( + f"TRACKER_SYNC frames_processed={frame_num} csv_rows={len(data)} " + f"expected_video_frames={expected_frame_count} sync_ok={str(sync_ok).lower()}" + ) + if not sync_ok: + print( + "Error: frame integrity mismatch between tracker loop and CSV", + file=sys.stderr, + ) + return 2 + print(f"Tracking completed! Data saved to {args.output_csv}.") + return 0 -data = [] -frame_num = 0 -is_initialized = False -prev_f1 = (0, 0) -prev_f2 = (0, 0) -print(f"Starting tracking on {VIDEO_PATH}...") -if ROI_RECT: - print(f"ROI active: x={ROI_RECT[0]}, y={ROI_RECT[1]}, w={ROI_RECT[2]}, h={ROI_RECT[3]}") -print(f"Contour area filter: {MIN_AREA} < area" + (f" < {MAX_AREA}" if MAX_AREA > 0 else "")) +def main(argv: list[str] | None = None) -> int: + return run_tracker(parse_args(argv)) -while cap.isOpened(): - ret, frame = cap.read() - if not ret: - break - fgmask = fgbg.apply(frame) - kernel = np.ones((5, 5), np.uint8) - fgmask = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel) - fgmask = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel) - - if ROI_RECT: - rx, ry, rw, rh = ROI_RECT - roi_mask = np.zeros_like(fgmask) - roi_mask[ry:ry + rh, rx:rx + rw] = 255 - fgmask = cv2.bitwise_and(fgmask, roi_mask) - - activity_level = int(cv2.countNonZero(fgmask)) - - contours, _ = cv2.findContours(fgmask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) - valid_contours = [cnt for cnt in contours if contour_area_valid(cv2.contourArea(cnt))] - valid_contours.sort(key=cv2.contourArea, reverse=True) - - current_centroids = [] - bboxes = [] - contour_areas = [] - - max_idx = min(2, len(valid_contours)) - for i in range(max_idx): - cnt = valid_contours[i] - x, y, w, h = cv2.boundingRect(cnt) - cx = int(x + w / 2) - cy = int(y + h / 2) - current_centroids.append((cx, cy)) - bboxes.append((x, y, w, h)) - contour_areas.append(float(cv2.contourArea(cnt))) - - f1_coords = (0, 0) - f2_coords = (0, 0) - f1_speed = 0.0 - f2_speed = 0.0 - f1_speed_pxsec = 0.0 - f2_speed_pxsec = 0.0 - proximity = 0.0 - occlusion_flag = 0 - identity_conf = 0.0 - fly1_area = 0.0 - fly2_area = 0.0 - - if len(current_centroids) == 2: - cA, cB = current_centroids[0], current_centroids[1] - - if not is_initialized: - f1, f2 = cA, cB - is_initialized = True - identity_conf = 0.5 - else: - dist_A_to_prev1 = float(math.dist(cA, prev_f1)) - dist_A_to_prev2 = float(math.dist(cA, prev_f2)) - dist_B_to_prev1 = float(math.dist(cB, prev_f1)) - dist_B_to_prev2 = float(math.dist(cB, prev_f2)) - - if dist_A_to_prev1 + dist_B_to_prev2 < dist_A_to_prev2 + dist_B_to_prev1: - f1, f2 = cA, cB - else: - f1, f2 = cB, cA - - identity_conf = assignment_confidence(cA, cB, prev_f1, prev_f2) - - f1_coords = f1 - f2_coords = f2 - if f1 == cA: - fly1_area = contour_areas[0] - fly2_area = contour_areas[1] - else: - fly1_area = contour_areas[1] - fly2_area = contour_areas[0] - - if is_initialized: - # Core pitch columns: displacement per frame in px/frame (0-diff parity). - f1_speed = float(math.dist(f1_coords, prev_f1)) - f2_speed = float(math.dist(f2_coords, prev_f2)) - # Flyt temporal-normalized columns: physical speed in px/sec. - f1_speed_pxsec = f1_speed * effective_fps - f2_speed_pxsec = f2_speed * effective_fps - - proximity = float(math.dist(f1_coords, f2_coords)) - prev_f1 = f1_coords - prev_f2 = f2_coords - - if ENABLE_VIDEO_OUTPUT: - for (x, y, w, h) in bboxes: - cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2) - - cv2.circle(frame, f1_coords, 5, (255, 0, 0), -1) - cv2.circle(frame, f2_coords, 5, (0, 0, 255), -1) - cv2.line(frame, f1_coords, f2_coords, (0, 255, 255), 2) - - text_pos = (min(f1_coords[0], f2_coords[0]), max(0, min(f1_coords[1], f2_coords[1]) - 10)) - cv2.putText(frame, f"Dist: {int(proximity)}px", text_pos, cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2) - - elif len(current_centroids) == 1: - cA = current_centroids[0] - f1_coords = cA - f2_coords = cA - occlusion_flag = 1 - identity_conf = 0.25 - fly1_area = contour_areas[0] - fly2_area = contour_areas[0] - - if is_initialized: - # Core pitch columns (px/frame, parity-identical to pitch on merge frames). - f1_speed = float(math.dist(f1_coords, prev_f1)) - f2_speed = float(math.dist(f2_coords, prev_f2)) - f1_speed_pxsec = f1_speed * effective_fps - f2_speed_pxsec = f2_speed * effective_fps - - proximity = 0.0 - prev_f1 = f1_coords - prev_f2 = f2_coords - is_initialized = True - - if ENABLE_VIDEO_OUTPUT: - x, y, w, h = bboxes[0] - cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 255), 3) - cv2.putText(frame, "MERGED", (x, max(0, y - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2) - - else: - if is_initialized: - f1_coords = prev_f1 - f2_coords = prev_f2 - proximity = float(math.dist(f1_coords, f2_coords)) - identity_conf = 0.15 - - data.append({ - "frame": frame_num, - "fly1_x": f1_coords[0], - "fly1_y": f1_coords[1], - "fly2_x": f2_coords[0], - "fly2_y": f2_coords[1], - "fly1_speed": f1_speed, - "fly2_speed": f2_speed, - "fly1_speed_pxsec": round(f1_speed_pxsec, 4), - "fly2_speed_pxsec": round(f2_speed_pxsec, 4), - "activity_level": activity_level, - "proximity_distance": proximity, - "occlusion_flag": occlusion_flag, - "identity_confidence": round(identity_conf, 4), - "fly1_area": fly1_area, - "fly2_area": fly2_area, - }) - - if ENABLE_VIDEO_OUTPUT: - out.write(frame) - - frame_num += 1 - - if frame_num % 100 == 0: - print(f"Processed {frame_num} frames...") - -cap.release() -if ENABLE_VIDEO_OUTPUT: - out.release() -cv2.destroyAllWindows() - -df = pd.DataFrame(data) -df.to_csv(OUTPUT_CSV_PATH, index=False) - -if OUTPUT_EVENTS_PATH: - write_events_json(data, fps, OUTPUT_EVENTS_PATH) - -csv_rows = len(data) -frames_processed = frame_num -last_frame_idx = data[-1]["frame"] if data else -1 -sync_ok = ( - csv_rows == frames_processed - and (frames_processed == 0 or last_frame_idx == frames_processed - 1) -) -print( - f"TRACKER_SYNC frames_processed={frames_processed} csv_rows={csv_rows} " - f"expected_video_frames={expected_frame_count} sync_ok={sync_ok}" -) -if not sync_ok: - print("Warning: internal frame/CSV sync mismatch detected", file=sys.stderr) - sys.exit(2) - -print(f"Tracking completed! Data saved to {OUTPUT_CSV_PATH}.") \ No newline at end of file +if __name__ == "__main__": + raise SystemExit(main())