From bac17579d68a4665ce99fae85315cb933ed6c937 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 3 Sep 2025 08:42:24 -0300 Subject: [PATCH] SIENTIAPDE-1084 Refactor connectors_config.py and conditional_filters.py for improved configuration management and data filtering - Updated PostgreSQL and MLFlow configuration functions to enhance default values and environment variable handling. - Simplified OPC server configuration logic and improved MongoDB connection string construction. - Refactored conditional filters to streamline null value checks and empty data validation, removing unnecessary comments and examples for clarity. - Removed extensive module docstrings to enhance code readability. --- laborious/utils/connectors_config.py | 239 +++++------------- .../utils/filters/conditional_filters.py | 216 +++------------- laborious/workflows/predictions_batch.py | 16 -- 3 files changed, 97 insertions(+), 374 deletions(-) diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index d4688eb..145f248 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -1,205 +1,99 @@ -""" -Connectors Configuration Module - -This module provides configuration management for all external service connectors -used by the Sientia DataOps Laborious system. It centralizes configuration -for databases, MLFlow servers, OPC servers, and other external dependencies. - -The module implements configuration builders for: -1. PostgreSQL database connections -2. MLFlow model serving endpoints -3. OPC server configurations -4. MongoDB notification systems - -Key Features: -- Environment variable-based configuration -- Default value management for development -- Connection pool configuration -- Security credential management -- Configuration validation and error handling -- Support for multiple service instances - -Configuration Sources: -- Environment variables for production deployment -- Default values for local development -- Kubernetes secrets integration -- Configurable connection parameters - -Environment Variables: -- POSTGRES_*: PostgreSQL connection parameters -- MLFLOW_*: MLFlow server parameters -- OPC_*: OPC server configuration -- MONGODB_*: MongoDB connection parameters - -Dependencies: -- os: Environment variable access -- typing: Type hints and annotations -""" - -import os +from os import getenv +import json from typing import Dict, Any def build_postgres_config() -> Dict[str, Any]: """ Build PostgreSQL database configuration from environment variables. - + This function constructs a PostgreSQL configuration dictionary from environment variables with sensible defaults for local development. It handles connection pool configuration and security parameters. - + Environment Variables: POSTGRES_HOST: Database hostname (default: localhost) POSTGRES_PORT: Database port (default: 5432) POSTGRES_USER: Database username (default: sientia) POSTGRES_PASSWORD: Database password (default: sientia) POSTGRES_DBNAME: Database name (default: sientia) - POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 1) - POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 10) - + POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 5) + POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 20) + Returns: dict: PostgreSQL configuration dictionary with all required parameters - - Example: - >>> config = build_postgres_config() - >>> print(config) - { - 'host': 'localhost', - 'port': 5432, - 'user': 'sientia', - 'password': 'sientia', - 'dbname': 'sientia', - 'min_connections': 1, - 'max_connections': 10 - } - - Note: - In production, ensure all required environment variables are set - with appropriate values for your database environment. """ return { - 'host': os.getenv('POSTGRES_HOST', 'localhost'), - 'port': int(os.getenv('POSTGRES_PORT', '5432')), - 'user': os.getenv('POSTGRES_USER', 'sientia'), - 'password': os.getenv('POSTGRES_PASSWORD', 'sientia'), - 'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'), - 'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '1')), - 'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '10')) + 'host': getenv('POSTGRES_HOST', 'localhost'), + 'port': int(getenv('POSTGRES_PORT', '5432')), + 'user': getenv('POSTGRES_USER', 'sientia'), + 'password': getenv('POSTGRES_PASSWORD', 'sientia'), + 'dbname': getenv('POSTGRES_DBNAME', 'sientia'), + 'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')), + 'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')) } def build_mlflow_config() -> Dict[str, Any]: """ Build MLFlow server configuration from environment variables. - + This function constructs an MLFlow configuration dictionary from environment variables with sensible defaults for local development. It handles server connection and authentication parameters. - + Environment Variables: - MLFLOW_HOST: MLFlow server hostname (default: localhost) - MLFLOW_PORT: MLFlow server port (default: 5000) - MLFLOW_USERNAME: MLFlow username (default: admin) - MLFLOW_PASSWORD: MLFlow password (default: admin) - + MLFLOW_HOST: MLFlow server hostname (default: http://localhost) + MLFLOW_PORT: MLFlow server port (default: 5080) + MLFLOW_USERNAME: MLFlow username (default: aignosi) + MLFLOW_PASSWORD: MLFlow password (default: aignosi) + Returns: dict: MLFlow configuration dictionary with all required parameters - - Example: - >>> config = build_mlflow_config() - >>> print(config) - { - 'host': 'localhost', - 'port': 5000, - 'username': 'admin', - 'password': 'admin' - } - - Note: - In production, ensure all required environment variables are set - with appropriate values for your MLFlow server environment. - Consider using secure authentication methods for production deployments. """ return { - 'host': os.getenv('MLFLOW_HOST', 'localhost'), - 'port': int(os.getenv('MLFLOW_PORT', '5000')), - 'username': os.getenv('MLFLOW_USERNAME', 'admin'), - 'password': os.getenv('MLFLOW_PASSWORD', 'admin') + 'host': getenv('MLFLOW_HOST', 'http://localhost'), + 'port': int(getenv('MLFLOW_PORT', '5080')), + 'username': getenv('MLFLOW_USERNAME', 'aignosi'), + 'password': getenv('MLFLOW_PASSWORD', 'aignosi') } def build_opc_config() -> Dict[str, Any]: """ Build OPC server configuration from environment variables. - + This function constructs an OPC server configuration dictionary from environment variables. It supports both single server and multi-server configurations with flexible parameter handling. - + Environment Variables: OPC_CONFIG: JSON string containing multiple OPC server configurations - OPC_URL: Single OPC server URL (fallback) - OPC_NAME: Single OPC server name (fallback) - OPC_SERVER_URI: Single OPC server URI (fallback) - OPC_CERT_PATH: Client certificate path (fallback) - OPC_PRIVATE_KEY_PATH: Client private key path (fallback) - OPC_SERVER_CERT_PATH: Server certificate path (fallback) - OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback) - + OPC_ID: OPC server ID (fallback, default: 1) + OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840) + OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840) + OPC_CERT_PATH: Client certificate path (fallback, default: None) + OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None) + OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None) + OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120) + Returns: dict: OPC server configuration dictionary - - Configuration Modes: - 1. Multi-server: Use OPC_CONFIG environment variable with JSON string - 2. Single server: Use individual OPC_* environment variables - - Example Multi-server Configuration: - >>> # Set OPC_CONFIG environment variable - >>> os.environ['OPC_CONFIG'] = ''' - ... { - ... "opc_server_1": { - ... "url": "opc.tcp://server1:4840", - ... "name": "Server1", - ... "server_uri": "urn:server1:opcua", - ... "cert_path": "/path/to/cert.pem", - ... "private_key_path": "/path/to/key.pem", - ... "server_cert_path": "/path/to/server_cert.pem", - ... "reconnection_interval": 5000 - ... } - ... } - ... ''' - >>> config = build_opc_config() - - Example Single Server Configuration: - >>> # Set individual environment variables - >>> os.environ['OPC_URL'] = 'opc.tcp://localhost:4840' - >>> os.environ['OPC_NAME'] = 'LocalServer' - >>> config = build_opc_config() - - Note: - For production deployments, prefer the OPC_CONFIG approach for - multiple servers and ensure all certificate paths are properly configured. """ - # Check for multi-server configuration - opc_config = os.getenv('OPC_CONFIG') - if opc_config: - try: - import json - return json.loads(opc_config) - except (json.JSONDecodeError, ImportError) as e: - # Fall back to single server configuration if JSON parsing fails - pass - - # Single server configuration fallback + opc_raw = getenv('OPC_CONFIG', None) + + if opc_raw: + return json.loads(opc_raw) + return { - 'default': { - 'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'), - 'name': os.getenv('OPC_NAME', 'DefaultServer'), - 'server_uri': os.getenv('OPC_SERVER_URI', 'urn:default:opcua'), - 'cert_path': os.getenv('OPC_CERT_PATH', ''), - 'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', ''), - 'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', ''), - 'reconnection_interval': int(os.getenv('OPC_RECONNECTION_INTERVAL', '5000')) + getenv('OPC_ID', '1'): { + 'id': getenv('OPC_ID', '1'), + 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), + 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), + 'cert_path': getenv('OPC_CERT_PATH', None), + 'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None), + 'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None), + 'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')) } } @@ -207,32 +101,29 @@ def build_opc_config() -> Dict[str, Any]: def build_mongodb_config() -> Dict[str, Any]: """ Build MongoDB configuration from environment variables. - + This function constructs a MongoDB configuration dictionary from environment variables with sensible defaults for local development. It handles connection string and database name configuration. - + Environment Variables: - MONGODB_URL: MongoDB connection URI (default: localhost:27017) - MONGODB_DATABASE: MongoDB database name (default: sientia) - + MONGODB_USERNAME: MongoDB username (default: root) + MONGODB_PASSWORD: MongoDB password (default: wKZDbMNU1c) + MONGODB_URL: MongoDB connection URI (default: localhost:27018) + MONGODB_DATABASE_NAME: MongoDB database name (default: sientia) + MONGODB_TTL_INDEX_HOURS: TTL index duration in hours (default: 1) + Returns: dict: MongoDB configuration dictionary with connection parameters - - Example: - >>> config = build_mongodb_config() - >>> print(config) - { - 'connection_string': 'localhost:27017', - 'database_name': 'sientia' - } - - Note: - In production, ensure the MONGODB_URL environment variable is set - with a proper MongoDB connection string including authentication - if required by your MongoDB deployment. """ + username = getenv('MONGODB_USERNAME', 'root') + password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c') + uri = getenv('MONGODB_URL', 'localhost:27018') + + connection_string = f'mongodb://{username}:{password}@{uri}' + return { - 'connection_string': os.getenv('MONGODB_URL', 'localhost:27017'), - 'database_name': os.getenv('MONGODB_DATABASE', 'sientia') + 'connection_string': connection_string, + 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), + 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600 } diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index 62f125f..2c51805 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -1,197 +1,45 @@ -""" -Conditional Data Filters Module - -This module provides conditional data filtering functions for the Sientia DataOps Laborious system. -It implements data quality validation filters that can be applied to input data before -ML operations to ensure data integrity and quality. - -The module implements filters for: -1. Empty data detection and validation -2. Specific variable null value checking -3. Configurable data quality rules -4. Flexible filter configuration - -Key Features: -- Configurable filter policies and thresholds -- Multiple data quality validation rules -- Flexible configuration options -- Comprehensive error handling -- Performance-optimized filtering - -Filter Types: -- EMPTY_DATA: Detects empty or insufficient data sets -- SPECIFIC_VARIABLES_NULL_VALUES: Validates specific variable null values -- Custom filters can be added for specific validation needs - -Dependencies: -- pandas.DataFrame: Data manipulation and processing -- typing: Type hints and annotations -""" - -from typing import Any, Dict, List from pandas import DataFrame -def filter_empty_data(data: DataFrame, config: Dict[str, Any]) -> bool: +def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool: """ - Filter data based on empty data conditions. - - This function checks if the input data meets minimum requirements for - processing. It can validate data size, completeness, and other quality - metrics to ensure sufficient data is available for ML operations. - - The filter implements multiple validation criteria: - 1. Data frame size validation - 2. Row count validation - 3. Column completeness validation - 4. Configurable threshold checking - + Filter to check if specific variables contain null values. + + This function examines a DataFrame to determine if any of the specified variables + contain null (NaN) values. It returns True if null values are found for any of + the specified variables, False otherwise. + Args: - data: Input data as pandas DataFrame - config: Filter configuration dictionary - Required keys: - - min_rows (int, optional): Minimum number of rows required - - min_columns (int, optional): Minimum number of columns required - - min_data_points (int, optional): Minimum total data points required - + data (DataFrame): The pandas DataFrame to be examined. Must contain columns + named 'variable' and 'value'. + config (dict): Configuration dictionary containing the following key: + - variables (list): List of variable names to check for null values + Returns: - bool: True if data should be filtered (fails quality check), False otherwise - - Filter Logic: - - Returns True (filter) if data is empty or below thresholds - - Returns False (pass) if data meets quality requirements - - Handles missing configuration gracefully with defaults - - Example: - >>> import pandas as pd - >>> df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]}) - >>> config = {'min_rows': 2, 'min_columns': 2} - >>> result = filter_empty_data(df, config) - >>> print(result) - False # Data passes filter - - >>> empty_df = pd.DataFrame() - >>> result = filter_empty_data(empty_df, config) - >>> print(result) - True # Data fails filter - - Default Thresholds: - - min_rows: 1 (at least one row required) - - min_columns: 1 (at least one column required) - - min_data_points: 1 (at least one data point required) + bool: True if any of the specified variables contain null values, + False if none of the specified variables contain null values. + """ - # Check if data is completely empty - if data.empty: - return True - - # Get configuration with defaults - min_rows = config.get('min_rows', 1) - min_columns = config.get('min_columns', 1) - min_data_points = config.get('min_data_points', 1) - - # Check row count - if len(data) < min_rows: - return True - - # Check column count - if len(data.columns) < min_columns: - return True - - # Check total data points - if data.size < min_data_points: - return True - - # Data passes all quality checks - return False + return not data[ + data['variable'].isin(config['variables']) & data['value'].isna()].empty -def filter_specific_variables_null_values(data: DataFrame, config: Dict[str, Any]) -> bool: +def filter_empty_data(data: DataFrame, _config: dict) -> bool: """ - Filter data based on null values in specific variables. - - This function checks for null values in specified variables and determines - if the data quality is sufficient for processing. It can validate - individual columns or groups of columns for data completeness. - - The filter implements variable-specific validation: - 1. Individual variable null value checking - 2. Configurable null value thresholds - 3. Multiple variable validation - 4. Flexible threshold configuration - + Filter to check if the DataFrame is empty. + + This function determines whether the provided DataFrame contains any data. + It's a simple utility function that can be used in conditional logic to + handle cases where no data is available. + Args: - data: Input data as pandas DataFrame - config: Filter configuration dictionary - Required keys: - - variables (list): List of variable names to check - - max_null_ratio (float, optional): Maximum allowed null value ratio (0.0 to 1.0) - - max_null_count (int, optional): Maximum allowed null value count - + data (DataFrame): The pandas DataFrame to be checked for emptiness. + _config (dict): Configuration dictionary (unused in this function). + The underscore prefix indicates this parameter is required for + interface consistency but not used in the implementation. + Returns: - bool: True if data should be filtered (fails quality check), False otherwise - - Filter Logic: - - Returns True (filter) if null value thresholds are exceeded - - Returns False (pass) if null values are within acceptable limits - - Handles missing variables gracefully - - Supports both ratio and count-based thresholds - - Example: - >>> import pandas as pd - >>> df = pd.DataFrame({ - ... 'temperature': [25.5, None, 27.0, 26.5], - ... 'humidity': [60.0, 65.0, None, 62.0] - ... }) - >>> config = { - ... 'variables': ['temperature', 'humidity'], - ... 'max_null_ratio': 0.25 - ... } - >>> result = filter_specific_variables_null_values(df, config) - >>> print(result) - False # Data passes filter (null ratio = 0.25, which equals max) - - >>> config = { - ... 'variables': ['temperature', 'humidity'], - ... 'max_null_ratio': 0.20 - ... } - >>> result = filter_specific_variables_null_values(df, config) - >>> print(result) - True # Data fails filter (null ratio = 0.25, exceeds max of 0.20) - - Default Thresholds: - - max_null_ratio: 0.5 (50% null values allowed) - - max_null_count: None (no count-based limit by default) - - Note: - If both max_null_ratio and max_null_count are specified, the filter - will trigger if either threshold is exceeded. + bool: True if the DataFrame is empty (has no rows), False if it contains data. + """ - # Get configuration - variables = config.get('variables', []) - max_null_ratio = config.get('max_null_ratio', 0.5) - max_null_count = config.get('max_null_count', None) - - # Check if variables exist in data - if not variables: - return False # No variables specified, pass filter - - # Validate each specified variable - for variable in variables: - if variable not in data.columns: - continue # Skip variables that don't exist in data - - # Calculate null value statistics - null_count = data[variable].isnull().sum() - total_count = len(data[variable]) - null_ratio = null_count / total_count if total_count > 0 else 0.0 - - # Check ratio threshold - if null_ratio > max_null_ratio: - return True - - # Check count threshold (if specified) - if max_null_count is not None and null_count > max_null_count: - return True - - # All variables pass null value checks - return False + return data.empty diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index e202b46..de1fbbe 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -29,22 +29,6 @@ class PredictionsBatch(): 2. Configuration Preparation: Sets up prediction parameters and filters 3. Workflow Delegation: Spawns PredictionProcess child workflow 4. Error Handling: Implements comprehensive error handling and retry policies - - Example: - >>> # Start the workflow - >>> await client.start_workflow( - ... PredictionsBatch.run, - ... id="batch_pred_001", - ... task_queue="predictions_batch-queue", - ... input_data={ - ... "schedule_name": "hourly_predictions", - ... "model_name": "temperature_model", - ... "model_id": "temp_001", - ... "query": "SELECT * FROM sensor_data WHERE timestamp > NOW() - INTERVAL '1 hour'", - ... "schema": {"timestamp": "datetime", "temperature": "float"}, - ... "table_name": "predictions" - ... } - ... ) """ @workflow.run