Feature/cep r peak

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.classpath

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="impl/java/bin" path="impl/java/src"/>
+	<classpathentry exported="true" kind="con" path="com.ibm.streams.java/com.ibm.streams.operator"/>
+	<classpathentry exported="true" kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
+	<classpathentry kind="src" path=".apt_generated">
+		<attributes>
+			<attribute name="optional" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="var" path="STREAMS_JAVA_FUNCTIONAL_API"/>
+	<classpathentry kind="output" path="impl/java/bin"/>
+</classpath>

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.project

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>com.ibm.streamsx.health.analyze.rpeak.cep</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.xtext.ui.shared.xtextBuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>com.ibm.streams.studio.splproject.builder.SPLProjectBuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.xtext.ui.shared.xtextNature</nature>
+		<nature>com.ibm.streams.studio.splproject.SPLProjectNature</nature>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+	</natures>
+</projectDescription>

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.settings/application.RPeakDetectCep-BuildConfig.splbuild

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
+<properties>
+<comment>SPL Build Configuration: BuildConfig</comment>
+<entry key="com.ibm.streams.studio.splproject:STATIC_LINKING">F</entry>
+<entry key="com.ibm.streams.studio.splproject:OPTIMIZED_CODE">T</entry>
+<entry key="com.ibm.streams.studio.splproject:FUSION_MODE">FDEF</entry>
+<entry key="com.ibm.streams.studio.splproject:MAIN_COMPOSITE">application::RPeakDetectCep</entry>
+<entry key="com.ibm.streams.studio.splproject:NAME">BuildConfig</entry>
+<entry key="com.ibm.streams.studio.splproject:GCC_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:SC_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:ACTIVE">T</entry>
+<entry key="com.ibm.streams.studio.splproject:LD_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:PRE_BUILD_CMD"/>
+<entry key="com.ibm.streams.studio.splproject:DATA_DIR"/>
+<entry key="com.ibm.streams.studio.splproject:POST_BUILD_CMD"/>
+<entry key="com.ibm.streams.studio.splproject:REPLACE_ENVIRONMENT">F</entry>
+<entry key="com.ibm.streams.studio.splproject:USE_PROJECT_DATA_DIR">T</entry>
+<entry key="com.ibm.streams.studio.splproject:PROFILING"/>
+<entry key="com.ibm.streams.studio.splproject:SDB">F</entry>
+<entry key="com.ibm.streams.studio.splproject:DEFAULT_POOL_SIZE"/>
+<entry key="com.ibm.streams.studio.splproject:OUTPUT_DIR">BuildConfig</entry>
+<entry key="com.ibm.streams.studio.splproject:SAMPLING_RATE"/>
+<entry key="com.ibm.streams.studio.splproject:STANDALONE">F</entry>
+</properties>

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.settings/com.ibm.streamsx.health.analyze.rpeak.cep.service.RPeakDetectCep-BuildConfig.splbuild

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
+<properties>
+<comment>SPL Build Configuration: BuildConfig</comment>
+<entry key="com.ibm.streams.studio.splproject:STATIC_LINKING">F</entry>
+<entry key="com.ibm.streams.studio.splproject:OPTIMIZED_CODE">T</entry>
+<entry key="com.ibm.streams.studio.splproject:FUSION_MODE">FDEF</entry>
+<entry key="com.ibm.streams.studio.splproject:MAIN_COMPOSITE">com.ibm.streamsx.health.analyze.rpeak.cep.service::RPeakDetectCep</entry>
+<entry key="com.ibm.streams.studio.splproject:NAME">BuildConfig</entry>
+<entry key="com.ibm.streams.studio.splproject:GCC_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:SC_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:ACTIVE">T</entry>
+<entry key="com.ibm.streams.studio.splproject:LD_OPTS"/>
+<entry key="com.ibm.streams.studio.splproject:PRE_BUILD_CMD"/>
+<entry key="com.ibm.streams.studio.splproject:DATA_DIR"/>
+<entry key="com.ibm.streams.studio.splproject:POST_BUILD_CMD"/>
+<entry key="com.ibm.streams.studio.splproject:REPLACE_ENVIRONMENT">F</entry>
+<entry key="com.ibm.streams.studio.splproject:USE_PROJECT_DATA_DIR">T</entry>
+<entry key="com.ibm.streams.studio.splproject:PROFILING"/>
+<entry key="com.ibm.streams.studio.splproject:SDB">F</entry>
+<entry key="com.ibm.streams.studio.splproject:DEFAULT_POOL_SIZE"/>
+<entry key="com.ibm.streams.studio.splproject:OUTPUT_DIR">BuildConfig</entry>
+<entry key="com.ibm.streams.studio.splproject:SAMPLING_RATE"/>
+<entry key="com.ibm.streams.studio.splproject:STANDALONE">F</entry>
+</properties>

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.settings/org.eclipse.core.runtime.prefs

@@ -0,0 +1,2 @@
+eclipse.preferences.version=1
+line.separator=\n

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.settings/org.eclipse.jdt.apt.core.prefs

@@ -0,0 +1,3 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.apt.aptEnabled=true
+org.eclipse.jdt.apt.reconcileEnabled=false

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/.settings/org.eclipse.jdt.core.prefs

@@ -0,0 +1,2 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.processAnnotations=enabled

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/build.gradle

@@ -0,0 +1,95 @@
+apply plugin: 'java'
+apply plugin: 'eclipse'
+apply from: "$rootProject.projectDir/common.gradle"
+
+sourceSets {
+    main {
+        java {
+            srcDirs = ['impl/java/src']
+        }
+        resources {
+            srcDirs = ['etc']
+        }
+    }
+}
+
+ext {
+	STREAMS_INSTALL = System.getenv('STREAMS_INSTALL')
+}
+
+
+repositories {
+    mavenCentral()
+
+   flatDir {
+       dirs STREAMS_INSTALL+'/lib',STREAMS_INSTALL+'/ext/lib',STREAMS_INSTALL+'/toolkits/com.ibm.streamsx.topology/lib'
+   }
+}
+
+
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
+
+dependencies {
+	// compile any project that this service may need
+	compile project(':ingest:common:com.ibm.streamsx.health.ingest')
+
+	// Streams dependencies
+	compile name: 'com.ibm.streams.operator'
+	compile name: 'com.ibm.streams.operator.samples'
+	compile name: 'com.ibm.streamsx.topology'
+	compile name: 'JSON4J'
+	compile name: 'commons-math-2.2'	
+}
+
+task buildToolkit {
+    doLast {
+        splMakeToolkit()
+    }
+}
+
+build.doLast {
+	tasks.buildToolkit.execute()
+}
+
+task copyLib(type: Copy) {
+    into "$projectDir/release/opt/lib"
+    from configurations.runtime
+}
+
+task copyJar(type: Copy) {
+    into "$projectDir/release"
+    from "$projectDir/build/libs/"
+    include '**/*.jar'
+}
+
+task rmRelease(type: Delete) {
+	delete 'release'
+}
+
+task copyLibDev(type: Copy) {
+    into "$projectDir/opt/lib"
+    from configurations.runtime
+}
+
+task deps << {
+	tasks.copyLibDev.execute()
+	tasks.copyLib.execute()
+}
+
+
+jar.doLast {
+    tasks.copyLib.execute()
+    tasks.copyJar.execute()
+}
+
+clean.doLast{
+	tasks.rmRelease.execute()
+}
+
+task execute(type: JavaExec) {
+    main='com.ibm.streamsx.health.analyze.rpeak.cep.service.RPeakDetectCepService'
+    classpath = configurations.runtime + files("$buildDir/libs/com.ibm.streamsx.health.analyze.rpeak.cep.jar")
+}
+
+

analyze/ecg/com.ibm.streamsx.health.analyze.rpeak.cep/com.ibm.streamsx.health.analyze.rpeak.cep.service/RPeakDetectCep.spl

@@ -0,0 +1,111 @@
+//*******************************************************************************
+//* Copyright (C) 2017 International Business Machines Corporation
+//* All Rights Reserved
+//*******************************************************************************
+namespace com.ibm.streamsx.health.analyze.rpeak.cep.service ;
+
+use com.ibm.streamsx.health.ingest.connector::IngestSubscribe ;
+use com.ibm.streamsx.health.ingest.types::* ;
+use com.ibm.streams.cep::MatchRegex ;
+use com.ibm.streamsx.topology.topic::Publish ;
+use com.ibm.streamsx.json::TupleToJSON ;
+
+
+type PeakEvent = rstring patientId, int64 ts, float64 data, float64 max,
+	float64 min, int32 count, list<float64> eventData, list<int64> eventTs ;
+type RPeakEvent = rstring patientId, int64 ts, float64 data ;
+type RREvent = rstring patientId, int64 rr, list<RPeakEvent> events ;
+
+composite RPeakDetectCep()
+{
+	param
+		expression<rstring> $subTopic : getSubmissionTimeValue("topic") ;
+		expression<rstring> $pubTopicRPeak : "analyze/rpeak/cep/r" ;
+		expression<rstring> $pubTopicRR : "analyze/rpeak/cep/rr" ;
+		expression<rstring> $readingCode : getSubmissionTimeValue("readingCode") ;
+		expression<float64> $peakThreshold :(float64)
+			getSubmissionTimeValue("peakThreshold", "0.8") ;
+	graph
+		(stream<Observation> SubscribeToObservations) =
+			IngestSubscribe()
+		{
+			param
+				topic : $subTopic ;
+		}
+
+		(stream<PeakEvent> PeakDetectEvents) =
+			MatchRegex(ECGSignal as I0)
+		{
+			param
+				pattern : ". rise+ drop drop drop" ;
+				predicates : { rise = data >= Last(data), drop = data < Last(data) } ;
+				partitionBy : I0.patientId ;
+			output
+				PeakDetectEvents : max = Max(data), min = Min(data), count = Count(),
+					eventData = Collect(data), eventTs = Collect(ts) ;
+		}
+
+		(stream<rstring patientId, int64 ts, float64 data> ECGSignal) = Functor(SubscribeToObservations as I0)
+		{
+			param
+				filter : I0.reading.readingType.code == $readingCode ;
+			output
+				ECGSignal : patientId = I0.patientId, ts = I0.reading.ts, data =
+					I0.reading.value ;
+		}
+
+		() as PublishR = Publish(RPeakToJson as inPort0Alias)
+		{
+			param
+				topic : $pubTopicRPeak ;
+		}
+
+		(stream<rstring jsonString> RPeakToJson)=
+			TupleToJSON(RPeakEvents)
+		{
+		}
+
+		(stream<RPeakEvent> RPeakEvents) = Custom(PeakDetectEvents as
+			I0)
+		{
+			logic
+				onTuple I0 : if(max - eventData [ 0 ] > $peakThreshold)
+				{
+					submit({ patientId = patientId, ts = eventTs [ size(eventTs) - 4 ], data =
+						max }, RPeakEvents) ;
+				}
+
+		}
+
+		(stream<list<RPeakEvent> events> RPeakEventList)  =
+			Aggregate(RPeakEvents as I0)
+		{
+			window
+				I0 : sliding, count(2), count(1), partitioned ;
+			param
+				partitionBy : I0.patientId ;
+			output
+				RPeakEventList : events = Collect(I0) ;
+		}
+
+		() as PublishRR = Publish(RRToJson)
+		{
+			param
+				topic : $pubTopicRR ;
+		}
+
+		(stream<rstring jsonString> RRToJson) =
+			TupleToJSON(RREvents as inPort0Alias)
+		{
+		}
+
+		(stream<RREvent> RREvents)  =
+			Functor(RPeakEventList as I0)
+		{
+			output
+				RREvents: patientId = I0.events[0].patientId, rr=I0.events[1].ts-I0.events[0].ts;
+
+		}
+
+}
+