feat(#3172): Implement processing element to enrich control limits (#…

…3175)

streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/EnricherExtensionModuleExport.java

@@ -23,6 +23,7 @@
 import org.apache.streampipes.extensions.api.migration.IModelMigrator;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement;
 import org.apache.streampipes.processors.enricher.jvm.processor.jseval.JSEvalProcessor;
+import org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment.QualityControlLimitsEnrichment;
 import org.apache.streampipes.processors.enricher.jvm.processor.math.MathOpProcessor;
 import org.apache.streampipes.processors.enricher.jvm.processor.math.staticmathop.StaticMathOpProcessor;
 import org.apache.streampipes.processors.enricher.jvm.processor.trigonometry.TrigonometryProcessor;
@@ -41,6 +42,7 @@ public List<StreamPipesAdapter> adapters() {
   public List<IStreamPipesPipelineElement<?>> pipelineElements() {
     return List.of(
         new JSEvalProcessor(),
+        new QualityControlLimitsEnrichment(),
         new MathOpProcessor(),
         new StaticMathOpProcessor(),
         new TrigonometryProcessor(),

streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/limitsenrichment/QualityControlLimitsEnrichment.java

@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.extensions.api.pe.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpProperties;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.vocabulary.SO;
+import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+public class QualityControlLimitsEnrichment extends StreamPipesDataProcessor {
+
+
+  protected static final String UPPER_CONTROL_LIMIT_LABEL = "upperControlLimitInput";
+  protected static final String UPPER_WARNING_LIMIT_LABEL = "upperWarningLimitInput";
+  protected static final String LOWER_WARNING_LIMIT_LABEL = "lowerWarningLimitInput";
+  protected static final String LOWER_CONTROL_LIMIT_LABEL = "lowerControlLimitInput";
+
+  protected static final String UPPER_CONTROL_LIMIT = "upperControlLimit";
+  protected static final String UPPER_WARNING_LIMIT = "upperWarningLimit";
+  protected static final String LOWER_WARNING_LIMIT = "lowerWarningLimit";
+  protected static final String LOWER_CONTROL_LIMIT = "lowerControlLimit";
+
+  private double upperControlLimitValue;
+  private double upperWarningLimitValue;
+  private double lowerWarningLimitValue;
+  private double lowerControlLimitValue;
+
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder
+        .create("org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment", 0)
+        .category(DataProcessorType.ENRICH)
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
+        .withLocales(Locales.EN)
+        .requiredStream(StreamRequirementsBuilder
+                            .create()
+                            .requiredProperty(EpRequirements.anyProperty())
+                            .build())
+        .requiredFloatParameter(Labels.withId(UPPER_CONTROL_LIMIT_LABEL))
+        .requiredFloatParameter(Labels.withId(UPPER_WARNING_LIMIT_LABEL))
+        .requiredFloatParameter(Labels.withId(LOWER_WARNING_LIMIT_LABEL))
+        .requiredFloatParameter(Labels.withId(LOWER_CONTROL_LIMIT_LABEL))
+        .outputStrategy(
+            OutputStrategies.append(
+                EpProperties.doubleEp(Labels.empty(), UPPER_CONTROL_LIMIT, SO.NUMBER),
+                EpProperties.doubleEp(Labels.empty(), UPPER_WARNING_LIMIT, SO.NUMBER),
+                EpProperties.doubleEp(Labels.empty(), LOWER_WARNING_LIMIT, SO.NUMBER),
+                EpProperties.doubleEp(Labels.empty(), LOWER_CONTROL_LIMIT, SO.NUMBER)
+            ))
+        .build();
+  }
+
+  @Override
+  public void onInvocation(
+      ProcessorParams parameters,
+      SpOutputCollector spOutputCollector,
+      EventProcessorRuntimeContext runtimeContext
+  ) throws SpRuntimeException {
+    this.upperControlLimitValue = parameters.extractor()
+                                            .singleValueParameter(UPPER_CONTROL_LIMIT_LABEL, Double.class);
+    this.upperWarningLimitValue = parameters.extractor()
+                                            .singleValueParameter(UPPER_WARNING_LIMIT_LABEL, Double.class);
+    this.lowerWarningLimitValue = parameters.extractor()
+                                            .singleValueParameter(LOWER_WARNING_LIMIT_LABEL, Double.class);
+    this.lowerControlLimitValue = parameters.extractor()
+                                            .singleValueParameter(LOWER_CONTROL_LIMIT_LABEL, Double.class);
+  }
+
+  @Override
+  public void onDetach() {
+
+  }
+
+  @Override
+  public void onEvent(Event event, SpOutputCollector collector) {
+    event.addField(UPPER_CONTROL_LIMIT, upperControlLimitValue);
+    event.addField(UPPER_WARNING_LIMIT, upperWarningLimitValue);
+    event.addField(LOWER_WARNING_LIMIT, lowerWarningLimitValue);
+    event.addField(LOWER_CONTROL_LIMIT, lowerControlLimitValue);
+
+    collector.collect(event);
+  }
+}

streampipes-extensions/streampipes-processors-enricher-jvm/src/main/resources/org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment/documentation.md

@@ -0,0 +1,98 @@
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+## Quality Control Limits Enrichment
+
+<p align="center"> 
+    <img src="icon.png" width="150px;" class="pe-image-documentation"/>
+</p>
+
+***
+
+## Description
+
+The Quality Control Limits Enrichment processor appends user-defined control and warning limits to incoming events.
+These limits can be used in quality control charts to monitor sensor values.
+
+***
+
+## Required Input
+
+This processor works with any event stream. It adds predefined limit values to the events, which are later used for
+quality control purposes.
+
+***
+
+## Configuration
+
+#### Upper Control Limit
+
+Specify the upper control limit for the quality control process. This value defines the maximum threshold for acceptable
+process behavior.
+
+#### Upper Warning Limit
+
+Specify the upper warning limit for the quality control process. This value indicates when the process is approaching
+the upper control limit.
+
+#### Lower Warning Limit
+
+Specify the lower warning limit for the quality control process. This value indicates when the process is approaching
+the lower control limit.
+
+#### Lower Control Limit
+
+Specify the lower control limit for the quality control process. This value defines the minimum threshold for acceptable
+process behavior.
+
+***
+
+## Output
+
+The processor appends the specified control and warning limits to each input event. These enriched events can be used in
+downstream processing to create quality control charts or other monitoring tools.
+
+***
+
+## Example
+
+### User Configuration
+- **Upper Control Limit**: `80.0`
+- **Upper Warning Limit**: `70.0`
+- **Lower Warning Limit**: `30.0`
+- **Lower Control Limit**: `20.0`
+
+### Input Event
+```
+{
+  "timestamp": 1627891234000,
+  "temperature": 65.0
+}
+```
+
+### Output Event
+```
+{
+  "timestamp": 1627891234000,
+  "temperature": 65.0,
+  "upperControlLimit": 80.0,
+  "upperWarningLimit": 70.0,
+  "lowerWarningLimit": 30.0,
+  "lowerControlLimit": 20.0
+}
+```

streampipes-extensions/streampipes-processors-enricher-jvm/src/main/resources/org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment/strings.en

@@ -0,0 +1,33 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment.title=Quality Control Limits Enrichment
+org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment.description=Appends user-defined control and warning limits to incoming events for use in quality control charts.
+
+upperControlLimitInput.title=Upper Control Limit
+upperControlLimitInput.description=Specify the upper control limit for the quality control process.
+
+upperWarningLimitInput.title=Upper Warning Limit
+upperWarningLimitInput.description=Specify the upper warning limit for the quality control process.
+
+lowerWarningLimitInput.title=Lower Warning Limit
+lowerWarningLimitInput.description=Specify the lower warning limit for the quality control process.
+
+lowerControlLimitInput.title=Lower Control Limit
+lowerControlLimitInput.description=Specify the lower control limit for the quality control process.
+

streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/limitsenrichment/QualityControlLimitsEnrichmentTest.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.processors.enricher.jvm.processor.limitsenrichment;
+
+import org.apache.streampipes.test.executors.PrefixStrategy;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+class QualityControlLimitsEnrichmentTest {
+
+  @Test
+  void onEvent() {
+
+    var processor = new QualityControlLimitsEnrichment();
+
+    List<Map<String, Object>> inputEvents = List.of(Map.of(
+        "temperature", 10.1
+    ));
+
+    List<Map<String, Object>> outputEvents = List.of(Map.of(
+        "temperature", 10.1,
+        QualityControlLimitsEnrichment.UPPER_CONTROL_LIMIT, 80.0,
+        QualityControlLimitsEnrichment.UPPER_WARNING_LIMIT, 70.0,
+        QualityControlLimitsEnrichment.LOWER_CONTROL_LIMIT, 30.0,
+        QualityControlLimitsEnrichment.LOWER_WARNING_LIMIT, 20.0
+    ));
+
+    var configuration = TestConfiguration
+        .builder()
+        .config(QualityControlLimitsEnrichment.UPPER_CONTROL_LIMIT_LABEL, 80.0)
+        .config(QualityControlLimitsEnrichment.UPPER_WARNING_LIMIT_LABEL, 70.0)
+        .config(QualityControlLimitsEnrichment.LOWER_CONTROL_LIMIT_LABEL, 30.0)
+        .config(QualityControlLimitsEnrichment.LOWER_WARNING_LIMIT_LABEL, 20.0)
+        .prefixStrategy(PrefixStrategy.SAME_PREFIX)
+        .build();
+
+    var testExecutor = new ProcessingElementTestExecutor(processor, configuration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
+}