Fix embedded graph time axis possible wrap with Measured Time (#224)

Author: pylessard

scrutiny/server/datalogging/datalogging_manager.py

@@ -110,6 +110,37 @@ def request_acquisition(self, request: api_datalogging.AcquisitionRequest, callb
             entry_signal_map=entry_signal_map,
             callback=callback), block=False)
 
+    @classmethod
+    def make_xaxis_indexed(cls, nb_points:int) -> List[float]:
+        return [i for i in range(nb_points)]
+
+    @classmethod
+    def make_xaxis_ideal_time(cls, nb_points:int, sampling_rate:api_datalogging.SamplingRate, decimation:int) -> List[float]:
+        if sampling_rate.frequency is None:
+            raise ValueError('Ideal time X-Axis is not possible with variable frequency loops')
+        timestep = 1 / sampling_rate.frequency
+        timestep *= decimation
+        return [round(i * timestep, cls.TIME_PRECISION_DIGIT) for i in range(nb_points)]
+    
+    @classmethod
+    def make_xaxis_measured_time(cls, time_data:List[bytes]) -> List[float]:
+        if (len(time_data)) < 1:
+            raise ValueError('Bad measured time')
+        time_codec = Codecs.get(EmbeddedDataType.uint32, endianness=Endianness.Big)
+        first_sample = time_codec.decode(time_data[0])
+        previous_sample = first_sample
+        offset = 0
+        out_data:List[float] = [0]
+        for sample in time_data[1:]:
+            v = time_codec.decode(sample)
+            if v < 0x80000000 and previous_sample > 0x80000000:
+                offset += 0x100000000
+            previous_sample = v
+            v_sec = (v - first_sample + offset) * 1e-7
+            out_data.append(v_sec)
+
+        return out_data
+    
     def acquisition_complete_callback(self, success: bool, detail_msg: str, data: Optional[List[List[bytes]]], metadata: Optional[device_datalogging.AcquisitionMetadata]) -> None:
         """Callback called by the device handler when the acquisition finally gets triggered and data has finished downloaded."""
         if self.active_request is None:
@@ -174,27 +205,20 @@ def acquisition_complete_callback(self, success: bool, detail_msg: str, data: Op
                 # Add the X-Axis. Either use a measured signal or use a generated one of the user wants IdealTime
                 xaxis = DataSeries()
                 if self.active_request.api_request.x_axis_type == api_datalogging.XAxisType.Indexed:
-                    xaxis.set_data([i for i in range(nb_points)])
+                    xaxis.set_data(self.make_xaxis_indexed(nb_points))
                     xaxis.name = 'Index'
                     xaxis.logged_watchable = None
                 elif self.active_request.api_request.x_axis_type == api_datalogging.XAxisType.IdealTime:
                     # Ideal time : Generate a time X-Axis based on the sampling rate. Assume the device is running the loop at a reliable fixed rate
                     sampling_rate = self.get_sampling_rate(self.active_request.api_request.rate_identifier)
-                    if sampling_rate.frequency is None:
-                        raise ValueError('Ideal time X-Axis is not possible with variable frequency loops')
-                    timestep = 1 / sampling_rate.frequency
-                    timestep *= self.active_request.api_request.decimation
-                    xaxis.set_data([round(i * timestep, self.TIME_PRECISION_DIGIT) for i in range(nb_points)])
+                    xaxis_data = self.make_xaxis_ideal_time(nb_points, sampling_rate, self.active_request.api_request.decimation)
+                    xaxis.set_data(xaxis_data)
                     xaxis.name = 'Time (ideal)'
                     xaxis.logged_watchable = None
                 elif self.active_request.api_request.x_axis_type == api_datalogging.XAxisType.MeasuredTime:
                     # Measured time is appended at the end of the signal list. See make_device_config_from_request
-                    time_data = data[-1]
-                    if (len(time_data)) < 1:
-                        raise ValueError('Bad measured time')
-                    time_codec = Codecs.get(EmbeddedDataType.uint32, endianness=Endianness.Big)
-                    first_sample = time_codec.decode(time_data[0])
-                    xaxis.set_data([(time_codec.decode(sample) - first_sample) * 1e-7 for sample in time_data])
+                    xaxis_data=self.make_xaxis_measured_time(data[-1])
+                    xaxis.set_data(xaxis_data)
                     xaxis.name = 'Time (measured)'
                     xaxis.logged_watchable = None
                 elif self.active_request.api_request.x_axis_type == api_datalogging.XAxisType.Signal:

test/server/datalogging/test_datalogging_manager.py

@@ -6,6 +6,8 @@
 #
 #   Copyright (c) 2023 Scrutiny Debugger
 
+import struct
+
 import scrutiny.server.datalogging.definitions.api as api_datalogging
 import scrutiny.server.datalogging.definitions.device as device_datalogging
 from scrutiny.core import datalogging as core_datalogging
@@ -632,6 +634,24 @@ def assert_state_change_and_clear(state: api_datalogging.DataloggingState, compl
             self.datalogging_manager.process()
             assert_state_change_and_clear(api_datalogging.DataloggingState.Standby, None)
 
+    def test_make_xaxis(self):
+
+        data = DataloggingManager.make_xaxis_indexed(10)
+        self.assertEqual(data, [0,1,2,3,4,5,6,7,8,9])
+
+        rate = api_datalogging.SamplingRate("test", 1000, api_datalogging.ExecLoopType.FIXED_FREQ, 0)
+        data = DataloggingManager.make_xaxis_ideal_time(10, rate, 2)
+        self.assertEqual(data, [0, 0.002, 0.004, 0.006, 0.008, 0.010, 0.012, 0.014, 0.016, 0.018])
+
+        raw_data = [ struct.pack('>L', v) for v in [0, 10000, 20000, 30000]]
+        data = DataloggingManager.make_xaxis_measured_time(raw_data)
+        self.assertEqual(data, [0, 0.001, 0.002, 0.003])
+
+        limit = 0xFFFFFFFF
+        raw_data = [ struct.pack('>L', v & 0xFFFFFFFF) for v in [limit-20000, limit-10000, limit, limit+10000, limit+20000]]
+        data = DataloggingManager.make_xaxis_measured_time(raw_data)
+        self.assertEqual(data, [0, 0.001, 0.002, 0.003, 0.004])
+
 
 if __name__ == '__main__':
     import unittest