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:23 UTC
[incubator-streampipes-extensions] 01/04: [STREAMPIPES-350]
conflate common codebase
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);
}
}