openvibers/openvibe.ai Commits
media.analyze transcribes long recordings in windows (WS-O task 5). The whisper provider caps a call at its own 10-minute timeout, so a whole hour-long VOD would have come back without a transcript. Speech-to-text now runs in windows (AI_MEDIA_STT_WINDOW_SEC, 10 minutes, about 2 minutes of work each on this host); the whisper provider decodes from startSec (-ss) and offsets the timestamps. A failed window is named in gaps and the others still count. The transcript's language is the one most windows detected. The stub places its synthetic segments at the window's offset. Tests: 10 s windows on the 28 s fixture give segments from 0, 10 and 20 s; a whisper window from 20 s keeps the recording's times.
Changes
@@@ -131,6 +131,8 @@ function load(env = process.env) {reserveBytes: int(env.AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES, 3 * 1024 * 1024 * 1024),fetchTimeoutMs: int(env.AI_MEDIA_ANALYSIS_FETCH_TIMEOUT_MS, 20 * 60000),sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),++ // Speech-to-text in windows: each call stays well inside the whisper provider's timeout (10 min).++ sttWindowSec: Math.max(10, int(env.AI_MEDIA_STT_WINDOW_SEC, 600)),ffmpeg: env.FFMPEG_BIN || 'ffmpeg',ffprobe: env.FFPROBE_BIN || 'ffprobe',},@@@ -125,7 +125,8 @@ function createStubProvider(record = { key: 'stub', metadata: {} }, { stats = nu}if (op === 'transcribe') {const n = 2 + Math.floor(rnd() * 3);-- const segments = Array.from({ length: n }, (_, i) => ({ start: i * 4, end: i * 4 + 3.5, text: `(synthetic) ${phrase(rnd, 5)}` }));++ const at = Number(req.offsetSec) || 0; // a window of a longer recording keeps the recording's times++ const segments = Array.from({ length: n }, (_, i) => ({ start: at + i * 4, end: at + i * 4 + 3.5, text: `(synthetic) ${phrase(rnd, 5)}` }));return { text: segments.map(s => s.text).join(' '), segments, model, synthetic: true, usage: { input: 0, output: 0, cached: 0, estimated: true } };}let json = null;@@@ -81,10 +81,11 @@ function createWhisperProvider(record, { config }) {return ch;}-- function toWav(src, seconds, signal, timeoutMs) {++ function toWav(src, seconds, signal, timeoutMs, startSec = 0) {return new Promise((resolve, reject) => {const wav = path.join(os.tmpdir(), `openvibe-ai-tx-${Date.now()}-${Math.floor(Math.random() * 1e6)}.wav`);-- const args = ['-y', '-nostdin', '-i', src];++ // startSec: one window of a long recording (media.analyze), decoded from there only.++ const args = ['-y', '-nostdin', ...(startSec > 0 ? ['-ss', String(startSec)] : []), '-i', src];if (seconds > 0) args.push('-t', String(seconds));args.push('-vn', '-ac', '1', '-ar', '16000', '-f', 'wav', wav);let ff;@@@ -142,7 +143,7 @@ function createWhisperProvider(record, { config }) {if (!available()) throw new ProviderError('whisper.cpp is not installed on this host', { code: 'provider.unavailable' });const lane = req.live ? 'live' : 'batch';const timeoutMs = req.timeoutMs || 300000;-- const wav = await toWav(req.filePath, req.seconds || 0, req.signal, timeoutMs);++ const wav = await toWav(req.filePath, req.seconds || 0, req.signal, timeoutMs, req.startSec || 0);try {await acquire(lane);try {@@@ -339,15 +339,29 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spaconst route = registry.resolveRoute(stepDef.stt_route || 'live.stt');if (!route || route.disabled) gaps.push('No speech-to-text route (live.stt): no transcript.');else {-- try {-- sttExec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: seconds || 0, offsetSec: 0, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });-- const r = sttExec.result;-- const segments = (r.segments || []).slice(0, LIMIT.segments).map((g) => ({ start: r2(g.start), end: r2(g.end), text: String(g.text || '').slice(0, 2000) }));-- transcript = { available: true, language: r.language || input.language || 'en', text: String(r.text || '').slice(0, LIMIT.text), segments };-- } catch (e) {-- if (ctx.signal && ctx.signal.aborted) throw e;-- gaps.push(`Speech-to-text did not run (${e.code || e.message}): no transcript.`);++ // In windows (sttWindowSec, 10 min): each call stays inside the provider's timeout, and one++ // failed window costs that window only.++ const win = ma.sttWindowSec || 600;++ const segs = []; const langs = {}; const failed = []; let lastErr = null;++ for (let start = 0; start < duration; start += win) {++ const len = Math.min(win, duration - start);++ try {++ const ex = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: len, startSec: start, offsetSec: start, timeoutMs: 600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });++ sttExec = ex;++ const r = ex.result;++ if (r.language) langs[r.language] = (langs[r.language] || 0) + (r.segments || []).length + 1;++ for (const g of r.segments || []) segs.push({ start: r2(g.start), end: r2(g.end), text: String(g.text || '').slice(0, 2000) });++ } catch (e) {++ if (ctx.signal && ctx.signal.aborted) throw e;++ failed.push(`${clock(start)}–${clock(start + len)}`); lastErr = e;++ }}++ if (sttExec) {++ const segments = segs.sort((a, b) => a.start - b.start).slice(0, LIMIT.segments);++ const language = Object.keys(langs).sort((a, b) => langs[b] - langs[a])[0] || input.language || 'en';++ transcript = { available: true, language, text: segments.map((g) => g.text).join(' ').replace(/\s+/g, ' ').trim().slice(0, LIMIT.text), segments };++ if (failed.length) gaps.push(`Speech-to-text failed for ${failed.slice(0, 5).join(', ')}${failed.length > 5 ? ` and ${failed.length - 5} more` : ''} (${lastErr.code || lastErr.message}).`);++ } else gaps.push(`Speech-to-text did not run (${(lastErr && (lastErr.code || lastErr.message)) || 'no window'}): no transcript.`);}}const ratio = transcript.available ? speechRatio(transcript.segments, duration) : null;@@@ -84,6 +84,8 @@ t.test('whisper reports the language it detected, not "auto"', async () => {const r = await w.transcribe({ filePath: clip, language: 'auto', seconds: 3, timeoutMs: 30000 });assert.deepStrictEqual([r.language, r.text], ['ja', 'こんにちは、みなさん']);assert.strictEqual((await w.transcribe({ filePath: clip, language: 'en', seconds: 3, timeoutMs: 30000 })).language, 'en');++ const later = await w.transcribe({ filePath: clip, language: 'auto', seconds: 5, startSec: 20, offsetSec: 20, timeoutMs: 30000 });++ assert.strictEqual(later.segments[0].start, 20, 'a window from 20 s: its timestamps are the recording\'s');} finally { fs.rmSync(dir, { recursive: true, force: true }); }});@@@ -93,13 +95,15 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local modawait new Promise((r) => media.listen(0, '127.0.0.1', r));mediaBase = `http://127.0.0.1:${media.address().port}`;const local = await seamServer(() => completion('A short test recording: a quiet tone, a pause, then a loud burst at 0:16.'));-- const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: mediaBase, AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'qwen-test' } });++ const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: mediaBase, AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'qwen-test', AI_MEDIA_STT_WINDOW_SEC: '10' } });try {const r = await run(h, {});assert.strictEqual(r.status, 201, r.text.slice(0, 300));const out = r.body.run.output;assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run.error));assert.ok(Math.abs(out.duration_seconds - 28) < 0.2);++ // Speech-to-text in 10 s windows: 0, 10 and 20 s, each placed at its own time.++ assert.ok([0, 10, 20].every((t) => out.transcript.segments.some((g) => g.start === t)), JSON.stringify(out.transcript.segments));assert.deepStrictEqual(out.streams, { video: true, audio: true });assert.deepStrictEqual(out.signals.scene_changes.map((c) => c.t), [8, 16, 20]);assert.deepStrictEqual(out.signals.black, [{ start: 16, end: 20 }]);
@@ -131,6 +131,8 @@ function load(env = process.env) {
@@ -131,6 +131,8 @@ function load(env = process.env) {
reserveBytes: int(env.AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES, 3 * 1024 * 1024 * 1024),
reserveBytes: int(env.AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES, 3 * 1024 * 1024 * 1024),
fetchTimeoutMs: int(env.AI_MEDIA_ANALYSIS_FETCH_TIMEOUT_MS, 20 * 60000),
fetchTimeoutMs: int(env.AI_MEDIA_ANALYSIS_FETCH_TIMEOUT_MS, 20 * 60000),
sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),
sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),
+ // Speech-to-text in windows: each call stays well inside the whisper provider's timeout (10 min).
+ sttWindowSec: Math.max(10, int(env.AI_MEDIA_STT_WINDOW_SEC, 600)),
ffmpeg: env.FFMPEG_BIN || 'ffmpeg',
ffmpeg: env.FFMPEG_BIN || 'ffmpeg',
ffprobe: env.FFPROBE_BIN || 'ffprobe',
ffprobe: env.FFPROBE_BIN || 'ffprobe',
},
},
@@ -125,7 +125,8 @@ function createStubProvider(record = { key: 'stub', metadata: {} }, { stats = nu
@@ -125,7 +125,8 @@ function createStubProvider(record = { key: 'stub', metadata: {} }, { stats = nu
}
}
if (op === 'transcribe') {
if (op === 'transcribe') {
const n = 2 + Math.floor(rnd() * 3);
const n = 2 + Math.floor(rnd() * 3);
- const segments = Array.from({ length: n }, (_, i) => ({ start: i * 4, end: i * 4 + 3.5, text: `(synthetic) ${phrase(rnd, 5)}` }));
+ const at = Number(req.offsetSec) || 0; // a window of a longer recording keeps the recording's times
+ const segments = Array.from({ length: n }, (_, i) => ({ start: at + i * 4, end: at + i * 4 + 3.5, text: `(synthetic) ${phrase(rnd, 5)}` }));
return { text: segments.map(s => s.text).join(' '), segments, model, synthetic: true, usage: { input: 0, output: 0, cached: 0, estimated: true } };
return { text: segments.map(s => s.text).join(' '), segments, model, synthetic: true, usage: { input: 0, output: 0, cached: 0, estimated: true } };
}
}
let json = null;
let json = null;
@@ -81,10 +81,11 @@ function createWhisperProvider(record, { config }) {
@@ -81,10 +81,11 @@ function createWhisperProvider(record, { config }) {
return ch;
return ch;
}
}
- function toWav(src, seconds, signal, timeoutMs) {
+ function toWav(src, seconds, signal, timeoutMs, startSec = 0) {
return new Promise((resolve, reject) => {
return new Promise((resolve, reject) => {
const wav = path.join(os.tmpdir(), `openvibe-ai-tx-${Date.now()}-${Math.floor(Math.random() * 1e6)}.wav`);
const wav = path.join(os.tmpdir(), `openvibe-ai-tx-${Date.now()}-${Math.floor(Math.random() * 1e6)}.wav`);
- const args = ['-y', '-nostdin', '-i', src];
+ // startSec: one window of a long recording (media.analyze), decoded from there only.
+ const args = ['-y', '-nostdin', ...(startSec > 0 ? ['-ss', String(startSec)] : []), '-i', src];
if (seconds > 0) args.push('-t', String(seconds));
if (seconds > 0) args.push('-t', String(seconds));
args.push('-vn', '-ac', '1', '-ar', '16000', '-f', 'wav', wav);
args.push('-vn', '-ac', '1', '-ar', '16000', '-f', 'wav', wav);
let ff;
let ff;
@@ -142,7 +143,7 @@ function createWhisperProvider(record, { config }) {
@@ -142,7 +143,7 @@ function createWhisperProvider(record, { config }) {
if (!available()) throw new ProviderError('whisper.cpp is not installed on this host', { code: 'provider.unavailable' });
if (!available()) throw new ProviderError('whisper.cpp is not installed on this host', { code: 'provider.unavailable' });
const lane = req.live ? 'live' : 'batch';
const lane = req.live ? 'live' : 'batch';
const timeoutMs = req.timeoutMs || 300000;
const timeoutMs = req.timeoutMs || 300000;
- const wav = await toWav(req.filePath, req.seconds || 0, req.signal, timeoutMs);
+ const wav = await toWav(req.filePath, req.seconds || 0, req.signal, timeoutMs, req.startSec || 0);
try {
try {
await acquire(lane);
await acquire(lane);
try {
try {
@@ -339,15 +339,29 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
@@ -339,15 +339,29 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
const route = registry.resolveRoute(stepDef.stt_route || 'live.stt');
const route = registry.resolveRoute(stepDef.stt_route || 'live.stt');
if (!route || route.disabled) gaps.push('No speech-to-text route (live.stt): no transcript.');
if (!route || route.disabled) gaps.push('No speech-to-text route (live.stt): no transcript.');
else {
else {
- try {
- sttExec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: seconds || 0, offsetSec: 0, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });
- const r = sttExec.result;
- const segments = (r.segments || []).slice(0, LIMIT.segments).map((g) => ({ start: r2(g.start), end: r2(g.end), text: String(g.text || '').slice(0, 2000) }));
- transcript = { available: true, language: r.language || input.language || 'en', text: String(r.text || '').slice(0, LIMIT.text), segments };
- } catch (e) {
- if (ctx.signal && ctx.signal.aborted) throw e;
- gaps.push(`Speech-to-text did not run (${e.code || e.message}): no transcript.`);
+ // In windows (sttWindowSec, 10 min): each call stays inside the provider's timeout, and one
+ // failed window costs that window only.
+ const win = ma.sttWindowSec || 600;
+ const segs = []; const langs = {}; const failed = []; let lastErr = null;
+ for (let start = 0; start < duration; start += win) {
+ const len = Math.min(win, duration - start);
+ try {
+ const ex = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: len, startSec: start, offsetSec: start, timeoutMs: 600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });
+ sttExec = ex;
+ const r = ex.result;
+ if (r.language) langs[r.language] = (langs[r.language] || 0) + (r.segments || []).length + 1;
+ for (const g of r.segments || []) segs.push({ start: r2(g.start), end: r2(g.end), text: String(g.text || '').slice(0, 2000) });
+ } catch (e) {
+ if (ctx.signal && ctx.signal.aborted) throw e;
+ failed.push(`${clock(start)}–${clock(start + len)}`); lastErr = e;
+ }
}
}
+ if (sttExec) {
+ const segments = segs.sort((a, b) => a.start - b.start).slice(0, LIMIT.segments);
+ const language = Object.keys(langs).sort((a, b) => langs[b] - langs[a])[0] || input.language || 'en';
+ transcript = { available: true, language, text: segments.map((g) => g.text).join(' ').replace(/\s+/g, ' ').trim().slice(0, LIMIT.text), segments };
+ if (failed.length) gaps.push(`Speech-to-text failed for ${failed.slice(0, 5).join(', ')}${failed.length > 5 ? ` and ${failed.length - 5} more` : ''} (${lastErr.code || lastErr.message}).`);
+ } else gaps.push(`Speech-to-text did not run (${(lastErr && (lastErr.code || lastErr.message)) || 'no window'}): no transcript.`);
}
}
}
}
const ratio = transcript.available ? speechRatio(transcript.segments, duration) : null;
const ratio = transcript.available ? speechRatio(transcript.segments, duration) : null;
@@ -84,6 +84,8 @@ t.test('whisper reports the language it detected, not "auto"', async () => {
@@ -84,6 +84,8 @@ t.test('whisper reports the language it detected, not "auto"', async () => {
const r = await w.transcribe({ filePath: clip, language: 'auto', seconds: 3, timeoutMs: 30000 });
const r = await w.transcribe({ filePath: clip, language: 'auto', seconds: 3, timeoutMs: 30000 });
assert.deepStrictEqual([r.language, r.text], ['ja', 'こんにちは、みなさん']);
assert.deepStrictEqual([r.language, r.text], ['ja', 'こんにちは、みなさん']);
assert.strictEqual((await w.transcribe({ filePath: clip, language: 'en', seconds: 3, timeoutMs: 30000 })).language, 'en');
assert.strictEqual((await w.transcribe({ filePath: clip, language: 'en', seconds: 3, timeoutMs: 30000 })).language, 'en');
+ const later = await w.transcribe({ filePath: clip, language: 'auto', seconds: 5, startSec: 20, offsetSec: 20, timeoutMs: 30000 });
+ assert.strictEqual(later.segments[0].start, 20, 'a window from 20 s: its timestamps are the recording\'s');
} finally { fs.rmSync(dir, { recursive: true, force: true }); }
} finally { fs.rmSync(dir, { recursive: true, force: true }); }
});
});
@@ -93,13 +95,15 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local mod
@@ -93,13 +95,15 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local mod
await new Promise((r) => media.listen(0, '127.0.0.1', r));
await new Promise((r) => media.listen(0, '127.0.0.1', r));
mediaBase = `http://127.0.0.1:${media.address().port}`;
mediaBase = `http://127.0.0.1:${media.address().port}`;
const local = await seamServer(() => completion('A short test recording: a quiet tone, a pause, then a loud burst at 0:16.'));
const local = await seamServer(() => completion('A short test recording: a quiet tone, a pause, then a loud burst at 0:16.'));
- const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: mediaBase, AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'qwen-test' } });
+ const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: mediaBase, AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'qwen-test', AI_MEDIA_STT_WINDOW_SEC: '10' } });
try {
try {
const r = await run(h, {});
const r = await run(h, {});
assert.strictEqual(r.status, 201, r.text.slice(0, 300));
assert.strictEqual(r.status, 201, r.text.slice(0, 300));
const out = r.body.run.output;
const out = r.body.run.output;
assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run.error));
assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run.error));
assert.ok(Math.abs(out.duration_seconds - 28) < 0.2);
assert.ok(Math.abs(out.duration_seconds - 28) < 0.2);
+ // Speech-to-text in 10 s windows: 0, 10 and 20 s, each placed at its own time.
+ assert.ok([0, 10, 20].every((t) => out.transcript.segments.some((g) => g.start === t)), JSON.stringify(out.transcript.segments));
assert.deepStrictEqual(out.streams, { video: true, audio: true });
assert.deepStrictEqual(out.streams, { video: true, audio: true });
assert.deepStrictEqual(out.signals.scene_changes.map((c) => c.t), [8, 16, 20]);
assert.deepStrictEqual(out.signals.scene_changes.map((c) => c.t), [8, 16, 20]);
assert.deepStrictEqual(out.signals.black, [{ start: 16, end: 20 }]);
assert.deepStrictEqual(out.signals.black, [{ start: 16, end: 20 }]);