Stream Processor

mCerebrum is a configurable smartphone software platform for mobile and wearable sensors. It provides support for reliable data collection from mobile and wearable sensors, and offers real-time processing of these data.

Stream Processor is a library and data processing tool that contains online implementations of algorithms designed to run on mCerebrum. This codebase also be run in a standalone fashion on most computing platforms.

It contains implementations of the following algorithms:

Algorithms

cStress: A continuous stress assessment algorithm

You can find more information about MD2K software on our software website or the MD2K organization on our MD2K website.

Install

Clone repository git clone https://github.com/MD2Korg/stream-processor

or import into Intellij IDEA through New->Project from Version Control->Github

Use this url https://github.com/MD2Korg/stream-processor
Check Use auto-import
Check Create directories ofr empyt content roots automatically
Choose Use gralde wrapper task configuration
Specify a Gradle JVM (jdk>=1.7)
Wait for Gradle to resolve dependencies and build project
Define the Project SDK and add the same JDK you are utilizing from the previous step
Open Edit Configurations
Add Application
- Main class: CC_Main or Main
- Program Arguments:
  - Directory to Cerebral Cortex data files
  - Path to cStressModelV5.json
  - Path to cStressModelRIPv4.json
  - Path to model_puffmarker.json
- Specify classpath of module: streamm-processor_main

Examples

Import data for replay through Stream Processor

CSVParser tp = new CSVParser();
tp.importData(path + "/rip.txt", AUTOSENSE.CHEST_RIP);
tp.importData(path + "/ecg.txt", AUTOSENSE.CHEST_ECG);
tp.importData(path + "/accelx.txt", AUTOSENSE.CHEST_ACCEL_X);
tp.importData(path + "/accely.txt", AUTOSENSE.CHEST_ACCEL_Y);
tp.importData(path + "/accelz.txt", AUTOSENSE.CHEST_ACCEL_Z);

tp.sort();

Setup the Stream Processor object with a 60 second window, define a path for exporting data streams, and load the cStress model file.

int windowSize = 60000;

StreamProcessor streamProcessor = new StreamProcessor(windowSize);
streamProcessor.setPath(path);
streamProcessor.loadModel(cStressModelPath);

Define datapoint callbacks for DataPoint and DataPointArray.

streamProcessor.dpInterface = new DataPointInterface() {
    @Override
    public void dataPointHandler(String stream, DataPoint dp) {
        System.out.println(path + "/" + stream + " " + dp);
    }

    @Override
    public void dataPointArrayHandler(String stream, DataPointArray dp) {
        System.out.println(path + "/" + stream + " " + dp);
    }
};

streamProcessor.registerCallbackDataArrayStream(StreamConstants.ORG_MD2K_CSTRESS_FV);
streamProcessor.registerCallbackDataStream(StreamConstants.ORG_MD2K_CSTRESS_DATA_ACCEL_ACTIVITY);
streamProcessor.registerCallbackDataStream(StreamConstants.ORG_MD2K_CSTRESS_PROBABILITY);
streamProcessor.registerCallbackDataStream(StreamConstants.ORG_MD2K_CSTRESS_STRESSLABEL);

Replay logic for sending data through Stream Processor and processing windows streamProcessor.go().

    long windowStartTime = -1;
    long st = -1;
    int count = 0;
    for (CSVDataPoint ap : tp) {
        DataPoint dp = new DataPoint(ap.timestamp, ap.value);

        if (windowStartTime < 0) {
            windowStartTime = Time.nextEpochTimestamp(dp.timestamp, windowSize);
            st = System.currentTimeMillis();
        }

        if ((dp.timestamp - windowStartTime) >= windowSize) { //Process the buffer every windowSize milliseconds
            streamProcessor.go();
            windowStartTime += windowSize;
        }

        streamProcessor.add(ap.channel, dp);

    }

References

UbiComp 2015 cStress: Towards a Gold Standard for Continuous Stress Assessment in the Mobile Environment Karen Hovsepian, Mustafa al'absi, Emre Ertin, Thomas Kamarck, Motoshiro Nakajima, Santosh Kumar pdf

Contributing

Please read our Contributing Guidelines for details on the process for submitting pull requests to us.

We use the Google Java Style Guide.

Our Code of Conduct is the Contributor Covenant.

Bug reports can be submitted through JIRA.

Our discussion forum can be found here.

Versioning

We use Semantic Versioning for versioning the software which is based on the following guidelines.

MAJOR.MINOR.PATCH (example: 3.0.12)

MAJOR version when incompatible API changes are made,
MINOR version when functionality is added in a backwards-compatible manner, and
PATCH version when backwards-compatible bug fixes are introduced.

For the versions available, see this repository's tags.

Contributors

Link to the list of contributors who participated in this project.

License

This project is licensed under the BSD 2-Clause - see the license file for details.

Acknowledgments

National Institutes of Health - Big Data to Knowledge Initiative
- Grants: R01MD010362, 1UG1DA04030901, 1U54EB020404, 1R01CA190329, 1R01DE02524, R00MD010468, 3UH2DA041713, 10555SC
National Science Foundation
- Grants: 1640813, 1722646
Intelligence Advanced Research Projects Activity
- Contract: 2017-17042800006

md2korg / stream-processor Goto Github PK