diff --git a/CloudServer/server.js b/CloudServer/server.js index e70cbab..d796e83 100644 --- a/CloudServer/server.js +++ b/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 (!fs.existsSync(sourcePath)) throw new Error(`Live DFPWM source missing: ${sourcePath}`); + if (!sourcePath) throw new Error('Live WAV sourcePath required'); + 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, - channels: 1, - audioFormat: 'dfpwm', + sampleRate, + channels, + 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'); - log('INFO', `Live DFPWM music stream start for ${stationName}: ${path.basename(sourcePath)}${loop ? ' (loop)' : ''}`); + ffmpegArgs.push('-re', '-i', sourcePath, '-f', 's16le', '-acodec', 'pcm_s16le', '-ac', String(channels), '-ar', String(sampleRate), 'pipe:1'); + 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) => { - if (!chunk || !chunk.length || settled) return; - if (token && token.cancelled) { - cancelStream(); - return; - } - + const flushChunk = (pcmChunk) => { + if (!pcmChunk || !pcmChunk.length) 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)); - if (encoded && encoded.length > 0) { - ws.send(encoded); - } + ws.send(buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample)); } 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' - ? await sendLiveDfpwmToStation(cleanStation, fallbackWs, task) + const result = task.audioFormat === 'wav-stream-live' + ? 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, diff --git a/station_broadcast.lua b/station_broadcast.lua index 70f1bb5..94fe81e 100644 --- a/station_broadcast.lua +++ b/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,