openvibers/openvibe.ai Commits
live.media.transcribe runs one window of a recording where it lies (roadmap WS-O task 2, so Live's VOD and clip transcripts can move here). The transcribe step streams through the loopback reader, now shared with media.analyze (one per engine), instead of downloading; AI_MEDIA_STREAM=0 still downloads. start_sec seeks and offsets the timestamps, unless offset_sec says otherwise; before this, offset_sec shifted timestamps without seeking. low_power caps whisper at two threads, as Live does while streams are live. test/transcribe-window.test.js: a window's segments sit at start_sec, offset_sec overrides it, nothing is written to disk, a URL outside the rules is refused. npm test passes.
Changes
@@@ -103,11 +103,12 @@ function createWhisperProvider(record, { config }) {});}-- function decode(wavPath, { language = 'en', live = false, offsetSec = 0, signal, timeoutMs }) {++ function decode(wavPath, { language = 'en', live = false, lowPower = false, offsetSec = 0, signal, timeoutMs }) {return new Promise((resolve, reject) => {const lang = language && language !== 'en' && multi() ? String(language) : 'en';const model = lang !== 'en' ? multi() : (live && exists(MODEL_LIVE) ? MODEL_LIVE : MODEL);-- const threads = live ? Math.max(1, Math.min(2, w.threads)) : w.threads;++ // Live segments, and batch work while the caller says streams are live (low power), use at most 2 threads.++ const threads = live || lowPower ? Math.max(1, Math.min(2, w.threads)) : w.threads;const outBase = `${wavPath}.out`;const jsonPath = `${outBase}.json`;const args = ['-m', model, '-f', wavPath, '-oj', '-of', outBase, '-t', String(threads), '-l', lang];@@@ -147,7 +148,7 @@ function createWhisperProvider(record, { config }) {try {await acquire(lane);try {-- const r = await decode(wav, { language: req.language, live: req.live, offsetSec: req.offsetSec || 0, signal: req.signal, timeoutMs });++ const r = await decode(wav, { language: req.language, live: req.live, lowPower: req.lowPower === true, offsetSec: req.offsetSec || 0, signal: req.signal, timeoutMs });return { ...r, usage: { input: 0, output: 0, cached: 0 } };} finally { release(lane); }} finally { try { fs.unlinkSync(wav); } catch { /* */ } }@@@ -97,7 +97,10 @@ function normMessages(messages, user) {}function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spawnImpl, log = console }) {-- const mediaAnalysis = require('./media-analysis').createMediaAnalysis({ registry, pool, fetcher, quotas, config, spawnImpl, log });++ // One loopback reader for everything that streams media (server/media-proxy.js): transcripts and media.analyze.++ let proxy = null;++ const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));++ const mediaAnalysis = require('./media-analysis').createMediaAnalysis({ registry, pool, fetcher, quotas, config, spawnImpl, log, reader });function routeFor(key) {const r = registry.resolveRoute(key);if (!r) throw new AiError(503, 'route.unavailable', `no route ${key}`);@@@ -186,13 +189,20 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spaasync function transcribeStep(step, input, wf, run, ctx) {const route = routeFor(step.route || 'live.stt');-- const { file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal });++ // Read where it lies (the loopback reader), one window at a time: start_sec seeks with a range request, and++ // timestamps are offset by start_sec unless offset_sec says otherwise. AI_MEDIA_STREAM=0 downloads first.++ const streamed = !config.mediaAnalysis || config.mediaAnalysis.stream !== false;++ const handle = streamed ? await reader().register(fetcher.mediaUrlOf(input)) : null;++ const file = handle ? handle.url : (await fetcher.loadMediaToFile(input, { signal: ctx.signal })).file;try {-- const exec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: input.seconds || 0, offsetSec: input.offset_sec || 0, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });++ const startSec = Number(input.start_sec) || 0;++ const offsetSec = input.offset_sec != null ? Number(input.offset_sec) : startSec;++ const exec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: input.seconds || 0, startSec, offsetSec, lowPower: input.low_power === true, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });const r = exec.result;return { output: { text: r.text || '', language: r.language || input.language || 'en', segments: r.segments || [] }, exec, route };} finally {-- try { fs.unlinkSync(file); } catch { /* */ }++ if (handle) handle.release();++ else { try { fs.unlinkSync(file); } catch { /* */ } }}}@@@ -536,7 +536,7 @@ const structured = [{key: 'live.media.transcribe', name: 'Speech-to-text for a VOD, clip or audio file', namespace: 'live',description: 'whisper.cpp on this host (Live\'s transcribe.js), media fetched only from allow-listed OpenVibe hosts.',-- input_schema: TEXT_OUT({ media: MEDIA_REF, media_url: STR(2048), language: LANG, seconds: { type: 'integer', minimum: 0, maximum: 36000 }, offset_sec: { type: 'number', minimum: 0 } }, []),++ input_schema: TEXT_OUT({ media: MEDIA_REF, media_url: STR(2048), language: LANG, seconds: { type: 'integer', minimum: 0, maximum: 36000 }, start_sec: { type: 'number', minimum: 0 }, offset_sec: { type: 'number', minimum: 0 }, low_power: { type: 'boolean' } }, []),output_schema: TEXT_OUT({ text: { type: 'string' }, language: STR(8), segments: { type: 'array', items: { type: 'object', required: ['start', 'end', 'text'], properties: { start: { type: 'number' }, end: { type: 'number' }, text: { type: 'string' } } } } }, ['text', 'segments']),steps: [{ kind: 'transcribe', route: 'live.stt' }],cache_mode: 'private',@@@ -183,10 +183,10 @@ function extractive(facts, highlights) {return parts.join(' ').slice(0, 1500);}--function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console }) {++function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console, reader: shared = null }) {// Long recordings are streamed, not downloaded (WS-O task 2): ffmpeg reads them through a loopback reader.let proxy = null;-- const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));++ const reader = () => (shared && shared()) || proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));const ma = config.mediaAnalysis || {};const ffmpeg = ma.ffmpeg || 'ffmpeg';const ffprobe = ma.ffprobe || 'ffprobe';nnew file mode 100644@@@ -0,0 +1,37 @@++'use strict';++// live.media.transcribe by window (roadmap WS-O task 2: Live's VOD and clip transcripts run here). start_sec seeks++// (the loopback reader streams the recording; nothing is downloaded) and offsets the timestamps unless offset_sec++// says otherwise; low_power is accepted; a URL outside the rules is refused before anything runs.++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const nodeHttp = require('http');++const { boot, request, token, suite, ALL } = require('./helpers');++++const t = suite('transcribe-window');++const tok = token('live', ALL);++let h; let media; let base; let hits = 0;++++t.test('boot with an internal Media stand-in', async () => {++ media = nodeHttp.createServer((req, res) => { hits++; res.writeHead(200, { 'Content-Type': 'video/mp4', 'Accept-Ranges': 'bytes' }); res.end(Buffer.alloc(4096, 1)); });++ await new Promise((r) => media.listen(0, '127.0.0.1', r));++ base = `http://127.0.0.1:${media.address().port}`;++ h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base } });++});++++t.test('a window starts at start_sec and its timestamps follow', async () => {++ const run = (input) => request(h.base, 'POST', '/api/v1/runs?wait=20000', { tok, body: { workflow: 'live.media.transcribe', input: { media_url: `${base}/v/7`, language: 'en', ...input } } });++ let r = await run({ start_sec: 300, seconds: 60, low_power: true });++ assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run.error || r.body));++ const segs = r.body.run.output.segments;++ assert.ok(segs.length && segs.every((s) => s.start >= 300 && s.start < 360), JSON.stringify(segs));++ r = await run({ start_sec: 300, seconds: 60, offset_sec: 0 });++ assert.ok(r.body.run.output.segments.every((s) => s.start < 60), 'offset_sec overrides');++ const work = path.join(h.dir, 'media-tmp');++ assert.deepStrictEqual(fs.existsSync(work) ? fs.readdirSync(work) : [], [], 'nothing downloaded');++ r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'live.media.transcribe', input: { media_url: 'https://evil.example/x.mp4' } } });++ assert.strictEqual(r.body.run ? r.body.run.error.code : r.body.code, 'fetch.refused');++ await h.stop(); media.close();++});++++t.run();
@@ -103,11 +103,12 @@ function createWhisperProvider(record, { config }) {
@@ -103,11 +103,12 @@ function createWhisperProvider(record, { config }) {
});
});
}
}
- function decode(wavPath, { language = 'en', live = false, offsetSec = 0, signal, timeoutMs }) {
+ function decode(wavPath, { language = 'en', live = false, lowPower = false, offsetSec = 0, signal, timeoutMs }) {
return new Promise((resolve, reject) => {
return new Promise((resolve, reject) => {
const lang = language && language !== 'en' && multi() ? String(language) : 'en';
const lang = language && language !== 'en' && multi() ? String(language) : 'en';
const model = lang !== 'en' ? multi() : (live && exists(MODEL_LIVE) ? MODEL_LIVE : MODEL);
const model = lang !== 'en' ? multi() : (live && exists(MODEL_LIVE) ? MODEL_LIVE : MODEL);
- const threads = live ? Math.max(1, Math.min(2, w.threads)) : w.threads;
+ // Live segments, and batch work while the caller says streams are live (low power), use at most 2 threads.
+ const threads = live || lowPower ? Math.max(1, Math.min(2, w.threads)) : w.threads;
const outBase = `${wavPath}.out`;
const outBase = `${wavPath}.out`;
const jsonPath = `${outBase}.json`;
const jsonPath = `${outBase}.json`;
const args = ['-m', model, '-f', wavPath, '-oj', '-of', outBase, '-t', String(threads), '-l', lang];
const args = ['-m', model, '-f', wavPath, '-oj', '-of', outBase, '-t', String(threads), '-l', lang];
@@ -147,7 +148,7 @@ function createWhisperProvider(record, { config }) {
@@ -147,7 +148,7 @@ function createWhisperProvider(record, { config }) {
try {
try {
await acquire(lane);
await acquire(lane);
try {
try {
- const r = await decode(wav, { language: req.language, live: req.live, offsetSec: req.offsetSec || 0, signal: req.signal, timeoutMs });
+ const r = await decode(wav, { language: req.language, live: req.live, lowPower: req.lowPower === true, offsetSec: req.offsetSec || 0, signal: req.signal, timeoutMs });
return { ...r, usage: { input: 0, output: 0, cached: 0 } };
return { ...r, usage: { input: 0, output: 0, cached: 0 } };
} finally { release(lane); }
} finally { release(lane); }
} finally { try { fs.unlinkSync(wav); } catch { /* */ } }
} finally { try { fs.unlinkSync(wav); } catch { /* */ } }
@@ -97,7 +97,10 @@ function normMessages(messages, user) {
@@ -97,7 +97,10 @@ function normMessages(messages, user) {
}
}
function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spawnImpl, log = console }) {
function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spawnImpl, log = console }) {
- const mediaAnalysis = require('./media-analysis').createMediaAnalysis({ registry, pool, fetcher, quotas, config, spawnImpl, log });
+ // One loopback reader for everything that streams media (server/media-proxy.js): transcripts and media.analyze.
+ let proxy = null;
+ const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));
+ const mediaAnalysis = require('./media-analysis').createMediaAnalysis({ registry, pool, fetcher, quotas, config, spawnImpl, log, reader });
function routeFor(key) {
function routeFor(key) {
const r = registry.resolveRoute(key);
const r = registry.resolveRoute(key);
if (!r) throw new AiError(503, 'route.unavailable', `no route ${key}`);
if (!r) throw new AiError(503, 'route.unavailable', `no route ${key}`);
@@ -186,13 +189,20 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
@@ -186,13 +189,20 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
async function transcribeStep(step, input, wf, run, ctx) {
async function transcribeStep(step, input, wf, run, ctx) {
const route = routeFor(step.route || 'live.stt');
const route = routeFor(step.route || 'live.stt');
- const { file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal });
+ // Read where it lies (the loopback reader), one window at a time: start_sec seeks with a range request, and
+ // timestamps are offset by start_sec unless offset_sec says otherwise. AI_MEDIA_STREAM=0 downloads first.
+ const streamed = !config.mediaAnalysis || config.mediaAnalysis.stream !== false;
+ const handle = streamed ? await reader().register(fetcher.mediaUrlOf(input)) : null;
+ const file = handle ? handle.url : (await fetcher.loadMediaToFile(input, { signal: ctx.signal })).file;
try {
try {
- const exec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: input.seconds || 0, offsetSec: input.offset_sec || 0, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });
+ const startSec = Number(input.start_sec) || 0;
+ const offsetSec = input.offset_sec != null ? Number(input.offset_sec) : startSec;
+ const exec = await pool.execute(route, 'transcribe', { filePath: file, language: input.language || 'en', seconds: input.seconds || 0, startSec, offsetSec, lowPower: input.low_power === true, timeoutMs: 3600000 }, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: null });
const r = exec.result;
const r = exec.result;
return { output: { text: r.text || '', language: r.language || input.language || 'en', segments: r.segments || [] }, exec, route };
return { output: { text: r.text || '', language: r.language || input.language || 'en', segments: r.segments || [] }, exec, route };
} finally {
} finally {
- try { fs.unlinkSync(file); } catch { /* */ }
+ if (handle) handle.release();
+ else { try { fs.unlinkSync(file); } catch { /* */ } }
}
}
}
}
@@ -536,7 +536,7 @@ const structured = [
@@ -536,7 +536,7 @@ const structured = [
{
{
key: 'live.media.transcribe', name: 'Speech-to-text for a VOD, clip or audio file', namespace: 'live',
key: 'live.media.transcribe', name: 'Speech-to-text for a VOD, clip or audio file', namespace: 'live',
description: 'whisper.cpp on this host (Live\'s transcribe.js), media fetched only from allow-listed OpenVibe hosts.',
description: 'whisper.cpp on this host (Live\'s transcribe.js), media fetched only from allow-listed OpenVibe hosts.',
- input_schema: TEXT_OUT({ media: MEDIA_REF, media_url: STR(2048), language: LANG, seconds: { type: 'integer', minimum: 0, maximum: 36000 }, offset_sec: { type: 'number', minimum: 0 } }, []),
+ input_schema: TEXT_OUT({ media: MEDIA_REF, media_url: STR(2048), language: LANG, seconds: { type: 'integer', minimum: 0, maximum: 36000 }, start_sec: { type: 'number', minimum: 0 }, offset_sec: { type: 'number', minimum: 0 }, low_power: { type: 'boolean' } }, []),
output_schema: TEXT_OUT({ text: { type: 'string' }, language: STR(8), segments: { type: 'array', items: { type: 'object', required: ['start', 'end', 'text'], properties: { start: { type: 'number' }, end: { type: 'number' }, text: { type: 'string' } } } } }, ['text', 'segments']),
output_schema: TEXT_OUT({ text: { type: 'string' }, language: STR(8), segments: { type: 'array', items: { type: 'object', required: ['start', 'end', 'text'], properties: { start: { type: 'number' }, end: { type: 'number' }, text: { type: 'string' } } } } }, ['text', 'segments']),
steps: [{ kind: 'transcribe', route: 'live.stt' }],
steps: [{ kind: 'transcribe', route: 'live.stt' }],
cache_mode: 'private',
cache_mode: 'private',
@@ -183,10 +183,10 @@ function extractive(facts, highlights) {
@@ -183,10 +183,10 @@ function extractive(facts, highlights) {
return parts.join(' ').slice(0, 1500);
return parts.join(' ').slice(0, 1500);
}
}
-function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console }) {
+function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console, reader: shared = null }) {
// Long recordings are streamed, not downloaded (WS-O task 2): ffmpeg reads them through a loopback reader.
// Long recordings are streamed, not downloaded (WS-O task 2): ffmpeg reads them through a loopback reader.
let proxy = null;
let proxy = null;
- const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));
+ const reader = () => (shared && shared()) || proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));
const ma = config.mediaAnalysis || {};
const ma = config.mediaAnalysis || {};
const ffmpeg = ma.ffmpeg || 'ffmpeg';
const ffmpeg = ma.ffmpeg || 'ffmpeg';
const ffprobe = ma.ffprobe || 'ffprobe';
const ffprobe = ma.ffprobe || 'ffprobe';
new file mode 100644
new file mode 100644
@@ -0,0 +1,37 @@
@@ -0,0 +1,37 @@
+'use strict';
+// live.media.transcribe by window (roadmap WS-O task 2: Live's VOD and clip transcripts run here). start_sec seeks
+// (the loopback reader streams the recording; nothing is downloaded) and offsets the timestamps unless offset_sec
+// says otherwise; low_power is accepted; a URL outside the rules is refused before anything runs.
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const nodeHttp = require('http');
+const { boot, request, token, suite, ALL } = require('./helpers');
+
+const t = suite('transcribe-window');
+const tok = token('live', ALL);
+let h; let media; let base; let hits = 0;
+
+t.test('boot with an internal Media stand-in', async () => {
+ media = nodeHttp.createServer((req, res) => { hits++; res.writeHead(200, { 'Content-Type': 'video/mp4', 'Accept-Ranges': 'bytes' }); res.end(Buffer.alloc(4096, 1)); });
+ await new Promise((r) => media.listen(0, '127.0.0.1', r));
+ base = `http://127.0.0.1:${media.address().port}`;
+ h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base } });
+});
+
+t.test('a window starts at start_sec and its timestamps follow', async () => {
+ const run = (input) => request(h.base, 'POST', '/api/v1/runs?wait=20000', { tok, body: { workflow: 'live.media.transcribe', input: { media_url: `${base}/v/7`, language: 'en', ...input } } });
+ let r = await run({ start_sec: 300, seconds: 60, low_power: true });
+ assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run.error || r.body));
+ const segs = r.body.run.output.segments;
+ assert.ok(segs.length && segs.every((s) => s.start >= 300 && s.start < 360), JSON.stringify(segs));
+ r = await run({ start_sec: 300, seconds: 60, offset_sec: 0 });
+ assert.ok(r.body.run.output.segments.every((s) => s.start < 60), 'offset_sec overrides');
+ const work = path.join(h.dir, 'media-tmp');
+ assert.deepStrictEqual(fs.existsSync(work) ? fs.readdirSync(work) : [], [], 'nothing downloaded');
+ r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'live.media.transcribe', input: { media_url: 'https://evil.example/x.mp4' } } });
+ assert.strictEqual(r.body.run ? r.body.run.error.code : r.body.code, 'fetch.refused');
+ await h.stop(); media.close();
+});
+
+t.run();