diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 89fe12a25a0..66110cb4f7b 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -214,6 +214,7 @@ export { isVideoPastTimelineEnd, resolveTimelineExtractionWindow, resolveVideoExtractionWindow, + rebaseVideoToWindow, resolveFinalFrameExtractionWindow, resolveVideoExtractionDuration, resolvePlayableVideoDuration, diff --git a/packages/engine/src/services/audioMixer.ts b/packages/engine/src/services/audioMixer.ts index 8771d3382af..d141543ddf3 100644 --- a/packages/engine/src/services/audioMixer.ts +++ b/packages/engine/src/services/audioMixer.ts @@ -203,6 +203,8 @@ function legacyFilterScriptOptionIsUnsupported(stderr: string): boolean { */ // Caps the ffmpeg runs that prepare clips; all of a long mix's clips at once starved the host (200 clips, 200 ffmpeg). const inPrepareSlot = createConcurrencyLimit(Math.max(2, Math.floor(cpus().length / 2))); +// A clip read from a URL spends its time waiting on the network, so it needs no CPU-sized limit. +const inRemotePrepareSlot = createConcurrencyLimit(16); const MAX_VOLUME_SEGMENTS = 32; @@ -757,6 +759,23 @@ async function extractAudioSegment( }; } +function isRemoteElement( + element: { src: string }, + baseDir: string, + compiledDir: string | undefined, + config?: { urlByAbsentLocalPath?: ReadonlyMap }, +): boolean { + if (isHttpUrl(element.src)) return true; + if (!config?.urlByAbsentLocalPath?.size) return false; + try { + return config.urlByAbsentLocalPath.has( + resolveProjectRelativeSrc(element.src, baseDir, compiledDir), + ); + } catch { + return false; + } +} + async function generateSilence( outputPath: string, duration: number, @@ -1357,7 +1376,9 @@ export async function processCompositionAudio( await Promise.all( elements.map((element, index) => - inPrepareSlot(async () => { + (isRemoteElement(element, baseDir, compiledDir, config) + ? inRemotePrepareSlot + : inPrepareSlot)(async () => { if (effectiveSignal.aborted) { failures.push({ stage: "cancelled", diff --git a/packages/engine/src/services/audioMixer.urlRead.test.ts b/packages/engine/src/services/audioMixer.urlRead.test.ts index e9fbbb92240..eb30e61a2a5 100644 --- a/packages/engine/src/services/audioMixer.urlRead.test.ts +++ b/packages/engine/src/services/audioMixer.urlRead.test.ts @@ -3,7 +3,7 @@ import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "nod import { createServer, type Server, type ServerResponse } from "node:http"; import { once } from "node:events"; import type { AddressInfo } from "node:net"; -import { tmpdir } from "node:os"; +import { cpus, tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { getFfmpegBinary } from "../utils/ffmpegBinaries.js"; @@ -231,4 +231,50 @@ describe.skipIf(!HAS_FFMPEG)("audio cut from a cloud-held file's URL", { timeout await new Promise((resolve) => sidecarServer.close(resolve)); } }); + + it("reads many clips of one URL at once, since the wait is on the network and not the CPU", async () => { + const body = makeVoice(); + let open = 0; + let peak = 0; + server = createServer((req, res) => { + const start = Number(/bytes=(\d+)-/.exec(req.headers.range ?? "")?.[1] ?? 0); + open++; + peak = Math.max(peak, open); + res.on("close", () => open--); + setTimeout(() => { + res.writeHead(206, { + "Accept-Ranges": "bytes", + "Content-Length": body.length - start, + "Content-Range": `bytes ${start}-${body.length - 1}/${body.length}`, + }); + res.end(body.subarray(start)); + }, 300); + }); + await new Promise((resolve) => server!.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + const baseDir = join(dir, "project"); + const local = join(baseDir, "takes", "voice.wav"); + + const result = await processCompositionAudio( + Array.from({ length: 24 }, (_, i) => ({ + id: `clip-${i}`, + src: "takes/voice.wav", + start: i, + end: i + 1, + mediaStart: i % 10, + layer: i, + volume: 1, + type: "audio" as const, + })), + baseDir, + join(dir, "work"), + join(dir, "out.m4a"), + 24, + abort.signal, + { urlByAbsentLocalPath: new Map([[local, `http://127.0.0.1:${port}/voice.wav`]]) }, + ); + + expect(result.success).toBe(true); + expect(peak).toBeGreaterThan(Math.max(2, Math.floor(cpus().length / 2))); + }); }); diff --git a/packages/producer/src/services/distributed/plan.test.ts b/packages/producer/src/services/distributed/plan.test.ts index 77908536661..a0b151f1a12 100644 --- a/packages/producer/src/services/distributed/plan.test.ts +++ b/packages/producer/src/services/distributed/plan.test.ts @@ -1131,48 +1131,53 @@ describe.skipIf(spawnSync(getFfmpegBinary(), ["-version"]).status !== 0)( } }, 60_000); - it("gives buildPlanAudio the sound of a video whose src the scene script sets", async () => { - const project = join(runRoot, "project-script-src"); - mkdirSync(project, { recursive: true }); - const clip = spawnSync(getFfmpegBinary(), [ - "-nostdin", - "-v", - "error", - "-f", - "lavfi", - "-i", - "testsrc=s=320x240:r=30:d=1", - "-f", - "lavfi", - "-i", - "sine=frequency=440:sample_rate=48000:d=1", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - "-c:a", - "aac", - "-shortest", - join(project, "clip.mp4"), - ]); - if (clip.status !== 0) throw new Error(`ffmpeg clip failed: ${clip.stderr}`); - writeFileSync( - join(project, "index.html"), - ` + for (const slot of [ + { name: "starts on the timeline", start: 0, duration: 1 }, + { name: "starts before the timeline", start: -0.5, duration: 1.5 }, + ]) { + it(`gives buildPlanAudio the sound of a video whose src the scene script sets and that ${slot.name}`, async () => { + const project = join(runRoot, `project-script-src-${slot.start}`); + mkdirSync(project, { recursive: true }); + const clip = spawnSync(getFfmpegBinary(), [ + "-nostdin", + "-v", + "error", + "-f", + "lavfi", + "-i", + "testsrc=s=320x240:r=30:d=1", + "-f", + "lavfi", + "-i", + "sine=frequency=440:sample_rate=48000:d=1", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + "-c:a", + "aac", + "-shortest", + join(project, "clip.mp4"), + ]); + if (clip.status !== 0) throw new Error(`ffmpeg clip failed: ${clip.stderr}`); + writeFileSync( + join(project, "index.html"), + `
- +
`, - ); - const config = { fps: 30, width: 320, height: 240, format: "mp4" } as const; - const carried = join(runRoot, "plan-script-src"); - await buildLocalExecutionPlan(project, config, carried, { chunksCarryAudio: true }); - const work = join(runRoot, "script-src-work"); - mkdirSync(work, { recursive: true }); - const audio = join(work, "audio.m4a"); - expect(await buildPlanAudio(project, config, work, audio)).toEqual({ hasAudio: true }); - expect(readFileSync(audio).equals(readFileSync(join(carried, "audio.m4a")))).toBe(true); - }, 60_000); + ); + const config = { fps: 30, width: 320, height: 240, format: "mp4" } as const; + const carried = join(runRoot, `plan-script-src-${slot.start}`); + await buildLocalExecutionPlan(project, config, carried, { chunksCarryAudio: true }); + const work = join(runRoot, `script-src-work-${slot.start}`); + mkdirSync(work, { recursive: true }); + const audio = join(work, "audio.m4a"); + expect(await buildPlanAudio(project, config, work, audio)).toEqual({ hasAudio: true }); + expect(readFileSync(audio).equals(readFileSync(join(carried, "audio.m4a")))).toBe(true); + }, 60_000); + } it("keeps assembled loudness within 0.1 LUFS of the frame-fitted float delivery", async () => { const project = audioProject("project-loudness-delivery", 0.3, 3.01); diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts index 7e3dbe41c32..2ae8b034d2a 100644 --- a/packages/producer/src/services/distributed/plan.ts +++ b/packages/producer/src/services/distributed/plan.ts @@ -64,6 +64,7 @@ import { measurePlanSecondsPerFrame } from "./planFrameCost.js"; import { runAudioStage } from "../render/stages/audioStage.js"; import { runCompileStage } from "../render/stages/compileStage.js"; import { + appendVideoAudioFromProbe, assertVideoExtractionSucceeded, runExtractVideosStage, } from "../render/stages/extractVideosStage.js"; @@ -1727,7 +1728,7 @@ export async function buildPlanAudio( }); try { let stageStarted = Date.now(); - await extractPlanVideos(c, true); + await appendVideoAudioFromProbe(c); logPlanStage(c.log, "extract", stageStarted); stageStarted = Date.now(); const mixed = await mixPlanAudio(c, c.compiledDir); diff --git a/packages/producer/src/services/render/stages/extractVideosStage.ts b/packages/producer/src/services/render/stages/extractVideosStage.ts index ece85cb34e6..b69f45bf5a8 100644 --- a/packages/producer/src/services/render/stages/extractVideosStage.ts +++ b/packages/producer/src/services/render/stages/extractVideosStage.ts @@ -33,7 +33,6 @@ import { join } from "node:path"; import { type CaptureVideoMetadataHint, type EngineConfig, - type ExtractedFrames, type ExtractionResult, type FrameLookupTable, type HdrTransfer, @@ -47,6 +46,10 @@ import { extractMediaMetadata, isVideoPastTimelineEnd, isHdrColorSpace, + rebaseVideoToWindow, + resolveFinalFrameExtractionWindow, + resolveVideoExtractionWindow, + isHttpUrl, resolveProjectRelativeSrc, runVideoExtractionWithRetry, safeVideoExtractionSourceIdentity, @@ -568,7 +571,7 @@ export async function runExtractVideosStage( */ export function appendAutoDetectedVideoAudio( composition: Pick, - extracted: ExtractedFrames[], + extracted: ReadonlyArray<{ videoId: string; metadata: { hasAudio: boolean } }>, ): void { const existingAudioSrcs = new Set(composition.audios.map((a) => a.src)); for (const ext of extracted) { @@ -588,3 +591,42 @@ export function appendAutoDetectedVideoAudio( existingAudioSrcs.add(video.src); } } + +/** + * The audio-only form of the extract stage: which videos carry sound takes a probe per source, not their frames. + * Each video is moved to its extraction window first, as extraction does, so its sound lands where its picture does. + */ +export async function appendVideoAudioFromProbe(input: { + projectDir: string; + compiledDir: string; + composition: CompositionMetadata; + urlByAbsentLocalPath?: ReadonlyMap; +}): Promise { + const { projectDir, compiledDir, composition, urlByAbsentLocalPath } = input; + const videos = composition.videos.filter( + (video) => !isVideoPastTimelineEnd(video, composition.duration), + ); + const probed = await Promise.all( + videos.map(async (video) => { + const path = isHttpUrl(video.src) + ? video.src + : resolveProjectRelativeSrc(video.src, projectDir, compiledDir); + const readFrom = urlByAbsentLocalPath?.get(path) ?? path; + if (!isHttpUrl(readFrom) && !existsSync(readFrom)) return null; + const metadata = await extractMediaMetadata(readFrom); + const window = await resolveFinalFrameExtractionWindow( + readFrom, + video, + metadata, + resolveVideoExtractionWindow(video, metadata, composition.duration), + ); + if (window.durationSeconds <= 0) return null; + rebaseVideoToWindow(video, window); + return { videoId: video.id, metadata }; + }), + ); + appendAutoDetectedVideoAudio( + composition, + probed.filter((entry) => entry !== null), + ); +} diff --git a/packages/producer/src/services/render/stages/videoAudioProbe.test.ts b/packages/producer/src/services/render/stages/videoAudioProbe.test.ts new file mode 100644 index 00000000000..7674fcb6114 --- /dev/null +++ b/packages/producer/src/services/render/stages/videoAudioProbe.test.ts @@ -0,0 +1,56 @@ +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { spawnSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, readdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getFfmpegBinary } from "@hyperframes/engine"; +import { collectRenderMedia } from "../../renderMediaCollector.js"; +import { appendVideoAudioFromProbe } from "./extractVideosStage.js"; + +const HAS_FFMPEG = spawnSync(getFfmpegBinary(), ["-version"]).status === 0; + +describe.skipIf(!HAS_FFMPEG)("appendVideoAudioFromProbe", () => { + let dir: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "hf-video-audio-probe-")); + mkdirSync(join(dir, "compiled")); + }); + afterEach(() => rmSync(dir, { recursive: true, force: true })); + + const make = (name: string, withAudio: boolean) => { + const args = [ + "-v", + "error", + "-y", + "-f", + "lavfi", + "-i", + "testsrc2=size=64x64:rate=25:duration=4", + ]; + if (withAudio) + args.push("-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=4"); + args.push("-c:v", "mpeg4", ...(withAudio ? ["-c:a", "aac"] : []), join(dir, name)); + expect(spawnSync(getFfmpegBinary(), args).status).toBe(0); + }; + + it("adds the sound of a video that has some, once per source, without extracting any frames", async () => { + make("talk.mp4", true); + make("silent.mp4", false); + const composition = + collectRenderMedia(`
+ + + +
`); + composition.audios.length = 0; + + await appendVideoAudioFromProbe({ + projectDir: dir, + compiledDir: join(dir, "compiled"), + composition, + }); + + expect(composition.audios.map((audio) => audio.id)).toEqual(["a-audio"]); + expect(readdirSync(join(dir, "compiled"))).toEqual([]); + }); +});