Code import - branch feature/SIENTIAPDE-1646
This commit is contained in:
319
scouter/activities/gates.py
Normal file
319
scouter/activities/gates.py
Normal file
@@ -0,0 +1,319 @@
|
||||
from collections.abc import Hashable
|
||||
|
||||
from sientia_do.observability.metrics_controller import MetricsController
|
||||
from temporalio import activity, workflow
|
||||
|
||||
with workflow.unsafe.imports_passed_through():
|
||||
import traceback
|
||||
from typing import Any
|
||||
|
||||
from pandas import DataFrame
|
||||
from sientia_do.notifications.handlers import NotificationHandler
|
||||
from sientia_do.notifications.models import NotificationLevel
|
||||
from sientia_do.observability.logger import Logger
|
||||
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
||||
|
||||
from scouter import metrics
|
||||
from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
|
||||
|
||||
quality_gate_filters = {
|
||||
'NULL_VALUES_FILTER': null_values_filter,
|
||||
'OUT_OF_BOUNDS_FILTER': out_of_bounds_filter,
|
||||
}
|
||||
|
||||
|
||||
class Gates(SientiaMonitoring):
|
||||
"""
|
||||
Data quality gates and filtering operations.
|
||||
|
||||
This class implements data quality validation and filtering for industrial
|
||||
time-series data. It provides:
|
||||
- Configurable data quality filters
|
||||
- Data aggregation functions for time-series data
|
||||
- Comprehensive error handling and notification
|
||||
- Metrics collection for quality monitoring
|
||||
|
||||
The class supports multiple aggregation strategies and quality filters to
|
||||
ensure data integrity and enable flexible data processing workflows.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
logger: Logger,
|
||||
notification_handler: NotificationHandler,
|
||||
metrics_controller: MetricsController,
|
||||
):
|
||||
"""
|
||||
Initialize the Gates class with logging and notification services.
|
||||
|
||||
Args:
|
||||
logger (Logger): Logger instance for operation logging
|
||||
notification_handler (NotificationHandler): Handler for system notifications
|
||||
metrics_controller (MetricsController): Metrics controller instance
|
||||
"""
|
||||
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
|
||||
|
||||
def close(self):
|
||||
"""
|
||||
Close the Gates class.
|
||||
"""
|
||||
SientiaMonitoring.shutdown(self)
|
||||
|
||||
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.
|
||||
|
||||
This method applies the specified aggregation function to a group of
|
||||
data points. It handles edge cases and provides comprehensive error
|
||||
reporting for invalid aggregation functions.
|
||||
|
||||
Args:
|
||||
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)
|
||||
metadata (dict[str, Any]): Workflow metadata for error reporting
|
||||
|
||||
Returns:
|
||||
float | None | str: Aggregated value, None if no valid data, or 'continue' for errors
|
||||
|
||||
Raises:
|
||||
NotificationError: If invalid aggregation function is specified
|
||||
"""
|
||||
aggregation_map = {
|
||||
'lts': lambda x: x.iloc[-1],
|
||||
'avg': lambda x: x.mean(),
|
||||
'mdn': lambda x: x.median(),
|
||||
'max': lambda x: x.max(),
|
||||
'min': lambda x: x.min(),
|
||||
}
|
||||
|
||||
if aggr_function not in aggregation_map:
|
||||
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'
|
||||
|
||||
if len(values) == 1:
|
||||
return values['value'].iloc[0]
|
||||
|
||||
if aggr_function == 'lts':
|
||||
return aggregation_map['lts'](values['value'])
|
||||
|
||||
clean_values = values['value'].dropna()
|
||||
|
||||
if clean_values.empty:
|
||||
return None
|
||||
|
||||
return aggregation_map[aggr_function](clean_values)
|
||||
|
||||
@activity.defn(name='aggregate_data')
|
||||
def aggregate_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
|
||||
"""
|
||||
Aggregate time-series data by tag and name using specified functions.
|
||||
|
||||
This activity processes time-series data by grouping it by tag and name,
|
||||
then applying the configured aggregation functions. It handles data
|
||||
validation and provides comprehensive error reporting.
|
||||
|
||||
Args:
|
||||
input_data (dict[str, Any]): Activity input parameters.
|
||||
Required fields:
|
||||
- data (dict[str, Any]): Time-series data to aggregate
|
||||
- model_tags (dict[str, Any]): Tag configuration with aggregation functions
|
||||
|
||||
Returns:
|
||||
dict[Hashable, Any]: Aggregated data organized by tag and name
|
||||
|
||||
Raises:
|
||||
Exception: If aggregation operation fails
|
||||
"""
|
||||
|
||||
metadata = input_data['metadata']
|
||||
|
||||
try:
|
||||
# Convert input data to DataFrame
|
||||
df = DataFrame(input_data['data'])
|
||||
|
||||
self.info(f'Aggregating time series data for {len(df)} rows', metadata=metadata)
|
||||
|
||||
# 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
|
||||
# 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 = model_tags.get(name, {}).get('aggr_func', 'lts')
|
||||
|
||||
# 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)
|
||||
|
||||
if aggr_value == 'continue':
|
||||
continue
|
||||
|
||||
# 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,
|
||||
}
|
||||
)
|
||||
|
||||
self.info(f'Aggregated data has {len(results)} rows', metadata=metadata)
|
||||
|
||||
# Convert to DataFrame only once at the end if we have results
|
||||
if results:
|
||||
result_df = DataFrame(results)
|
||||
|
||||
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()
|
||||
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='AGGREGATION_ISSUES',
|
||||
message=f'Error aggregating data: {e}',
|
||||
block='aggregate_data',
|
||||
level=NotificationLevel.ERROR,
|
||||
attachment_content=trace,
|
||||
)
|
||||
|
||||
self.error(trace, metadata=metadata)
|
||||
raise e
|
||||
|
||||
@activity.defn(name='data_quality_gate')
|
||||
def data_quality_gate(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
|
||||
"""
|
||||
Apply data quality filters to incoming data.
|
||||
|
||||
This activity applies configurable quality filters to validate incoming
|
||||
data. It supports multiple filter types and provides comprehensive
|
||||
error reporting for quality issues.
|
||||
|
||||
Args:
|
||||
input_data (dict[str, Any]): Activity input parameters.
|
||||
Required fields:
|
||||
- data (dict[str, Any]): Data to validate
|
||||
- filters (dict[str, str]): Filter configuration
|
||||
- model_tags (dict[str, Any]): Tag-specific validation rules
|
||||
|
||||
Returns:
|
||||
dict[Hashable, Any]: Filtered data that passes quality validation
|
||||
|
||||
Raises:
|
||||
Exception: If quality validation fails
|
||||
"""
|
||||
|
||||
metadata = input_data['metadata']
|
||||
|
||||
filters = input_data['filters']
|
||||
data = DataFrame(input_data['data'])
|
||||
model_tags = input_data['model_tags']
|
||||
|
||||
self.info(f'Applying quality gate to data to {len(data)} rows', metadata=metadata)
|
||||
|
||||
tags = list(model_tags.keys())
|
||||
|
||||
data = data[data['name'].isin(tags)]
|
||||
|
||||
for filter_name, config in filters.items():
|
||||
policy = config['policy']
|
||||
if filter_name not in quality_gate_filters:
|
||||
self.warning(f'Filter {filter_name} not found', metadata=metadata)
|
||||
continue
|
||||
|
||||
try:
|
||||
filtered_data = quality_gate_filters[filter_name](data, model_tags)
|
||||
|
||||
except Exception as e:
|
||||
trace = traceback.format_exc()
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='DATA_QUALITY_GATE_ISSUES',
|
||||
message=f'Error applying filter {filter_name}: {e}',
|
||||
block='data_quality_gate',
|
||||
level=NotificationLevel.ERROR,
|
||||
attachment_content=trace,
|
||||
)
|
||||
|
||||
self.error(trace, metadata=metadata)
|
||||
|
||||
else:
|
||||
if filtered_data.empty:
|
||||
continue
|
||||
|
||||
message = f'{len(filtered_data)} rows has quality issues: {filter_name}: {policy}'
|
||||
attachment = filtered_data.to_string()
|
||||
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id=f'DATA_QUALITY_GATE_ISSUES__{filter_name}',
|
||||
message=message,
|
||||
block='data_quality_gate',
|
||||
level=NotificationLevel.WARNING,
|
||||
attachment_content=attachment,
|
||||
)
|
||||
|
||||
if policy == 'DISCARD':
|
||||
data = data[~data.index.isin(filtered_data.index)]
|
||||
|
||||
self.info(f'Data quality gate applied, final data has {len(data)} rows', metadata=metadata)
|
||||
|
||||
return data.to_dict()
|
||||
|
||||
@activity.defn(name='write_metrics')
|
||||
def write_metrics(self, input_data: dict[str, Any]) -> None:
|
||||
"""
|
||||
Write metrics to the database.
|
||||
input_data:
|
||||
metadata: dict[str, Any]
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
tag_values = DataFrame(input_data['tag_values'])
|
||||
|
||||
self.info(f'Writing metrics for {metadata["model_name"]}', metadata=metadata)
|
||||
|
||||
metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels(
|
||||
pod_id=self.pod_id,
|
||||
model_name=metadata['model_name'],
|
||||
workflow_name=metadata['workflow_name'],
|
||||
).inc()
|
||||
|
||||
# Register metrics
|
||||
for _, row in tag_values.iterrows():
|
||||
value = row['value']
|
||||
if value is not None:
|
||||
metrics.TAG_CHANGES_MONITOR.labels(
|
||||
pod_id=self.pod_id,
|
||||
model_name=metadata['model_name'],
|
||||
workflow_name=metadata['workflow_name'],
|
||||
tag_name=row['variable'],
|
||||
).set(row['value'])
|
||||
|
||||
self.info(f'Metrics written for {metadata["model_name"]}', metadata=metadata)
|
||||
Reference in New Issue
Block a user