Implement publishing for TimeseriesDataTimestamp (#130)

harisbotic · web-flow · commit f6d904091793 · 2023-05-31T17:43:49.000+01:00
* Add publish API for TimeseriesDataTimestamp
* Add missing buffer publish api for TimeseriesDataRaw
diff --git a/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/StreamTimeseriesConsumerShould.cs b/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/StreamTimeseriesConsumerShould.cs
@@ -7,7 +7,7 @@
 
 namespace QuixStreams.Streaming.UnitTests.Models
 {
-    public class StreamParametersConsumerShould
+    public class StreamTimeseriesConsumerShould
     {
 
         [Fact]
diff --git a/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/StreamTimeseriesProducerShould.cs b/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/StreamTimeseriesProducerShould.cs
@@ -10,7 +10,7 @@
 
 namespace QuixStreams.Streaming.UnitTests.Models
 {
-    public class StreamParametersProducerShould
+    public class StreamTimeseriesProducerShould
     {
 
         [Fact]
diff --git a/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/TimeseriesDataTimestampShould.cs b/src/CsharpClient/QuixStreams.Streaming.UnitTests/Models/TimeseriesDataTimestampShould.cs
@@ -49,5 +49,35 @@ public void AddTags_WithTags_ShouldNotThrowException()
             tsdts.Tags["test4"].Should().Be("val4");
 
         }
+        
+        [Fact]
+        public void ConvertToTimeseriesDataRaw_ReturnsWithSingleTimestamp()
+        {
+            // Arrange
+            var timeseriesData = new TimeseriesData();
+            timeseriesData.AddTimestampNanoseconds(100)
+                .AddValue("string", "1")
+                .AddTag("1", "tag1");    
+            
+            var timestampToConvert = timeseriesData.AddTimestampNanoseconds(200)
+                .AddValue("string", "2")
+                .AddValue("double", 2)
+                .AddValue("byte", new byte[]{0x02})
+                .AddTag("1", "tag1")
+                .AddTag("2", "tag2");    
+            
+            
+
+            // Act
+            var raw = timestampToConvert.ConvertToTimeseriesDataRaw();
+
+            // Assert
+            raw.Timestamps.Length.Should().Be(1);
+            raw.Timestamps[0].Should().Be(200);
+            raw.StringValues.Count.Should().Be(1);
+            raw.NumericValues.Count.Should().Be(1);
+            raw.BinaryValues.Count.Should().Be(1);
+            raw.TagValues.Count.Should().Be(2);
+        }
     }
 }
diff --git a/src/CsharpClient/QuixStreams.Streaming/Models/StreamProducer/StreamTimeseriesProducer.cs b/src/CsharpClient/QuixStreams.Streaming/Models/StreamProducer/StreamTimeseriesProducer.cs
@@ -73,7 +73,7 @@ public void Publish(TimeseriesData data)
 
             this.streamProducer.Publish(data.ConvertToTimeseriesDataRaw());
         }
-
+        
         /// <summary>
         /// Publish data in TimeseriesDataRaw format without any buffering
         /// </summary>
@@ -100,7 +100,7 @@ public void Publish(QuixStreams.Telemetry.Models.TimeseriesDataRaw data)
                 updatedTimestamps[i] = data.Timestamps[i] + epochDiff;
             }
 
-            QuixStreams.Telemetry.Models.TimeseriesDataRaw new_data = new QuixStreams.Telemetry.Models.TimeseriesDataRaw(
+            QuixStreams.Telemetry.Models.TimeseriesDataRaw newData = new QuixStreams.Telemetry.Models.TimeseriesDataRaw(
                 data.Epoch, 
                 updatedTimestamps, 
                 data.NumericValues, 
@@ -109,10 +109,29 @@ public void Publish(QuixStreams.Telemetry.Models.TimeseriesDataRaw data)
                 data.TagValues
             );
 
-            this.streamProducer.Publish(new_data);
+            this.streamProducer.Publish(newData);
         }
 
+        /// <summary>
+        /// Publish single timestamp to stream without any buffering
+        /// </summary>
+        /// <param name="timestamp">Timeseries timestamp to publish</param>
+        public void Publish(TimeseriesDataTimestamp timestamp)
+        {
+            if (isDisposed)
+            {
+                throw new ObjectDisposedException(nameof(StreamTimeseriesProducer));
+            }
+            
+            if (!timestamp.EpochIncluded)
+            {
+                timestamp.TimestampNanoseconds += this.streamProducer.Epoch.ToUnixNanoseconds();
+                timestamp.EpochIncluded = true;
+            }
 
+            this.streamProducer.Publish(timestamp.ConvertToTimeseriesDataRaw());
+        }
+        
         /// <summary>
         /// Default Location of the parameters. Parameter definitions added with <see cref="AddDefinition"/> will be inserted at this location.
         /// See <see cref="AddLocation"/> for adding definitions at a different location without changing default.
diff --git a/src/CsharpClient/QuixStreams.Streaming/Models/StreamProducer/TimeseriesBufferProducer.cs b/src/CsharpClient/QuixStreams.Streaming/Models/StreamProducer/TimeseriesBufferProducer.cs
@@ -93,7 +93,7 @@ private TimeseriesDataBuilder AddTimestampNanoseconds(long timestampNanoseconds,
 
 
         /// <summary>
-        /// Publish the provided timeseries data to the buffer.
+        /// Publish timeseries data to the buffer.
         /// </summary>
         /// <param name="data">Data to publish</param>
         public void Publish(TimeseriesData data)
@@ -119,6 +119,78 @@ public void Publish(TimeseriesData data)
 
             this.WriteChunk(data.ConvertToTimeseriesDataRaw(false, false)); // use merge & clean of Buffer is more efficient
         }
+        
+        /// <summary>
+        /// Publish timeseries data raw to the buffer.
+        /// </summary>
+        /// <param name="data">Data to publish</param>
+        public void Publish(TimeseriesDataRaw data)
+        {
+            long epochDifference = this.streamProducer.Epoch.ToUnixNanoseconds();
+            long[] updatedTimestamps = null;
+            
+            if (epochDifference != 0)
+            {
+                updatedTimestamps = new long[data.Timestamps.Length];
+                for (int i = 0; i < updatedTimestamps.Length; i++)
+                {
+                    updatedTimestamps[i] = data.Timestamps[i] + epochDifference;
+                }
+            }
+            
+            Dictionary<string, string[]> updatedTagValues = null;
+
+            if (this.DefaultTags.Count > 0)
+            {
+                updatedTagValues = new Dictionary<string, string[]>(data.TagValues);
+                foreach (var tag in this.DefaultTags)
+                {
+                    if (!updatedTagValues.TryGetValue(tag.Key, out var tagValues))
+                    {
+                        tagValues = new string[data.Timestamps.Length];
+                        updatedTagValues[tag.Key] = tagValues;
+                    }
+
+                    for (int i = 0; i < data.Timestamps.Length; i++)
+                    {
+                        tagValues[i] = tag.Value;
+                    }
+                }
+            }
+
+            var newData = new QuixStreams.Telemetry.Models.TimeseriesDataRaw(
+                data.Epoch, 
+                updatedTimestamps ?? data.Timestamps, 
+                data.NumericValues, 
+                data.StringValues, 
+                data.BinaryValues, 
+                updatedTagValues ?? data.TagValues
+            );
+
+            this.WriteChunk(newData);
+        }
+        
+        /// <summary>
+        /// Publish single timestamp to the buffer.
+        /// </summary>
+        /// <param name="timestamp">Timeseries timestamp to publish</param>
+        public void Publish(TimeseriesDataTimestamp timestamp)
+        {
+            if (!timestamp.EpochIncluded)
+            {
+                timestamp.TimestampNanoseconds += this.Epoch.ToUnixNanoseconds();
+                timestamp.EpochIncluded = true;
+            }
+
+            foreach (var kv in this.DefaultTags)
+            {
+                if (!timestamp.Tags.ContainsKey(kv.Key))
+                {
+                    timestamp.AddTag(kv.Key, kv.Value);
+                }
+            }
+            this.WriteChunk(timestamp.ConvertToTimeseriesDataRaw());
+        }
 
 
         /// <summary>
diff --git a/src/CsharpClient/QuixStreams.Streaming/Models/TimeseriesDataTimestamp.cs b/src/CsharpClient/QuixStreams.Streaming/Models/TimeseriesDataTimestamp.cs
@@ -1,6 +1,7 @@
 ﻿using System;
 using System.Collections.Generic;
 using QuixStreams.Streaming.Utils;
+using QuixStreams.Telemetry.Models;
 using QuixStreams.Telemetry.Models.Utility;
 
 namespace QuixStreams.Streaming.Models
@@ -222,5 +223,10 @@ public TimeseriesDataTimestamp RemoveTag(string tagId)
 
             return this;
         }
+        
+        internal TimeseriesDataRaw ConvertToTimeseriesDataRaw()
+        {
+            return new TimeseriesData(new List<TimeseriesDataTimestamp>{this}).rawData;
+        }
     }
 }
diff --git a/src/PythonClient/src/quixstreams/models/streamproducer/streamtimeseriesproducer.py b/src/PythonClient/src/quixstreams/models/streamproducer/streamtimeseriesproducer.py
@@ -6,7 +6,7 @@
 from .timeseriesbufferproducer import TimeseriesBufferProducer
 from ...builders import ParameterDefinitionBuilder
 from ...helpers.nativedecorator import nativedecorator
-from ...models import TimeseriesData, TimeseriesDataRaw
+from ...models import TimeseriesData, TimeseriesDataRaw, TimeseriesDataTimestamp
 from ...native.Python.QuixStreamsStreaming.Models.StreamProducer.StreamTimeseriesProducer import StreamTimeseriesProducer as stspi
 
 
@@ -102,12 +102,12 @@ def buffer(self) -> TimeseriesBufferProducer:
             self._buffer = TimeseriesBufferProducer(self._stream_producer, self._interop.get_Buffer())
         return self._buffer
 
-    def publish(self, packet: Union[TimeseriesData, pd.DataFrame, TimeseriesDataRaw]) -> None:
+    def publish(self, packet: Union[TimeseriesData, pd.DataFrame, TimeseriesDataRaw, TimeseriesDataTimestamp]) -> None:
         """
         Publish the given packet to the stream without any buffering.
 
         Args:
-            packet: The packet containing TimeseriesData, TimeseriesDataRaw, or pandas DataFrame.
+            packet: The packet containing TimeseriesData, TimeseriesDataRaw, TimeseriesDataTimestamp, or pandas DataFrame.
 
         Note:
             - Pandas DataFrame should contain 'time' label, else the first integer label will be taken as time.
@@ -141,6 +141,9 @@ def publish(self, packet: Union[TimeseriesData, pd.DataFrame, TimeseriesDataRaw]
         if isinstance(packet, TimeseriesDataRaw):
             self._interop.Publish2(packet.get_net_pointer())
             return
+        if isinstance(packet, TimeseriesDataTimestamp):
+            self._interop.Publish3(packet.get_net_pointer())
+            return
         if isinstance(packet, pd.DataFrame):
             data = TimeseriesDataRaw.from_dataframe(packet)
             with data:
diff --git a/src/PythonClient/src/quixstreams/models/streamproducer/timeseriesbufferproducer.py b/src/PythonClient/src/quixstreams/models/streamproducer/timeseriesbufferproducer.py
@@ -7,7 +7,7 @@
 
 from ..netdict import NetDict
 from ..timeseriesbuffer import TimeseriesBuffer
-from ... import TimeseriesData
+from ... import TimeseriesData, TimeseriesDataRaw, TimeseriesDataTimestamp
 from ...builders import TimeseriesDataBuilder
 from ...helpers.dotnet.datetimeconverter import DateTimeConverter as dtc
 from ...helpers.nativedecorator import nativedecorator
@@ -117,12 +117,12 @@ def flush(self):
         """
         self._interop.Flush()
 
-    def publish(self, packet: Union[TimeseriesData, pd.DataFrame]) -> None:
+    def publish(self, packet: Union[TimeseriesData, pd.DataFrame, TimeseriesDataRaw, TimeseriesDataTimestamp]) -> None:
         """
         Publish the provided timeseries packet to the buffer.
 
         Args:
-            packet: The packet containing TimeseriesData or panda DataFrame
+            packet: The packet containing TimeseriesData, TimeseriesDataRaw, TimeseriesDataTimestamp, or pandas DataFrame.
                 - packet type panda.DataFrame:
                     * Note 1: panda data frame should contain 'time' label, else the first integer label will be taken as time.
                     * Note 2: Tags should be prefixed by TAG__ or they will be treated as timeseries parameters
@@ -155,6 +155,12 @@ def publish(self, packet: Union[TimeseriesData, pd.DataFrame]) -> None:
         if isinstance(packet, TimeseriesData):
             self._interop.Publish(packet.get_net_pointer())
             return
+        if isinstance(packet, TimeseriesDataRaw):
+            self._interop.Publish2(packet.get_net_pointer())
+            return
+        if isinstance(packet, TimeseriesDataTimestamp):
+            self._interop.Publish3(packet.get_net_pointer())
+            return
         if isinstance(packet, pd.DataFrame):
             data = TimeseriesData.from_panda_dataframe(packet)
             with data:
diff --git a/src/PythonClient/tests/quixstreams/integrationtests/test_integration.py b/src/PythonClient/tests/quixstreams/integrationtests/test_integration.py
diff --git a/src/PythonClient/tests/quixstreams/unittests/models/test_timeseriesdata.py b/src/PythonClient/tests/quixstreams/unittests/models/test_timeseriesdata.py

Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,7 @@`
`7`	`7`
`8`	`8`	`namespace QuixStreams.Streaming.UnitTests.Models`
`9`	`9`	`{`
`10`		`- public class StreamParametersConsumerShould`
	`10`	`+ public class StreamTimeseriesConsumerShould`
`11`	`11`	`{`
`12`	`12`
`13`	`13`	`[Fact]`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@`
`10`	`10`
`11`	`11`	`namespace QuixStreams.Streaming.UnitTests.Models`
`12`	`12`	`{`
`13`		`- public class StreamParametersProducerShould`
	`13`	`+ public class StreamTimeseriesProducerShould`
`14`	`14`	`{`
`15`	`15`
`16`	`16`	`[Fact]`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Generic;`
`3`	`3`	`using QuixStreams.Streaming.Utils;`
	`4`	`+using QuixStreams.Telemetry.Models;`
`4`	`5`	`using QuixStreams.Telemetry.Models.Utility;`
`5`	`6`
`6`	`7`	`namespace QuixStreams.Streaming.Models`
`@@ -222,5 +223,10 @@ public TimeseriesDataTimestamp RemoveTag(string tagId)`
`222`	`223`
`223`	`224`	`return this;`
`224`	`225`	`}`
	`226`	`+`
	`227`	`+ internal TimeseriesDataRaw ConvertToTimeseriesDataRaw()`
	`228`	`+ {`
	`229`	`+ return new TimeseriesData(new List<TimeseriesDataTimestamp>{this}).rawData;`
	`230`	`+ }`
`225`	`231`	`}`
`226`	`232`	`}`