gemma/arcana-fx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add continuous mixer stream and sound sets

Browse Source
This commit is contained in:
Hermes Agent
2026-05-21 00:16:13 +02:00
parent 232e6923e3
commit 23391a2d3e
8 changed files with 680 additions and 570 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
src/routes.js
View File

@@ -4,11 +4,11 @@ const express = require('express');
const { isSafeRel } = require('./effects');
/**
* Builds the API router. Each route is intentionally narrow: clients can
* list effects, play a known effect path, stop, or pull a known effect
* via the media proxy. No arbitrary HA service / Filestash path passthrough.
* Builds the API router. Routes stay narrow: clients can list indexed effects,
* inject known effects into the backend mixer, control the one allowed Home
* Assistant media-player stream, and fetch/proxy only known Filestash paths.
*/
function createApiRouter({ config, effects, filestash, ha }) {
function createApiRouter({ config, effects, filestash, ha, mixer }) {
const router = express.Router();
const allowedEntity = config.homeAssistant.allowedMediaPlayer;
const publicBaseUrl = config.env.publicBaseUrl;
@@ -21,6 +21,8 @@ function createApiRouter({ config, effects, filestash, ha }) {
cacheAgeSec: effects.cache
? Math.round((Date.now() - effects.cacheLoadedAt) / 1000)
: null,
streamClients: mixer ? mixer.stats().clients : 0,
activeEffects: mixer ? mixer.stats().active.length : 0,
});
});
@@ -29,11 +31,7 @@ function createApiRouter({ config, effects, filestash, ha }) {
const force = req.query.refresh === '1' || req.query.refresh === 'true';
const list = await effects.list({ force });
const categories = [...new Set(list.map((e) => e.category))].sort();
res.json({
count: list.length,
categories,
effects: list,
});
res.json({ count: list.length, categories, effects: list });
} catch (err) {
next(err);
}
@@ -43,78 +41,90 @@ function createApiRouter({ config, effects, filestash, ha }) {
res.json({
mediaPlayer: allowedEntity,
publicBaseUrlConfigured: Boolean(publicBaseUrl),
streamUrl: publicBaseUrl ? `${publicBaseUrl}/stream/${encodeURIComponent(allowedEntity)}` : null,
lightControl: config.lightControl,
});
});
router.post('/api/play', express.json({ limit: '4kb' }), async (req, res, next) => {
router.get('/api/stream/status', (_req, res) => {
res.json({ ok: true, stream: mixer.stats(), entity: allowedEntity });
});
router.post('/api/stream/start', async (_req, res, next) => {
try {
const { path: effectPath } = req.body || {};
if (!isSafeRel(effectPath)) {
return res.status(400).json({ error: 'invalid effect path' });
}
const effect = await effects.findByPath(effectPath);
if (!effect) {
return res.status(404).json({ error: 'effect not found' });
}
if (!publicBaseUrl) {
return res.status(503).json({
error: 'PUBLIC_BASE_URL is not configured; Home Assistant cannot fetch /media',
error: 'PUBLIC_BASE_URL is not configured; Home Assistant cannot fetch /stream',
});
}
const mediaUrl = `${publicBaseUrl}/media?path=${encodeURIComponent(effect.path)}`;
const mediaContentType = effects.contentTypeFor(effect.fileName);
const streamUrl = `${publicBaseUrl}/stream/${encodeURIComponent(allowedEntity)}`;
await ha.playMedia({
entityId: allowedEntity,
mediaContentId: mediaUrl,
mediaContentType,
});
res.json({
ok: true,
played: {
name: effect.name,
path: effect.path,
category: effect.category,
mediaContentType,
mediaUrl,
entity: allowedEntity,
},
mediaContentId: streamUrl,
mediaContentType: 'audio/wav',
});
res.json({ ok: true, entity: allowedEntity, streamUrl });
} catch (err) {
next(err);
}
});
router.post('/api/stop', async (_req, res, next) => {
router.post('/api/stream/stop', async (_req, res, next) => {
try {
const stoppedEffects = mixer.stopAll();
await ha.stop({ entityId: allowedEntity });
res.json({ ok: true, stopped: allowedEntity });
res.json({ ok: true, stopped: allowedEntity, stoppedEffects });
} catch (err) {
next(err);
}
});
router.post('/api/play', express.json({ limit: '8kb' }), async (req, res, next) => {
try {
const { path: effectPath, volume } = req.body || {};
if (!isSafeRel(effectPath)) return res.status(400).json({ error: 'invalid effect path' });
const effect = await effects.findByPath(effectPath);
if (!effect) return res.status(404).json({ error: 'effect not found' });
const injected = await mixer.play(effect, { volume });
res.json({ ok: true, injected, stream: mixer.stats(), entity: allowedEntity });
} catch (err) {
next(err);
}
});
router.post('/api/play/stop', express.json({ limit: '4kb' }), async (req, res) => {
const id = req.body && req.body.id;
if (!id || typeof id !== 'string') return res.status(400).json({ error: 'id is required' });
const stopped = mixer.stopOne(id);
res.status(stopped ? 200 : 404).json({ ok: stopped, id });
});
router.post('/api/stop', async (_req, res, next) => {
try {
const stoppedEffects = mixer.stopAll();
res.json({ ok: true, stoppedEffects, stream: mixer.stats() });
} catch (err) {
next(err);
}
});
router.get('/stream/:entity', (req, res) => {
const entity = decodeURIComponent(req.params.entity || '');
if (entity !== allowedEntity) return res.status(403).json({ error: 'entity not permitted' });
mixer.stream(res);
});
router.get('/media', async (req, res, next) => {
try {
const effectPath = typeof req.query.path === 'string' ? req.query.path : '';
if (!isSafeRel(effectPath)) {
return res.status(400).json({ error: 'invalid path' });
}
if (!isSafeRel(effectPath)) return res.status(400).json({ error: 'invalid path' });
const effect = await effects.findByPath(effectPath);
if (!effect) {
return res.status(404).json({ error: 'effect not found' });
}
if (!effect) return res.status(404).json({ error: 'effect not found' });
const upstream = await filestash.streamFile(effect.path, {
range: req.headers.range,
});
const upstream = await filestash.streamFile(effect.path, { range: req.headers.range });
if (!upstream.ok && upstream.status !== 206) {
return res
.status(upstream.status || 502)
.json({ error: 'upstream fetch failed', status: upstream.status });
return res.status(upstream.status || 502).json({ error: 'upstream fetch failed', status: upstream.status });
}
const passthrough = ['content-length', 'content-range', 'accept-ranges', 'last-modified', 'etag'];
@@ -125,10 +135,7 @@ function createApiRouter({ config, effects, filestash, ha }) {
res.setHeader('Content-Type', effects.contentTypeFor(effect.fileName));
res.status(upstream.status === 206 ? 206 : 200);
if (!upstream.body) {
return res.end();
}
// upstream.body is a Web ReadableStream; pipe it to the Node response.
if (!upstream.body) return res.end();
const reader = upstream.body.getReader();
res.on('close', () => {
try { reader.cancel(); } catch { /* noop */ }
@@ -136,9 +143,7 @@ function createApiRouter({ config, effects, filestash, ha }) {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!res.write(Buffer.from(value))) {
await new Promise((resolve) => res.once('drain', resolve));
}
if (!res.write(Buffer.from(value))) await new Promise((resolve) => res.once('drain', resolve));
}
res.end();
} catch (err) {

10
src/server.js
View File

@@ -7,6 +7,7 @@ const { loadConfig } = require('./config');
const { FilestashClient } = require('./filestash');
const { EffectsService } = require('./effects');
const { HomeAssistantClient } = require('./homeassistant');
const { StreamMixer } = require('./streamMixer');
const { createApiRouter } = require('./routes');
function buildApp(config) {
@@ -48,6 +49,11 @@ function buildApp(config) {
};
}
const mixer = new StreamMixer({
effects,
localMediaBaseUrl: `http://127.0.0.1:${config.env.port}`,
});
const app = express();
app.disable('x-powered-by');
@@ -66,7 +72,7 @@ function buildApp(config) {
});
}
app.use(createApiRouter({ config, effects, filestash, ha }));
app.use(createApiRouter({ config, effects, filestash, ha, mixer }));
const publicDir = path.join(__dirname, '..', 'public');
app.use(express.static(publicDir, { fallthrough: true, index: 'index.html' }));
@@ -82,7 +88,7 @@ function buildApp(config) {
res.status(status).json({ error: err.message || 'internal error' });
});
return { app, effects, filestash };
return { app, effects, filestash, mixer };
}
function main() {

252
src/streamMixer.js Normal file
View File

@@ -0,0 +1,252 @@
'use strict';
const { spawn } = require('node:child_process');
const crypto = require('node:crypto');
const SAMPLE_RATE = 44100;
const CHANNELS = 2;
const BYTES_PER_SAMPLE = 2;
const FRAME_MS = 20;
const FRAME_BYTES = Math.floor(SAMPLE_RATE * CHANNELS * BYTES_PER_SAMPLE * FRAME_MS / 1000);
const MAX_ACTIVE = 24;
const MAX_QUEUE_BYTES = FRAME_BYTES * 250; // 5 seconds of decoded audio buffering per effect.
class StreamMixer {
constructor({ effects, localMediaBaseUrl }) {
if (!effects) throw new Error('StreamMixer requires effects');
if (!localMediaBaseUrl) throw new Error('StreamMixer requires localMediaBaseUrl');
this.effects = effects;
this.localMediaBaseUrl = localMediaBaseUrl.replace(/\/+$/, '');
this.clients = new Set();
this.active = new Map();
this.timer = null;
this.sequence = 0;
}
stream(res) {
res.status(200);
res.setHeader('Content-Type', 'audio/wav');
res.setHeader('Cache-Control', 'no-store, no-cache, must-revalidate, proxy-revalidate');
res.setHeader('Pragma', 'no-cache');
res.setHeader('Expires', '0');
res.setHeader('Connection', 'keep-alive');
res.write(wavHeader());
const client = { res, connectedAt: Date.now(), mutedUntil: Date.now() + 80 };
this.clients.add(client);
this._ensureTimer();
res.on('close', () => {
this.clients.delete(client);
this._maybeStopTimer();
});
}
async play(effect, { volume = 1 } = {}) {
if (!effect || !effect.path) {
const err = new Error('effect is required');
err.status = 400;
throw err;
}
if (this.active.size >= MAX_ACTIVE) {
const oldest = [...this.active.values()].sort((a, b) => a.startedAt - b.startedAt)[0];
if (oldest) this.stopOne(oldest.id);
}
const id = `${Date.now().toString(36)}-${(++this.sequence).toString(36)}-${crypto.randomBytes(3).toString('hex')}`;
const mediaUrl = `${this.localMediaBaseUrl}/media?path=${encodeURIComponent(effect.path)}`;
const item = {
id,
path: effect.path,
name: effect.name,
category: effect.category,
type: effect.type,
volume: clamp(Number(volume) || 1, 0, 2),
startedAt: Date.now(),
queue: [],
queueBytes: 0,
ended: false,
error: null,
proc: null,
};
const args = [
'-hide_banner', '-loglevel', 'error',
'-i', mediaUrl,
'-f', 's16le',
'-acodec', 'pcm_s16le',
'-ac', String(CHANNELS),
'-ar', String(SAMPLE_RATE),
'pipe:1',
];
const proc = spawn('ffmpeg', args, { stdio: ['ignore', 'pipe', 'pipe'] });
item.proc = proc;
proc.stdout.on('data', (chunk) => {
if (!this.active.has(id)) return;
if (item.queueBytes > MAX_QUEUE_BYTES) {
proc.stdout.pause();
setTimeout(() => proc.stdout.resume(), 25);
return;
}
item.queue.push(Buffer.from(chunk));
item.queueBytes += chunk.length;
});
proc.stderr.on('data', (chunk) => {
item.error = String(chunk).slice(0, 500);
});
proc.on('close', (code) => {
item.ended = true;
if (code && code !== 0 && !item.error) item.error = `ffmpeg exited ${code}`;
});
proc.on('error', (err) => {
item.error = err.message;
item.ended = true;
});
this.active.set(id, item);
this._ensureTimer();
return publicItem(item);
}
listActive() {
return [...this.active.values()].map(publicItem);
}
stopOne(id) {
const item = this.active.get(id);
if (!item) return false;
this.active.delete(id);
if (item.proc && !item.proc.killed) item.proc.kill('SIGTERM');
this._maybeStopTimer();
return true;
}
stopAll() {
const count = this.active.size;
for (const item of this.active.values()) {
if (item.proc && !item.proc.killed) item.proc.kill('SIGTERM');
}
this.active.clear();
this._maybeStopTimer();
return count;
}
stats() {
return {
sampleRate: SAMPLE_RATE,
channels: CHANNELS,
clients: this.clients.size,
active: this.listActive(),
};
}
_ensureTimer() {
if (this.timer) return;
this.timer = setInterval(() => this._tick(), FRAME_MS);
this.timer.unref?.();
}
_maybeStopTimer() {
if (this.clients.size || this.active.size) return;
if (this.timer) clearInterval(this.timer);
this.timer = null;
}
_tick() {
const output = Buffer.alloc(FRAME_BYTES);
const mix = new Int32Array(FRAME_BYTES / BYTES_PER_SAMPLE);
const now = Date.now();
for (const item of [...this.active.values()]) {
const frame = consumeFrame(item, FRAME_BYTES);
if (frame) {
for (let i = 0; i < mix.length; i += 1) {
mix[i] += Math.round(frame.readInt16LE(i * 2) * item.volume);
}
}
if (item.ended && item.queueBytes === 0) {
this.active.delete(item.id);
}
}
for (let i = 0; i < mix.length; i += 1) {
output.writeInt16LE(clamp(mix[i], -32768, 32767), i * 2);
}
for (const client of [...this.clients]) {
if (now < client.mutedUntil) continue;
if (client.res.destroyed || client.res.writableEnded) {
this.clients.delete(client);
continue;
}
try {
client.res.write(output);
} catch {
this.clients.delete(client);
}
}
this._maybeStopTimer();
}
}
function consumeFrame(item, size) {
if (!item.queueBytes) return null;
const out = Buffer.alloc(size);
let offset = 0;
while (offset < size && item.queue.length) {
const chunk = item.queue[0];
const take = Math.min(size - offset, chunk.length);
chunk.copy(out, offset, 0, take);
offset += take;
item.queueBytes -= take;
if (take === chunk.length) {
item.queue.shift();
} else {
item.queue[0] = chunk.subarray(take);
}
}
return out;
}
function wavHeader() {
const h = Buffer.alloc(44);
h.write('RIFF', 0);
h.writeUInt32LE(0xffffffff, 4);
h.write('WAVE', 8);
h.write('fmt ', 12);
h.writeUInt32LE(16, 16);
h.writeUInt16LE(1, 20);
h.writeUInt16LE(CHANNELS, 22);
h.writeUInt32LE(SAMPLE_RATE, 24);
h.writeUInt32LE(SAMPLE_RATE * CHANNELS * BYTES_PER_SAMPLE, 28);
h.writeUInt16LE(CHANNELS * BYTES_PER_SAMPLE, 32);
h.writeUInt16LE(16, 34);
h.write('data', 36);
h.writeUInt32LE(0xffffffff, 40);
return h;
}
function publicItem(item) {
return {
id: item.id,
path: item.path,
name: item.name,
category: item.category,
type: item.type,
volume: item.volume,
startedAt: item.startedAt,
ageMs: Date.now() - item.startedAt,
error: item.error,
};
}
function clamp(n, min, max) {
return Math.max(min, Math.min(max, n));
}
module.exports = { StreamMixer };