From 38d3465bf873a3871320fcfcefde86bcda5f545d Mon Sep 17 00:00:00 2001 From: legop3 Date: Thu, 8 Oct 2026 00:11:32 -0400 Subject: [PATCH 1/8] room audio --- compose.yaml | 6 +- server/src/configuration/definition.js | 2 + server/src/services/authService/index.js | 3 +- server/src/services/embedHttpService/index.js | 2 +- server/src/services/mediaMtxService/config.js | 1 + server/src/services/replayEngineV2/index.js | 8 +- .../services/replayEngineV2/replayBuilder.js | 24 ++- .../services/replayEngineV2/segmentStore.js | 28 ++- server/src/services/replayEngineV2/sources.js | 9 +- .../services/replayEngineV2/workerManager.js | 46 +++-- .../roomAudioService/configuration.js | 14 ++ server/src/services/roomAudioService/index.js | 171 ++++++++++++++++++ server/src/services/sessionService/index.js | 4 + .../services/videoAuthService/httpRoute.js | 11 +- server/src/services/videoAuthService/index.js | 3 + .../src/services/videoAuthService/policy.js | 6 + .../videoAuthService/streamParsing.js | 2 + .../src/services/videoSocketService/index.js | 7 +- .../src/components/RoomAudioPlayer/index.jsx | 69 +++++++ webui/src/components/SettingsPanel/index.jsx | 6 + .../ServerDisplayApp/ServerDisplayAppRoot.jsx | 8 +- webui/src/driver/DriverPageRoot.jsx | 7 +- webui/src/lib/socket.js | 1 + webui/src/main.jsx | 2 + webui/src/mic/MicApp.jsx | 111 ++++++++++++ .../MiniSummaryApp/MiniSummaryAppRoot.jsx | 2 + webui/src/ptz/PtzAppRoot.jsx | 2 + webui/src/settings/namespaces.js | 1 + .../SpectatorApp/SpectatorAppRoot.jsx | 8 +- 29 files changed, 530 insertions(+), 34 deletions(-) create mode 100644 server/src/services/roomAudioService/configuration.js create mode 100644 server/src/services/roomAudioService/index.js create mode 100644 webui/src/components/RoomAudioPlayer/index.jsx create mode 100644 webui/src/mic/MicApp.jsx diff --git a/compose.yaml b/compose.yaml index 869878a9..8ce2a558 100644 --- a/compose.yaml +++ b/compose.yaml @@ -23,9 +23,13 @@ services: # Required for Bluetooth hardware such as the Balance Board. - /run/dbus/system_bus_socket:/run/dbus/system_bus_socket:ro - # Required for Kinect USB access. + # Kinect USB access. devices: - /dev/bus/usb:/dev/bus/usb + # For a server-local microphone, uncomment this device and group_add below. + # - /dev/snd:/dev/snd + # group_add: + # - "${AUDIO_GID:?Set AUDIO_GID to the host sound device group ID}" cap_add: - NET_ADMIN diff --git a/server/src/configuration/definition.js b/server/src/configuration/definition.js index fb0de1c6..dc27d605 100644 --- a/server/src/configuration/definition.js +++ b/server/src/configuration/definition.js @@ -13,6 +13,7 @@ const audioForward = require('../services/audioForwardService/configuration'); const audioLevels = require('../services/audioLevelsService/configuration'); const homeAssistant = require('../services/homeAssistantService/configuration'); const homeAssistantActivities = require('../services/homeAssistantActivitiesService/configuration'); +const roomAudio = require('../services/roomAudioService/configuration'); const roomCameras = require('../services/roomCameraService/configuration'); const ptzCamera = require('../services/ptzCameraService/configuration'); const kinect = require('../services/kinectService/configuration'); @@ -46,6 +47,7 @@ const definitions = [ // Activities own a separate catalog even though the HA connection is shared. homeAssistantActivities, roomCameras, + roomAudio, ptzCamera, kinect, balanceBoard, diff --git a/server/src/services/authService/index.js b/server/src/services/authService/index.js index b78315e9..39a43ebe 100644 --- a/server/src/services/authService/index.js +++ b/server/src/services/authService/index.js @@ -124,7 +124,7 @@ function grantExternalSpectatorAccessAfterAdminLogin(socket) { } io.on('connection', (socket) => { - const requestedRole = socket.handshake?.query?.role; + const requestedRole = socket.handshake?.auth?.role || socket.handshake?.query?.role; /* Role is assigned before the browser's full identity heartbeat has completed. For admin-gated external spectators, fail closed here; the spectator page can @@ -188,4 +188,5 @@ module.exports = { isAdmin, isLockdownAdmin, authenticate, + canBecomeSpectator, }; diff --git a/server/src/services/embedHttpService/index.js b/server/src/services/embedHttpService/index.js index ce8e7023..0cdacedb 100644 --- a/server/src/services/embedHttpService/index.js +++ b/server/src/services/embedHttpService/index.js @@ -16,7 +16,7 @@ const { renderIndexHtml, renderOgImage, renderWebManifest } = require('../embedS a direct browser load. Keeping the setup and admin routes in this explicit allowlist prevents them from working only after client-side navigation. */ -app.get(['/', '/old', '/spectate', '/mini', '/display', '/scanner', '/database', '/ptz', '/reports', '/setup', '/admin'], async (req, res) => { +app.get(['/', '/old', '/spectate', '/mini', '/display', '/scanner', '/database', '/ptz', '/reports', '/setup', '/admin', '/mic'], async (req, res) => { try { const html = await renderIndexHtml(req); res.type('html').send(html); diff --git a/server/src/services/mediaMtxService/config.js b/server/src/services/mediaMtxService/config.js index 96964091..175206ab 100644 --- a/server/src/services/mediaMtxService/config.js +++ b/server/src/services/mediaMtxService/config.js @@ -81,6 +81,7 @@ function buildMediaMtxConfig({ config, serverPort, snapshotWriterPath }) { { action: 'pprof' }, ], paths: { + '~^room-audio/': { source: 'publisher' }, all: { source: 'publisher', sourceOnDemand: false, diff --git a/server/src/services/replayEngineV2/index.js b/server/src/services/replayEngineV2/index.js index 33110e8e..a4d8b7af 100644 --- a/server/src/services/replayEngineV2/index.js +++ b/server/src/services/replayEngineV2/index.js @@ -6,6 +6,7 @@ const { promisify } = require('util'); const fsp = require('fs/promises'); const logger = require('../../globals/logger').child('replayEngineV2'); const roverManager = require('../roverManager'); +const { roomAudioEvents } = require('../roomAudioService'); const { getRoomCameras, roomCameraEvents } = require('../roomCameraService'); const { @@ -40,8 +41,8 @@ function getActiveSegmentRoot() { return runtime.activeSegmentRoot; } -const workerManager = createWorkerManager({ getActiveSegmentRoot }); const segmentStore = createSegmentStore({ getActiveSegmentRoot }); +const workerManager = createWorkerManager({ getActiveSegmentRoot, onWorkerClosed: segmentStore.refreshIndexForWorker }); const sidebarRenderer = createSidebarRenderer({ execFileAsync, ensureDir }); const replayBuilder = createReplayBuilder({ execFileAsync, @@ -50,6 +51,7 @@ const replayBuilder = createReplayBuilder({ renderSidebarVideo: sidebarRenderer.renderSidebarVideo, getVideoEntriesForSource: segmentStore.getVideoEntriesForSource, getAudioEntriesForSource: segmentStore.getAudioEntriesForSource, + getRoomAudioEntries: segmentStore.getRoomAudioEntries, overlapping: segmentStore.overlapping, }); registerReplaySocketHooks({ tryTriggerReplay, validateSources, getDefaultWebSources }); @@ -88,6 +90,10 @@ async function start() { }, CLEANUP_INTERVAL_MS); } +roomAudioEvents.on('update', () => { + tick().catch((err) => logger.warn('room audio update tick failed', err.message)); +}); + roomCameraEvents.on('update', () => { tick().catch((err) => logger.warn('room camera update tick failed', err.message)); }); diff --git a/server/src/services/replayEngineV2/replayBuilder.js b/server/src/services/replayEngineV2/replayBuilder.js index 257c4231..cd25d642 100644 --- a/server/src/services/replayEngineV2/replayBuilder.js +++ b/server/src/services/replayEngineV2/replayBuilder.js @@ -51,7 +51,7 @@ function buildChatEventsForWindow(startMs, endMs, limit = 22, preWindowCount = 1 return [...beforeWindow, ...inWindow].sort((a, b) => a.ts - b.ts); } -function createReplayBuilder({ execFileAsync, fsp, ensureDir, renderSidebarVideo, getVideoEntriesForSource, getAudioEntriesForSource, overlapping }) { +function createReplayBuilder({ execFileAsync, fsp, ensureDir, renderSidebarVideo, getVideoEntriesForSource, getAudioEntriesForSource, getRoomAudioEntries, overlapping }) { function resolveReplayWindow({ sources = [], nowMs, guardMs, durationMs }) { const tentativeEnd = nowMs - guardMs; const sourceEnds = []; @@ -186,6 +186,28 @@ function createReplayBuilder({ execFileAsync, fsp, ensureDir, renderSidebarVideo if (!normalizedVideos.length) throw new Error('No replay segments available for selected sources'); + const roomAudioEntries = await pinSegmentFiles( + overlapping(getRoomAudioEntries(), tStart, tEnd), tmpDir, 'room-audio-seg', + ); + if (roomAudioEntries.length) { + // Place each segment at its recorded time. A mic joining late or + // reconnecting must leave silence in the gap, not shift its speech earlier. + const audioInputs = roomAudioEntries.flatMap((entry) => ['-i', entry.filePath]); + const audioFilters = roomAudioEntries.map((entry, index) => { + const start = Math.max(0, (tStart - entry.startMs) / 1000); + const duration = (Math.min(tEnd, entry.endMs) - Math.max(tStart, entry.startMs)) / 1000; + const delay = Math.max(0, Math.round(entry.startMs - tStart)); + return `[${index}:a]atrim=start=${start.toFixed(3)}:duration=${duration.toFixed(3)},asetpts=PTS-STARTPTS,adelay=${delay}:all=1[room${index}]`; + }); + const refs = roomAudioEntries.map((_, index) => `[room${index}]`).join(''); + audioFilters.push(`${refs}amix=inputs=${roomAudioEntries.length}:normalize=0[roomout]`); + const roomAudioPath = path.join(tmpDir, 'room-audio.m4a'); + await execFileAsync(FFMPEG_BIN, ['-y', '-hide_banner', '-loglevel', 'error', ...audioInputs, + '-filter_complex', audioFilters.join(';'), '-map', '[roomout]', '-vn', '-ac', '1', '-ar', '48000', + '-c:a', 'aac', '-b:a', '96k', roomAudioPath]); + normalizedAudios.push(roomAudioPath); + } + const layout = buildGridLayout(normalizedVideos.length); const { maxWidth, maxHeight } = await probeMaxFrameSize(normalizedVideos.map((v) => v.path)); let tileWidth = maxWidth || 640; diff --git a/server/src/services/replayEngineV2/segmentStore.js b/server/src/services/replayEngineV2/segmentStore.js index 87e25b71..af752e09 100644 --- a/server/src/services/replayEngineV2/segmentStore.js +++ b/server/src/services/replayEngineV2/segmentStore.js @@ -6,9 +6,10 @@ const fs = require('fs'); const path = require('path'); const logger = require('../../globals/logger').child('replayEngineV2'); const { BUFFER_SECONDS, SEGMENT_SECONDS } = require('./constants'); -const { workers, segmentIndex } = require('./state'); +const { workers, pendingWorkerStarts, segmentIndex } = require('./state'); const { sourceKey, sourceDirForKey } = require('./sources'); const ptzCameraService = require('../ptzCameraService'); +const SEGMENT_FILENAME_PATTERN = /^seg-[a-f0-9-]+-\d{6}\.mp4$/; async function ensureDir(dir) { await fsp.mkdir(dir, { recursive: true }); @@ -25,7 +26,7 @@ function createSegmentStore({ getActiveSegmentRoot }) { return; } - const segmentFiles = files.filter((entry) => entry.isFile() && /^seg-\d{6}\.mp4$/.test(entry.name)).map((entry) => entry.name).sort(); + const segmentFiles = files.filter((entry) => entry.isFile() && SEGMENT_FILENAME_PATTERN.test(entry.name)).map((entry) => entry.name).sort(); if (!segmentFiles.length) { segmentIndex.set(key, []); return; @@ -64,6 +65,13 @@ function createSegmentStore({ getActiveSegmentRoot }) { async function cleanupOldFiles() { const cutoff = Date.now() - BUFFER_SECONDS * 1000; + // Retention applies to stopped workers too; dynamic microphone sessions + // must not leave permanent index entries after their audio expires. + for (const [key, entries] of segmentIndex) { + const retained = entries.filter((entry) => entry.endMs >= cutoff); + if (!retained.length && !workers.has(key) && !pendingWorkerStarts.has(key)) segmentIndex.delete(key); + else segmentIndex.set(key, retained); + } const root = getActiveSegmentRoot(); try { await ensureDir(root); @@ -81,6 +89,10 @@ function createSegmentStore({ getActiveSegmentRoot }) { if (stat.mtimeMs < cutoff) await fsp.unlink(filePath); } catch {} } + if (!workers.has(dirent.name) && !pendingWorkerStarts.has(dirent.name)) { + // rmdir only removes empty directories, preserving buffered or newly written files. + try { await fsp.rmdir(dirPath); } catch {} + } } } catch (err) { logger.warn('cleanup failed', err.message); @@ -115,6 +127,14 @@ function createSegmentStore({ getActiveSegmentRoot }) { return []; } + function getRoomAudioEntries() { + // Include retained buffers from microphones that disconnected during the + // replay window, rather than limiting history to currently live streams. + return [...segmentIndex.entries()] + .filter(([key]) => key.startsWith('roomAudio__audio__')) + .flatMap(([, entries]) => entries); + } + function overlapping(entries, startMs, endMs) { return entries.filter((entry) => entry.endMs > startMs && entry.startMs < endMs); } @@ -153,7 +173,7 @@ function createSegmentStore({ getActiveSegmentRoot }) { try { const files = fs.readdirSync(dir); for (const name of files) { - if (!/^seg-\d{6}\.mp4$/.test(name)) continue; + if (!SEGMENT_FILENAME_PATTERN.test(name)) continue; const full = path.join(dir, name); let stat; try { stat = fs.statSync(full); } catch { continue; } @@ -172,10 +192,12 @@ function createSegmentStore({ getActiveSegmentRoot }) { return { refreshSegmentIndex, + refreshIndexForWorker, cleanupOldFiles, getVideoEntriesForSource, getAudioEntriesForRover, getAudioEntriesForSource, + getRoomAudioEntries, overlapping, bootstrapIndexFromDisk, getReplayHealthSnapshot, diff --git a/server/src/services/replayEngineV2/sources.js b/server/src/services/replayEngineV2/sources.js index a1143711..19d78058 100644 --- a/server/src/services/replayEngineV2/sources.js +++ b/server/src/services/replayEngineV2/sources.js @@ -2,8 +2,10 @@ // Purpose: Resolves replay-capable media sources and ffmpeg worker arguments. // Scope: Converts live rover/room state into stable source descriptors and stream worker config. const path = require('path'); +const { randomUUID } = require('crypto'); const roverManager = require('../roverManager'); const { getRoomCameras } = require('../roomCameraService'); +const { getActiveRoomAudioStreams } = require('../roomAudioService'); const ptzCameraService = require('../ptzCameraService'); const { FFMPEG_BIN, SEGMENT_SECONDS, TARGET_FPS } = require('./constants'); @@ -51,12 +53,17 @@ function listDesiredSources() { // both come from the same live MediaMTX path, unlike rovers where audio is a // separate published stream. sources.push(...ptzCameraService.getReplayWorkerSources()); + for (const stream of getActiveRoomAudioStreams()) { + sources.push({ id: stream.id, sourceType: 'roomAudio', kind: 'audio', label: stream.name, + inputUrl: `rtsp://127.0.0.1:8554/room-audio/${encodeURIComponent(stream.id)}` }); + } return sources; } function buildWorkerArgs(activeSegmentRoot, source) { const dir = sourceDirForKey(activeSegmentRoot, sourceKey(source)); - const pattern = path.join(dir, 'seg-%06d.mp4'); + // Each capture process owns its filenames so restarts preserve buffered segments. + const pattern = path.join(dir, `seg-${randomUUID()}-%06d.mp4`); // MediaMTX and RTSP cameras use TCP so replay capture has one reliable // transport and never falls back to separate RTP/RTCP UDP listeners. const inputTransport = /^rtsps?:\/\//i.test(source.inputUrl) diff --git a/server/src/services/replayEngineV2/workerManager.js b/server/src/services/replayEngineV2/workerManager.js index a3298fea..a04e3e77 100644 --- a/server/src/services/replayEngineV2/workerManager.js +++ b/server/src/services/replayEngineV2/workerManager.js @@ -12,7 +12,7 @@ async function ensureDir(dir) { await fsp.mkdir(dir, { recursive: true }); } -function createWorkerManager({ getActiveSegmentRoot }) { +function createWorkerManager({ getActiveSegmentRoot, onWorkerClosed }) { function startWorker(source) { const key = sourceKey(source); if (workers.has(key) || pendingWorkerStarts.has(key)) return; @@ -21,26 +21,36 @@ function createWorkerManager({ getActiveSegmentRoot }) { const dir = sourceDirForKey(getActiveSegmentRoot(), key); ensureDir(dir) .then(() => { - if (workers.has(key)) { + // Directory creation can finish after a microphone has disconnected. + if (workers.has(key) || !listDesiredSources().some((entry) => sourceKey(entry) === key)) { pendingWorkerStarts.delete(key); return; } const proc = spawn(FFMPEG_BIN, args, { stdio: ['ignore', 'ignore', 'pipe'] }); - workers.set(key, { key, source, proc }); + const worker = { key, source, proc, stopping: false }; + workers.set(key, worker); pendingWorkerStarts.delete(key); 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); + proc.on('error', (err) => logger.warn('worker process failed', { key, error: err.message })); + worker.finished = new Promise((resolve) => { + proc.once('close', async (code, signal) => { + clearTimeout(worker.killTimer); + // FFmpeg writes the final MP4 trailer before close. Index it before + // releasing this worker so a disconnect cannot drop its last audio. + try { await onWorkerClosed(worker); } + catch (err) { logger.warn('final worker indexing failed', { key, error: err.message }); } + if (workers.get(key) === worker) workers.delete(key); + resolve(); + logger.warn('worker exited', { key, code, signal }); + if (!worker.stopping) setTimeout(() => { + const desired = listDesiredSources().find((entry) => sourceKey(entry) === key); + if (desired && !workers.has(key)) startWorker(desired); + }, 1500); + }); }); }) .catch((err) => { @@ -52,8 +62,14 @@ function createWorkerManager({ getActiveSegmentRoot }) { function stopWorker(key) { const worker = workers.get(key); if (!worker) return; - try { worker.proc.kill('SIGTERM'); } catch {} - workers.delete(key); + if (!worker.stopping) { + worker.stopping = true; + worker.proc.kill('SIGTERM'); + // Bound shutdown if a capture is blocked waiting on its network input. + worker.killTimer = setTimeout(() => worker.proc.kill('SIGKILL'), 3000); + worker.killTimer.unref(); + } + return worker.finished; } async function syncWorkers() { @@ -63,9 +79,11 @@ function createWorkerManager({ getActiveSegmentRoot }) { const key = sourceKey(source); if (!workers.has(key)) startWorker(source); } + const stops = []; for (const key of Array.from(workers.keys())) { - if (!desiredKeys.has(key)) stopWorker(key); + if (!desiredKeys.has(key)) stops.push(stopWorker(key)); } + await Promise.all(stops); } return { startWorker, stopWorker, syncWorkers }; diff --git a/server/src/services/roomAudioService/configuration.js b/server/src/services/roomAudioService/configuration.js new file mode 100644 index 00000000..f822b392 --- /dev/null +++ b/server/src/services/roomAudioService/configuration.js @@ -0,0 +1,14 @@ +const { strictObject, string, boolean } = require('../../configuration/schemaHelpers'); +module.exports = { + key: 'roomAudio', feature: true, + defaultValue: { enabled: false, sources: [] }, + schema: strictObject({ + enabled: boolean({ description: 'Enable room audio and local browser microphone publishing.' }), + sources: { type: 'array', items: strictObject({ + id: string({ minLength: 1, maxLength: 80, pattern: '^[a-zA-Z0-9_-]+$' }), + name: string({ minLength: 1, maxLength: 120 }), + type: { type: 'string', enum: ['device', 'stream'] }, + input: string({ minLength: 1, maxLength: 2048, description: 'ALSA device (for example default or plughw:CARD=USB,DEV=0), or a network stream URL.' }), + }, { required: ['id', 'name', 'type', 'input'] }) }, + }, { title: 'Room audio', required: ['enabled', 'sources'] }), +}; diff --git a/server/src/services/roomAudioService/index.js b/server/src/services/roomAudioService/index.js new file mode 100644 index 00000000..680f580f --- /dev/null +++ b/server/src/services/roomAudioService/index.js @@ -0,0 +1,171 @@ +const { spawn } = require('child_process'); +const EventEmitter = require('events'); +const io = require('../../globals/io'); +const logger = require('../../globals/logger').child('roomAudio'); +const { loadConfig, registerConfigurationHandler } = require('../../configuration'); +const { getSocketIp, isLocalNetwork } = require('../../helpers/ipResolver'); +const { getMode, MODES } = require('../modeManager'); +const { getRole, isAdmin, isLockdownAdmin } = require('../roleService'); +const { canBecomeSpectator } = require('../authService'); +const { isDeterred } = require('../verificationService'); +const videoSessions = require('../videoSessions'); +const roomAudioEvents = new EventEmitter(); +const sources = new Map(); +const mediaSessions = new Map(); +let enabled = false; +let active = []; +let polling = false; +let generation = 0; + +function canListen(socket) { + if (!enabled || !socket) return false; + const mode = getMode(); + if (mode === MODES.LOCKDOWN) return isLockdownAdmin(socket); + if (isAdmin(socket)) return true; + if (getRole(socket) === 'spectator') return canBecomeSpectator(socket); + // Room audio follows general driving eligibility, without tying listening + // to rover assignment, private-rover access, or the current driving turn. + return getRole(socket) === 'user' && mode !== MODES.ADMIN && !isDeterred(socket); +} +function canPublish(socket, id) { + return enabled && isLocalNetwork(getSocketIp(socket)) && sources.get(id)?.socketId === socket.id; +} +function getRoomAudioStreams(socket) { return canListen(socket) ? active : []; } +function getActiveRoomAudioStreams() { return active; } +function trackMediaSession(id, token) { if (id) mediaSessions.set(id, token); } +function hasStream(id) { return active.some((source) => source.id === id); } +function stopSource(id) { + const source = sources.get(id); + if (!source) return; + sources.delete(id); + clearTimeout(source.timer); + source.child?.kill('SIGTERM'); + videoSessions.revokeWhere((info) => ['roomAudio', 'roomMic'].includes(info.sourceType) && info.sourceId === id); + active = active.filter((entry) => entry.id !== id); + roomAudioEvents.emit('update'); +} +function startWorker(source) { + if (sources.get(source.id) !== source) return; + const args = ['-nostdin', '-hide_banner', '-loglevel', 'warning']; + if (source.type === 'device') args.push('-f', 'alsa'); + else if (/^rtsps?:/i.test(source.input)) args.push('-rtsp_transport', 'tcp', '-timeout', '10000000'); + else args.push('-rw_timeout', '10000000'); + args.push('-i', source.input, '-map', '0:a:0', '-vn', '-c:a', 'libopus', '-ar', '48000', '-ac', '1', '-b:a', '64k', '-f', 'rtsp', '-rtsp_transport', 'tcp', `rtsp://127.0.0.1:8554/room-audio/${source.id}`); + const child = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] }); + source.child = child; + child.stderr.on('data', (data) => logger.warn('Room audio %s: %s', source.name, String(data).trim())); + child.on('error', (error) => logger.warn('Room audio %s: %s', source.name, error.message)); + child.on('close', () => { + if (sources.get(source.id) !== source) return; + source.child = null; + source.timer = setTimeout(() => startWorker(source), 2000); + }); +} +function applyConfig(config = {}) { + const ids = (config.sources || []).map((source) => source.id); + if (new Set(ids).size !== ids.length) throw new Error('Room audio source IDs must be unique'); + generation += 1; + enabled = Boolean(config.enabled); + const configured = new Map((config.sources || []).map((entry) => { + const source = { ...entry, id: `static-${entry.id}` }; + return [source.id, source]; + })); + for (const [id, source] of sources) { + const next = configured.get(id); + // Static-source edits should not interrupt browser microphones or restart + // unchanged captures. Disabling the feature still stops every source. + if (!enabled || (!source.socketId && (!next || next.type !== source.type || next.input !== source.input))) stopSource(id); + } + if (enabled) for (const [id, source] of configured) { + const existing = sources.get(id); + if (existing) existing.name = source.name; + else { + sources.set(id, source); + startWorker(source); + } + } + roomAudioEvents.emit('update'); +} +async function listMediaMtxItems(collection) { + const items = []; + let page = 0; + let pageCount = 1; + while (page < pageCount) { + const response = await fetch(`http://127.0.0.1:9997/v3/${collection}/list?page=${page}`, { signal: AbortSignal.timeout(1500) }); + if (!response.ok) throw new Error('MediaMTX unavailable'); + const data = await response.json(); + items.push(...(data.items || [])); + pageCount = data.pageCount || 1; + page += 1; + } + return items; +} + +// Advertise actual ready MediaMTX paths, rather than a browser reservation or +// an ffmpeg process that may still be trying to connect to its input. +async function refreshStreams() { + if ((!enabled && !mediaSessions.size) || polling) return; + polling = true; + const currentGeneration = generation; + let ready = new Set(); + try { + // HTTP authorization is checked at connection time. Recheck established + // browser sessions too so lockdown, disable, and disconnect stop live audio. + const previousSessionIds = [...mediaSessions.keys()]; + const sessions = await listMediaMtxItems('webrtcsessions'); + const liveSessionIds = new Set(sessions.map((session) => session.id)); + for (const session of sessions) { + if (!session.path?.startsWith('room-audio/')) continue; + const info = videoSessions.getSession(mediaSessions.get(session.id)); + const socket = info && io.sockets.sockets.get(info.socketId); + const allowed = socket && (session.state === 'publish' + ? info.sourceType === 'roomMic' && canPublish(socket, info.sourceId) + : info.sourceType === 'roomAudio' && canListen(socket)); + if (!allowed) await fetch(`http://127.0.0.1:9997/v3/webrtcsessions/kick/${session.id}`, { method: 'POST', signal: AbortSignal.timeout(1500) }); + } + for (const id of previousSessionIds) if (!liveSessionIds.has(id)) mediaSessions.delete(id); + } catch (error) { + logger.warn('Room audio permission refresh failed: %s', error.message); + } + // Permission API failures must not make ready streams disappear. Only path + // discovery determines the catalog used by playback and replay capture. + try { + const paths = await listMediaMtxItems('paths'); + for (const path of paths) if (path.ready) ready.add(path.name); + } catch { ready = new Set(); } + finally { polling = false; } + if (generation !== currentGeneration) return; + const next = [...sources.values()].filter((source) => ready.has(`room-audio/${source.id}`)).map(({ id, name }) => ({ id, name })); + if (JSON.stringify(next) !== JSON.stringify(active)) { + active = next; + roomAudioEvents.emit('update'); + } +} +io.on('connection', (socket) => { + socket.on('roomAudio:start', (_payload, cb = () => {}) => { + if (!enabled || !isLocalNetwork(getSocketIp(socket))) return cb({ error: 'Room microphone publishing requires a local visitor and enabled room audio.' }); + // One owned path per socket; retries replace credentials without accumulating streams. + const id = `mic-${socket.id}`; + videoSessions.revokeWhere((info) => info.sourceType === 'roomMic' && info.sourceId === id); + sources.set(id, { id, name: 'Room microphone', socketId: socket.id }); + const token = videoSessions.createSession(socket, { type: 'roomMic', id }); + cb({ id, token, url: `/video/room-audio/${id}/whip` }); + }); + socket.on('roomAudio:release', ({ token } = {}) => { + const info = videoSessions.getSession(token); + if (info?.socketId === socket.id && info.sourceType === 'roomAudio') videoSessions.revokeSession(token); + }); + socket.on('roomAudio:stop', (_payload, cb = () => {}) => { + stopSource(`mic-${socket.id}`); + cb({ success: true }); + }); + socket.on('disconnect', () => { + stopSource(`mic-${socket.id}`); + }); +}); +applyConfig(loadConfig().roomAudio); +registerConfigurationHandler('roomAudio', applyConfig); +const pollTimer = setInterval(refreshStreams, 2000); +pollTimer.unref(); +process.once('exit', () => { for (const source of sources.values()) source.child?.kill('SIGTERM'); }); +module.exports = { getActiveRoomAudioStreams, trackMediaSession, roomAudioEvents, getRoomAudioStreams, canListen, canPublish, hasStream }; diff --git a/server/src/services/sessionService/index.js b/server/src/services/sessionService/index.js index 2e5d2b3b..3bc4fcbe 100644 --- a/server/src/services/sessionService/index.js +++ b/server/src/services/sessionService/index.js @@ -12,6 +12,7 @@ const { managerEvents } = roverManager; const assignmentService = require('../assignmentService'); const { getActiveDrivers, getTurnQueues, canDrive, turnEvents } = require('../turnService'); const { buildRoverTurn } = require('../turnService/display'); +const { getRoomAudioStreams, roomAudioEvents } = require('../roomAudioService'); const { getRoomCameras, roomCameraEvents } = require('../roomCameraService'); const { getPublicState: getPtzCameraState, @@ -214,6 +215,7 @@ function buildSession(socket) { status: assignmentRoverId ? assignment.status : assignment.status === 'waiting' ? 'waiting' : null, }, roomCameras: getRoomCameras(), + roomAudioStreams: getRoomAudioStreams(socket), ptzCamera: getPtzCameraState(socket), homeAssistant: getHomeAssistantState(), homeAssistantActivities: getActivityState(), @@ -407,6 +409,8 @@ turnEvents.on('queue', (event = {}) => { syncAll(); }); +roomAudioEvents.on('update', () => syncAll()); + roomCameraEvents.on('update', () => { logger.info('Room camera change detected; syncing all clients'); syncAll(); diff --git a/server/src/services/videoAuthService/httpRoute.js b/server/src/services/videoAuthService/httpRoute.js index a2e2bd93..53780b50 100644 --- a/server/src/services/videoAuthService/httpRoute.js +++ b/server/src/services/videoAuthService/httpRoute.js @@ -11,6 +11,7 @@ function registerVideoAuthRoute(deps) { logAdminEvent, extractStreamInfoFromBody, canAccessStream, + trackRoomAudioSession, } = deps; app.post('/mediamtx/auth', (req, res) => { @@ -36,7 +37,8 @@ function registerVideoAuthRoute(deps) { const isLoopback = ip === '127.0.0.1' || ip === '::1'; const isRoverForwardAudioRead = action === 'read' && isRtspProtocol - && streamInfo?.id?.endsWith('-fwd'); + && streamInfo?.type === 'rover' + && streamInfo.id.endsWith('-fwd'); // Replay and snapshot workers read MediaMTX through loopback RTSP. They do // not represent a browser session. Rovers likewise read their dedicated // -fwd speaker feed without browser credentials; every other non-loopback @@ -50,7 +52,7 @@ function registerVideoAuthRoute(deps) { callback for RTSP, so explicitly admit that publish protocol while leaving WHEP reads under the existing session and role checks below. */ - if (action === 'publish' && isRtspProtocol) { + if (action === 'publish' && isRtspProtocol && (streamInfo?.type !== 'roomAudio' || isLoopback)) { return res.status(200).end(); } @@ -62,7 +64,9 @@ function registerVideoAuthRoute(deps) { const info = videoSessions.getSession(sessionId); const streamTypeMatches = info && - (info.sourceType === streamInfo.type || (info.sourceType === 'roverMic' && streamInfo.type === 'rover')); + (info.sourceType === streamInfo.type + || (info.sourceType === 'roomMic' && streamInfo.type === 'roomAudio') + || (info.sourceType === 'roverMic' && streamInfo.type === 'rover')); if (!info || !streamTypeMatches || info.sourceId !== streamInfo.id) { logger.warn('invalid session %s for stream %s:%s', sessionId, streamInfo.type, streamInfo.id); return res.status(401).end(); @@ -78,6 +82,7 @@ function registerVideoAuthRoute(deps) { return res.status(401).end(); } + if (streamInfo.type === 'roomAudio' && protocol === 'webrtc') trackRoomAudioSession(body.id, sessionId); return res.status(200).end(); }); } diff --git a/server/src/services/videoAuthService/index.js b/server/src/services/videoAuthService/index.js index 507ac433..4c6055bf 100644 --- a/server/src/services/videoAuthService/index.js +++ b/server/src/services/videoAuthService/index.js @@ -4,6 +4,7 @@ const { app } = require('../../globals/http'); const io = require('../../globals/io'); const logger = require('../../globals/logger').child('videoAuth'); +const roomAudioService = require('../roomAudioService'); const videoSessions = require('../videoSessions'); const { getMode, MODES } = require('../modeManager'); const { isAdmin, isLockdownAdmin, getRole } = require('../roleService'); @@ -28,6 +29,7 @@ const { canAccessStream } = createVideoAuthPolicy({ turnService, roverManager, ptzCameraService, + roomAudioService, getSocketIp, isLocalNetwork, io, @@ -42,6 +44,7 @@ registerVideoAuthRoute({ logAdminEvent, extractStreamInfoFromBody, canAccessStream, + trackRoomAudioSession: roomAudioService.trackMediaSession, }); module.exports = {}; diff --git a/server/src/services/videoAuthService/policy.js b/server/src/services/videoAuthService/policy.js index ec92f5a0..9fa6d943 100644 --- a/server/src/services/videoAuthService/policy.js +++ b/server/src/services/videoAuthService/policy.js @@ -17,6 +17,7 @@ function createVideoAuthPolicy(deps) { turnService, roverManager, ptzCameraService, + roomAudioService, getSocketIp, isLocalNetwork, io, @@ -58,6 +59,11 @@ function createVideoAuthPolicy(deps) { } function canAccessStream({ socket, streamInfo, action, sourceType }) { + if (streamInfo.type === 'roomAudio') { + return action === 'publish' + ? sourceType === 'roomMic' && roomAudioService.canPublish(socket, streamInfo.id) + : action === 'read' && sourceType === 'roomAudio' && roomAudioService.canListen(socket) && roomAudioService.hasStream(streamInfo.id); + } if (!canView(socket)) { return false; } diff --git a/server/src/services/videoAuthService/streamParsing.js b/server/src/services/videoAuthService/streamParsing.js index 10bbbffe..0cf5bbe6 100644 --- a/server/src/services/videoAuthService/streamParsing.js +++ b/server/src/services/videoAuthService/streamParsing.js @@ -33,6 +33,8 @@ function extractStreamInfo(path) { return { type: 'rover', id: rawId, baseId }; } + if (remaining.length === 2 && remaining[0] === 'room-audio') return { type: 'roomAudio', id: remaining[1] }; + if (remaining.length === 2 && remaining[0] === 'room') { return { type: 'room', id: remaining[1] || '' }; } diff --git a/server/src/services/videoSocketService/index.js b/server/src/services/videoSocketService/index.js index d5a9a210..cb25cbb7 100644 --- a/server/src/services/videoSocketService/index.js +++ b/server/src/services/videoSocketService/index.js @@ -18,7 +18,9 @@ const { function buildWhepUrlForSource(source) { const segments = []; - if (source.type === 'room') { + if (source.type === 'roomAudio') { + segments.push('room-audio', encodeURIComponent(source.id)); + } else if (source.type === 'room') { segments.push('room', encodeURIComponent(source.id)); } else if (source.type === 'ptz') { /* @@ -145,6 +147,9 @@ io.on('connection', (socket) => { */ throw new Error('Live video is limited to this rover queue'); } + } else if (target.type === 'roomAudio') { + const roomAudio = require('../roomAudioService'); + if (!roomAudio.canListen(socket) || !roomAudio.hasStream(target.id)) throw new Error('Room audio unavailable'); } else if (target.type === 'room') { throw new Error('Room cameras now use the snapshot feed'); } else if (target.type === 'ptz') { diff --git a/webui/src/components/RoomAudioPlayer/index.jsx b/webui/src/components/RoomAudioPlayer/index.jsx new file mode 100644 index 00000000..c65a11e7 --- /dev/null +++ b/webui/src/components/RoomAudioPlayer/index.jsx @@ -0,0 +1,69 @@ +import { useCallback, useEffect, useRef } from 'react'; +import { useSessionSelector } from '../../context/SessionContext.jsx'; +import { useSocket } from '../../context/SocketContext.jsx'; +import { useSettingsNamespace } from '../../settings/index.js'; +import { AUDIO_SETTINGS_DEFAULTS } from '../../settings/namespaces.js'; +import { RESTART_DELAY_MS, startWhepPlayback } from '../../lib/whepPlayback.js'; +import useWhepRestart from '../../hooks/useWhepRestart.js'; + +function StreamPlayer({ id, volume }) { + const socket = useSocket(); + const audio = useRef(null); + const { restartToken, scheduleRestart } = useWhepRestart(); + useEffect(() => { + let active = true; + let stop; + let token; + socket.timeout(5000).emit('video:request', { type: 'roomAudio', id }, (error, response = {}) => { + if (!active) { + if (response.token && socket.connected) socket.emit('roomAudio:release', { token: response.token }); + return; + } + token = response.token; + if (error || response.error || !response.url) { scheduleRestart(); return; } + stop = startWhepPlayback({ + ...response, video: audio.current, audioOnly: true, + onStatus: () => {}, onError: () => {}, scheduleRestart, + }); + }); + return () => { + active = false; stop?.(); + if (token && socket.connected) socket.emit('roomAudio:release', { token }); + }; + }, [id, socket, restartToken, scheduleRestart]); + useEffect(() => { if (audio.current) audio.current.volume = Math.max(0, Math.min(1, volume)); }, [volume]); + return