320 lines
11 KiB
Python
320 lines
11 KiB
Python
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)
|