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();
        //console.log("DataProcessing state change", this.state, newState);
        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);
        };
        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(this.newDeviceURL + "/api/session/data", this.onNewData);
    }

    onRunningChanged = (running, deviceURL) => {
        if (running) {
            //console.log("Starting session");
            let req = new XMLHttpRequest();
            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;
        }
    }

    onNewData = (data) => {
        data.values;
        data.sessionStartTime;
        data.startIndex;
        let success = false;
        if (data.sessionStartTime === this.sessionStartTime && data.startIndex === this.rawMeasurements.length) {
            // normal case, add received data to measurement array
            this.rawMeasurements.concat(List(data.values));
            this.analyzeNewMeasurements(data.startIndex);
            success = true;
        }
        else if (data.startIndex === 0) {
            // new start
            this.sessionStartTime = data.sessionStartTime;
            this.rawMeasurements = List(data.values);
            success = true;
        } else {
            // 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);
            this.store.dispatch(reportDeviceData(this.sessionStartTime, data.startIndex, this.rawMeasurements, analysis));
        }
        
    }

    analyzeNewMeasurements = (newDataStartIdx) => {
        const newPeaks = this.peakDetectorSimple.addVector(this.rawMeasurements.slice(newDataStartIdx));
        this.peaks = this.peaks.concat(List(newPeaks));
        const totalMomentum = this.rawMeasurements.reduce((sum, x) => sum + x, 0);
        const peakMax = this.rawMeasurements.reduce((running, x) => 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) => 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;