martin/swimtracker-app
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

New data processing

Browse Source
This commit is contained in:
Martin Bauer
2020-06-23 21:36:14 +02:00
parent 655d8b0dc8
commit 00be9e1db2
5 changed files with 61 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
data_processing/DataProcessing.js
View File

@@ -1,32 +1,24 @@
import DeviceHttpDataSource from './DeviceDataSource';
import { List } from 'immutable';
import { reportDeviceData } from '../state/ActionCreators';
import { PeakDetectorSimple} from './PeakDetection';
import { reportDeviceData, resetDeviceData } 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();
this.onDataSourceChanged(this.state.settings.deviceURL);
this.onRunningChanged(this.state.session.running, this.state.settings.deviceURL);
}
onStateChange = () => {
@@ -37,13 +29,23 @@ class DataProcessing {
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));
if (newState.settings.peakDetectorSimpleThreshold !== this.state.settings.peakDetectorSimpleThreshold) {
this.onAnalysisParameterChange();
};
this.state = newState;
}
resetAnalysis = () => {
this.peakDetectorSimple = new PeakDetectorSimple(this.state.settings.peakDetectorSimpleThreshold);
}
onAnalysisParameterChange = () => {
this.resetAnalysis();
this.peakDetectorSimple.addVector(this.state.session.rawData.toArray());
const analysis = this.analyzeNewMeasurements(data.values, List());
this.store.dispatch(reportDeviceData(this.state.session.sessionId, this.state.session.rawData.size, this.state.session.rawData, analysis));
}
onDataSourceChanged = (newDeviceURL) => {
if (this.dataSource !== null) {
this.dataSource.stop();
@@ -55,12 +57,12 @@ class DataProcessing {
onRunningChanged = (running, deviceURL) => {
let req = new XMLHttpRequest();
if (running) {
console.log("Starting session", deviceURL + "/api/session/start" );
//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");
//console.log("Stopping session");
req.open("GET", deviceURL + "/api/session/stop");
this.dataSource.stop();
this.dataSource.startIndex = 0;
@@ -71,55 +73,46 @@ class DataProcessing {
}
onNewData = (data) => {
data.values;
data.sessionStartTime;
data.startIndex;
let success = false;
if (data.sessionStartTime == this.sessionStartTime && data.startIndex == this.rawMeasurements.size) {
if (data.sessionStartTime == this.state.session.sessionId &&
data.startIndex == this.state.session.rawData.size) {
// normal case, add received data to measurement array
this.rawMeasurements = this.rawMeasurements.concat(List(data.values));
success = true;
const newData = this.state.session.rawData.concat(List(data.values));
const analysis = this.analyzeNewMeasurements(data.values, this.state.session.rawData);
this.store.dispatch(reportDeviceData(data.sessionStartTime, data.startIndex, newData, analysis));
}
else if (data.startIndex === 0) {
// new start
this.sessionStartTime = data.sessionStartTime;
this.rawMeasurements = List(data.values);
success = true;
this.resetAnalysis();
const newData = List(data.values);
const analysis = this.analyzeNewMeasurements(data.values, this.state.session.rawData);
this.store.dispatch(reportDeviceData(data.sessionStartTime, data.startIndex, newData, analysis));
} 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
console.log("Requery :(");
//console.log("Session times", data.sessionStartTime == this.state.session.sessionId, data.sessionStartTime, this.state.session.sessionId);
//console.log("Index ",data.startIndex == this.state.session.rawData.size, data.startIndex, this.state.session.rawData.size);
this.resetAnalysis();
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);
this.store.dispatch(resetDeviceData());
}
}
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);
analyzeNewMeasurements = (newData, oldData) => {
const newPeaks = this.peakDetectorSimple.addVector(newData);
const allPeaks = this.state.session.analysis.peaks.concat(List(newPeaks));
const allMeasurements = oldData.concat(List(newData));
const totalMomentum = allMeasurements.reduce((sum, x) => sum + x, 0);
const peakMax = allMeasurements.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 windowedSeq = allMeasurements.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,
peaks: allPeaks,
totalTime: allMeasurements.length / NUM_MEASUREMENTS_PER_SECOND,
activeTime: 0,
totalMomentum: totalMomentum,
peakFrequency: 0,
@@ -130,6 +123,6 @@ class DataProcessing {
peakMaxWindow: peakMaxWindow,
}
};
};
}
export default DataProcessing;

1
data_processing/DeviceDataSource.js
View File

@@ -45,7 +45,6 @@ class DeviceHttpDataSource {
const arrayBuffer = await this.getUrl(url);
const decoded = msgpack.decode(new Uint8Array(arrayBuffer), { codec: this.msgpackCodec });
this.startIndex += decoded["values"].length;
//"values", "sessionStartTime", "startIndex"
this.onNewData(decoded);
} catch (err) {
console.log(err);