feat(music): switch station music to live PCM streaming

Move music playback off the HLS/download-first path and stream 48k mono PCM chunks over WebSocket so stations can start playing immediately. Add station-side PCM chunk playback support and stream lifecycle logging to make no-audio issues easier to trace.

CloudServer/server.js

@@ -286,6 +286,122 @@ async function sendAudioBufferToStation(stationName, fallbackWs, task) {
     return { audioId, cancelled: false, durationSec };
 }
 
+async function sendPcmStreamToStation(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('PCM stream sourcePath required');
+    if (!fs.existsSync(sourcePath)) throw new Error(`PCM 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;
+
+    targetWs.send(JSON.stringify({
+        type: 'audio_start',
+        stationName,
+        audioId,
+        byteLength: 0,
+        sampleRate: 48000,
+        channels: 1,
+        audioFormat: 'pcm',
+        pcmEncoding: 's8',
+        loop,
+        priority: Number(task.priority || 0),
+        interruptMode: task.interruptMode || getInterruptModeForPriority(Number(task.priority || 0)),
+        source: task.source || 'unknown'
+    }));
+
+    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');
+    log('INFO', `PCM music stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''}`);
+
+    return await new Promise((resolve, reject) => {
+        let stderr = '';
+        let settled = false;
+        let proc = null;
+
+        const finish = (result, error) => {
+            if (settled) return;
+            settled = true;
+            if (task.onCancel === cancelStream) task.onCancel = null;
+            if (error) {
+                reject(error);
+                return;
+            }
+            try {
+                const ws = getStationWs(stationName, fallbackWs);
+                if (ws && ws.readyState === WebSocket.OPEN) {
+                    ws.send(JSON.stringify({
+                        type: 'audio_end',
+                        stationName,
+                        audioId,
+                        durationSec: result.durationSec
+                    }));
+                }
+            } catch {}
+            resolve(result);
+        };
+
+        const cancelStream = () => {
+            if (token) token.cancelled = true;
+            if (proc && !proc.killed) {
+                try { proc.kill('SIGKILL'); } catch {}
+            }
+        };
+
+        task.onCancel = cancelStream;
+
+        try {
+            proc = spawn('ffmpeg', ffmpegArgs, { stdio: ['ignore', 'pipe', 'pipe'] });
+        } catch (err) {
+            finish(null, err);
+            return;
+        }
+
+        proc.stdout.on('data', (chunk) => {
+            if (!chunk || !chunk.length || settled) return;
+            if (token && token.cancelled) {
+                cancelStream();
+                return;
+            }
+
+            const ws = getStationWs(stationName, fallbackWs);
+            if (!ws || ws.readyState !== WebSocket.OPEN) {
+                cancelStream();
+                return;
+            }
+
+            try {
+                ws.send(chunk);
+            } catch (err) {
+                finish(null, err);
+            }
+        });
+
+        proc.stderr.on('data', (chunk) => {
+            stderr += chunk.toString();
+        });
+
+        proc.on('error', (err) => {
+            finish(null, err);
+        });
+
+        proc.on('close', (code, signal) => {
+            const cancelled = token && token.cancelled;
+            log(cancelled ? 'INFO' : 'SUCCESS', `PCM music 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;
+            }
+            finish({ audioId, cancelled: !!cancelled, durationSec: loop ? 0 : durationSec });
+        });
+    });
+}
+
 const stationMusicStates = new Map();
 
 function normalizeBaseName(name) {
@@ -364,6 +480,7 @@ function getPlaybackAudioByBaseName(baseName) {
         if (fs.existsSync(p)) {
             const buffer = fs.readFileSync(p);
             return {
+                sourcePath: p,
                 audioFormat: 'stream',
                 streamUrl: getManagedMediaUrl(p.startsWith(audioStationDir) ? 'audio/station' : 'audio/general', path.basename(p)),
                 durationSec: getWavDurationSec(buffer) || 0
@@ -389,6 +506,18 @@ function getPlaybackAudioByBaseName(baseName) {
     throw new Error(`Audio not found: ${base}`);
 }
 
+function cancelStationAudioTask(task) {
+    if (!task) return;
+    if (task.cancelToken) task.cancelToken.cancelled = true;
+    if (typeof task.onCancel === 'function') {
+        try { task.onCancel(); } catch {}
+    }
+    if (task.endTimer) {
+        try { clearTimeout(task.endTimer); } catch {}
+        task.endTimer = null;
+    }
+}
+
 function stopStationMusic(stationName, fallbackWs) {
     const cleanStation = String(stationName || '').trim();
     const q = getStationAudioQueue(cleanStation);
@@ -396,11 +525,7 @@ function stopStationMusic(stationName, fallbackWs) {
     q.queue = q.queue.filter(t => t?.source !== 'music');
 
     if (q.current && q.current.source === 'music') {
-        if (q.current.cancelToken) q.current.cancelToken.cancelled = true;
-        if (q.current.endTimer) {
-            try { clearTimeout(q.current.endTimer); } catch {}
-            q.current.endTimer = null;
-        }
+        cancelStationAudioTask(q.current);
     }
     sendAudioControl(cleanStation, { type: 'audio_stop', scope: 'music', mode: 'current', reason: 'music_stop' }, fallbackWs);
 }
@@ -426,21 +551,13 @@ function enqueueStationAudioTask(stationName, fallbackWs, task) {
     const cur = q.current;
 
     if (task.interruptMode === 'all' || newP >= 2) {
-        if (cur && cur.cancelToken) cur.cancelToken.cancelled = true;
-        if (cur && cur.endTimer) {
-            try { clearTimeout(cur.endTimer); } catch {}
-            cur.endTimer = null;
-        }
+        cancelStationAudioTask(cur);
         q.queue.length = 0;
         sendAudioControl(cleanStation, { type: 'audio_stop', mode: 'all', reason: 'emergency' }, fallbackWs);
     } else if (cur) {
         const curP = Number(cur.priority || 0);
         if (newP > curP) {
-            if (cur.cancelToken) cur.cancelToken.cancelled = true;
-            if (cur.endTimer) {
-                try { clearTimeout(cur.endTimer); } catch {}
-                cur.endTimer = null;
-            }
+            cancelStationAudioTask(cur);
             sendAudioControl(cleanStation, { type: 'audio_stop', mode: 'current', flushBelow: newP, reason: 'preempt' }, fallbackWs);
             if (cur.requeueOnPreempt) {
                 const replay = { ...cur, cancelToken: null };
@@ -467,7 +584,9 @@ async function processStationAudioQueue(stationName, fallbackWs) {
 
             try {
                 if (task.consoleStreamStart) broadcastToConsole(task.consoleStreamStart);
-                const result = await sendAudioBufferToStation(cleanStation, fallbackWs, task);
+                const result = task.audioFormat === 'pcm'
+                    ? await sendPcmStreamToStation(cleanStation, fallbackWs, task)
+                    : await sendAudioBufferToStation(cleanStation, fallbackWs, task);
                 if (task.consoleStreamEnd) {
                     if (result.cancelled) {
                         broadcastToConsole(task.consoleStreamEnd);
@@ -2390,6 +2509,23 @@ app.post('/api/music/start', express.json(), (req, res) => {
                 return;
             }
 
+            if (playbackAudio.sourcePath) {
+                enqueueStationAudioTask(name, currentWs, {
+                    audioFormat: 'pcm',
+                    sourcePath: playbackAudio.sourcePath,
+                    durationSec: playbackAudio.durationSec,
+                    priority: -1,
+                    interruptMode: 'none',
+                    source: 'music',
+                    loop,
+                    audioId: `music-${track}-${name}-${Date.now()}`,
+                    requeueOnPreempt: true,
+                    consoleStreamStart: { type: 'stream_start', station: name, seq: [`MUSIC:${track}`], streamName, hlsUrl },
+                    consoleStreamEnd: { type: 'stream_end', station: name }
+                });
+                return;
+            }
+
             if (HLS_ENABLE) {
                 enqueueHlsPush(name, [track]);
                 enqueueStationAudioTask(name, currentWs, {

station_broadcast.lua

@@ -654,6 +654,19 @@ local function playChunkOnSpeakers(samples, volume, playSeq)
     return true
 end
 
+local function decodePcmS8Chunk(chunk)
+    if type(chunk) ~= "string" or #chunk == 0 then return nil end
+    local samples = {}
+    for i = 1, #chunk do
+        local b = string.byte(chunk, i)
+        if b >= 128 then
+            b = b - 256
+        end
+        samples[i] = b
+    end
+    return samples
+end
+
 local function areHqSpeakersBusy(speakers)
     for _, sp in ipairs(speakers) do
         local playing = false
@@ -824,10 +837,47 @@ function Audio.playDfpwmItem(item)
     end
 end
 
+function Audio.playPcmItem(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 break end
+        local chunk = source()
+        if chunk == nil then break end
+        if type(chunk) == "string" and #chunk > 0 then
+            local samples = decodePcmS8Chunk(chunk)
+            if samples and not playChunkOnSpeakers(samples, CONFIG.AUDIO_VOLUME, playSeq) then
+                break
+            end
+        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)
     end
     return Audio.playDfpwmItem(item)
 end
@@ -1160,9 +1210,12 @@ function Network.messageLoop()
                             audioId = data.audioId
                         })
                         State.audioRx = nil
-                    elseif audioFormat == "dfpwm" then
+                    elseif audioFormat == "dfpwm" or audioFormat == "pcm" then
                         local stream = Audio.createStream()
                         State.audioRx.stream = stream
+                        if audioFormat == "pcm" then
+                            Utils.log("INFO", "Receiving live PCM audio stream")
+                        end
                         Audio.enqueueItem({
                             audioData = stream,
                             audioFormat = audioFormat,
@@ -1204,7 +1257,8 @@ function Network.messageLoop()
             else
                     if State.audioRx then
                         -- Memory protection: limit to 10MB per audio item
-                        if State.audioRx.receivedBytes + #msg > 10 * 1024 * 1024 then
+                        local isLiveStream = State.audioRx.stream and (tonumber(State.audioRx.expectedBytes) or 0) <= 0
+                        if (not isLiveStream) and State.audioRx.receivedBytes + #msg > 10 * 1024 * 1024 then
                             Utils.log("ERROR", "Audio item too large, discarding")
                             if State.audioRx.stream then
                                 State.audioRx.stream.finish()