SIENTIAPDE-1316

SIENTIAPDE-1084: Refactor apply_aggregation method in Gates class for improved performance and clarity. Changed parameter name from 'group' to 'values', optimized aggregation logic, and enhanced error handling for invalid functions. Streamlined data processing by reducing unnecessary operations and added batch debug logging.

scouter/activities/gates.py

@@ -44,7 +44,7 @@ class Gates(BaseActivity):
         BaseActivity.__init__(
             self, logger, notification_handler, set_error_counter=True)
 
-    def apply_aggregation(self, group: DataFrame, aggr_function: str,
+    def apply_aggregation(self, values: DataFrame, aggr_function: str,
                           metadata: dict[str, Any]) -> float | None | str:
         """
         Apply aggregation function to a group of time-series data.
@@ -54,7 +54,7 @@ class Gates(BaseActivity):
         reporting for invalid aggregation functions.
 
         Args:
-            group (DataFrame): Group of data points to aggregate
+            values (DataFrame): Group of data points to aggregate (pre-sorted by timestamp)
             aggr_function (str): Aggregation function to apply.
                 Supported functions: 'lts' (latest), 'avg' (average), 'mdn' (median),
                 'max' (maximum), 'min' (minimum)
@@ -66,36 +66,39 @@ class Gates(BaseActivity):
         Raises:
             NotificationError: If invalid aggregation function is specified
         """
-        if len(group) == 1:
+        # Fast path for single value
-            return group['value'].item()
+        if len(values) == 1:
+            return values['value'].iloc[0]
 
-        # Apply aggregation function to value
         if aggr_function == 'lts':
-            return group['value'].iloc[-1]
+            return values['value'].iloc[-1]
+
+        # Remove NaN values without inplace operation
+        clean_values = values['value'].dropna()
+
+        if clean_values.empty:
+            return None
+
+        # Use dictionary lookup for aggregation functions (faster than if-elif chain)
+        aggregation_map = {
+            'avg': lambda x: x.mean(),
+            'mdn': lambda x: x.median(),
+            'max': lambda x: x.max(),
+            'min': lambda x: x.min()
+        }
+
+        if aggr_function in aggregation_map:
+            return aggregation_map[aggr_function](clean_values)
         else:
-            group.dropna(inplace=True, subset=['value'])
+            self.send_notification(
-
+                metadata=metadata,
-            if group.empty:
+                notification_id="AGGREGATION_ISSUES",
-                return None
+                message=f"Invalid aggregation function: {aggr_function}",
-
+                block="aggregate_data",
-            if aggr_function == 'avg':
+                level=NotificationLevel.ERROR,
-                return group['value'].mean()
+                attachment_content=traceback.format_exc()
-            elif aggr_function == 'mdn':
+            )
-                return group['value'].median()
+            return 'continue'
-            elif aggr_function == 'max':
-                return group['value'].max()
-            elif aggr_function == 'min':
-                return group['value'].min()
-            else:
-                self.send_notification(
-                    metadata=metadata,
-                    notification_id="AGGREGATION_ISSUES",
-                    message=f"Invalid aggregation function: {aggr_function}",
-                    block="aggregate_data",
-                    level=NotificationLevel.ERROR,
-                    attachment_content=traceback.format_exc()
-                )
-                return 'continue'
 
     @activity.defn(name="aggregate_data")
     async def aggregate_data(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -130,21 +133,25 @@ class Gates(BaseActivity):
                 metadata=metadata
             )
 
-            # Initialize result dictionary
+            # Sort once by timestamp for all data (more efficient than sorting each group)
-            result = {}
+            df = df.sort_values(['tag', 'name', 'timestamp'])
 
             # Group by tag and name
-            grouped = df.groupby(['tag', 'name'])
+            # sort=False since we already sorted
+            grouped = df.groupby(['tag', 'name'], sort=False)
 
+            # Prepare aggregation functions mapping
+            model_tags = input_data['model_tags']
+
+            # Process groups efficiently
+            results = []
             for (tag, name), group in grouped:
                 # Get the aggregation function from model_tags
-                aggr_function = input_data['model_tags'].get(
+                aggr_function = model_tags.get(
                     name, {}).get('aggr_func', 'lts')
 
-                group.sort_values(by='timestamp', inplace=True)
+                # Get the latest timestamp (last row since data is sorted)
-
+                latest_timestamp = group['timestamp'].iloc[-1]
-                # Get the latest timestamp
-                latest_timestamp = group['timestamp'].max()
 
                 aggr_value = self.apply_aggregation(
                     group, aggr_function, metadata)
@@ -152,48 +159,42 @@ class Gates(BaseActivity):
                 if aggr_value == 'continue':
                     continue
 
-                self.debug(
+                # Batch debug logging to reduce overhead
-                    f"Aggregated data: {aggr_value}",
+                if self.logger.level <= 10:  # DEBUG level
-                    metadata=metadata
+                    self.debug(
-                )
+                        f"Processed {tag}_{name}: value={aggr_value}, "
-                self.debug(
+                        f"timestamp={latest_timestamp}, func={aggr_function}",
-                    f"Latest timestamp: {latest_timestamp}",
+                        metadata=metadata
-                    metadata=metadata
+                    )
-                )
-                self.debug(
-                    f"Groups: {group.to_string()}",
-                    metadata=metadata
-                )
-                self.debug(
-                    f"group name: {name}",
-                    metadata=metadata
-                )
-                self.debug(
-                    f"group tag: {tag}",
-                    metadata=metadata
-                )
 
-                # Store the result
+                # Store the result directly in list for better performance
-                result[f"{tag}_{name}"] = {
+                results.append({
                     'tag': tag,
                     'name': name,
                     'value': aggr_value,
                     'timestamp': latest_timestamp,
                     'aggregation_function': aggr_function
-                }
+                })
 
-            result_df = DataFrame(list(result.values()))
             self.info(
-                f"Aggregated data has {len(result_df)} rows",
+                f"Aggregated data has {len(results)} rows",
                 metadata=metadata
             )
 
-            self.debug(
+            # Convert to DataFrame only once at the end if we have results
-                f"Aggregated data: {result_df.to_string()}",
+            if results:
-                metadata=metadata
+                result_df = DataFrame(results)
-            )
 
-            return result_df.to_dict()
+                if self.logger.level <= 10:  # DEBUG level
+                    self.debug(
+                        f"Final aggregated data:\n{result_df.to_string()}",
+                        metadata=metadata
+                    )
+
+                return result_df.to_dict()
+            else:
+                # Return empty DataFrame dict structure
+                return DataFrame().to_dict()
 
         except Exception as e:
             trace = traceback.format_exc()