From 8bdbf049b8a2f5e776c12e8536693fee56a58e90 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 16 Oct 2025 13:06:44 -0300 Subject: [PATCH] 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 | 135 ++++++++++++++++++------------------ 1 file changed, 68 insertions(+), 67 deletions(-) diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index de8bf30..53f2732 100644 --- a/scouter/activities/gates.py +++ b/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: - return group['value'].item() + # Fast path for single value + 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']) - - if group.empty: - return None - - if aggr_function == 'avg': - return group['value'].mean() - elif aggr_function == 'mdn': - return group['value'].median() - 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' + 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 - result = {} + # Sort once by timestamp for all data (more efficient than sorting each group) + 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 - latest_timestamp = group['timestamp'].max() + # Get the latest timestamp (last row since data is sorted) + latest_timestamp = group['timestamp'].iloc[-1] aggr_value = self.apply_aggregation( group, aggr_function, metadata) @@ -152,48 +159,42 @@ class Gates(BaseActivity): if aggr_value == 'continue': continue - self.debug( - f"Aggregated data: {aggr_value}", - metadata=metadata - ) - self.debug( - f"Latest timestamp: {latest_timestamp}", - 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 - ) + # Batch debug logging to reduce overhead + if self.logger.level <= 10: # DEBUG level + self.debug( + f"Processed {tag}_{name}: value={aggr_value}, " + f"timestamp={latest_timestamp}, func={aggr_function}", + metadata=metadata + ) - # Store the result - result[f"{tag}_{name}"] = { + # Store the result directly in list for better performance + 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( - f"Aggregated data: {result_df.to_string()}", - metadata=metadata - ) + # Convert to DataFrame only once at the end if we have results + if results: + 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()