swimtracker-app/SwimTracker/data_processing/SwimTrackerWebsocketConnection.js

import ReconnectingWebSocket from 'reconnecting-websocket';
import * as msgpack from 'msgpack-lite';


const OpCodes = {
    // from swim tracker device to frontend
    ERROR: 1,

    INITIAL_INFO: 2,
    SESSION_STARTED: 3,
    SESSION_STOPPED: 4,
    SESSION_NEW_DATA: 5,
    ANSWER_USER_LIST: 6,
    ANSWER_SESSION_LIST: 7,
    WIFI_STATE_RESPONSE: 8,
    WIFI_SCAN_RESPONSE: 9,
    APP_LAYER_PING: 10,
    LOG_UPDATE: 11,

    // from frontend to device
    START_SESSION: 128,
    STOP_SESSION: 129,
    TARE: 130,
    QUERY_USER_LIST: 131,
    QUERY_SESSION_LIST: 132,
    WIFI_STATE_SET: 133,
    WIFI_STATE_GET: 134,
    WIFI_TRIGGER_SCAN: 135,
    LOG_STREAMING_START: 136,
    LOG_STREAMING_STOP: 137
};

const HEARTBEAT_TIMEOUT = 3000;
const PROVISIONING_IP = "192.168.42.1";

export default class SwimTrackerWebsocketConnection {
    /**
     * Creates a new persistent websocket connection to a swimtracker device 
     * 
     * @param {string} swimTrackerHost hostname or ip of the swimtracker device
     * @param {(data: Uint16Array) => any} onData called whenever new measurement data is available
     * @param {(sessionId: number) => any} onStarted  called when a new measurement session was started
     * @param {() => any} onStopped called when session was stopped
     * @param {(wifistate : object) => any} onWifiStateInfo  wifi state contains "state" (STATION_MODE|AP_PROVISIONING|AP_SECURE) and "hostname"
     * @param {() => any} onConnect called when websocket connection was established
     * @param {() => any} onDisconnect called when websocket disconnected
     */
    constructor(swimTrackerHost, onData, onStarted, onStopped, onWifiStateInfo, onConnect, onDisconnect, reconnectingWsOptions={}) {
        this.swimTrackerHost = swimTrackerHost;

        this.onData = onData;
        this.onStarted = onStarted;
        this.onStopped = onStopped;
        this.onWifiStateInfo = onWifiStateInfo;
        this.onConnect = onConnect;
        this.onDisconnect = onDisconnect;
        this.onLogMessage = () => {};
        
        // try configured URL and provisioning URL
        const urls = [`ws://${swimTrackerHost}:81`, `ws://${PROVISIONING_IP}:81`];
        let urlIndex = 0;
        const urlProvider = () => urls[urlIndex++ % urls.length]; // round robin url provider
        
        this.ws = new ReconnectingWebSocket(urlProvider, [], { ...reconnectingWsOptions,  maxReconnectionDelay: 3000 });
        this.ws.onmessage = this._onMessage;
        this.ws.onopen = this.onConnect;
        this.ws.onclose = this.onDisconnect;
        this.ws.onerror = this._onError;
        this.ws.binaryType = 'arraybuffer';

        this.msgpackCodec = msgpack.createCodec();
        this.msgpackCodec.addExtUnpacker(205, function (byteArr) {
            const buffer = byteArr.buffer.slice(byteArr.byteOffset, byteArr.byteLength + byteArr.byteOffset);
            const result = new Int16Array(buffer);
            return result;
        });

        this._wifiScanPromises = [];
        this.pingTimeout = null;
    }

    heartbeat() {
        clearTimeout(this.pingTimeout);

        let connection = this;
        this.pingTimeout = setTimeout(() => {
            if(connection.ws !== null)
                connection.ws.reconnect();            
        }, HEARTBEAT_TIMEOUT);
    }

    close() {
        if (this.ws !== null) {
            this.ws.onmessage = null;
            this.ws.onopen = null;
            this.ws.onclose = null;
            this.ws.onerror = null;
            this.ws.close();
            this.ws = null;
        }
    }

    sendStartCommand() {
        this._sendMsg(OpCodes.START_SESSION);
    }

    sendStopCommand() {
        this._sendMsg(OpCodes.STOP_SESSION);
    }

    sendTareCommand = () => {
        this._sendMsg(OpCodes.TARE);
    }

    sendLogStreamStartCommand = () => {
        this._sendMsg(OpCodes.LOG_STREAMING_START);
    }

    sendLogStreamStopCommand = () => {
        this._sendMsg(OpCodes.LOG_STREAMING_STOP);
    }

    scanWifiNetworks() {
        console.log("Trigger wifi scan");
        this._sendMsg(OpCodes.WIFI_TRIGGER_SCAN);

        let conn = this;
        return new Promise((resolve, reject) => {
            conn._wifiScanPromises.push({ resolve: resolve, reject: reject });
        });

    }

    sendTareCommand = () => {
        this._sendMsg(OpCodes.WIFI_STATE_SET, {
            "reset_to_provisioning": true,
        });

    }

    wifiSetModeAP(password) {
        this._sendMsg(OpCodes.WIFI_STATE_SET, {
            "ap_password": password,
        });
    }

    wifiSetModeSTA(ssid, password) {
        console.log("Setting sta mode", ssid, password);
        this._sendMsg(OpCodes.WIFI_STATE_SET, {
            "sta_ssid": ssid,
            "sta_password": password,
        });
    }

    _sendMsg(code, data) {
        let msg = undefined;
        if (data) {
            const serializedData = msgpack.encode(data);
            msg = new Uint8Array([code, ...serializedData]);
        } else {
            msg = new Uint8Array(1);
            msg[0] = code;
        }
        this.ws.send(msg);
    }

    _onMessage = (e) => {
        const dv = new DataView(e.data);
        const opCode = dv.getInt8(0);
        const payload = new Uint8Array(e.data).slice(1);
        
        this.heartbeat();

        if (opCode === OpCodes.INITIAL_INFO) {
            const headerSize = 6;
            const running = Boolean(dv.getInt8(1));
            const sessionId = dv.getUint32(2);
            if (running && e.data.byteLength > headerSize) {
                const data = new Uint16Array(e.data.slice(headerSize));
                this.onStarted(sessionId);
                this.onData(data);
            } else
                this.onStopped();
        } else if (opCode === OpCodes.SESSION_STARTED) {
            const sessionId = dv.getUint32(1);
            this.onStarted(sessionId);
        } else if (opCode === OpCodes.SESSION_STOPPED) {
            this.onStopped();
        } else if (opCode === OpCodes.SESSION_NEW_DATA) {
            const data = new Uint16Array(e.data.slice(1));
            this.onData(data);
        } else if (opCode === OpCodes.WIFI_SCAN_RESPONSE) {
            const scanResult = msgpack.decode(payload, { codec: this.msgpackCodec });

            for (let i = 0; i < this._wifiScanPromises.length; ++i) {
                this._wifiScanPromises[i].resolve(scanResult);
            }
            this._wifiScanPromises.length = 0;
        } else if (opCode === OpCodes.WIFI_STATE_RESPONSE) {
            const wifiInfo = msgpack.decode(payload, { codec: this.msgpackCodec });
            this.onWifiStateInfo(wifiInfo);
        } else if (opCode === OpCodes.LOG_UPDATE) {
            const logMsg = msgpack.decode(payload, { codec: this.msgpackCodec });
            this.onLogMessage(logMsg);
        } else if (opCode === OpCodes.APP_LAYER_PING) {
            //console.log("got heartbeat");
        }
    }

    _onError = (ev) => {
        console.log("Websocket error", ev, ev.error, ev.message);
    }
};