Needed pipeable PcmFileSink for my project - contributing it back

[parity-error]

I needed to write to a continuous PCM (audio) stream for a 24/7 application that produces mp3 output chunks. But the stock WAVFileSink is not capable of being used with pipes since it uses file seek operations. So I created a new PcmFileSink that does not write any headers and does not do any seeking. Making it friendly for stream processing with pipes. I am including the file here since it is tested and working fine for me. The comments include instructions for how to register it in the init.lua along with examples for how it can be used. Feel free to add this in if you think it is useful. Thank you all for building and maintaining such a great radio processing framework!

---
-- Sink one or more real-valued signals to a PCM file (no headers). The supported sample
-- formats are 8-bit unsigned integer, 16-bit signed integer, and 32-bit signed
-- integer.  This sink is safe to use with all types of pipes unlike its WAV counterpart 
-- as it uses no seek operations. 
--
-- You can pipe the output of this sink to processes that can take streaming pcm data
-- like sox.   For example to have sox create 3 minute mp3 chunks continuously from 
-- your output stream you would do this (adjusting for your own bitrate, # channels,  etc) 
-- <your.lua.process> | sox -r 100k -t raw -e s -b 16 -c 1 -L -V1 - "ouput_.mp3" trim 0 180 : newfile : restart &
--
-- registering this new sink:  To register this add this line
-- PcmFileSink = require('radio.blocks.sinks.pcmfile'),
-- to the     -- Sink Blocks section of your init.lua 
-- Then place this file in the ./radio/blocks/sinks directory with the other sinks.
-- To use this with a pipe then initialize like this: 
-- local sink = radio.PcmFileSink(io.stdout, 1)
-- 
--
-- @category Sinks
-- @block PcmFileSink
-- @tparam string|file|int file Filename, file object, or file descriptor
-- @tparam int num_channels Number of channels (e.g. 1 for mono, 2 for stereo, etc.)
-- @tparam[opt=16] int bits_per_sample Bits per sample, choice of 8, 16, or 32
--
-- @signature in:Float32 >
-- @signature in1:Float32, in2:Float32, ... >
--
-- @usage
-- -- Sink to a one channel WAV file
-- local snk = radio.PcmFileSink('test.wav', 1)
-- top:connect(src, snk)
--
-- -- Sink to a two channel WAV file
-- local snk = radio.PcmFileSink('test.wav', 2)
-- top:connect(src1, 'out', snk, 'in1')
-- top:connect(src2, 'out', snk, 'in2')

local ffi = require('ffi')
local block = require('radio.core.block')
local vector = require('radio.core.vector')
local types = require('radio.types')
local format_utils = require('radio.utilities.format_utils')
local PcmFileSink = block.factory("PcmFileSink")


local wave_formats = {
    [8]     = format_utils.formats.u8,
    [16]    = format_utils.formats.s16le,
    [32]    = format_utils.formats.s32le,
}

function PcmFileSink:instantiate(file, num_channels, bits_per_sample)
    if type(file) == "string" then
        self.filename = file
    elseif type(file) == "number" then
        self.fd = file
    else
        self.file = assert(file, "Missing argument #1 (file)")
    end

    self.num_channels = assert(num_channels, "Missing argument #2 (num_channels)")
    self.bits_per_sample = bits_per_sample or 16
    self.format = assert(wave_formats[self.bits_per_sample], string.format("Unsupported bits per sample (%s)", tostring(bits_per_sample)))

    self.num_samples = 0
    self.count = 0

    -- Build type signature
    if num_channels == 1 then
        self:add_type_signature({block.Input("in", types.Float32)}, {})
    else
        local block_inputs = {}
        for i = 1, num_channels do
            block_inputs[#block_inputs+1] = block.Input("in" .. i, types.Float32)
        end
        self:add_type_signature(block_inputs, {})
    end
end

-- Header endianness conversion

local function bswap32(x)
    return bit.bswap(x)
end

local function bswap16(x)
    return bit.rshift(bit.bswap(x), 16)
end


function PcmFileSink:initialize()
    if self.filename then
        self.file = ffi.C.fopen(self.filename, "wb")
        if self.file == nil then
            error("fopen(): " .. ffi.string(ffi.C.strerror(ffi.errno())))
        end
    elseif self.fd then
        self.file = ffi.C.fdopen(self.fd, "wb")
        if self.file == nil then
            error("fdopen(): " .. ffi.string(ffi.C.strerror(ffi.errno())))
        end
    end

    -- Allocate raw samples vector
    self.raw_samples = vector.Vector(self.format.real_ctype)

    -- Register open file
    self.files[self.file] = true
end

function PcmFileSink:process(...)
    local samples = {...}
    local num_samples_per_channel = samples[1].length

    self.count = self.count + samples[1].length

    -- Resize raw samples vector
    self.raw_samples:resize(num_samples_per_channel * self.num_channels)

    -- Convert float32 samples to raw samples
    for i = 0, num_samples_per_channel-1 do
        for j = 1, self.num_channels do
            self.raw_samples.data[i*self.num_channels + (j-1)].value = (samples[j].data[i].value*self.format.scale) + self.format.offset
        end
    end

    -- Perform byte swap for endianness if needed
    if self.format.swap then
        for i = 0, (self.num_channels*num_samples_per_channel)-1 do
            format_utils.swap_bytes(self.raw_samples.data[i])
        end
    end

    -- Write to file
    local num_samples = ffi.C.fwrite(self.raw_samples.data, ffi.sizeof(self.format.real_ctype), num_samples_per_channel * self.num_channels, self.file)
    if num_samples ~= num_samples_per_channel * self.num_channels then
        error("fwrite(): " .. ffi.string(ffi.C.strerror(ffi.errno())))
    end

    -- Update our sample count
    self.num_samples = self.num_samples + num_samples_per_channel
end

function PcmFileSink:cleanup()
    if self.filename then
        if ffi.C.fclose(self.file) ~= 0 then
            error("fclose(): " .. ffi.string(ffi.C.strerror(ffi.errno())))
        end
    else
        if ffi.C.fflush(self.file) ~= 0 then
            error("fflush(): " .. ffi.string(ffi.C.strerror(ffi.errno())))
        end
    end
end

return PcmFileSink

[cdgraff]

Amazing job! but just for curiosity, all these is not similar to the capability of:

-- Sink raw samples to stdout
local sink = radio.RealFileSink(1, 's16le')

I'm using them for doing the same you need, just piping to LIQUIDSOAP process and LS is responsable for all the recording in filesystem, but basically LUARADIO pipe to other process.

[parity-error]

Interesting. Before I wrote this I tried to get it work with RawFileSink, which failed with sox (raw) because of the extra data that is peppered in the stream with that method. But somehow I failed to notice the RealFileSink option. ( I am still new to luaradio) I will give this a try since I would rather use supported off the shelf code rather than my own mods. That will make distribution and deployment easier. Thanks for the tip @cdgraff .