[UPDATE] Support data fetching density from trace server

Previously, TMLL used to retrieve data (XY and TIMEGRAPH) from trace server
with the parameters of experiment's start to end and 65536 items in that specific timerange.
Now, based on the indicated resampling frequency (e.g., 1s, 5ms, 100us, etc.), the client
can indicate how many datapoints it's required to be fetched from the server.

Signed-off-by: Kaveh Shahedi <kaveh.shahedi@ericsson.com>

Author: kavehshahedi

tmll/ml/modules/base_module.py

@@ -41,12 +41,12 @@ def _plot(self, plots: List[Dict[str, Any]], plot_size: Tuple[float, float] = (1
         Plot the given plots. 
 
         :param plots: The plots to plot. Each plot should be a dictionary with the following keys:
-            - plot_type (PLOT_TYPES): The type of the plot (e.g., 'time_series' or 'scatter')
+            - plot_type (PLOT_TYPES): The type of the plot (e.g., "time_series" or "scatter")
             - data (Any): The data to plot
             - x (str): The name of the x-axis column
             - y (str): The name of the y-axis column
             - hue (str, optional): The name of the hue column. Default is None
-            - color (str, optional): The color of the plot. Default is 'blue'
+            - color (str, optional): The color of the plot. Default is "blue"
         :type plots: List[Dict[str, Any]]
         :param plot_size: The size of the plot. Default is (15, 10)
         :type plot_size: Tuple[int, int], optional
@@ -62,31 +62,31 @@ def _plot(self, plots: List[Dict[str, Any]], plot_size: Tuple[float, float] = (1
         :return: None
         """
         # Create a new figure and axis
-        fig, ax = plt.subplots(figsize=plot_size, dpi=kwargs.get('dpi', 100))
+        fig, ax = plt.subplots(figsize=plot_size, dpi=kwargs.get("dpi", 100))
 
         # Plot each plot
         for plot_info in plots:
-            plot_type = plot_info.get('plot_type', None)
-            data = plot_info.get('data', None)
+            plot_type = plot_info.get("plot_type", None)
+            data = plot_info.get("data", None)
 
             # Create the plot
             plot_strategy = PlotFactory.create_plot(plot_type)
-            plot_strategy.plot(ax, data, **{k: v for k, v in plot_info.items() if k not in ['plot_type', 'data']})
+            plot_strategy.plot(ax, data, **{k: v for k, v in plot_info.items() if k not in ["plot_type", "data"]})
 
         # Set the title, x-axis label, and y-axis label of the plot
-        ax.set_title(kwargs.get('fig_title', ''))
-        ax.set_xlabel(kwargs.get('fig_xlabel', ''))
-        ax.set_ylabel(kwargs.get('fig_ylabel', ''))
+        ax.set_title(kwargs.get("fig_title", ""))
+        ax.set_xlabel(kwargs.get("fig_xlabel", ""))
+        ax.set_ylabel(kwargs.get("fig_ylabel", ""))
 
-        if kwargs.get('fig_xticks', None) is not None:
-            ax.set_xticks(kwargs.get('fig_xticks'))  # type: ignore
+        if kwargs.get("fig_xticks", None) is not None:
+            ax.set_xticks(kwargs.get("fig_xticks"))  # type: ignore
 
-        if kwargs.get('fig_yticks', None) is not None:
-            ax.set_yticks(kwargs.get('fig_yticks'))  # type: ignore
+        if kwargs.get("fig_yticks", None) is not None:
+            ax.set_yticks(kwargs.get("fig_yticks"))  # type: ignore
         else:
             y_min, y_max = ax.get_ylim()
             if isinstance(y_min, (int, float)) and isinstance(y_max, (int, float)):
-                num_yticks = kwargs.get('fig_num_yticks', 6)
+                num_yticks = kwargs.get("fig_num_yticks", 6)
                 yticks = PlotUtils.get_formatted_ticks(y_min, y_max, num_yticks)
                 ax.set_yticks(yticks)
 
@@ -95,27 +95,27 @@ def _plot(self, plots: List[Dict[str, Any]], plot_size: Tuple[float, float] = (1
                 ax.set_ylim(yticks[0] - padding, yticks[-1] + padding)
                 ax.set_yticklabels([f"{val:.2f}{unit}" for tick in yticks for val, unit in [Formatter.format_large_number(tick)]])
 
-        if kwargs.get('fig_xticklabels', None) is not None:
-            ax.set_xticklabels(kwargs.get('fig_xticklabels'))  # type: ignore
-        if kwargs.get('fig_yticklabels', None) is not None:
-            ax.set_yticklabels(kwargs.get('fig_yticklabels'))  # type: ignore
-        if kwargs.get('fig_xticklabels_rotation', None) is not None:
+        if kwargs.get("fig_xticklabels", None) is not None:
+            ax.set_xticklabels(kwargs.get("fig_xticklabels"))  # type: ignore
+        if kwargs.get("fig_yticklabels", None) is not None:
+            ax.set_yticklabels(kwargs.get("fig_yticklabels"))  # type: ignore
+        if kwargs.get("fig_xticklabels_rotation", None) is not None:
             ax.set_xticks(ax.get_xticks())
-            ax.set_xticklabels(ax.get_xticklabels(), rotation=kwargs.get('fig_xticklabels_rotation'))
-        if kwargs.get('fig_yticklabels_rotation', None) is not None:
+            ax.set_xticklabels(ax.get_xticklabels(), rotation=kwargs.get("fig_xticklabels_rotation"))
+        if kwargs.get("fig_yticklabels_rotation", None) is not None:
             ax.set_yticks(ax.get_yticks())
-            ax.set_yticklabels(ax.get_yticklabels(), rotation=kwargs.get('fig_yticklabels_rotation'))
+            ax.set_yticklabels(ax.get_yticklabels(), rotation=kwargs.get("fig_yticklabels_rotation"))
 
         # Add the legend to the plot (remove duplicates)
-        if kwargs.get('legend', True):
+        if kwargs.get("legend", True):
             handles, labels = plt.gca().get_legend_handles_labels()
             by_label = dict(zip(labels, handles))
-            PlotUtils.set_standard_legend_style(ax, by_label.values(), by_label.keys(), title=kwargs.get('legend_title', None))
+            PlotUtils.set_standard_legend_style(ax, by_label.values(), by_label.keys(), title=kwargs.get("legend_title", None))
         else:
             if ax.get_legend():
                 ax.get_legend().remove()
 
-        ax.grid(kwargs.get('grid', True))
+        ax.grid(kwargs.get("grid", True))
 
         # Display the plot
         plt.tight_layout()
@@ -134,7 +134,8 @@ def _process(self, outputs: Optional[List[Output]] = None, **kwargs) -> None:
         data, outputs = self.data_fetcher.fetch_data(
             experiment=self.experiment,
             target_outputs=outputs,
-            **kwargs.get('fetch_params', {}))
+            resample_freq=kwargs.get("resample_freq", "1s"),
+            **kwargs.get("fetch_params", {}))
 
         if data is None:
             self.logger.error("No data fetched")
@@ -153,26 +154,26 @@ def _process(self, outputs: Optional[List[Output]] = None, **kwargs) -> None:
 
                 if converted and converted.type == "TIME_GRAPH":
                     df = df.rename({"start_time": "timestamp"}, axis=1)
-                    df['end_time'] = pd.to_datetime(df['end_time'])
+                    df["end_time"] = pd.to_datetime(df["end_time"])
 
                 # Apply common preprocessing steps
-                if kwargs.get('normalize', True):
+                if kwargs.get("normalize", True):
                     df = self.data_preprocessor.normalize(df)
-                if kwargs.get('convert_datetime', True):
+                if kwargs.get("convert_datetime", True):
                     df = self.data_preprocessor.convert_to_datetime(df)
-                if kwargs.get('resample', True):
-                    df = self.data_preprocessor.resample(df, frequency=kwargs.get('resample_freq', '1s'))
-                if kwargs.get('remove_minimum', False):
+                if kwargs.get("resample", True):
+                    df = self.data_preprocessor.resample(df, frequency=kwargs.get("resample_freq", "1s"))
+                if kwargs.get("remove_minimum", False):
                     df = self.data_preprocessor.remove_minimum(df)
 
                 self.dataframes[shortened] = df
 
         # Filter out dataframes with less than min_size instances
-        min_size = kwargs.get('min_size', 1)
+        min_size = kwargs.get("min_size", 1)
         self.dataframes = {k: v for k, v in self.dataframes.items() if len(v) >= min_size}
 
         # Align timestamps if needed
-        if kwargs.get('align_timestamps', True) and self.dataframes:
+        if kwargs.get("align_timestamps", True) and self.dataframes:
             self.dataframes, self.timestamps = DataPreprocessor.align_timestamps(self.dataframes)
 
         # Call module-specific post-processing

tmll/tmll_client.py

@@ -8,7 +8,7 @@
 import pandas as pd
 import numpy as np
 
-from typing import Dict, List, Optional, Union, cast
+from typing import Dict, List, Optional, Tuple, Union, cast
 
 from tmll.common.models.data.table.column import TableDataColumn
 from tmll.common.models.data.table.table import TableData
@@ -324,16 +324,18 @@ def fetch_data(self, experiment: Experiment, outputs: List[Dict[str, Union[Outpu
                     item_ids = list(map(int, [node.id for node in o_tree.nodes]))
                     time_range_start = kwargs.get("start", experiment.start)
                     time_range_end = kwargs.get("end", experiment.end)
-                    time_range_num_times = kwargs.get("num_times", 65536)
+                    resample_freq = kwargs.get("resample_freq", "1s")
 
-                    while True:
+                    intervals = self._calculate_intervals(time_range_start, time_range_end, resample_freq)
+
+                    for interval in intervals:
                         parameters = {
                             TspClient.PARAMETERS_KEY: {
                                 TspClient.REQUESTED_ITEM_KEY: item_ids,
                                 TspClient.REQUESTED_TIME_RANGE_KEY: {
-                                    TspClient.REQUESTED_TIME_RANGE_START_KEY: time_range_start,
-                                    TspClient.REQUESTED_TIME_RANGE_END_KEY: time_range_end,
-                                    TspClient.REQUESTED_TIME_RANGE_NUM_TIMES_KEY: time_range_num_times
+                                    TspClient.REQUESTED_TIME_RANGE_START_KEY: interval[0],
+                                    TspClient.REQUESTED_TIME_RANGE_END_KEY: interval[1],
+                                    TspClient.REQUESTED_TIME_RANGE_NUM_TIMES_KEY: interval[2]
                                 }
                             }
                         }
@@ -374,31 +376,23 @@ def fetch_data(self, experiment: Experiment, outputs: List[Dict[str, Union[Outpu
                             datasets[o_output.id] = datasets.get(o_output.id, {})
                             datasets[o_output.id][series_name] = pd.concat([datasets[o_output.id].get(series_name, pd.DataFrame()), dataset])
 
-                        # Update the time_range_start for the next iteration
-                        if x and len(x) > 0:
-                            time_range_start = x[-1] + 1  # Start from the next timestamp after the last received
-
-                            # Check if the time_range_start is greater than the time_range_end
-                            if time_range_start > time_range_end:
-                                break
-                        else:
-                            break  # Exit if no data was received in this iteration
-
                 case "TIME_GRAPH":
                     items = list(map(int, [node.id for node in o_tree.nodes]))
                     time_range_start = kwargs.get("start", experiment.start)
                     time_range_end = kwargs.get("end", experiment.end)
-                    time_range_num_times = kwargs.get("num_times", 65536)
+                    resample_freq = kwargs.get("resample_freq", "1s")
                     strategy = kwargs.get("strategy", "DEEP")
 
-                    while True:
+                    intervals = self._calculate_intervals(time_range_start, time_range_end, resample_freq)
+
+                    for interval in intervals:
                         parameters = {
                             TspClient.PARAMETERS_KEY: {
                                 TspClient.REQUESTED_ITEM_KEY: items,
                                 TspClient.REQUESTED_TIME_RANGE_KEY: {
-                                    TspClient.REQUESTED_TIME_RANGE_START_KEY: time_range_start,
-                                    TspClient.REQUESTED_TIME_RANGE_END_KEY: time_range_end,
-                                    TspClient.REQUESTED_TIME_RANGE_NUM_TIMES_KEY: time_range_num_times
+                                    TspClient.REQUESTED_TIME_RANGE_START_KEY: interval[0],
+                                    TspClient.REQUESTED_TIME_RANGE_END_KEY: interval[1],
+                                    TspClient.REQUESTED_TIME_RANGE_NUM_TIMES_KEY: interval[2]
                                 },
                                 "filter_query_parameters": {
                                     "strategy": strategy
@@ -442,16 +436,6 @@ def fetch_data(self, experiment: Experiment, outputs: List[Dict[str, Union[Outpu
 
                         datasets[o_output.id] = pd.concat([datasets[o_output.id], dataset])
 
-                        # Update the time_range_start for the next iteration
-                        if data and len(data) > 0:
-                            time_range_start = data[-1]["end_time"] + 1
-
-                            # Check if the time_range_start is greater than the time_range_end
-                            if time_range_start > time_range_end:
-                                break
-                        else:
-                            break
-
                 case "TABLE" | "DATA_TREE":
                     columns = self.tsp_client.fetch_virtual_table_columns(exp_uuid=experiment.uuid, output_id=o_output.id)
                     if columns.status_code != 200:
@@ -532,6 +516,54 @@ def fetch_data(self, experiment: Experiment, outputs: List[Dict[str, Union[Outpu
 
         return datasets
 
+    def _calculate_intervals(self, start: int, end: int, resample_freq: str) -> List[Tuple[int, int, int]]:
+        """
+        Based on the resampling frequency, we should indicate how many items we want to fetch for XY/TIMEGRAPH data
+        from the TSP server. For example, if the resampling frequency is 1s, we should fetch 1 item per second.
+
+        Now, based on the start and end timestamps, we should calculate the number of items
+        to fetch from the TSP server. For example, if the start timestamp is 0 and the end timestamp is 1000,
+        and the resampling frequency is 1s, we should fetch 1000 items.
+
+        The formula to calculate the number of items is:
+            num_items = (end - start) / resample_freq
+        where:
+            - start: Start timestamp in nano-seconds
+            - end: End timestamp in nano-seconds
+            - resample_freq: Resampling frequency in string format (e.g., "1s", "1ms", "1us", etc.)
+
+        The catch is that, the tsp server can handle up to 65536 items at a time. So, we need to return the list
+        of intervals to fetch the XY data from the TSP server. For example, if the number of items is 100000, we should
+        return a list of two intervals with the following values:
+            - start: the start timestamp of the interval
+            - end: the end timestamp of the interval
+            - num_items: the number of items to fetch from the TSP server
+        So, the list should be like:
+            [(start1, end1, 65536), (start2, end2, 34464)]
+
+        :param start: Start timestamp
+        :type start: int
+        :param end: End timestamp
+        :type end: int
+        :param resample_freq: Resampling frequency
+        :type resample_freq: str
+        :return: List of intervals
+        :rtype: List[Tuple[int, int, int]]
+        """
+        resample_freq_timedelta = pd.to_timedelta(resample_freq).value
+
+        num_items = int((end - start) / resample_freq_timedelta)
+        intervals = []
+
+        while num_items > 0:
+            batch_size = min(num_items, 65536)
+            interval_end = start + batch_size * resample_freq_timedelta
+            intervals.append((start, interval_end, batch_size))
+            start = interval_end
+            num_items -= batch_size
+
+        return intervals
+
     @staticmethod
     def enable_instrumentation(instrumentation_file: Optional[str] = None, instrument_kernel: bool = False, verbose: bool = True) -> None:
         """