feat(music): stream live wav chunks to HQ speakers

Slice realtime WAV audio into small self-contained chunks on the server and feed them into the HQ speaker WAV queue on stations so music can start playing before the full file finishes transferring.

CloudServer/server.js

@@ -286,27 +286,55 @@ async function sendAudioBufferToStation(stationName, fallbackWs, task) {
     return { audioId, cancelled: false, durationSec };
 }
 
-async function sendLiveDfpwmToStation(stationName, fallbackWs, task) {
+function buildWavChunkFromPcm(buffer, sampleRate = 48000, channels = 1, bitsPerSample = 16) {
+    const dataSize = buffer.length;
+    const header = Buffer.alloc(44);
+    const byteRate = sampleRate * channels * (bitsPerSample / 8);
+    const blockAlign = channels * (bitsPerSample / 8);
+
+    header.write('RIFF', 0, 4, 'ascii');
+    header.writeUInt32LE(36 + dataSize, 4);
+    header.write('WAVE', 8, 4, 'ascii');
+    header.write('fmt ', 12, 4, 'ascii');
+    header.writeUInt32LE(16, 16);
+    header.writeUInt16LE(1, 20);
+    header.writeUInt16LE(channels, 22);
+    header.writeUInt32LE(sampleRate, 24);
+    header.writeUInt32LE(byteRate, 28);
+    header.writeUInt16LE(blockAlign, 32);
+    header.writeUInt16LE(bitsPerSample, 34);
+    header.write('data', 36, 4, 'ascii');
+    header.writeUInt32LE(dataSize, 40);
+
+    return Buffer.concat([header, buffer]);
+}
+
+async function sendLiveWavChunksToStation(stationName, fallbackWs, task) {
     const targetWs = getStationWs(stationName, fallbackWs);
     if (!targetWs) throw new Error(`Station "${stationName}" not connected`);
 
     const sourcePath = String(task.sourcePath || '').trim();
-    if (!sourcePath) throw new Error('Live DFPWM sourcePath required');
+    if (!sourcePath) throw new Error('Live WAV sourcePath required');
-    if (!fs.existsSync(sourcePath)) throw new Error(`Live DFPWM source missing: ${sourcePath}`);
+    if (!fs.existsSync(sourcePath)) throw new Error(`Live WAV source missing: ${sourcePath}`);
 
     const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`);
     const durationSec = Number(task.durationSec || 0);
     const loop = !!task.loop;
     const token = task.cancelToken;
+    const sampleRate = 48000;
+    const channels = 1;
+    const bitsPerSample = 16;
+    const chunkDurationSec = 0.25;
+    const chunkBytes = Math.max(4096, Math.floor(sampleRate * channels * (bitsPerSample / 8) * chunkDurationSec));
 
     targetWs.send(JSON.stringify({
         type: 'audio_start',
         stationName,
         audioId,
         byteLength: 0,
-        sampleRate: 48000,
+        sampleRate,
-        channels: 1,
+        channels,
-        audioFormat: 'dfpwm',
+        audioFormat: 'wav-stream',
         loop,
         priority: Number(task.priority || 0),
         interruptMode: task.interruptMode || getInterruptModeForPriority(Number(task.priority || 0)),
@@ -315,14 +343,14 @@ async function sendLiveDfpwmToStation(stationName, fallbackWs, task) {
 
     const ffmpegArgs = ['-hide_banner', '-loglevel', 'error'];
     if (loop) ffmpegArgs.push('-stream_loop', '-1');
-    ffmpegArgs.push('-re', '-i', sourcePath, '-f', 's8', '-acodec', 'pcm_s8', '-ac', '1', '-ar', '48000', 'pipe:1');
+    ffmpegArgs.push('-re', '-i', sourcePath, '-f', 's16le', '-acodec', 'pcm_s16le', '-ac', String(channels), '-ar', String(sampleRate), 'pipe:1');
-    log('INFO', `Live DFPWM music stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''}`);
+    log('INFO', `Live WAV chunk stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''}`);
 
     return await new Promise((resolve, reject) => {
         let stderr = '';
         let settled = false;
         let proc = null;
-        const encoder = new dfpwm.Encoder();
+        let pcmPending = Buffer.alloc(0);
 
         const finish = (result, error) => {
             if (settled) return;
@@ -362,13 +390,8 @@ async function sendLiveDfpwmToStation(stationName, fallbackWs, task) {
             return;
         }
 
-        proc.stdout.on('data', (chunk) => {
+        const flushChunk = (pcmChunk) => {
-            if (!chunk || !chunk.length || settled) return;
+            if (!pcmChunk || !pcmChunk.length) return;
-            if (token && token.cancelled) {
-                cancelStream();
-                return;
-            }
-
             const ws = getStationWs(stationName, fallbackWs);
             if (!ws || ws.readyState !== WebSocket.OPEN) {
                 cancelStream();
@@ -376,13 +399,25 @@ async function sendLiveDfpwmToStation(stationName, fallbackWs, task) {
             }
 
             try {
-                const encoded = encoder.encode(Buffer.from(chunk));
+                ws.send(buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample));
-                if (encoded && encoded.length > 0) {
-                    ws.send(encoded);
-                }
             } catch (err) {
                 finish(null, err);
             }
+        };
+
+        proc.stdout.on('data', (chunk) => {
+            if (!chunk || !chunk.length || settled) return;
+            if (token && token.cancelled) {
+                cancelStream();
+                return;
+            }
+
+            pcmPending = pcmPending.length > 0 ? Buffer.concat([pcmPending, chunk]) : Buffer.from(chunk);
+            while (pcmPending.length >= chunkBytes && !settled) {
+                const part = pcmPending.subarray(0, chunkBytes);
+                pcmPending = pcmPending.subarray(chunkBytes);
+                flushChunk(part);
+            }
         });
 
         proc.stderr.on('data', (chunk) => {
@@ -395,7 +430,11 @@ async function sendLiveDfpwmToStation(stationName, fallbackWs, task) {
 
         proc.on('close', (code, signal) => {
             const cancelled = token && token.cancelled;
-            log(cancelled ? 'INFO' : 'SUCCESS', `Live DFPWM music stream end for ${stationName}: ${path.basename(sourcePath)}${cancelled ? ' (cancelled)' : ''}`);
+            if (!cancelled && pcmPending.length > 0) {
+                flushChunk(pcmPending);
+                pcmPending = Buffer.alloc(0);
+            }
+            log(cancelled ? 'INFO' : 'SUCCESS', `Live WAV chunk stream end for ${stationName}: ${path.basename(sourcePath)}${cancelled ? ' (cancelled)' : ''}`);
             if (!cancelled && code !== 0) {
                 finish(null, new Error((stderr || `ffmpeg exited with code ${code ?? 'unknown'} signal ${signal ?? 'none'}`).trim()));
                 return;
@@ -587,8 +626,8 @@ async function processStationAudioQueue(stationName, fallbackWs) {
 
             try {
                 if (task.consoleStreamStart) broadcastToConsole(task.consoleStreamStart);
-                const result = task.audioFormat === 'dfpwm-live'
+                const result = task.audioFormat === 'wav-stream-live'
-                    ? await sendLiveDfpwmToStation(cleanStation, fallbackWs, task)
+                    ? await sendLiveWavChunksToStation(cleanStation, fallbackWs, task)
                     : await sendAudioBufferToStation(cleanStation, fallbackWs, task);
                 if (task.consoleStreamEnd) {
                     if (result.cancelled) {
@@ -2386,7 +2425,7 @@ app.post('/api/music/start', express.json(), (req, res) => {
 
             if (playbackAudio.sourcePath) {
                 enqueueStationAudioTask(name, currentWs, {
-                    audioFormat: 'dfpwm-live',
+                    audioFormat: 'wav-stream-live',
                     sourcePath: playbackAudio.sourcePath,
                     durationSec: playbackAudio.durationSec,
                     priority: -1,

station_broadcast.lua

@@ -796,6 +796,49 @@ local function playHqAudioItem(item)
     end
 end
 
+local function pushWavChunkToSpeakers(payload, playSeq)
+    local controller = Peripherals.speakers and Peripherals.speakers[1] or nil
+    if controller and type(controller.speakWavAll) == "function" then
+        while true do
+            if shouldStop(playSeq) then return false end
+            local ok = false
+            pcall(function()
+                ok = controller.speakWavAll(payload, CONFIG.AUDIO_VOLUME)
+            end)
+            if ok then
+                return true
+            end
+            if not waitForInterruptibleDelay(playSeq, 0.05) then
+                return false
+            end
+        end
+    end
+
+    local pending = {}
+    for _, sp in ipairs(Peripherals.speakers or {}) do
+        local name = peripheral.getName(sp)
+        if name then pending[name] = sp end
+    end
+
+    while next(pending) do
+        if shouldStop(playSeq) then return false end
+        for name, sp in pairs(pending) do
+            local ok = false
+            pcall(function()
+                if type(sp.speakWav) == "function" then
+                    ok = sp.speakWav(payload, CONFIG.AUDIO_VOLUME)
+                end
+            end)
+            if ok then pending[name] = nil end
+        end
+        if next(pending) and not waitForInterruptibleDelay(playSeq, 0.05) then
+            return false
+        end
+    end
+
+    return true
+end
+
 function Audio.playDfpwmItem(item)
     if not Peripherals.speakers or #Peripherals.speakers == 0 then
         Utils.log("WARN", "No speakers found; cannot play audio")
@@ -868,12 +911,64 @@ function Audio.playPcmItem(item)
     end
 end
 
+function Audio.playWavStreamItem(item)
+    if not Peripherals.speakers or #Peripherals.speakers == 0 then
+        Utils.log("WARN", "No speakers found; cannot play audio")
+        return
+    end
+
+    local playSeq = State.interruptSeq
+    local source = item.audioData
+
+    if type(source) == "table" and source.read then
+        source = source.read
+    end
+
+    if type(source) ~= "function" then
+        local data = source
+        source = function()
+            local chunk = data
+            data = nil
+            return chunk
+        end
+    end
+
+    while true do
+        if shouldStop(playSeq) then
+            Audio.stop("current")
+            return
+        end
+
+        local chunk = source()
+        if chunk == nil then break end
+        if type(chunk) == "string" and #chunk > 0 then
+            if not pushWavChunkToSpeakers(chunk, playSeq) then
+                Audio.stop("current")
+                return
+            end
+        end
+    end
+
+    while areHqSpeakersBusy(Peripherals.speakers) do
+        if shouldStop(playSeq) then
+            Audio.stop("current")
+            return
+        end
+        if not waitForInterruptibleDelay(playSeq, 0.05) then
+            Audio.stop("current")
+            return
+        end
+    end
+end
+
 function Audio.playItem(item)
     local format = tostring(item.audioFormat or "dfpwm"):lower()
     if format == "wav" or format == "file" or format == "stream" or format == "hls" then
         return playHqAudioItem(item)
     elseif format == "pcm" then
         return Audio.playPcmItem(item)
+    elseif format == "wav-stream" then
+        return Audio.playWavStreamItem(item)
     end
     return Audio.playDfpwmItem(item)
 end
@@ -1084,11 +1179,13 @@ function Network.messageLoop()
                             audioId = data.audioId
                         })
                         State.audioRx = nil
-                    elseif audioFormat == "dfpwm" or audioFormat == "pcm" then
+                    elseif audioFormat == "dfpwm" or audioFormat == "pcm" or audioFormat == "wav-stream" then
                         local stream = Audio.createStream()
                         State.audioRx.stream = stream
                         if audioFormat == "pcm" then
                             Utils.log("INFO", "Receiving live PCM audio stream")
+                        elseif audioFormat == "wav-stream" then
+                            Utils.log("INFO", "Receiving live WAV chunk stream")
                         end
                         Audio.enqueueItem({
                             audioData = stream,