Electric Imp streaming audio player demo (imp001/002/003/004)

agent.nut

// Streaming audio player - agent side
// 20180909 hugo@electricimp.com

// ToDo: lots. Like, parsing the WAV header, transcoding in the agent, etc

// Summary:
// - device asks agent to play a URL
// - agent fetches AGENT_CHUNK sized blocks from the URL using range fetches
// - agent chops these up into DEVICE_CHUNK sized pieces and puts them in
//   send_queue. It then refills send_queue every time it gets down to half
//   empty.
// - device consumes these pieces, requesting a new one each time it has
//   queued one for the DAC
// - initially, we send 4 DEVICE_CHUNKs to the device; when it has got all
//   4 of these locally it starts playing (consuming data and hence requesting
//   more buffers)
// - in this way, the device always has ~3x DEVICE_CHUNK samples on hand
//   (~1.5s buffering with 8192 byte chunks and 16kHz sample rate)
// - the data from the device to the agent is essentially also operating in a
//   "sliding window" mode, with up to 4 buffers in flight at any time

// Buffer we will keep locally to minimize number of fetches. We will refill this
// when it gets half-empty, so we will use 1.5x this amount of memory at peak
const AGENT_CHUNK = 262144;
agent_buffer <- null;

// Chunk size we are going to send to the device
const DEVICE_CHUNK = 8192;

// URL we're fetching from & offset through the fetch
// fetching indicates if an async http request is in progress (we only do one
// at a time). done indicates that the fetching is complete, though the buffer
// may not be all sent to the device yet
url <- null;
fetch_offset <- 0;
fetching <- false;
done <- false;

// Queue of buffers to device
send_queue <- [];

// Fetch a chunk from the server and queue it for device
function fetch_chunk(ready_cb = null) {
    if (fetching) return;
    
    // Note range is inclusive
    local req = http.get(url, { "Content-Type": "application/octet-stream",
                                "Range": format("Bytes=%d-%d", fetch_offset, fetch_offset + AGENT_CHUNK - 1)});
    fetching = true;
    
    // Issue this request; when it returns, send the fetched data to the device
    req.sendasync(function(response) {
        fetching = false;
        if (response.statuscode == 200 || response.statuscode == 206) {
            local length = response.body.len();

            server.log("Fetched "+fetch_offset+"-"+(fetch_offset + length));
            fetch_offset += length;
            if (length > 0) {
                // Split it into bite size chunks & queue them ready for sending
                for(local o = 0; o < length; o += DEVICE_CHUNK) {
                    // Last chunk will be small
                    local chunklen = (length - o) > DEVICE_CHUNK ? DEVICE_CHUNK : (length - o);
                    send_queue.append(response.body.slice(o, o + chunklen));
                }
            }
            
            if (length < AGENT_CHUNK || length == 0) {
                // End of file; append a zero length sentinel
                server.log("Agent reached EOF");
                done = true;
            }
            
            // If we had a callback passed, trigger it now buffer is filled
            if (ready_cb != null) ready_cb();
        } else {
            // We'll generally get error 416 when we are beyond the end
            done = true;
            server.log(format("Got error %d: %s", response.statuscode, response.body));
        }
    });
}

// Handler for when device requests another buffer
device.on("next", function(v) {
    // Anything left?
    if (send_queue.len() == 0) {
        // Send null to finish playback
        // Generally this won't get hit - this only happens if the HTTP fetch
        // is slower than the device's audio consumption
        device.send("buffer", null);
        done = true;
    } else {
        // Send next chunk
        device.send("buffer", send_queue[0]);
        send_queue.remove(0);
        
        // Refill the agent-side buffer when we are halfway down (and not done!)
        if (!done && send_queue.len() < (AGENT_CHUNK / DEVICE_CHUNK / 2)) {
            fetch_chunk();
        }
    }
});

// When device tells us a URL to stream, start!
device.on("fetch", function(file) {
    server.log("Streaming "+file+" to device");

    // Reset offset
    fetch_offset = 0;
    done = false;
    url = file;

    fetch_chunk(function() {
        // Send first four buffers to get started
        device.send("buffer", send_queue.remove(0));
        device.send("buffer", send_queue.remove(0));
        device.send("buffer", send_queue.remove(0));
        device.send("buffer", send_queue.remove(0));
    });    
});

device.nut

// Streaming audio player - device side
// 20180909 hugo@electricimp.com

const SAMPLERATE = 16000;

// DAC pin to use on imp
dac <- hardware.fixedfrequencydac;
dacpin <- null;
switch(imp.info().type) {
    case "imp001":
    case "imp002":
        dacpin = hardware.pin5;
        break;
        
    case "imp003":
        dacpin = hardware.pinC;
        break;
        
    case "imp004m":
        // See https://developer.electricimp.com/resources/imp004maudio
        dacpin = hardware.pwmpairKD;
        break;
    
    default:
        server.log("imp does not support ffdac");
        return;
}

// Audio queue; this holds buffers from the agent, of size DEVICE_CHUNK
// We have 4 of these queued initially
q <- [];
playing <- false;

// Handler to deal with the DAC having consumed a buffer
function bufferEmpty(buffer)
{
    // Refill buffer from queue
    if (q.len() > 0) {
        local newbuffer = q.remove(0);
        if (newbuffer != null) {
            // Queue the buffer
            dac.addbuffer(newbuffer);

            // Tell agent we need another one        
            agent.send("next", 1);
            return;
        }
    }

    // If we got here, we either underran or we hit the null buffer
    // that signifies EOF
    if (playing) {
        playing = false;
        server.log("Device stopped playing");
    }
}

// Handler to deal with new audio data from agent
agent.on("buffer", function(buffer) {
    // put it in the queue
    q.append(buffer);
    
    // If we're not playing, and we have enough buffered to be safe, start
    if (!playing && q.len() >= 4) {
        playing = true;
        server.log("Device started playing");
        dac.start();
    }
});

// Configure DAC
//
// Note we pass two blobs in here; these are empty but as each is consumed 
// we will get an empty callback, which will queue more real audio data, making
// us double buffered. We can add more here for more buffer depth and hence
// ability to block in user code without affecting playback
dac.configure(dacpin, SAMPLERATE, [blob(1024), blob(1024)], bufferEmpty, AUDIO | A_LAW_DECOMPRESS);

// Tell agent we're ready to play something!
// This is an old mp3.com free MP3 "Anezal" from "Strange Angel" transcoded to 16kHz A-Law
agent.send("fetch", "http://utter.chaos.org.uk/~altman/strange_angel_anezal.alaw");