From 709388ec3ab25e63ef65fdd9ddf4b170e50e1b9e Mon Sep 17 00:00:00 2001 From: HenryDu8133 <813367384@qq.com> Date: Fri, 31 Jul 2026 21:00:29 +0800 Subject: [PATCH] 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 | 168 ++++++++++++++++++++++++++++++++++++++---- station_broadcast.lua | 58 ++++++++++++++- 2 files changed, 208 insertions(+), 18 deletions(-) diff --git a/CloudServer/server.js b/CloudServer/server.js index 4409a2d..c7f9a7c 100644 --- a/CloudServer/server.js +++ b/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, { diff --git a/station_broadcast.lua b/station_broadcast.lua index 5f6c878..a18fec2 100644 --- a/station_broadcast.lua +++ b/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()