mirror of
https://github.com/legop3/MultiRoombaRover.git
synced 2026-09-16 01:21:20 -04:00
reapaltle
This commit is contained in:
legop3
@@ -3,8 +3,7 @@ const path = require('path');
|
||||
const roverManager = require('./roverManager');
|
||||
const { getRoomCameras } = require('./roomCameraService');
|
||||
const { getRoomCameraState } = require('./roomCameraSnapshotService');
|
||||
const { getReplaySources } = require('./replaySourceService');
|
||||
const { replaySegmentsDir, segmentSeconds, bufferSeconds } = require('./replaySegmentManager');
|
||||
const { getReplayHealthSnapshot } = require('./replayEngineV2');
|
||||
|
||||
const ROVER_SNAPSHOT_DIR = process.env.ROVER_SNAPSHOT_DIR || '/var/lib/rover-snapshots';
|
||||
const HEALTH_INTERVAL_MS = 5000;
|
||||
@@ -17,44 +16,8 @@ let latest = {
|
||||
snapshots: { rovers: [], rooms: [] },
|
||||
};
|
||||
|
||||
async function collectReplayHealth(now) {
|
||||
const neededCount = Math.max(1, Math.ceil(20000 / (segmentSeconds * 1000)));
|
||||
const sources = getReplaySources();
|
||||
const list = [];
|
||||
let readyCount = 0;
|
||||
for (const source of sources) {
|
||||
const key = `${source.type}__${source.id}`;
|
||||
const dir = path.join(replaySegmentsDir, key);
|
||||
let lastSegmentAt = null;
|
||||
let recentCount = 0;
|
||||
try {
|
||||
const entries = await fsp.readdir(dir, { withFileTypes: true });
|
||||
for (const entry of entries) {
|
||||
if (!entry.isFile() || !entry.name.endsWith('.mp4')) continue;
|
||||
const stat = await fsp.stat(path.join(dir, entry.name));
|
||||
if (stat.mtimeMs > (lastSegmentAt || 0)) {
|
||||
lastSegmentAt = stat.mtimeMs;
|
||||
}
|
||||
if (now - stat.mtimeMs <= bufferSeconds * 1000) {
|
||||
recentCount += 1;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// directory missing or unreadable
|
||||
}
|
||||
const ready = recentCount >= neededCount;
|
||||
if (ready) readyCount += 1;
|
||||
list.push({
|
||||
type: source.type,
|
||||
id: source.id,
|
||||
label: source.label || `${source.type}:${source.id}`,
|
||||
recentCount,
|
||||
neededCount,
|
||||
lastSegmentAt,
|
||||
ready,
|
||||
});
|
||||
}
|
||||
return { sources: list, readyCount, totalCount: list.length };
|
||||
function collectReplayHealth() {
|
||||
return getReplayHealthSnapshot();
|
||||
}
|
||||
|
||||
async function collectSnapshotHealth(now) {
|
||||
@@ -109,7 +72,7 @@ async function collectSnapshotHealth(now) {
|
||||
|
||||
async function refreshHealth() {
|
||||
const now = Date.now();
|
||||
const replay = await collectReplayHealth(now);
|
||||
const replay = collectReplayHealth(now);
|
||||
const snapshots = await collectSnapshotHealth(now);
|
||||
latest = { updatedAt: now, replay, snapshots };
|
||||
}
|
||||
|
||||
@@ -22,8 +22,8 @@ const ACTIVITY_WINDOW_MS = 60000;
|
||||
const ACTIVITY_BUCKET_MS = 1000;
|
||||
const ACTIVITY_SCORE_WINDOW_MS = 30000;
|
||||
const SELF_TALK_WINDOW_MS = 30 * 60 * 1000;
|
||||
const MAX_CONTEXT_EVENTS = 15;
|
||||
const MAX_RUN_HISTORY = 30;
|
||||
const MAX_CONTEXT_EVENTS = 8;
|
||||
const MAX_RUN_HISTORY = 100;
|
||||
const MAX_ROVER_EVENTS = 400;
|
||||
const POST_COOLDOWN_MS = 10000;
|
||||
|
||||
|
||||
@@ -1,218 +1,4 @@
|
||||
const { execFile } = require('child_process');
|
||||
const fsp = require('fs/promises');
|
||||
const os = require('os');
|
||||
const path = require('path');
|
||||
const { promisify } = require('util');
|
||||
const logger = require('../globals/logger').child('replayBuild');
|
||||
const { replaySegmentsDir, segmentSeconds } = require('./replaySegmentManager');
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
const REPLAY_DURATION_MS = 20000;
|
||||
const REPLAY_FPS = 15;
|
||||
const MAX_REPLAY_WIDTH = 1280;
|
||||
const MAX_REPLAY_HEIGHT = 720;
|
||||
const REPLAY_MAX_BYTES = Math.floor(9.5 * 1024 * 1024);
|
||||
|
||||
function buildGridLayout(count) {
|
||||
const cols = Math.ceil(Math.sqrt(count));
|
||||
const rows = Math.ceil(count / cols);
|
||||
return { cols, rows };
|
||||
}
|
||||
|
||||
function clampEven(value) {
|
||||
return Math.max(2, Math.floor(value / 2) * 2);
|
||||
}
|
||||
|
||||
function buildScalePadFilter(tileWidth, tileHeight) {
|
||||
return `scale=${tileWidth}:${tileHeight}:force_original_aspect_ratio=decrease:flags=lanczos,pad=${tileWidth}:${tileHeight}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1`;
|
||||
}
|
||||
|
||||
async function probeMaxFrameSize(paths) {
|
||||
let maxWidth = 0;
|
||||
let maxHeight = 0;
|
||||
for (const filePath of paths) {
|
||||
try {
|
||||
const { stdout } = await execFileAsync('ffprobe', [
|
||||
'-v',
|
||||
'error',
|
||||
'-select_streams',
|
||||
'v:0',
|
||||
'-show_entries',
|
||||
'stream=width,height',
|
||||
'-of',
|
||||
'csv=p=0',
|
||||
filePath,
|
||||
]);
|
||||
const [widthRaw, heightRaw] = stdout.trim().split(',');
|
||||
const width = Number(widthRaw);
|
||||
const height = Number(heightRaw);
|
||||
if (Number.isFinite(width) && Number.isFinite(height)) {
|
||||
maxWidth = Math.max(maxWidth, width);
|
||||
maxHeight = Math.max(maxHeight, height);
|
||||
}
|
||||
} catch (err) {
|
||||
logger.warn('Failed to probe replay clip size', err.message);
|
||||
}
|
||||
}
|
||||
return { maxWidth, maxHeight };
|
||||
}
|
||||
|
||||
async function listLatestSegments(sourceKey, neededCount) {
|
||||
const dir = path.join(replaySegmentsDir, sourceKey);
|
||||
const entries = await fsp.readdir(dir, { withFileTypes: true });
|
||||
const files = [];
|
||||
const now = Date.now();
|
||||
for (const entry of entries) {
|
||||
if (!entry.isFile() || !entry.name.endsWith('.mp4')) continue;
|
||||
const filePath = path.join(dir, entry.name);
|
||||
const stat = await fsp.stat(filePath);
|
||||
if (stat.size < 16 * 1024) {
|
||||
continue;
|
||||
}
|
||||
if (now - stat.mtimeMs < segmentSeconds * 1000) {
|
||||
continue;
|
||||
}
|
||||
files.push({ filePath, mtimeMs: stat.mtimeMs });
|
||||
}
|
||||
files.sort((a, b) => a.mtimeMs - b.mtimeMs);
|
||||
return files.slice(-neededCount).map((file) => file.filePath);
|
||||
}
|
||||
|
||||
async function buildReplayVideo({ sources = [] } = {}) {
|
||||
if (!sources.length) {
|
||||
throw new Error('No replay sources selected');
|
||||
}
|
||||
const segmentCount = Math.max(1, Math.ceil(REPLAY_DURATION_MS / (segmentSeconds * 1000)));
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'rover-replay-'));
|
||||
try {
|
||||
const clipPaths = [];
|
||||
const usedSources = [];
|
||||
const missingSources = [];
|
||||
for (let i = 0; i < sources.length; i += 1) {
|
||||
const source = sources[i];
|
||||
const key = `${source.type}__${source.id}`;
|
||||
let segmentPaths;
|
||||
try {
|
||||
segmentPaths = await listLatestSegments(key, segmentCount);
|
||||
} catch (err) {
|
||||
missingSources.push({ ...source, reason: err.message || 'missing segments' });
|
||||
continue;
|
||||
}
|
||||
if (segmentPaths.length < segmentCount) {
|
||||
missingSources.push({
|
||||
...source,
|
||||
reason: `only ${segmentPaths.length}/${segmentCount} segments available`,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const concatPath = path.join(tmpDir, `concat-${clipPaths.length}.txt`);
|
||||
const concatBody = segmentPaths.map((file) => `file '${file}'`).join('\n');
|
||||
await fsp.writeFile(concatPath, concatBody);
|
||||
const clipPath = path.join(tmpDir, `clip-${clipPaths.length}.mp4`);
|
||||
await execFileAsync('ffmpeg', [
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-f',
|
||||
'concat',
|
||||
'-safe',
|
||||
'0',
|
||||
'-i',
|
||||
concatPath,
|
||||
'-c',
|
||||
'copy',
|
||||
clipPath,
|
||||
]);
|
||||
clipPaths.push(clipPath);
|
||||
usedSources.push(source);
|
||||
}
|
||||
if (!clipPaths.length) {
|
||||
throw new Error('No replay segments available for selected sources');
|
||||
}
|
||||
if (missingSources.length) {
|
||||
logger.warn('Replay sources missing segments', {
|
||||
missing: missingSources.map((source) => `${source.type}:${source.id}`),
|
||||
});
|
||||
}
|
||||
|
||||
const { maxWidth, maxHeight } = await probeMaxFrameSize(clipPaths);
|
||||
const layout = buildGridLayout(clipPaths.length);
|
||||
let tileWidth = maxWidth || 640;
|
||||
let tileHeight = maxHeight || 360;
|
||||
let outputWidth = tileWidth * layout.cols;
|
||||
let outputHeight = tileHeight * layout.rows;
|
||||
if (outputWidth > MAX_REPLAY_WIDTH || outputHeight > MAX_REPLAY_HEIGHT) {
|
||||
const scale = Math.min(MAX_REPLAY_WIDTH / outputWidth, MAX_REPLAY_HEIGHT / outputHeight);
|
||||
tileWidth *= scale;
|
||||
tileHeight *= scale;
|
||||
outputWidth = tileWidth * layout.cols;
|
||||
outputHeight = tileHeight * layout.rows;
|
||||
}
|
||||
tileWidth = clampEven(tileWidth);
|
||||
tileHeight = clampEven(tileHeight);
|
||||
outputWidth = clampEven(tileWidth * layout.cols);
|
||||
outputHeight = clampEven(tileHeight * layout.rows);
|
||||
|
||||
const inputArgs = [];
|
||||
const filterParts = [];
|
||||
const layoutParts = [];
|
||||
for (let i = 0; i < clipPaths.length; i += 1) {
|
||||
inputArgs.push('-i', clipPaths[i]);
|
||||
filterParts.push(`[${i}:v]${buildScalePadFilter(tileWidth, tileHeight)}[v${i}]`);
|
||||
const x = (i % layout.cols) * tileWidth;
|
||||
const y = Math.floor(i / layout.cols) * tileHeight;
|
||||
layoutParts.push(`${x}_${y}`);
|
||||
}
|
||||
if (clipPaths.length === 1) {
|
||||
filterParts.push('[v0]null[v]');
|
||||
} else {
|
||||
filterParts.push(
|
||||
`${clipPaths.map((_, i) => `[v${i}]`).join('')}` +
|
||||
`xstack=inputs=${clipPaths.length}:layout=${layoutParts.join('|')}:fill=black[v]`,
|
||||
);
|
||||
}
|
||||
|
||||
const durationSec = Math.max(1, REPLAY_DURATION_MS / 1000);
|
||||
const targetBitrateKbps = Math.max(300, Math.floor((REPLAY_MAX_BYTES * 8) / durationSec / 1000));
|
||||
const maxrateKbps = Math.floor(targetBitrateKbps * 1.1);
|
||||
const bufsizeKbps = Math.floor(targetBitrateKbps * 2);
|
||||
|
||||
const outPath = path.join(tmpDir, 'replay.mp4');
|
||||
await execFileAsync('ffmpeg', [
|
||||
'-y',
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
...inputArgs,
|
||||
'-filter_complex',
|
||||
filterParts.join(';'),
|
||||
'-map',
|
||||
'[v]',
|
||||
'-r',
|
||||
String(REPLAY_FPS),
|
||||
'-c:v',
|
||||
'libx264',
|
||||
'-b:v',
|
||||
`${targetBitrateKbps}k`,
|
||||
'-maxrate',
|
||||
`${maxrateKbps}k`,
|
||||
'-bufsize',
|
||||
`${bufsizeKbps}k`,
|
||||
'-pix_fmt',
|
||||
'yuv420p',
|
||||
outPath,
|
||||
]);
|
||||
const buffer = await fsp.readFile(outPath);
|
||||
return { buffer, usedSources, missingSources };
|
||||
} finally {
|
||||
try {
|
||||
await fsp.rm(tmpDir, { recursive: true, force: true });
|
||||
} catch (err) {
|
||||
logger.warn('Failed to cleanup replay temp dir', err.message);
|
||||
}
|
||||
}
|
||||
}
|
||||
const { buildReplayVideo } = require('./replayEngineV2');
|
||||
|
||||
module.exports = {
|
||||
buildReplayVideo,
|
||||
|
||||
@@ -0,0 +1,718 @@
|
||||
const { spawn, execFile } = require('child_process');
|
||||
const fsp = require('fs/promises');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
const path = require('path');
|
||||
const { promisify } = require('util');
|
||||
const EventEmitter = require('events');
|
||||
const logger = require('../globals/logger').child('replayEngineV2');
|
||||
const roverManager = require('./roverManager');
|
||||
const { getRoomCameras, roomCameraEvents } = require('./roomCameraService');
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
const FFMPEG_BIN = process.env.FFMPEG_BIN || 'ffmpeg';
|
||||
const SEGMENT_ROOT = process.env.REPLAY_SEGMENT_DIR || '/var/lib/replay-segments-v2';
|
||||
const SEGMENT_SECONDS = Math.max(1, Number.parseInt(process.env.REPLAY_SEGMENT_SECONDS || '1', 10));
|
||||
const BUFFER_SECONDS = Math.max(20, Number.parseInt(process.env.REPLAY_BUFFER_SECONDS || '45', 10));
|
||||
const CLEANUP_INTERVAL_MS = 10_000;
|
||||
const BUILD_DURATION_MS = Math.max(5000, Number.parseInt(process.env.REPLAY_DURATION_MS || '20000', 10));
|
||||
const BUILD_GUARD_MS = Math.max(200, Number.parseInt(process.env.REPLAY_GUARD_MS || '1200', 10));
|
||||
const TARGET_FPS = Math.max(10, Number.parseInt(process.env.REPLAY_TARGET_FPS || '30', 10));
|
||||
const MAX_WIDTH = Math.max(320, Number.parseInt(process.env.REPLAY_MAX_WIDTH || '1280', 10));
|
||||
const MAX_HEIGHT = Math.max(180, Number.parseInt(process.env.REPLAY_MAX_HEIGHT || '720', 10));
|
||||
const MAX_BYTES = Math.floor(Number.parseFloat(process.env.REPLAY_MAX_OUTPUT_MB || '9.5') * 1024 * 1024);
|
||||
|
||||
const events = new EventEmitter();
|
||||
|
||||
const workers = new Map(); // key -> worker
|
||||
const segmentIndex = new Map(); // key -> [{filePath,startMs,endMs,mtimeMs,size,kind,sourceType,sourceId}]
|
||||
let cleanupTimer = null;
|
||||
|
||||
function sourceKey(source) {
|
||||
return `${source.sourceType}__${source.kind}__${source.id}`;
|
||||
}
|
||||
|
||||
function sourceDirForKey(key) {
|
||||
return path.join(SEGMENT_ROOT, key);
|
||||
}
|
||||
|
||||
function toSrtReadPath(streamId) {
|
||||
return `srt://127.0.0.1:9000?streamid=read:${encodeURIComponent(streamId)}`;
|
||||
}
|
||||
|
||||
function getRoomCameraStream(camera) {
|
||||
if (camera?.streamUrl) return camera.streamUrl;
|
||||
const url = String(camera?.url || '');
|
||||
if (url.includes('.mjpg') || url.includes('mjpeg') || url.includes('stream')) return url;
|
||||
return null;
|
||||
}
|
||||
|
||||
function listDesiredSources() {
|
||||
const sources = [];
|
||||
for (const rover of roverManager.getRoster()) {
|
||||
if (!roverManager.canReplayRoverId(rover.id)) continue;
|
||||
const roverId = String(rover.id);
|
||||
sources.push({
|
||||
id: roverId,
|
||||
sourceType: 'rover',
|
||||
kind: 'video',
|
||||
label: rover.name || roverId,
|
||||
inputUrl: toSrtReadPath(roverId),
|
||||
});
|
||||
if (rover?.media?.audioPublishUrl) {
|
||||
sources.push({
|
||||
id: `${roverId}-audio`,
|
||||
sourceType: 'rover',
|
||||
roverId,
|
||||
kind: 'audio',
|
||||
label: `${rover.name || roverId} audio`,
|
||||
inputUrl: toSrtReadPath(`${roverId}-audio`),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
for (const camera of getRoomCameras()) {
|
||||
const streamUrl = getRoomCameraStream(camera);
|
||||
if (!streamUrl) continue;
|
||||
sources.push({
|
||||
id: String(camera.id),
|
||||
sourceType: 'room',
|
||||
kind: 'video',
|
||||
label: camera.name || camera.id,
|
||||
inputUrl: streamUrl,
|
||||
});
|
||||
}
|
||||
|
||||
return sources;
|
||||
}
|
||||
|
||||
function buildWorkerArgs(source) {
|
||||
const dir = sourceDirForKey(sourceKey(source));
|
||||
const pattern = path.join(dir, 'seg-%06d.mp4');
|
||||
const listPath = path.join(dir, 'index.csv');
|
||||
const listSize = Math.ceil((BUFFER_SECONDS + 10) / SEGMENT_SECONDS);
|
||||
|
||||
const common = [
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'warning',
|
||||
'-y',
|
||||
'-fflags',
|
||||
'+genpts',
|
||||
'-use_wallclock_as_timestamps',
|
||||
'1',
|
||||
'-i',
|
||||
source.inputUrl,
|
||||
];
|
||||
|
||||
if (source.kind === 'audio') {
|
||||
return [
|
||||
...common,
|
||||
'-vn',
|
||||
'-ac',
|
||||
'1',
|
||||
'-ar',
|
||||
'48000',
|
||||
'-af',
|
||||
'aresample=48000',
|
||||
'-c:a',
|
||||
'aac',
|
||||
'-b:a',
|
||||
'96k',
|
||||
'-f',
|
||||
'segment',
|
||||
'-segment_time',
|
||||
String(SEGMENT_SECONDS),
|
||||
'-segment_atclocktime',
|
||||
'1',
|
||||
'-segment_list',
|
||||
listPath,
|
||||
'-segment_list_type',
|
||||
'csv',
|
||||
'-segment_list_size',
|
||||
String(listSize),
|
||||
'-segment_list_flags',
|
||||
'+live',
|
||||
'-reset_timestamps',
|
||||
'1',
|
||||
pattern,
|
||||
];
|
||||
}
|
||||
|
||||
return [
|
||||
...common,
|
||||
'-an',
|
||||
'-vf',
|
||||
`fps=${TARGET_FPS}`,
|
||||
'-c:v',
|
||||
'libx264',
|
||||
'-preset',
|
||||
'veryfast',
|
||||
'-tune',
|
||||
'zerolatency',
|
||||
'-pix_fmt',
|
||||
'yuv420p',
|
||||
'-g',
|
||||
String(TARGET_FPS * SEGMENT_SECONDS),
|
||||
'-keyint_min',
|
||||
String(TARGET_FPS * SEGMENT_SECONDS),
|
||||
'-sc_threshold',
|
||||
'0',
|
||||
'-f',
|
||||
'segment',
|
||||
'-segment_time',
|
||||
String(SEGMENT_SECONDS),
|
||||
'-segment_atclocktime',
|
||||
'1',
|
||||
'-segment_list',
|
||||
listPath,
|
||||
'-segment_list_type',
|
||||
'csv',
|
||||
'-segment_list_size',
|
||||
String(listSize),
|
||||
'-segment_list_flags',
|
||||
'+live',
|
||||
'-reset_timestamps',
|
||||
'1',
|
||||
pattern,
|
||||
];
|
||||
}
|
||||
|
||||
async function ensureDir(dir) {
|
||||
await fsp.mkdir(dir, { recursive: true });
|
||||
}
|
||||
|
||||
function startWorker(source) {
|
||||
const key = sourceKey(source);
|
||||
if (workers.has(key)) return;
|
||||
const dir = sourceDirForKey(key);
|
||||
ensureDir(dir)
|
||||
.then(() => {
|
||||
const proc = spawn(FFMPEG_BIN, buildWorkerArgs(source), { stdio: ['ignore', 'ignore', 'pipe'] });
|
||||
const worker = {
|
||||
key,
|
||||
source,
|
||||
proc,
|
||||
startedAtMs: Date.now(),
|
||||
firstSegmentStartSec: null,
|
||||
restarting: false,
|
||||
};
|
||||
workers.set(key, worker);
|
||||
proc.stderr.on('data', (chunk) => {
|
||||
const text = String(chunk || '').trim();
|
||||
if (!text) return;
|
||||
logger.warn('worker stderr', { key, text: text.slice(0, 500) });
|
||||
});
|
||||
proc.on('exit', (code, signal) => {
|
||||
const current = workers.get(key);
|
||||
if (current?.proc === proc) {
|
||||
workers.delete(key);
|
||||
}
|
||||
logger.warn('worker exited', { key, code, signal });
|
||||
setTimeout(() => {
|
||||
const desired = listDesiredSources().find((entry) => sourceKey(entry) === key);
|
||||
if (desired && !workers.has(key)) {
|
||||
startWorker(desired);
|
||||
}
|
||||
}, 1500);
|
||||
});
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.warn('failed to start worker', { key, error: err.message });
|
||||
});
|
||||
}
|
||||
|
||||
function stopWorker(key) {
|
||||
const worker = workers.get(key);
|
||||
if (!worker) return;
|
||||
try {
|
||||
worker.proc.kill('SIGTERM');
|
||||
} catch {
|
||||
// noop
|
||||
}
|
||||
workers.delete(key);
|
||||
}
|
||||
|
||||
async function syncWorkers() {
|
||||
const desired = listDesiredSources();
|
||||
const desiredKeys = new Set(desired.map(sourceKey));
|
||||
|
||||
for (const source of desired) {
|
||||
const key = sourceKey(source);
|
||||
if (!workers.has(key)) {
|
||||
startWorker(source);
|
||||
}
|
||||
}
|
||||
|
||||
for (const key of Array.from(workers.keys())) {
|
||||
if (!desiredKeys.has(key)) {
|
||||
stopWorker(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function parseCsvLine(line) {
|
||||
const trimmed = String(line || '').trim();
|
||||
if (!trimmed) return null;
|
||||
const parts = trimmed.split(',');
|
||||
if (parts.length < 3) return null;
|
||||
return {
|
||||
filename: parts[0],
|
||||
startSec: Number(parts[1]),
|
||||
endSec: Number(parts[2]),
|
||||
};
|
||||
}
|
||||
|
||||
async function refreshIndexForWorker(worker) {
|
||||
const key = worker.key;
|
||||
const dir = sourceDirForKey(key);
|
||||
const csvPath = path.join(dir, 'index.csv');
|
||||
let csv;
|
||||
try {
|
||||
csv = await fsp.readFile(csvPath, 'utf8');
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
const lines = csv.split(/\r?\n/).map(parseCsvLine).filter(Boolean);
|
||||
if (!lines.length) return;
|
||||
const first = lines[0];
|
||||
if (!Number.isFinite(first.startSec) || !Number.isFinite(first.endSec)) return;
|
||||
|
||||
if (worker.firstSegmentStartSec == null) {
|
||||
worker.firstSegmentStartSec = first.startSec;
|
||||
}
|
||||
|
||||
const baseMs = worker.startedAtMs - Math.max(0, (worker.firstSegmentStartSec || 0) * 1000);
|
||||
const cutoffMs = Date.now() - BUFFER_SECONDS * 1000 - 5000;
|
||||
|
||||
const entries = [];
|
||||
for (const row of lines) {
|
||||
if (!Number.isFinite(row.startSec) || !Number.isFinite(row.endSec)) continue;
|
||||
const startMs = Math.round(baseMs + row.startSec * 1000);
|
||||
const endMs = Math.round(baseMs + row.endSec * 1000);
|
||||
const filePath = path.join(dir, row.filename);
|
||||
let stat;
|
||||
try {
|
||||
stat = await fsp.stat(filePath);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (!stat.isFile() || stat.size < 4096) continue;
|
||||
if (endMs < cutoffMs) continue;
|
||||
entries.push({
|
||||
filePath,
|
||||
startMs,
|
||||
endMs,
|
||||
mtimeMs: stat.mtimeMs,
|
||||
size: stat.size,
|
||||
kind: worker.source.kind,
|
||||
sourceType: worker.source.sourceType,
|
||||
sourceId: worker.source.id,
|
||||
roverId: worker.source.roverId || null,
|
||||
});
|
||||
}
|
||||
|
||||
segmentIndex.set(key, entries.sort((a, b) => a.startMs - b.startMs));
|
||||
}
|
||||
|
||||
async function refreshSegmentIndex() {
|
||||
const list = Array.from(workers.values());
|
||||
for (const worker of list) {
|
||||
await refreshIndexForWorker(worker);
|
||||
}
|
||||
}
|
||||
|
||||
async function cleanupOldFiles() {
|
||||
const cutoff = Date.now() - BUFFER_SECONDS * 1000;
|
||||
try {
|
||||
await ensureDir(SEGMENT_ROOT);
|
||||
const dirs = await fsp.readdir(SEGMENT_ROOT, { withFileTypes: true });
|
||||
for (const dirent of dirs) {
|
||||
if (!dirent.isDirectory()) continue;
|
||||
const dirPath = path.join(SEGMENT_ROOT, dirent.name);
|
||||
let files;
|
||||
try {
|
||||
files = await fsp.readdir(dirPath, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
for (const file of files) {
|
||||
if (!file.isFile() || !file.name.endsWith('.mp4')) continue;
|
||||
const filePath = path.join(dirPath, file.name);
|
||||
try {
|
||||
const stat = await fsp.stat(filePath);
|
||||
if (stat.mtimeMs < cutoff) {
|
||||
await fsp.unlink(filePath);
|
||||
}
|
||||
} catch {
|
||||
// noop
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.warn('cleanup failed', err.message);
|
||||
}
|
||||
}
|
||||
|
||||
function getVideoEntriesForSource(source) {
|
||||
const key = sourceKey({ sourceType: String(source.type), kind: 'video', id: String(source.id) });
|
||||
return segmentIndex.get(key) || [];
|
||||
}
|
||||
|
||||
function getAudioEntriesForRover(roverId) {
|
||||
const key = sourceKey({ sourceType: 'rover', kind: 'audio', id: `${String(roverId)}-audio` });
|
||||
return segmentIndex.get(key) || [];
|
||||
}
|
||||
|
||||
function overlapping(entries, startMs, endMs) {
|
||||
return entries.filter((entry) => entry.endMs > startMs && entry.startMs < endMs);
|
||||
}
|
||||
|
||||
function buildGridLayout(count) {
|
||||
const cols = Math.ceil(Math.sqrt(count));
|
||||
const rows = Math.ceil(count / cols);
|
||||
return { cols, rows };
|
||||
}
|
||||
|
||||
function clampEven(value) {
|
||||
return Math.max(2, Math.floor(value / 2) * 2);
|
||||
}
|
||||
|
||||
function scalePadFilter(tileWidth, tileHeight) {
|
||||
return `scale=${tileWidth}:${tileHeight}:force_original_aspect_ratio=decrease:flags=lanczos,pad=${tileWidth}:${tileHeight}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1`;
|
||||
}
|
||||
|
||||
async function concatFiles(inputPaths, outPath) {
|
||||
const listPath = `${outPath}.concat.txt`;
|
||||
const body = inputPaths.map((file) => `file '${file.replace(/'/g, "'\\''")}'`).join('\n');
|
||||
await fsp.writeFile(listPath, `${body}\n`, 'utf8');
|
||||
await execFileAsync(FFMPEG_BIN, [
|
||||
'-y',
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-f',
|
||||
'concat',
|
||||
'-safe',
|
||||
'0',
|
||||
'-i',
|
||||
listPath,
|
||||
'-c',
|
||||
'copy',
|
||||
outPath,
|
||||
]);
|
||||
}
|
||||
|
||||
async function probeMaxFrameSize(paths) {
|
||||
let maxWidth = 0;
|
||||
let maxHeight = 0;
|
||||
for (const filePath of paths) {
|
||||
try {
|
||||
const { stdout } = await execFileAsync('ffprobe', [
|
||||
'-v',
|
||||
'error',
|
||||
'-select_streams',
|
||||
'v:0',
|
||||
'-show_entries',
|
||||
'stream=width,height',
|
||||
'-of',
|
||||
'csv=p=0',
|
||||
filePath,
|
||||
]);
|
||||
const [wRaw, hRaw] = stdout.trim().split(',');
|
||||
const w = Number(wRaw);
|
||||
const h = Number(hRaw);
|
||||
if (Number.isFinite(w) && Number.isFinite(h)) {
|
||||
maxWidth = Math.max(maxWidth, w);
|
||||
maxHeight = Math.max(maxHeight, h);
|
||||
}
|
||||
} catch {
|
||||
// noop
|
||||
}
|
||||
}
|
||||
return { maxWidth, maxHeight };
|
||||
}
|
||||
|
||||
async function buildReplayVideo({ sources = [] } = {}) {
|
||||
if (!Array.isArray(sources) || !sources.length) {
|
||||
throw new Error('No replay sources selected');
|
||||
}
|
||||
|
||||
const tEnd = Date.now() - BUILD_GUARD_MS;
|
||||
const tStart = tEnd - BUILD_DURATION_MS;
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'mrr-replay-v2-'));
|
||||
|
||||
try {
|
||||
await refreshSegmentIndex();
|
||||
|
||||
const usedSources = [];
|
||||
const missingSources = [];
|
||||
const normalizedVideos = [];
|
||||
const normalizedAudios = [];
|
||||
|
||||
for (let i = 0; i < sources.length; i += 1) {
|
||||
const source = sources[i];
|
||||
const sourceId = String(source.id);
|
||||
const videoEntries = overlapping(getVideoEntriesForSource({ id: sourceId }), tStart, tEnd);
|
||||
if (!videoEntries.length) {
|
||||
missingSources.push({ ...source, reason: 'no video coverage in replay window' });
|
||||
continue;
|
||||
}
|
||||
|
||||
const videoConcat = path.join(tmpDir, `video-${i}.mp4`);
|
||||
await concatFiles(videoEntries.map((entry) => entry.filePath), videoConcat);
|
||||
const videoTrimmed = path.join(tmpDir, `video-${i}.trim.mp4`);
|
||||
const firstStartMs = videoEntries[0].startMs;
|
||||
const ss = Math.max(0, (tStart - firstStartMs) / 1000);
|
||||
const to = Math.max(ss + 0.1, (tEnd - firstStartMs) / 1000);
|
||||
await execFileAsync(FFMPEG_BIN, [
|
||||
'-y',
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-ss',
|
||||
ss.toFixed(3),
|
||||
'-to',
|
||||
to.toFixed(3),
|
||||
'-i',
|
||||
videoConcat,
|
||||
'-an',
|
||||
'-c:v',
|
||||
'libx264',
|
||||
'-preset',
|
||||
'veryfast',
|
||||
'-pix_fmt',
|
||||
'yuv420p',
|
||||
'-r',
|
||||
String(TARGET_FPS),
|
||||
videoTrimmed,
|
||||
]);
|
||||
|
||||
normalizedVideos.push({ path: videoTrimmed, source });
|
||||
usedSources.push(source);
|
||||
|
||||
if (source.type === 'rover') {
|
||||
const audioEntries = overlapping(getAudioEntriesForRover(sourceId), tStart, tEnd);
|
||||
if (audioEntries.length) {
|
||||
const audioConcat = path.join(tmpDir, `audio-${i}.m4a`);
|
||||
await concatFiles(audioEntries.map((entry) => entry.filePath), audioConcat);
|
||||
const audioTrimmed = path.join(tmpDir, `audio-${i}.trim.m4a`);
|
||||
const firstAudioStartMs = audioEntries[0].startMs;
|
||||
const ass = Math.max(0, (tStart - firstAudioStartMs) / 1000);
|
||||
const ato = Math.max(ass + 0.1, (tEnd - firstAudioStartMs) / 1000);
|
||||
await execFileAsync(FFMPEG_BIN, [
|
||||
'-y',
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-ss',
|
||||
ass.toFixed(3),
|
||||
'-to',
|
||||
ato.toFixed(3),
|
||||
'-i',
|
||||
audioConcat,
|
||||
'-vn',
|
||||
'-ac',
|
||||
'1',
|
||||
'-ar',
|
||||
'48000',
|
||||
'-c:a',
|
||||
'aac',
|
||||
'-b:a',
|
||||
'96k',
|
||||
audioTrimmed,
|
||||
]);
|
||||
normalizedAudios.push(audioTrimmed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!normalizedVideos.length) {
|
||||
throw new Error('No replay segments available for selected sources');
|
||||
}
|
||||
|
||||
const layout = buildGridLayout(normalizedVideos.length);
|
||||
const { maxWidth, maxHeight } = await probeMaxFrameSize(normalizedVideos.map((v) => v.path));
|
||||
let tileWidth = maxWidth || 640;
|
||||
let tileHeight = maxHeight || 360;
|
||||
let outWidth = tileWidth * layout.cols;
|
||||
let outHeight = tileHeight * layout.rows;
|
||||
if (outWidth > MAX_WIDTH || outHeight > MAX_HEIGHT) {
|
||||
const scale = Math.min(MAX_WIDTH / outWidth, MAX_HEIGHT / outHeight);
|
||||
tileWidth *= scale;
|
||||
tileHeight *= scale;
|
||||
outWidth = tileWidth * layout.cols;
|
||||
outHeight = tileHeight * layout.rows;
|
||||
}
|
||||
tileWidth = clampEven(tileWidth);
|
||||
tileHeight = clampEven(tileHeight);
|
||||
|
||||
const inputArgs = [];
|
||||
const filterParts = [];
|
||||
const layoutParts = [];
|
||||
|
||||
for (let i = 0; i < normalizedVideos.length; i += 1) {
|
||||
inputArgs.push('-i', normalizedVideos[i].path);
|
||||
filterParts.push(`[${i}:v]${scalePadFilter(tileWidth, tileHeight)}[v${i}]`);
|
||||
const x = (i % layout.cols) * tileWidth;
|
||||
const y = Math.floor(i / layout.cols) * tileHeight;
|
||||
layoutParts.push(`${x}_${y}`);
|
||||
}
|
||||
|
||||
const audioInputStart = normalizedVideos.length;
|
||||
for (const audioPath of normalizedAudios) {
|
||||
inputArgs.push('-i', audioPath);
|
||||
}
|
||||
|
||||
if (normalizedVideos.length === 1) {
|
||||
filterParts.push('[v0]null[vout]');
|
||||
} else {
|
||||
filterParts.push(
|
||||
`${normalizedVideos.map((_, i) => `[v${i}]`).join('')}` +
|
||||
`xstack=inputs=${normalizedVideos.length}:layout=${layoutParts.join('|')}:fill=black[vout]`,
|
||||
);
|
||||
}
|
||||
|
||||
if (normalizedAudios.length) {
|
||||
const audioRefs = normalizedAudios.map((_, idx) => `[${audioInputStart + idx}:a]`).join('');
|
||||
filterParts.push(`${audioRefs}amix=inputs=${normalizedAudios.length}:normalize=0,alimiter=limit=0.9[aout]`);
|
||||
}
|
||||
|
||||
const durationSec = BUILD_DURATION_MS / 1000;
|
||||
const targetBitrateKbps = Math.max(400, Math.floor((MAX_BYTES * 8) / durationSec / 1000));
|
||||
const outPath = path.join(tmpDir, 'replay.mp4');
|
||||
|
||||
const args = [
|
||||
'-y',
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
...inputArgs,
|
||||
'-filter_complex',
|
||||
filterParts.join(';'),
|
||||
'-map',
|
||||
'[vout]',
|
||||
'-c:v',
|
||||
'libx264',
|
||||
'-preset',
|
||||
'veryfast',
|
||||
'-pix_fmt',
|
||||
'yuv420p',
|
||||
'-r',
|
||||
String(TARGET_FPS),
|
||||
'-b:v',
|
||||
`${targetBitrateKbps}k`,
|
||||
'-maxrate',
|
||||
`${Math.floor(targetBitrateKbps * 1.15)}k`,
|
||||
'-bufsize',
|
||||
`${Math.floor(targetBitrateKbps * 2)}k`,
|
||||
];
|
||||
|
||||
if (normalizedAudios.length) {
|
||||
args.push('-map', '[aout]', '-c:a', 'aac', '-b:a', '128k');
|
||||
}
|
||||
|
||||
args.push(outPath);
|
||||
|
||||
await execFileAsync(FFMPEG_BIN, args);
|
||||
const buffer = await fsp.readFile(outPath);
|
||||
return { buffer, usedSources, missingSources };
|
||||
} finally {
|
||||
try {
|
||||
await fsp.rm(tmpDir, { recursive: true, force: true });
|
||||
} catch {
|
||||
// noop
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function getReplayHealthSnapshot() {
|
||||
const now = Date.now();
|
||||
const neededWindowMs = BUILD_DURATION_MS;
|
||||
const sources = [];
|
||||
let readyCount = 0;
|
||||
|
||||
const replaySources = [];
|
||||
for (const rover of roverManager.getRoster().filter((entry) => roverManager.canReplayRoverId(entry.id))) {
|
||||
replaySources.push({ type: 'rover', id: String(rover.id), label: rover.name || rover.id });
|
||||
}
|
||||
for (const camera of getRoomCameras()) {
|
||||
replaySources.push({ type: 'room', id: String(camera.id), label: camera.name || camera.id });
|
||||
}
|
||||
|
||||
for (const source of replaySources) {
|
||||
const entries = getVideoEntriesForSource({ id: source.id });
|
||||
const inWindow = entries.filter((entry) => now - entry.endMs <= BUFFER_SECONDS * 1000);
|
||||
const newest = inWindow[inWindow.length - 1] || null;
|
||||
const oldest = inWindow[0] || null;
|
||||
const coveredMs = newest && oldest ? Math.max(0, newest.endMs - oldest.startMs) : 0;
|
||||
const ready = coveredMs >= neededWindowMs;
|
||||
if (ready) readyCount += 1;
|
||||
sources.push({
|
||||
type: source.type,
|
||||
id: source.id,
|
||||
label: source.label,
|
||||
recentCount: inWindow.length,
|
||||
neededCount: Math.ceil(neededWindowMs / 1000),
|
||||
lastSegmentAt: newest?.endMs || null,
|
||||
ready,
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
sources,
|
||||
readyCount,
|
||||
totalCount: sources.length,
|
||||
};
|
||||
}
|
||||
|
||||
async function bootstrapIndexFromDisk() {
|
||||
await ensureDir(SEGMENT_ROOT);
|
||||
const dirs = await fsp.readdir(SEGMENT_ROOT, { withFileTypes: true });
|
||||
for (const dirent of dirs) {
|
||||
if (!dirent.isDirectory()) continue;
|
||||
segmentIndex.set(dirent.name, []);
|
||||
}
|
||||
}
|
||||
|
||||
async function tick() {
|
||||
await syncWorkers();
|
||||
await refreshSegmentIndex();
|
||||
await cleanupOldFiles();
|
||||
events.emit('health', getReplayHealthSnapshot());
|
||||
}
|
||||
|
||||
async function start() {
|
||||
await ensureDir(SEGMENT_ROOT);
|
||||
await bootstrapIndexFromDisk();
|
||||
await tick();
|
||||
if (cleanupTimer) clearInterval(cleanupTimer);
|
||||
cleanupTimer = setInterval(() => {
|
||||
tick().catch((err) => logger.warn('tick failed', err.message));
|
||||
}, CLEANUP_INTERVAL_MS);
|
||||
}
|
||||
|
||||
roomCameraEvents.on('update', () => {
|
||||
tick().catch((err) => logger.warn('room camera update tick failed', err.message));
|
||||
});
|
||||
roverManager.managerEvents.on('rover', () => {
|
||||
tick().catch((err) => logger.warn('rover update tick failed', err.message));
|
||||
});
|
||||
roverManager.managerEvents.on('private', () => {
|
||||
tick().catch((err) => logger.warn('private update tick failed', err.message));
|
||||
});
|
||||
|
||||
start().catch((err) => {
|
||||
logger.warn('replay engine startup failed', err.message);
|
||||
});
|
||||
|
||||
module.exports = {
|
||||
buildReplayVideo,
|
||||
getReplayHealthSnapshot,
|
||||
replayEngineEvents: events,
|
||||
replaySegmentRootDir: SEGMENT_ROOT,
|
||||
replaySegmentSeconds: SEGMENT_SECONDS,
|
||||
replayBufferSeconds: BUFFER_SECONDS,
|
||||
};
|
||||
@@ -1,325 +0,0 @@
|
||||
const { spawn } = require('child_process');
|
||||
const fsp = require('fs/promises');
|
||||
const path = require('path');
|
||||
const logger = require('../globals/logger').child('replaySegments');
|
||||
const { getRoomCameras, roomCameraEvents } = require('./roomCameraService');
|
||||
const roverManager = require('./roverManager');
|
||||
|
||||
const SEGMENT_DIR = process.env.REPLAY_SEGMENT_DIR || '/var/lib/replay-segments';
|
||||
const SEGMENT_SECONDS = 2;
|
||||
const BUFFER_SECONDS = 40;
|
||||
const CLEANUP_INTERVAL_MS = 20000;
|
||||
const FPS = 15;
|
||||
const SCALE_WIDTH = 640;
|
||||
const MAX_BYTES = Number.parseInt(process.env.REPLAY_SEGMENT_MAX_BYTES || '0', 10);
|
||||
const FFMPEG_BIN = process.env.FFMPEG_BIN || 'ffmpeg';
|
||||
const ROVER_SNAPSHOT_DIR = process.env.ROVER_SNAPSHOT_DIR || '/var/lib/rover-snapshots';
|
||||
const ROVER_SNAPSHOT_FPS = 3;
|
||||
|
||||
const recorders = new Map(); // key -> { proc, source }
|
||||
let cleanupTimer = null;
|
||||
|
||||
function sourceKey(source) {
|
||||
return `${source.type}__${source.id}`;
|
||||
}
|
||||
|
||||
function getRoomCameraStream(camera) {
|
||||
if (camera.streamUrl) return camera.streamUrl;
|
||||
const url = String(camera.url || '');
|
||||
if (url.includes('.mjpg') || url.includes('mjpeg') || url.includes('stream')) {
|
||||
return url;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function listSources() {
|
||||
const rooms = getRoomCameras().map((camera) => ({
|
||||
type: 'room',
|
||||
id: String(camera.id),
|
||||
label: camera.name || camera.id,
|
||||
streamUrl: getRoomCameraStream(camera),
|
||||
}));
|
||||
const rovers = roverManager.getRoster().map((rover) => ({
|
||||
type: 'rover',
|
||||
id: String(rover.id),
|
||||
label: rover.name || rover.id,
|
||||
}));
|
||||
return [...rooms, ...rovers];
|
||||
}
|
||||
|
||||
function buildInputUrl(source) {
|
||||
if (source.type === 'room') {
|
||||
return source.streamUrl || null;
|
||||
}
|
||||
return `srt://127.0.0.1:9000?streamid=read:${encodeURIComponent(source.id)}`;
|
||||
}
|
||||
|
||||
async function ensureDir(dir) {
|
||||
await fsp.mkdir(dir, { recursive: true });
|
||||
}
|
||||
|
||||
function buildOutputPattern(key) {
|
||||
return path.join(SEGMENT_DIR, key, '%Y%m%d-%H%M%S.mp4');
|
||||
}
|
||||
|
||||
function spawnRecorder(source) {
|
||||
const key = sourceKey(source);
|
||||
const inputUrl = buildInputUrl(source);
|
||||
if (!inputUrl) {
|
||||
logger.warn('Replay recorder missing input URL', { key, source });
|
||||
return;
|
||||
}
|
||||
const outPattern = buildOutputPattern(key);
|
||||
const outputSnapshot = source.type === 'rover'
|
||||
? path.join(ROVER_SNAPSHOT_DIR, `${source.id}.jpg`)
|
||||
: null;
|
||||
ensureDir(path.dirname(outPattern))
|
||||
.then(async () => {
|
||||
if (outputSnapshot) {
|
||||
await ensureDir(ROVER_SNAPSHOT_DIR);
|
||||
}
|
||||
const args = [
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'info',
|
||||
'-y',
|
||||
'-fflags',
|
||||
'nobuffer',
|
||||
'-flags',
|
||||
'low_delay',
|
||||
'-i',
|
||||
inputUrl,
|
||||
];
|
||||
if (outputSnapshot) {
|
||||
args.push(
|
||||
'-filter_complex',
|
||||
`[0:v]fps=${FPS},scale=${SCALE_WIDTH}:-1[vseg];` +
|
||||
`[0:v]fps=${ROVER_SNAPSHOT_FPS},scale=${SCALE_WIDTH}:-1[vjpg]`,
|
||||
'-map',
|
||||
'[vseg]',
|
||||
'-an',
|
||||
'-c:v:0',
|
||||
'libx264',
|
||||
'-preset',
|
||||
'veryfast',
|
||||
'-tune',
|
||||
'zerolatency',
|
||||
'-g',
|
||||
String(FPS),
|
||||
'-keyint_min',
|
||||
String(FPS),
|
||||
'-sc_threshold',
|
||||
'0',
|
||||
'-f',
|
||||
'segment',
|
||||
'-segment_time',
|
||||
String(SEGMENT_SECONDS),
|
||||
'-segment_format',
|
||||
'mp4',
|
||||
'-reset_timestamps',
|
||||
'1',
|
||||
'-strftime',
|
||||
'1',
|
||||
outPattern,
|
||||
'-map',
|
||||
'[vjpg]',
|
||||
'-an',
|
||||
'-q:v:1',
|
||||
'8',
|
||||
'-f',
|
||||
'image2',
|
||||
'-update',
|
||||
'1',
|
||||
outputSnapshot,
|
||||
);
|
||||
} else {
|
||||
args.push(
|
||||
'-r',
|
||||
String(FPS),
|
||||
'-vf',
|
||||
`fps=${FPS},scale=${SCALE_WIDTH}:-1`,
|
||||
'-an',
|
||||
'-c:v',
|
||||
'libx264',
|
||||
'-preset',
|
||||
'veryfast',
|
||||
'-tune',
|
||||
'zerolatency',
|
||||
'-g',
|
||||
String(FPS),
|
||||
'-keyint_min',
|
||||
String(FPS),
|
||||
'-sc_threshold',
|
||||
'0',
|
||||
'-f',
|
||||
'segment',
|
||||
'-segment_time',
|
||||
String(SEGMENT_SECONDS),
|
||||
'-segment_format',
|
||||
'mp4',
|
||||
'-reset_timestamps',
|
||||
'1',
|
||||
'-strftime',
|
||||
'1',
|
||||
outPattern,
|
||||
);
|
||||
}
|
||||
const proc = spawn(FFMPEG_BIN, args, { stdio: ['ignore', 'ignore', 'pipe'] });
|
||||
const stderrChunks = [];
|
||||
let stderrSize = 0;
|
||||
proc.stderr.on('data', (chunk) => {
|
||||
if (!chunk || stderrSize > 8192) return;
|
||||
stderrChunks.push(chunk);
|
||||
stderrSize += chunk.length;
|
||||
});
|
||||
recorders.set(key, { proc, source });
|
||||
proc.on('exit', (code, signal) => {
|
||||
recorders.delete(key);
|
||||
const stderrBuffer = stderrChunks.length ? Buffer.concat(stderrChunks) : null;
|
||||
if (stderrBuffer && stderrBuffer.length) {
|
||||
const preview = stderrBuffer.toString('utf8', 0, 600).trim();
|
||||
logger.warn('Replay recorder stderr', {
|
||||
key,
|
||||
stderrBytes: stderrBuffer.length,
|
||||
stderrPreview: preview || '<non-utf8>',
|
||||
});
|
||||
} else {
|
||||
logger.warn('Replay recorder stderr', { key, stderrBytes: 0 });
|
||||
}
|
||||
if (!shouldRecord(source)) {
|
||||
return;
|
||||
}
|
||||
const delay = 2000;
|
||||
logger.warn('Replay recorder exited; restarting', { key, code, signal });
|
||||
setTimeout(() => {
|
||||
if (!recorders.has(key) && shouldRecord(source)) {
|
||||
spawnRecorder(source);
|
||||
}
|
||||
}, delay);
|
||||
});
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.warn('Replay recorder setup failed', { key, err: err.message });
|
||||
});
|
||||
}
|
||||
|
||||
function stopRecorder(key) {
|
||||
const entry = recorders.get(key);
|
||||
if (!entry) return;
|
||||
entry.proc.kill('SIGTERM');
|
||||
recorders.delete(key);
|
||||
}
|
||||
|
||||
async function removeSourceArtifacts(key, source) {
|
||||
try {
|
||||
await fsp.rm(path.join(SEGMENT_DIR, key), { recursive: true, force: true });
|
||||
} catch (err) {
|
||||
logger.warn('Failed to clear replay segments', { key, error: err.message });
|
||||
}
|
||||
if (source?.type === 'rover' && source?.id) {
|
||||
try {
|
||||
await fsp.rm(path.join(ROVER_SNAPSHOT_DIR, `${source.id}.jpg`), { force: true });
|
||||
} catch (err) {
|
||||
logger.warn('Failed to clear rover snapshot artifact', { roverId: source.id, error: err.message });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function shouldRecord(source) {
|
||||
if (source.type === 'room') {
|
||||
return Boolean(source.streamUrl);
|
||||
}
|
||||
return roverManager.canReplayRoverId(source.id);
|
||||
}
|
||||
|
||||
function syncRecorders() {
|
||||
const sources = listSources();
|
||||
const desiredKeys = new Set();
|
||||
sources.forEach((source) => {
|
||||
if (!shouldRecord(source)) return;
|
||||
const key = sourceKey(source);
|
||||
desiredKeys.add(key);
|
||||
if (!recorders.has(key)) {
|
||||
spawnRecorder(source);
|
||||
}
|
||||
});
|
||||
Array.from(recorders.keys()).forEach((key) => {
|
||||
if (!desiredKeys.has(key)) {
|
||||
const entry = recorders.get(key);
|
||||
stopRecorder(key);
|
||||
removeSourceArtifacts(key, entry?.source).catch(() => {});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function cleanupSegments() {
|
||||
try {
|
||||
await ensureDir(SEGMENT_DIR);
|
||||
const cutoff = Date.now() - BUFFER_SECONDS * 1000;
|
||||
const entries = await fsp.readdir(SEGMENT_DIR, { withFileTypes: true });
|
||||
let totalBytes = 0;
|
||||
const files = [];
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
const dirPath = path.join(SEGMENT_DIR, entry.name);
|
||||
const inner = await fsp.readdir(dirPath, { withFileTypes: true });
|
||||
for (const file of inner) {
|
||||
if (!file.isFile() || !file.name.endsWith('.mp4')) continue;
|
||||
const filePath = path.join(dirPath, file.name);
|
||||
const stats = await fsp.stat(filePath);
|
||||
totalBytes += stats.size;
|
||||
files.push({ filePath, mtimeMs: stats.mtimeMs, size: stats.size });
|
||||
if (stats.mtimeMs < cutoff) {
|
||||
await fsp.unlink(filePath);
|
||||
}
|
||||
}
|
||||
const remaining = await fsp.readdir(dirPath);
|
||||
if (!remaining.length) {
|
||||
await fsp.rmdir(dirPath);
|
||||
}
|
||||
}
|
||||
if (MAX_BYTES > 0 && totalBytes > MAX_BYTES) {
|
||||
const overBy = totalBytes - MAX_BYTES;
|
||||
let freed = 0;
|
||||
files.sort((a, b) => a.mtimeMs - b.mtimeMs);
|
||||
for (const file of files) {
|
||||
if (freed >= overBy) break;
|
||||
try {
|
||||
await fsp.unlink(file.filePath);
|
||||
freed += file.size;
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.warn('Replay cleanup failed', err.message);
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
syncRecorders();
|
||||
cleanupSegments();
|
||||
if (cleanupTimer) clearInterval(cleanupTimer);
|
||||
cleanupTimer = setInterval(cleanupSegments, CLEANUP_INTERVAL_MS);
|
||||
}
|
||||
|
||||
roomCameraEvents.on('update', () => {
|
||||
syncRecorders();
|
||||
});
|
||||
|
||||
roverManager.managerEvents.on('rover', () => {
|
||||
syncRecorders();
|
||||
});
|
||||
|
||||
roverManager.managerEvents.on('private', () => {
|
||||
syncRecorders();
|
||||
});
|
||||
|
||||
start();
|
||||
|
||||
module.exports = {
|
||||
replaySegmentsDir: SEGMENT_DIR,
|
||||
segmentSeconds: SEGMENT_SECONDS,
|
||||
bufferSeconds: BUFFER_SECONDS,
|
||||
};
|
||||
Reference in New Issue
Block a user