You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by bo...@apache.org on 2021/05/12 07:48:22 UTC
[incubator-streampipes-extensions] branch dev updated (2fcf125 ->
2847262)
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a change to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git.
from 2fcf125 Merge pull request #45 from grainier/STREAMPIPES-355
new fc6a180 [STREAMPIPES-350] conflate common codebase
new 6fdfa13 [STREAMPIPES-350] add label values as valueSpecification
new 34f507e [STREAMPIPES-343] remove redundancy
new 2847262 [STREAMPIPES-350] adapt to Double support
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../connect/protocol/set/FileProtocol.java | 4 +-
.../jvm/processor/state/StateUtils.java | 64 ++++++++++++++++++++++
.../jvm/processor/state/labeler/LabelerUtils.java | 3 +-
.../buffer/StateBufferLabelerController.java | 56 +++++--------------
.../labeler/number/NumberLabelerController.java | 57 +++++--------------
5 files changed, 95 insertions(+), 89 deletions(-)
create mode 100644 streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
[incubator-streampipes-extensions] 03/04: [STREAMPIPES-343] remove
redundancy
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 34f507edc1e6c5a3e380dff05666e24c916fac12
Author: bossenti <bo...@posteo.de>
AuthorDate: Thu May 6 17:30:58 2021 +0200
[STREAMPIPES-343] remove redundancy
---
.../org/apache/streampipes/connect/protocol/set/FileProtocol.java | 4 +---
1 file changed, 1 insertion(+), 3 deletions(-)
diff --git a/streampipes-connect-adapters/src/main/java/org/apache/streampipes/connect/protocol/set/FileProtocol.java b/streampipes-connect-adapters/src/main/java/org/apache/streampipes/connect/protocol/set/FileProtocol.java
index 062ef49..1a21497 100644
--- a/streampipes-connect-adapters/src/main/java/org/apache/streampipes/connect/protocol/set/FileProtocol.java
+++ b/streampipes-connect-adapters/src/main/java/org/apache/streampipes/connect/protocol/set/FileProtocol.java
@@ -97,10 +97,8 @@ public class FileProtocol extends Protocol {
SendToPipeline stk = new SendToPipeline(format, adapterPipeline);
try {
- InputStream in = Request.Get(fileFetchUrl).execute().returnContent().asStream();;
+ InputStream in = getDataFromEndpoint();
parser.parse(in, stk);
- } catch (IOException e) {
- e.printStackTrace();
} catch (ParseException e) {
logger.error("Error while parsing: " + e.getMessage());
}
[incubator-streampipes-extensions] 02/04: [STREAMPIPES-350] add
label values as valueSpecification
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 6fdfa136ff4c797a033f1879b7e3f36c4961342e
Author: bossenti <bo...@posteo.de>
AuthorDate: Fri Apr 30 08:26:53 2021 +0200
[STREAMPIPES-350] add label values as valueSpecification
---
.../transformation/jvm/processor/state/labeler/LabelerUtils.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/LabelerUtils.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/LabelerUtils.java
index 31db2f2..eba8d13 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/LabelerUtils.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/LabelerUtils.java
@@ -14,7 +14,7 @@ import java.util.List;
public class LabelerUtils {
- public static EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, String labelName) throws SpRuntimeException {
+ public static EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, String labelName, List<String> labelStrings) throws SpRuntimeException {
List<EventProperty> properties = processingElement
.getInputStreams()
@@ -24,6 +24,7 @@ public class LabelerUtils {
properties.add(PrimitivePropertyBuilder
.create(Datatypes.String, labelName)
+ .valueSpecification(labelName, "possible label values", labelStrings)
.domainProperty(SPSensor.STATE)
.scope(PropertyScope.DIMENSION_PROPERTY)
.build());
[incubator-streampipes-extensions] 04/04: [STREAMPIPES-350] adapt
to Double support
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 2847262368831e2a1bcfec1ab4f98e336e655fc1
Author: bossenti <bo...@posteo.de>
AuthorDate: Wed May 12 09:38:53 2021 +0200
[STREAMPIPES-350] adapt to Double support
---
.../processors/transformation/jvm/processor/state/StateUtils.java | 4 ++--
.../processor/state/labeler/buffer/StateBufferLabelerController.java | 2 +-
.../jvm/processor/state/labeler/number/NumberLabelerController.java | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
index 094f206..6019395 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
@@ -26,7 +26,7 @@ public class StateUtils {
return extractor.collectionMembersAsGroup(LABEL_COLLECTION_ID);
}
- public static List<Integer> getNumberValues(ProcessingElementParameterExtractor extractor) {
+ public static List<Double> getNumberValues(ProcessingElementParameterExtractor extractor) {
return getGroupItems(extractor)
.stream()
.map(group -> (
@@ -34,7 +34,7 @@ public class StateUtils {
.extractGroupMember(NUMBER_VALUE_ID, group)
.as(FreeTextStaticProperty.class))
.getValue())
- .map(Integer::parseInt)
+ .map(Double::parseDouble)
.collect(Collectors.toList());
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
index 19ef060..de538ec 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
@@ -97,7 +97,7 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla
String labelName = getLabelName(extractor);
- List<Integer> numberValues = getNumberValues(extractor);
+ List<Double> numberValues = getNumberValues(extractor);
List<String> labelStrings = getLabelStrings(extractor);
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
index 8de4d2e..971b121 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
@@ -75,7 +75,7 @@ public class NumberLabelerController extends StandaloneEventProcessingDeclarer<N
String labelName = getLabelName(extractor);
- List<Integer> numberValues = getNumberValues(extractor);
+ List<Double> numberValues = getNumberValues(extractor);
List<String> labelStrings = getLabelStrings(extractor);
[incubator-streampipes-extensions] 01/04: [STREAMPIPES-350]
conflate common codebase
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit fc6a180e145a91fc5accc6ebecc1aae0dcc54c86
Author: bossenti <bo...@posteo.de>
AuthorDate: Fri Apr 30 08:26:34 2021 +0200
[STREAMPIPES-350] conflate common codebase
---
.../jvm/processor/state/StateUtils.java | 64 ++++++++++++++++++++++
.../buffer/StateBufferLabelerController.java | 56 +++++--------------
.../labeler/number/NumberLabelerController.java | 57 +++++--------------
3 files changed, 92 insertions(+), 85 deletions(-)
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
new file mode 100644
index 0000000..094f206
--- /dev/null
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/StateUtils.java
@@ -0,0 +1,64 @@
+package org.apache.streampipes.processors.transformation.jvm.processor.state;
+
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.OneOfStaticProperty;
+import org.apache.streampipes.model.staticproperty.Option;
+import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
+import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class StateUtils {
+
+ public static final String LABEL_NAME = "labelName";
+ public static final String LABEL_COLLECTION_ID = "labelCollectionId";
+ public static final String NUMBER_VALUE_ID = "numberValueId";
+ public static final String COMPARATOR_ID = "comparatorId";
+ public static final String LABEL_STRING_ID = "labelStringId";
+
+
+ public static String getLabelName(ProcessingElementParameterExtractor extractor) {
+ return extractor.textParameter(LABEL_NAME);
+ }
+
+ public static List<StaticPropertyGroup> getGroupItems(ProcessingElementParameterExtractor extractor) {
+ return extractor.collectionMembersAsGroup(LABEL_COLLECTION_ID);
+ }
+
+ public static List<Integer> getNumberValues(ProcessingElementParameterExtractor extractor) {
+ return getGroupItems(extractor)
+ .stream()
+ .map(group -> (
+ extractor
+ .extractGroupMember(NUMBER_VALUE_ID, group)
+ .as(FreeTextStaticProperty.class))
+ .getValue())
+ .map(Integer::parseInt)
+ .collect(Collectors.toList());
+ }
+
+ public static List<String> getLabelStrings(ProcessingElementParameterExtractor extractor) {
+ return getGroupItems(extractor)
+ .stream()
+ .map(group -> (extractor
+ .extractGroupMember(LABEL_STRING_ID, group)
+ .as(FreeTextStaticProperty.class))
+ .getValue())
+ .collect(Collectors.toList());
+ }
+
+ public static List<String> getComparators(ProcessingElementParameterExtractor extractor) {
+ return getGroupItems(extractor)
+ .stream()
+ .map(group -> (extractor
+ .extractGroupMember(COMPARATOR_ID, group)
+ .as(OneOfStaticProperty.class))
+ .getOptions()
+ .stream()
+ .filter(Option::isSelected).findFirst().get().getName())
+ .collect(Collectors.toList());
+ }
+
+
+}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
index 16c85bf..19ef060 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java
@@ -24,10 +24,6 @@ import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
-import org.apache.streampipes.model.staticproperty.OneOfStaticProperty;
-import org.apache.streampipes.model.staticproperty.Option;
-import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
import org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.LabelerUtils;
import org.apache.streampipes.sdk.StaticProperties;
import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
@@ -42,17 +38,14 @@ import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcess
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.streampipes.processors.transformation.jvm.processor.state.StateUtils.*;
+
public class StateBufferLabelerController extends StandaloneEventProcessingDeclarer<StateBufferLabelerParameters> implements ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation, ProcessingElementParameterExtractor> {
public static final String STATE_FILTER_ID = "stateFilterId";
public static final String STATE_FIELD_ID = "stateFieldId";
public static final String OPERATIONS_ID = "operationsId";
public static final String SENSOR_VALUE_ID = "sensorValueId";
- public static final String LABEL_COLLECTION_ID = "labelCollectionId";
- public static final String COMPARATOR_ID = "comparatorId";
- public static final String NUMBER_VALUE_ID = "numberValueId";
- public static final String LABEL_STRING_ID = "labelStringId";
- public static final String LABEL_NAME = "labelName";
public static final String LABEL = "label";
@@ -101,36 +94,15 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla
String stateProperty = extractor.mappingPropertyValue(STATE_FIELD_ID);
String stateFilter = extractor.singleValueParameter(STATE_FILTER_ID, String.class);
String selectedOperation = extractor.selectedSingleValue(OPERATIONS_ID, String.class);
- String labelName = extractor.textParameter(LABEL_NAME);
-
- List<StaticPropertyGroup> groupItems = extractor.collectionMembersAsGroup(LABEL_COLLECTION_ID);
- List<Double> numberValues = groupItems
- .stream()
- .map(group -> (
- extractor
- .extractGroupMember(NUMBER_VALUE_ID, group)
- .as(FreeTextStaticProperty.class))
- .getValue())
- .map(Double::parseDouble)
- .collect(Collectors.toList());
-
- List<String> labelStrings = groupItems
- .stream()
- .map(group -> (extractor
- .extractGroupMember(LABEL_STRING_ID, group)
- .as(FreeTextStaticProperty.class))
- .getValue())
- .collect(Collectors.toList());
-
- List<String> comparators = groupItems
- .stream()
- .map(group -> (extractor
- .extractGroupMember(COMPARATOR_ID, group)
- .as(OneOfStaticProperty.class))
- .getOptions()
- .stream()
- .filter(Option::isSelected).findFirst().get().getName())
- .collect(Collectors.toList());
+
+ String labelName = getLabelName(extractor);
+
+ List<Integer> numberValues = getNumberValues(extractor);
+
+ List<String> labelStrings = getLabelStrings(extractor);
+
+ List<String> comparators = getComparators(extractor);
+
StateBufferLabelerParameters params = new StateBufferLabelerParameters(graph, sensorListValueProperty, stateProperty, stateFilter, selectedOperation, labelName, numberValues, labelStrings, comparators);
return new ConfiguredEventProcessor<>(params, StateBufferLabeler::new);
@@ -139,8 +111,10 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla
@Override
public EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, ProcessingElementParameterExtractor parameterExtractor) throws SpRuntimeException {
- String labelName = parameterExtractor.textParameter(LABEL_NAME);
+ String labelName = getLabelName(parameterExtractor);
+
+ List<String> labelStrings = getLabelStrings(parameterExtractor);
- return LabelerUtils.resolveOutputStrategy(processingElement, labelName);
+ return LabelerUtils.resolveOutputStrategy(processingElement, labelName, labelStrings);
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
index 63bbba3..8de4d2e 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/number/NumberLabelerController.java
@@ -22,34 +22,25 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.container.api.ResolvesContainerProvidedOutputStrategy;
import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.model.staticproperty.*;
import org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.LabelerUtils;
import org.apache.streampipes.sdk.StaticProperties;
-import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.sdk.helpers.*;
import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.sdk.utils.Datatypes;
-import org.apache.streampipes.vocabulary.SPSensor;
import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
import java.util.List;
-import java.util.stream.Collectors;
+
+import static org.apache.streampipes.processors.transformation.jvm.processor.state.StateUtils.*;
public class NumberLabelerController extends StandaloneEventProcessingDeclarer<NumberLabelerParameters> implements ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation, ProcessingElementParameterExtractor> {
public static final String SENSOR_VALUE_ID = "sensorValueId";
- public static final String LABEL_COLLECTION_ID = "labelCollectionId";
- public static final String COMPARATOR_ID = "comparatorId";
- public static final String NUMBER_VALUE_ID = "numberValueId";
- public static final String LABEL_STRING_ID = "labelStringId";
- public static final String LABEL_NAME = "labelName";
@Override
public DataProcessorDescription declareModel() {
@@ -82,37 +73,13 @@ public class NumberLabelerController extends StandaloneEventProcessingDeclarer<N
String sensorListValueProperty = extractor.mappingPropertyValue(SENSOR_VALUE_ID);
- String labelName = extractor.textParameter(LABEL_NAME);
-
- List<StaticPropertyGroup> groupItems = extractor.collectionMembersAsGroup(LABEL_COLLECTION_ID);
-
- List<Double> numberValues = groupItems
- .stream()
- .map(group -> (
- extractor
- .extractGroupMember(NUMBER_VALUE_ID, group)
- .as(FreeTextStaticProperty.class))
- .getValue())
- .map(Double::parseDouble)
- .collect(Collectors.toList());
-
- List<String> labelStrings = groupItems
- .stream()
- .map(group -> (extractor
- .extractGroupMember(LABEL_STRING_ID, group)
- .as(FreeTextStaticProperty.class))
- .getValue())
- .collect(Collectors.toList());
-
- List<String> comparators = groupItems
- .stream()
- .map(group -> (extractor
- .extractGroupMember(COMPARATOR_ID, group)
- .as(OneOfStaticProperty.class))
- .getOptions()
- .stream()
- .filter(Option::isSelected).findFirst().get().getName())
- .collect(Collectors.toList());
+ String labelName = getLabelName(extractor);
+
+ List<Integer> numberValues = getNumberValues(extractor);
+
+ List<String> labelStrings = getLabelStrings(extractor);
+
+ List<String> comparators = getComparators(extractor);
NumberLabelerParameters params = new NumberLabelerParameters(graph, sensorListValueProperty, labelName, numberValues, labelStrings, comparators);
@@ -123,8 +90,10 @@ public class NumberLabelerController extends StandaloneEventProcessingDeclarer<N
@Override
public EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, ProcessingElementParameterExtractor parameterExtractor) throws SpRuntimeException {
- String labelName = parameterExtractor.textParameter(LABEL_NAME);
+ String labelName = getLabelName(parameterExtractor);
+
+ List<String> labelStrings = getLabelStrings(parameterExtractor);
- return LabelerUtils.resolveOutputStrategy(processingElement, labelName);
+ return LabelerUtils.resolveOutputStrategy(processingElement, labelName, labelStrings);
}
}