swimtracker-app/data_processing/DataProcessing.js

import DeviceHttpDataSource from './DeviceDataSource';
import { List } from 'immutable';
import { reportDeviceData } from '../state/ActionCreators';
import { PeakDetectorSimple} from './PeakDetection';


// todo: put in settings?
const NUM_MEASUREMENTS_PER_SECOND = 10;
const WINDOW_SIZE_SECS = 5;

// This is the main data processing entry point, coupling between device and redux store
// - periodically fetch data
// - feeds them to analysis, (manages analysis classes)
// - adds them to redux store
class DataProcessing {
    constructor(reduxStore) {
        this.store = reduxStore;
        this.store.subscribe(this.onStateChange);
        this.state = this.store.getState();
        this.dataSource = null;
        //console.log("state", this.state);
        console.assert(this.state.session.running === false, "Created DataProcessing with running=True");
        this.onDataSourceChanged(this.state.settings.deviceURL);

        this.rawMeasurements = List();
        this.sessionStartTime = 0;

        this.peakDetectorSimple = new PeakDetectorSimple(this.state.settings.peakDetectorSimpleThreshold);
        this.peaks = List();
    }

    onStateChange = () => {
        const newState = this.store.getState();
        if (newState.settings.deviceURL !== this.state.settings.deviceURL)
            this.onDataSourceChanged(newState.settings.deviceURL);

        if (newState.session.running !== this.state.session.running) {
            this.onRunningChanged(newState.session.running, newState.settings.deviceURL);
        };
        if(newState.settings.peakDetectorSimpleThreshold !== this.state.settings.peakDetectorSimpleThreshold) {
            this.peakDetectorSimple = new PeakDetectorSimple(newState.settings.peakDetectorSimpleThreshold, this.onNewPeak);
            this.peaks = List(this.peakDetectorSimple.addVector(this.rawMeasurements));
        };
        this.state = newState;
    }

    onDataSourceChanged = (newDeviceURL) => {
        if (this.dataSource !== null) {
            this.dataSource.stop();
            this.dataSource = null;
        }
        this.dataSource = new DeviceHttpDataSource(newDeviceURL + "/api/session/data", this.onNewData);
    }

    onRunningChanged = (running, deviceURL) => {
        let req = new XMLHttpRequest();
        if (running) {
            console.log("Starting session", deviceURL + "/api/session/start" );
            req.open("GET", deviceURL + "/api/session/start");
            this.dataSource.startIndex = 0;
            this.dataSource.start();
        } else {
            console.log("Stopping session");
            req.open("GET", deviceURL + "/api/session/stop");
            this.dataSource.stop();
            this.dataSource.startIndex = 0;
        }
        req.addEventListener("error", evt => console.log(evt));
        req.addEventListener("abort", evt => console.log(evt));
        req.send();
    }

    onNewData = (data) => {
        data.values;
        data.sessionStartTime;
        data.startIndex;
        let success = false;
        if (data.sessionStartTime == this.sessionStartTime && data.startIndex == this.rawMeasurements.size) {
            // normal case, add received data to measurement array
            this.rawMeasurements = this.rawMeasurements.concat(List(data.values));
            success = true;
        }
        else if (data.startIndex === 0) {
            // new start
            this.sessionStartTime = data.sessionStartTime;
            this.rawMeasurements = List(data.values);
            success = true;
        } else {
            console.log("Requery :(");
            console.log("this.sessionStartTime", this.sessionStartTime);
            console.log("this.rawMeasurements", this.rawMeasurements.toArray());
            console.log("data", data);
            // missed some data -> re-query
            this.dataSource.startIndex = 0;
            this.sessionStartTime = 0;
            //console.log("Problem: got non-consequtive data. Received:", data,
            //    "Own state ", { startTime: this.sessionStartTime, values: this.rawMeasurements });
        }

        if (success) {
            const analysis = this.analyzeNewMeasurements(data.startIndex);
            const report = reportDeviceData(this.sessionStartTime, data.startIndex, this.rawMeasurements, analysis);
            this.store.dispatch(report);
        }
    }

    analyzeNewMeasurements = (newDataStartIdx) => {
        //TODO is ".toArray()" really necessary here?
        const newPeaks = this.peakDetectorSimple.addVector(this.rawMeasurements.slice(newDataStartIdx).toArray());
        this.peaks = this.peaks.concat(List(newPeaks));
        const totalMomentum = this.rawMeasurements.reduce((sum, x) => sum + x, 0);
        const peakMax = this.rawMeasurements.reduce((running, x) => Math.max(x, running), 0);

        // windowed quantities
        const windowSizeMeasurements = WINDOW_SIZE_SECS * NUM_MEASUREMENTS_PER_SECOND;
        const windowedSeq = this.rawMeasurements.slice(-windowSizeMeasurements);
        const peakMaxWindow = windowedSeq.reduce((running, x) => Math.max(x, running), 0);
        const momentumWindow = windowedSeq.reduce((sum, x) => sum + x, 0);

        return {
            peaks: this.peaks,
            totalTime: this.rawMeasurements.length / NUM_MEASUREMENTS_PER_SECOND,
            activeTime: 0,
            totalMomentum: totalMomentum,
            peakFrequency: 0,
            peakMax: peakMax,
            // windowed quantities
            momentumWindow: momentumWindow,
            frequencyWindow: 0,
            peakMaxWindow: peakMaxWindow,
        }
    };
};

export default DataProcessing;