piaudio.js

var DeviceClient = require('azure-iot-device').Client
var DeviceProtocol = require('azure-iot-device-amqp').AmqpWs;
var exec = require('child_process').exec;
var tempfile = require('tempfile');
var fs = require('fs');
var SerialPort = require("serialport");
var request = require('request');

var connectionString = "HostName=[...]";
var azureCognitiveServicesAuthEndpoint = 'https://northeurope.api.cognitive.microsoft.com/sts/v1.0/issueToken';
var azureCognitiveServicesMainEndpoint = 'https://northeurope.tts.speech.microsoft.com/cognitiveservices/v1';
var azureCognitiveServicesSubscriptionKey = "[...]";

var portdev = "/dev/ttyACM0";
var port = new SerialPort(portdev, {
    baudRate: 9600,
    parser: SerialPort.parsers.Readline
});

port.on('open', function() {
    port.write("\nCOLOR #000000\n");
});

port.on('disconnect', function(err) {
    console.log('Disconnect: ', err.message);
    setTimeout(function() {
        process.exit(1);
    }, 5000);
});

port.on('error', function(err) {
    console.log('Error: ', err.message);
    setTimeout(function() {
        process.exit(1);
    }, 5000);
})

port.on('data', function (data) {
    console.log('Data: ' + data);
});

// We'll process messages sequentially - build a queue
function MyController() {
    this.timeout = 100;
    this.queue = [];
    this.ready = true;
};

MyController.prototype.exec = function() {
    this.queue.push(arguments);
    this.process();
};

MyController.prototype.action = function(message) {
    var self = this;
    console.log("Processing message...");
    if (message.text) {
        azureCallApi(message.text, function(error, mp3file) {
            if (error) {
                console.log("Azure API error: " + ((typeof(error) == typeof({}))?JSON.stringify(error):error));
                self.ready = true;
                self.process();
                return;
            }
          
            if (message.color2 && message.color) {
                port.write("COLORS " + message.color + " " + message.color2 + "\n");
            }
            else if (message.color) {
                port.write("COLOR " + message.color + "\n");
            }
            port.write("HEAD MOVE\n");
            exec("omxplayer " + mp3file, function(err, stdout, stderr) {
                console.log('stdout: ' + stdout);
                console.log('stderr: ' + stderr);
                if (err !== null) {
                    console.log('exec error: ' + err);
                }
                fs.unlink(mp3file, function(err){});
                port.write("TAIL MOVE\n");
                setTimeout(function() {
                    port.write("TAIL RELEASE\n");
                    port.write("COLOR #000000\n");
                }, 500);
                setTimeout(function() {
                    self.ready = true;
                    self.process();
                }, 1500);
            });
        });
    }
    else {
        self.ready = true;
        self.process();
    }
};

MyController.prototype.process = function() {
    if (this.queue.length === 0) return;
    if (!this.ready) return;
    var self = this;
    this.ready = false;
    this.action.apply(this, this.queue.shift());
};

var messageController = new MyController();

// AMQP-specific factory function returns Client object from core package
var client = DeviceClient.fromConnectionString(connectionString, DeviceProtocol);
 
// use Message object from core package
var Message = require('azure-iot-device').Message;

var connectCallback = function (err) {
    if (err) {
        console.log('Could not connect: ' + err);
        connected = false;
    } else {
        console.log('Client connected');
        connected = true;
        client.on('message', onMessage);
    }
};

function onMessage(msg) {
    console.log('Id: ' + msg.messageId + ' Body: ' + msg.data);
    messageController.exec(JSON.parse(msg.data.toString()));
    client.complete(msg, printResultFor('completed'));
}

client.on('disconnect', function() {
    console.error('Client was disconnected');
    client.close(function(err, res) {
        connected = false;
    });
});

function iotConnect() {
    client.open(connectCallback);
}

setInterval(function () {
    if (!connected) {
        //console.log("Re-connecting to IoT Hub...");
        //iotConnect();
        // XXX I can't get messages to arrive after a reconnect so, for now,
        // just exit this program and have the looping script re-run it
        // so we make a fresh connection.
        console.log("Not connected, exiting...");
        process.exit();
    }
}, 60000);

iotConnect();

// Helper function to print results in the console
function printResultFor(op) {
    return function printResult(err, res) {
        if (err) console.log(op + ' error: ' + err.toString());
        if (res) console.log(op + ' status: ' + res.constructor.name);
    };
}

// Azure Cognitive Services interface
var _cachedAzureToken = null;
var _cachedAzureTokenExpires = null;
function azureGetAuthToken(callback) {
    if (_cachedAzureToken && _cachedAzureTokenExpires) {
        if ((new Date).getTime() < _cachedAzureTokenExpires) {
            console.log("Using cached auth token: " + _cachedAzureToken);
            callback(null, _cachedAzureToken);
            return;
        }        
    }
    request({
        url: azureCognitiveServicesAuthEndpoint,
        method: "POST",
        headers: {
            'Content-Type' : 'application/x-www-form-urlencoded',
            'Content-Length': 0,
            'Ocp-Apim-Subscription-Key': azureCognitiveServicesSubscriptionKey
        }
    }, function (error, response, body){
        if (error) {
            callback(error, null);
        }
        else if (response.statusCode != 200) {
            callback(body, null);
        }
        else {
            _cachedAzureToken = body;
            _cachedAzureTokenExpires = (new Date).getTime() + (9 * 60 * 1000); // Expires in 10 minutes, refresh in 9
            console.log("Got auth token: " + _cachedAzureToken);
            callback(null, _cachedAzureToken);
        }
    });
}

function azureCallApi(text, callback) {
    var ssml = "<speak version='1.0' xmlns=\"http://www.w3.org/2001/10/synthesis\" xml:lang='en-US'>" +
        "<voice name='Microsoft Server Speech Text to Speech Voice (en-GB, George, Apollo)'>" +
        "<prosody rate=\"slow\">" +
        text +
        "</prosody>" +
        "</voice>" +
        "</speak>";
    console.log(ssml);
    azureGetAuthToken(function (err, token) {
        if (err) {
            callback(err, token);
            return;
        }
        var mp3file = tempfile(".mp3");
        var x = fs.createWriteStream(mp3file);
        var r = request({
            url: azureCognitiveServicesMainEndpoint,
            method: "POST",
            body: ssml,
            headers: {
                'Content-Type' : 'application/ssml+xml',
                'X-Microsoft-OutputFormat': 'audio-24khz-48kbitrate-mono-mp3',
                'User-Agent': 'Talking fish',
                'Authorization': 'Bearer ' + token
            }
        })
        .on('error', function(err) {
            callback(err, null);
        })
        .on('response', function(response) {
            if (response.statusCode != 200) {
                callback(response, null);
                return
            }
        })
        .pipe(x)
        .on('finish', function() {
            x.close();
            callback(null, mp3file);
        });
    });
}
// TODO fix up error handling