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 <[email protected]> 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); } }
