Repository: apache/streampipes
Pull request #2232 (2023-11-28 13:58:48)

fix: retrieve release version for Python from GitHub tag
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/.github/workflows/pypi-deployment.yml b/.github/workflows/pypi-deployment.yml
index 948d837e0f..931dc683db 100644
--- a/.github/workflows/pypi-deployment.yml
+++ b/.github/workflows/pypi-deployment.yml
@@ -47,6 +47,7 @@ jobs:
       - name: Build package
         working-directory: ./streampipes-client-python
         run: |
+          poetry version $(echo ${{ github.ref_name }} | awk -F'/' '{print $2}')  # extract release version from tag name
           poetry build
           poetry run twine check --strict dist/*
 
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/SwingingDoorTrendingFilterProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/SwingingDoorTrendingFilterProcessor.java
index e13e52d4bb..5f2c3322dc 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/SwingingDoorTrendingFilterProcessor.java
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/SwingingDoorTrendingFilterProcessor.java
@@ -37,11 +37,11 @@ import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
 
 public class SwingingDoorTrendingFilterProcessor extends StreamPipesDataProcessor {
 
-  private static final String SDT_TIMESTAMP_FIELD_KEY = "sdt-timestamp-field";
-  private static final String SDT_VALUE_FIELD_KEY = "sdt-value-field";
-  private static final String SDT_COMPRESSION_DEVIATION_KEY = "sdt-compression-deviation";
-  private static final String SDT_COMPRESSION_MIN_INTERVAL_KEY = "sdt-compression-min-interval";
-  private static final String SDT_COMPRESSION_MAX_INTERVAL_KEY = "sdt-compression-max-interval";
+  public static final String SDT_TIMESTAMP_FIELD_KEY = "sdt-timestamp-field";
+  public static final String SDT_VALUE_FIELD_KEY = "sdt-value-field";
+  public static final String SDT_COMPRESSION_DEVIATION_KEY = "sdt-compression-deviation";
+  public static final String SDT_COMPRESSION_MIN_INTERVAL_KEY = "sdt-compression-min-interval";
+  public static final String SDT_COMPRESSION_MAX_INTERVAL_KEY = "sdt-compression-max-interval";
 
   private String sdtTimestampField;
   private String sdtValueField;


Test patch:
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java
new file mode 100644
index 0000000000..db2f93f783
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/sdt/TestSwingingDoorTrendingFilterProcessor.java
@@ -0,0 +1,182 @@
+/*
+ * 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.filters.jvm.processor.sdt;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.processors.filters.jvm.processor.merge.TestMergeByTimeProcessor;
+import org.apache.streampipes.test.extensions.api.StoreEventCollector;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class TestSwingingDoorTrendingFilterProcessor {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestMergeByTimeProcessor.class);
+
+  @org.junit.runners.Parameterized.Parameter
+  public String testName;
+  @org.junit.runners.Parameterized.Parameter(1)
+  public String sdtCompressionDeviation;
+  @org.junit.runners.Parameterized.Parameter(2)
+  public String sdtCompressionMinTimeInterval;
+  @org.junit.runners.Parameterized.Parameter(3)
+  public String sdtCompressionMaxTimeInterval;
+  @org.junit.runners.Parameterized.Parameter(4)
+  public int expectedFilteredCount;
+  @org.junit.runners.Parameterized.Parameter(5)
+  public List<Pair<Long, Double>> eventSettings;
+  @org.junit.runners.Parameterized.Parameter(6)
+  public boolean expectException;
+  @org.junit.runners.Parameterized.Parameter(7)
+  public String expectedErrorMessage;
+
+  private final String sdtTimestampField = "sdtTimestampField";
+  private final String sdtValueField = "sdtValueField";
+
+  @org.junit.runners.Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+
+    return Arrays.asList(new Object[][]{
+            {"testWithOneEvent", "10.0", "100", "500", 1, List.of(Pair.of(9, 50.0)), false, ""},
+            {"testFullFilter", "10.0", "100", "500", 4, List.of(
+                            Pair.of(0, 50.0),      //true
+                            Pair.of(50, 50.0),     //false
+                            Pair.of(200, 100.0),   //false
+                            Pair.of(270, 140.0),   //false
+                            Pair.of(300, 250.0),   //true
+                            Pair.of(900, 500.0),   //true
+                            Pair.of(1100, 800.0),  //false
+                            Pair.of(1250, 1600.0)  //true
+            )
+                    , false, ""},
+            {"testWithNegativeCompressionDeviation", "-10.0", "100", "500", 1, new ArrayList<>(), true
+                    , "Compression Deviation should be positive!"},
+            {"testWithNegativeMinInterval", "10.0", "-100", "500", 1, new ArrayList<>(), true
+                    , "Compression Minimum Time Interval should be >= 0!"},
+            {"testWithMinInterval>MaxInterval", "10.0", "1000", "500", 1, new ArrayList<>(), true
+                    , "Compression Minimum Time Interval should be < Compression Maximum Time Interval!"}
+    });
+  }
+
+  @Rule
+  public ExpectedException exceptionRule = ExpectedException.none();
+
+  @Test
+  public void testSdtFilter() {
+    LOG.info("Executing test: {}", testName);
+    SwingingDoorTrendingFilterProcessor processor = new SwingingDoorTrendingFilterProcessor();
+    DataProcessorDescription originalGraph = processor.declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+            InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    ProcessorParams params = new ProcessorParams(graph);
+    params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_TIMESTAMP_FIELD_KEY
+                    , MappingPropertyUnary.class)
+            .setSelectedProperty("test::" + sdtTimestampField);
+    params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_VALUE_FIELD_KEY
+                    , MappingPropertyUnary.class)
+            .setSelectedProperty("test::" + sdtValueField);
+    params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_DEVIATION_KEY
+                    , FreeTextStaticProperty.class)
+            .setValue(sdtCompressionDeviation);
+    params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MIN_INTERVAL_KEY
+                    , FreeTextStaticProperty.class)
+            .setValue(sdtCompressionMinTimeInterval);
+    params.extractor().getStaticPropertyByName(SwingingDoorTrendingFilterProcessor.SDT_COMPRESSION_MAX_INTERVAL_KEY
+                    , FreeTextStaticProperty.class)
+            .setValue(sdtCompressionMaxTimeInterval);
+
+    if (expectException){
+      LOG.info("Expecting Error Message: {}", expectedErrorMessage);
+      exceptionRule.expect(SpRuntimeException.class);
+      exceptionRule.expectMessage(expectedErrorMessage);
+    }
+
+    StoreEventCollector eventCollector = new StoreEventCollector();
+    processor.onInvocation(params, eventCollector, null);
+
+    int result = sendEvents(processor, eventCollector);
+
+    LOG.info("Expected SDT filtered count is: {}", expectedFilteredCount);
+    LOG.info("Actual SDT filtered count is: {}", result);
+    assertEquals(expectedFilteredCount, result);
+  }
+
+
+  private int sendEvents(SwingingDoorTrendingFilterProcessor processor, StoreEventCollector collector) {
+    List<Event> events = makeEvents();
+    for (Event event : events) {
+      LOG.info("Sending event with timestamp: "
+              + event.getFieldBySelector("test::" + sdtTimestampField).getAsPrimitive().getAsLong()
+              + ", and value: "
+              + event.getFieldBySelector("test::" + sdtValueField).getAsPrimitive().getAsFloat());
+      processor.onEvent(event, collector);
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+    return collector.getEvents().size();
+  }
+
+  private List<Event> makeEvents() {
+    List<Event> events = new ArrayList<>();
+    for (Pair<Long, Double> eventSetting: eventSettings) {
+      events.add(makeEvent(eventSetting));
+    }
+    return events;
+  }
+
+  private Event makeEvent(Pair<Long, Double> eventSetting) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(sdtTimestampField, eventSetting.getKey());
+    map.put(sdtValueField, eventSetting.getValue());
+    return EventFactory.fromMap(map, new SourceInfo("test" + "-topic", "test"),
+            new SchemaInfo(null, new ArrayList<>()));
+  }
+}
\ No newline at end of file