gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

SIENTIAPDE-1182

Browse Source 
Remove Docker configuration files and refactor project structure

- Deleted docker-compose.yml and Dockerfile as part of the project restructuring.
- Updated README.md to reflect changes in project setup and configuration.
- Introduced a new __init__.py file in the laborious package to provide an overview of the system.
- Enhanced documentation across various modules, including metrics, activities, and workflows, to improve clarity and usability.
- Added comprehensive docstrings and comments to key classes and methods for better maintainability.
This commit is contained in:
vitor-aignosi
2025-08-29 13:22:48 -03:00
parent 30c1d6746a
commit 995ba7900a
24 changed files with 2583 additions and 533 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
laborious/activities/__init__.py
View File

@@ -0,0 +1,10 @@
"""
Laborious Activities Package
This package contains all Temporal activity implementations for the Laborious system,
including data quality gates, MLFlow operations, OPC server integration, and
database operations.
Activities are the building blocks of workflows and implement the actual business
logic for data processing, ML model inference, and data export operations.
"""

52
laborious/activities/activities.py
View File

@@ -11,13 +11,52 @@ with workflow.unsafe.imports_passed_through():
class Activities(Postgres, MLFlow, Gates, OPC):
"""
Main activities orchestrator for the Laborious system.
This class combines functionality from multiple activity classes to provide
a unified interface for all workflow operations. It manages database connections,
MLFlow model interactions, data quality validation, and OPC server communications.
The class implements multiple inheritance to combine specialized functionality:
- Postgres: Database operations and data persistence
- MLFlow: Model inference and transformation operations
- Gates: Data quality validation and filtering mechanisms
- OPC: Real-time data export to OPC servers
Attributes:
postgres_config (dict): PostgreSQL connection configuration
mlflow_config (dict): MLFlow server configuration
opc_config (dict): OPC server configuration
logger (Logger): Logging and observability instance
notification_handler (NotificationHandler): Notification management instance
"""
def __init__(self,
postgres_config: dict[str, Any],
mlflow_config: dict[str, Any],
opc_config: dict[str, Any],
logger: Logger, notification_handler: NotificationHandler):
logger: Logger,
notification_handler: NotificationHandler):
"""
Initialize the Activities orchestrator with all required configurations.
This constructor initializes all parent classes with their respective
configurations and sets up the foundation for all activity operations.
Args:
postgres_config: PostgreSQL connection configuration dictionary
Required keys: host, port, user, password, dbname, min_connections, max_connections
mlflow_config: MLFlow server configuration dictionary
Required keys: host, port, username, password
opc_config: OPC server configuration dictionary
Can contain multiple server configurations
logger: Logger instance for observability and debugging
notification_handler: Notification handler for alerts and monitoring
Raises:
Exception: If any parent class initialization fails
"""
# Initialize parent classes
Postgres.__init__(self, host=postgres_config['host'],
port=postgres_config['port'],
@@ -45,5 +84,16 @@ class Activities(Postgres, MLFlow, Gates, OPC):
notification_handler=notification_handler)
async def shutdown(self):
"""
Gracefully shutdown all activities and clean up resources.
This method ensures proper cleanup of all resources including:
- PostgreSQL connection pools
- OPC server connections
- Any other resources that need explicit cleanup
The method should be called before the application terminates to ensure
proper resource cleanup and prevent resource leaks.
"""
Postgres.close(self)
await OPC.shutdown(self)

276
laborious/activities/gates.py
View File

@@ -17,6 +17,7 @@ with workflow.unsafe.imports_passed_through():
from pandas import DataFrame
from laborious import metrics
# Input filter function mappings
input_filter_functions = {
'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
'EMPTY_DATA': filter_empty_data,
@@ -27,6 +28,7 @@ input_filter_functions = {
}
}
# MLFlow response filter function mappings
mlflow_response_filter_functions = {
'API_ERROR': api_error_filter,
'path_confidence': {
@@ -36,6 +38,7 @@ mlflow_response_filter_functions = {
},
}
# MLFlow content filter function mappings
mlflow_content_filter_functions = {
'NAN_VALUES': nan_values_filter,
'path_confidence': {
@@ -47,24 +50,72 @@ mlflow_content_filter_functions = {
class Gates(BaseActivity):
"""
Data quality gates and filtering activities for the Laborious system.
This class implements comprehensive data quality validation and filtering
mechanisms that can be applied at different stages of the prediction pipeline.
It provides configurable filters with policy-based decision making to ensure
data integrity and quality throughout the ML workflow.
The class supports multiple filter types and implements a flexible policy
system that can be configured for different validation requirements. Each
filter returns a path decision (STOP, CONTINUE, REPEAT) along with confidence
scores and detailed comments for monitoring and debugging.
Attributes:
input_filter_functions (dict): Mapping of input filter names to functions
mlflow_response_filter_functions (dict): Mapping of MLFlow response filter names to functions
mlflow_content_filter_functions (dict): Mapping of MLFlow content filter names to functions
"""
def __init__(self, logger: Logger, notification_handler: NotificationHandler):
"""
Initialize data quality gates with logging and notification capabilities.
Args:
logger: Logger instance for observability and debugging
notification_handler: Notification handler for alerts and monitoring
Raises:
Exception: If BaseActivity initialization fails
"""
BaseActivity.__init__(
self, logger, notification_handler, set_error_counter=True)
@activity.defn(name="input_gate")
async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
"""
Filters the data based on the filters. The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Apply input data quality filters and validation.
This activity validates input data quality using configurable filters
before proceeding with ML operations. It applies multiple filter types
and returns a path decision based on the filter results and configured
policies.
The method implements a comprehensive filtering system that:
1. Applies configured filters to input data
2. Evaluates filter results against policy configurations
3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
4. Provides confidence scores and detailed comments
5. Handles errors gracefully with notification integration
Args:
- input_data (dict): The input data. Contains:
- filters (dict): The filters to apply.
The key is the filter name and the value is the filter configuration.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority.
input_data: Configuration and data for input validation
Required keys:
- metadata (dict): Workflow execution metadata
- filters (dict): Filter configuration and policies
- data (dict): Input data to validate
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
list and filter configuration and functions.
tuple: (path_flag, confidence, comment)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
Raises:
Exception: If filter execution fails or configuration is invalid
"""
metadata = input_data['metadata']
@@ -81,6 +132,7 @@ class Gates(BaseActivity):
self.debug(f"Input data:\n {data}", metadata)
self.debug(f"Filters: {filters}", metadata)
# Apply each configured filter
for fil, config in filters.items():
if fil not in input_filter_functions:
self.error(f"Filter {fil} not found", metadata)
@@ -113,20 +165,37 @@ class Gates(BaseActivity):
@activity.defn(name="mlflow_response_gate")
async def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
"""
Filters the data based on the mlflow response filters.
The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Args:
- input_data (dict): The input data. Contains:
- filters (dict): The filter configuration to apply.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority list.
- type (str): The type of the gate.
Returns:
tuple[str | None, int, str]: (policy, confidence, comments) based in priority list
and filter configuration and functions.
"""
Validate MLFlow API response quality and integrity.
This activity validates MLFlow API responses to ensure they meet quality
standards before proceeding with further processing. It applies response-specific
filters and determines appropriate path decisions based on response quality.
The method implements response validation that:
1. Applies MLFlow response-specific filters
2. Evaluates API response quality and integrity
3. Determines path decisions based on response validation results
4. Provides confidence scores and detailed validation comments
5. Handles API errors and response validation failures
Args:
input_data: Configuration and data for response validation
Required keys:
- metadata (dict): Workflow execution metadata
- filters (dict): Response filter configuration and policies
- data (dict): MLFlow API response data to validate
- type (str): Type of MLFlow operation (transform, predict)
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple: (path_flag, confidence, comment)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
Raises:
Exception: If response validation fails or configuration is invalid
"""
metadata = input_data['metadata']
self.info("Performing mlflow response gate...", metadata)
@@ -143,6 +212,7 @@ class Gates(BaseActivity):
comments = []
for fil, config in filters.items():
if fil not in mlflow_response_filter_functions:
self.error(f"Filter {fil} not found", metadata)
continue
try:
if mlflow_response_filter_functions[fil](data, config):
@@ -180,20 +250,37 @@ class Gates(BaseActivity):
@activity.defn(name="mlflow_content_gate")
async def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
"""
Filters the data based on the mlflow content filters.
The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Args:
- input_data (dict): The input data. Contains:
- filters (dict): The filter configuration to apply.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority list.
- type (str): The type of the gate.
Returns:
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
list and filter configuration and functions.
"""
Validate MLFlow prediction content quality and integrity.
This activity validates the content of MLFlow predictions to ensure they
meet quality standards before export and persistence. It applies content-specific
filters and determines appropriate path decisions based on content quality.
The method implements content validation that:
1. Applies MLFlow content-specific filters
2. Evaluates prediction content quality and integrity
3. Determines path decisions based on content validation results
4. Provides confidence scores and detailed validation comments
5. Handles content validation failures and quality issues
Args:
input_data: Configuration and data for content validation
Required keys:
- metadata (dict): Workflow execution metadata
- filters (dict): Content filter configuration and policies
- data (dict): MLFlow prediction content to validate
- type (str): Type of MLFlow operation (transform, predict)
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple: (path_flag, confidence, comment)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
Raises:
Exception: If content validation fails or configuration is invalid
"""
metadata = input_data['metadata']
self.info("Performing mlflow content gate...", metadata)
@@ -245,6 +332,27 @@ class Gates(BaseActivity):
def get_prediction_store_policy(self,
prediction_store_policy: str,
metadata: dict[str, Any]) -> tuple[str, int]:
"""
Parse and validate prediction store policy configuration.
This method parses prediction store policy strings in the format 'type:value'
and validates them against allowed policy types and values. It provides
sensible defaults for invalid configurations and logs policy validation
failures for operational monitoring.
Supported Policy Types:
- 'lts': Latest timestamp - sorts data by timestamp descending
- 'erl': Earliest timestamp - sorts data by timestamp ascending
Args:
prediction_store_policy (str): Policy string in format 'type:value'
metadata (dict[str, Any]): Context metadata for logging and notifications
Returns:
tuple[str, int]: (policy_type, policy_value)
- policy_type (str): Validated policy type ('lts' or 'erl')
- policy_value (int): Number of rows to retain
"""
policy_elements = prediction_store_policy.split(':')
if len(policy_elements) < 2:
@@ -267,16 +375,27 @@ class Gates(BaseActivity):
@activity.defn(name="format_prediction")
async def format_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
"""
Formats the prediction data.
Format prediction data according to configured storage policies.
This method formats prediction data for storage and export operations.
It applies timestamp-based sorting policies, adds metadata fields,
and ensures data consistency before persistence. The method supports
multiple storage policies for flexible data retention strategies.
Storage Policies:
- 'lts:N': Latest timestamp - retains N most recent predictions
- 'erl:N': Earliest timestamp - retains N oldest predictions
Args:
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to format.
- timestamp (str): The timestamp of the data.
- model_id (str): The id of the model.
- prediction_confidence (float): The confidence of the prediction.
- prediction_store_policy (str): The policy to store the prediction.
input_data (dict): Input data containing:
- data (dict[str, Any]): Raw prediction data to format
- timestamp (str): Default timestamp if data lacks timestamp column
- model_id (str): Unique identifier for the ML model
- prediction_confidence (float): Confidence score for the prediction
- prediction_store_policy (str): Storage policy in format 'type:value'
Returns:
dict: The formatted data.
dict: Formatted prediction data ready for storage and export
"""
metadata = input_data['metadata']
prediction_store_policy = input_data['prediction_store_policy']
@@ -332,17 +451,28 @@ class Gates(BaseActivity):
@activity.defn(name="format_default_prediction")
async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
"""
Creates and formats the default prediction data, with zero value in prediction,
and usefull information in the other fields.
Create and format default prediction data for error conditions.
This method generates default prediction data when the main prediction
pipeline encounters errors or quality issues. It creates a standardized
data structure with zero values for predictions and useful metadata
for operational monitoring and debugging.
The default prediction serves as a fallback mechanism to:
1. Maintain data pipeline continuity during failures
2. Provide operational visibility into prediction quality issues
3. Enable downstream systems to handle error conditions gracefully
4. Support debugging and troubleshooting efforts
Args:
- input_data (dict): The input data. Contains:
- timestamp (str): The timestamp of the data.
- model_id (str): The id of the model.
- prediction_confidence (float): The confidence of the prediction.
- comment (str): The comment of the prediction.
input_data (dict): Input data containing:
- timestamp (str): Timestamp for the default prediction
- model_id (str): Unique identifier for the ML model
- prediction_confidence (float): Confidence score (typically low for errors)
- comment (str): Error description or operational comment
Returns:
dict: The formatted data.
dict: Formatted default prediction data with error indicators
"""
metadata = input_data['metadata']
@@ -364,12 +494,25 @@ class Gates(BaseActivity):
@activity.defn(name="get_last_timestamp")
async def get_last_timestamp(self, input_data: dict[str, Any]) -> str:
"""
Gets the last timestamp of the data.
Extract the most recent timestamp from prediction data.
This method analyzes prediction data to find the latest timestamp,
enabling incremental processing and data continuity tracking.
It handles empty datasets gracefully by returning the current time
as a fallback timestamp.
The method is essential for:
1. Incremental data processing workflows
2. Data continuity validation
3. Timestamp-based data loading optimization
4. Workflow execution tracking
Args:
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to get the last timestamp from.
input_data (dict): Input data containing:
- data (dict[str, Any]): Prediction data to analyze
Returns:
str: The last timestamp of the data.
str: Formatted timestamp string in UTC with timezone
"""
metadata = input_data['metadata']
@@ -393,10 +536,25 @@ class Gates(BaseActivity):
@activity.defn(name="write_metrics")
async def write_metrics(self, input_data: dict[str, Any]):
"""
Write metrics to the database.
input_data:
metadata: dict[str, Any]
prediction: dict[str, Any]
Write prediction performance metrics to Prometheus monitoring system.
This method records comprehensive metrics for prediction operations,
enabling operational monitoring, performance analysis, and alerting.
It tracks prediction counts, confidence levels, and response times
for each model and pipeline combination.
Metrics Recorded:
1. Prediction Count: Incremental counter for successful predictions
2. Confidence Monitor: Current confidence level for predictions
3. Response Time Monitor: Histogram of prediction response times
Args:
input_data (dict): Input data containing:
- metadata (dict[str, Any]): Workflow execution metadata
- prediction (dict[str, Any]): Prediction data with metrics
Raises:
Exception: If metrics writing fails or configuration is invalid
"""
metadata = input_data['metadata']
prediction = DataFrame(input_data['prediction'])

168
laborious/activities/mlflow.py
View File

@@ -15,8 +15,40 @@ with workflow.unsafe.imports_passed_through():
class MLFlow(BaseActivity):
"""
MLFlow integration activities for model inference operations.
This class provides activities for interacting with MLFlow models, including
data transformation and prediction operations. It handles authentication,
data preprocessing, and model management with configurable retention policies.
The class implements comprehensive error handling and logging for all
MLFlow operations, ensuring reliable model inference in production environments.
Attributes:
mlflow_host (str): MLFlow server hostname
mlflow_port (int): MLFlow server port
mlflow_username (str): MLFlow authentication username
mlflow_password (str): MLFlow authentication password
model_monitoring_repository (MLFlowRepository): Repository for MLFlow operations
"""
def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str,
mlflow_password: str, logger: Logger, notification_handler: NotificationHandler):
"""
Initialize MLFlow activities with server configuration.
Args:
mlflow_host: MLFlow server hostname or IP address
mlflow_port: MLFlow server port number
mlflow_username: Username for MLFlow authentication
mlflow_password: Password for MLFlow authentication
logger: Logger instance for observability and debugging
notification_handler: Notification handler for alerts and monitoring
Raises:
Exception: If MLFlowRepository initialization fails
"""
BaseActivity.__init__(
self, logger, notification_handler, set_error_counter=True)
self.mlflow_host = mlflow_host
@@ -31,14 +63,32 @@ class MLFlow(BaseActivity):
@activity.defn(name="request_transform")
async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Access MLFlow model to get the transformed data.
Transform input data using MLFlow models.
This activity processes input data through MLFlow model transformation,
including data preprocessing, format conversion, and validation. It handles
data deduplication, pivoting, and cleanup to ensure optimal model performance.
The transformation process includes:
1. Data deduplication based on variable and timestamp
2. Data pivoting for model input format
3. Null value handling and cleanup
4. MLFlow model transformation request
5. Response validation and logging
Args:
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to transform.
- model_name (str): The name of the model.
- model_retention (int): The retention time of the model, in minutes.
input_data: Configuration and data for transformation
Required keys:
- metadata (dict): Workflow execution metadata
- data (dict): Input data for transformation
- model_name (str): Name of the MLFlow model to use
- model_retention (int): Model retention period in minutes
Returns:
dict[str, Any]: The transformed data.
dict: Transformed data from MLFlow model
Raises:
Exception: If transformation fails or MLFlow model is unavailable
"""
metadata = input_data['metadata']
self.info('Transforming data...', metadata)
@@ -54,6 +104,7 @@ class MLFlow(BaseActivity):
subset=['variable', 'timestamp'], keep='first'
)
# Pivot data for model input format
data = data.pivot(
index='timestamp', columns='variable',
values='value')
@@ -64,6 +115,7 @@ class MLFlow(BaseActivity):
self.debug("Processed input data:", metadata)
self.debug(data, metadata)
# Request transformation from MLFlow model
response_data = self.model_monitoring_repository.transform(
model_name, data, model_retention)
@@ -77,14 +129,32 @@ class MLFlow(BaseActivity):
@activity.defn(name="request_predict")
async def request_predict(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Access MLFlow model to get the predicted data.
Execute predictions using MLFlow models.
This activity performs ML model inference using MLFlow models with the
transformed data. It handles data format conversion, null value processing,
and model prediction requests with comprehensive error handling.
The prediction process includes:
1. Data format validation and cleanup
2. Null value handling for model compatibility
3. MLFlow model prediction request
4. Response validation and logging
5. Performance monitoring and metrics
Args:
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to predict.
- model_name (str): The name of the model.
- model_retention (int): The retention time of the model, in minutes.
input_data: Configuration and data for prediction
Required keys:
- metadata (dict): Workflow execution metadata
- data (dict): Transformed data for prediction
- model_name (str): Name of the MLFlow model to use
- model_retention (int): Model retention period in minutes
Returns:
dict[str, Any]: The predicted data.
dict: Prediction results from MLFlow model
Raises:
Exception: If prediction fails or MLFlow model is unavailable
"""
metadata = input_data['metadata']
self.info('Predicting data...', metadata)
@@ -94,26 +164,51 @@ class MLFlow(BaseActivity):
self.debug(data, metadata)
# Convert numpy.nan to None for model compatibility
data.replace(np.nan, None, inplace=True)
# Request prediction from MLFlow model
response_data = self.model_monitoring_repository.predict(
model_name, data, model_retention)
self.debug("Prediction response data:", metadata)
self.debug(json.dumps(response_data, indent=4), metadata)
self.info("Prediction completed successfully", metadata)
self.info("Data predicted successfully", metadata)
return response_data
@activity.defn(name="retrain_model")
async def retrain_model(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Retrain the model.
Retrain MLFlow models with updated training data.
This activity orchestrates the complete model retraining process,
including data preparation, model retraining execution, and result
validation. It handles data preprocessing, column cleanup, and
comprehensive error handling for production model management.
The retraining process includes:
1. Data timestamp extraction and validation
2. Column cleanup and data preparation
3. Data pivoting for model input format
4. MLFlow model retraining execution
5. Result validation and error handling
Args:
- input_data (dict): The input data. Contains:
- model_name (str): The name of the model.
- data (dict[str, Any]): The data to retrain the model.
input_data (dict): Input data containing:
- metadata (dict): Workflow execution metadata
- data (dict[str, Any]): Training data for model retraining
- model_name (str): Name of the MLFlow model to retrain
Returns:
dict: Retraining results containing:
- status (str): Retraining operation status
- timestamp (str): Timestamp of the retraining operation
- experiment (str): MLFlow experiment identifier
Raises:
Exception: If retraining fails or encounters critical errors
"""
metadata = input_data['metadata']
data = DataFrame(input_data['data'])
@@ -162,16 +257,39 @@ class MLFlow(BaseActivity):
@activity.defn(name="update_production_model")
async def update_production_model(self, input_data: dict[str, Any]) -> dict[Any, Any]:
"""
Update the production model.
Update production model with newly trained model version.
This activity manages the critical process of updating production
models with newly trained versions. It handles model deployment,
status tracking, and comprehensive reporting for operational
visibility and audit trails.
The update process includes:
1. Production model update execution
2. Status and metadata tracking
3. Comprehensive reporting and logging
4. Error handling and notification
5. Audit trail maintenance
Args:
- input_data (dict): The input data. Contains:
- model_name (str): The name of the model.
- experiment (str): The name of the experiment.
- model_id (str): The id of the model.
- timestamp (str): The timestamp of the model.
- status (str): The status of the model.
input_data (dict): Input data containing:
- metadata (dict): Workflow execution metadata
- model_name (str): Name of the MLFlow model to update
- experiment (str): MLFlow experiment identifier
- model_id (str): Unique identifier for the model version
- timestamp (str): Timestamp of the update operation
- status (str): Current status of the model update
Returns:
dict[Any, Any]: The report of the model.
dict[Any, Any]: Comprehensive update report containing:
- model_id (str): Model version identifier
- model_name (str): Name of the updated model
- timestamp (str): Update operation timestamp
- status (str): Update operation status
- Additional MLFlow response metadata
Raises:
Exception: If production model update fails
"""
metadata = input_data['metadata']
model_name = input_data['model_name']

136
laborious/activities/opc.py
View File

@@ -15,6 +15,24 @@ OPC_WRITTING_ERROR_CONFIDENCE = 12
class OPC(BaseActivity):
"""
OPC server integration activities for real-time data export.
This class provides comprehensive OPC UA client functionality for connecting
to multiple OPC servers and writing prediction data in real-time. It implements
secure communication with certificate-based authentication and automatic
reconnection capabilities.
The class supports multiple OPC servers with individual configurations and
provides robust error handling and monitoring for production environments.
Attributes:
opc_servers (dict): Configuration for multiple OPC servers
opc_repository (dict): Active OPC repository connections
logger (Logger): Logging and observability instance
notification_handler (NotificationHandler): Notification management instance
"""
def __init__(self, opc_servers: dict[str, dict[str, Any]],
logger: Logger, notification_handler: NotificationHandler):
@@ -29,7 +47,29 @@ class OPC(BaseActivity):
self.opc_servers = opc_servers
async def init_opc(self):
"""
Initialize OPC server connections and establish communication channels.
This method iterates through all configured OPC servers and attempts to
establish secure connections using certificate-based authentication.
Each server connection is managed independently, and connection failures
are reported through the notification system.
The method performs the following operations:
1. Creates OpcRepository instances for each configured server
2. Establishes secure connections with certificate validation
3. Reports connection success/failure through notifications
4. Logs connection status for operational visibility
Raises:
Exception: If OPC repository initialization fails or connection
establishment encounters critical errors
Note:
Connection failures are logged and reported but do not prevent
the initialization of other OPC servers. Each server is handled
independently to ensure maximum availability.
"""
self.logger.info("Initializing OPC servers...")
for id, server in self.opc_servers.items():
self.opc_repository[id] = OpcRepository(
@@ -67,7 +107,12 @@ class OPC(BaseActivity):
async def write_data(self, server_id: str, tag: str, data: Any,
data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool:
"""
Write data to OPC server.
Write data to a specific OPC server tag with comprehensive error handling.
This method provides a secure and reliable way to write data to OPC servers
with automatic error handling, notification integration, and detailed logging.
It validates server availability before attempting write operations and
provides comprehensive error reporting for operational monitoring.
Args:
- server_id (str): The id of the OPC server.
@@ -108,6 +153,26 @@ class OPC(BaseActivity):
raise e
def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
"""
Validate that an OPC server is available and configured for write operations.
This method checks if the specified OPC server exists in the active
repository and is available for data writing operations. It provides
immediate feedback for server availability and logs validation failures
for operational monitoring.
Args:
server_id (str): Unique identifier for the OPC server to validate
metadata (dict[str, Any]): Context metadata for logging and notifications
Returns:
bool: True if server is available, False otherwise
Note:
Server validation failures are automatically reported through the
notification system with detailed information about available servers.
This helps operators quickly identify configuration issues.
"""
if self.opc_repository.get(server_id) is None:
message = f"OPC server {server_id} not found to perform write operation."
self.send_notification(
@@ -124,6 +189,32 @@ class OPC(BaseActivity):
async def manage_output_tags(
self, server_id: str, config: dict[str, Any], data: DataFrame,
metadata: dict[str, Any], success: bool) -> tuple[bool, int]:
"""
Manage the writing of prediction and confidence data to OPC server tags.
This method orchestrates the writing of multiple data types to OPC servers
based on configuration. It handles both prediction data and confidence
values independently, allowing for flexible tag configuration and
comprehensive error handling.
The method supports two main tag types:
1. Prediction tags: Write actual prediction values to configured OPC tags
2. Confidence tags: Write confidence scores to separate OPC tags
Args:
server_id (str): Unique identifier for the target OPC server
config (dict[str, Any]): OPC tag configuration containing:
- prediction_tags (dict, optional): Prediction tag configurations
- confidence_tags (dict, optional): Confidence tag configurations
data (DataFrame): DataFrame containing prediction and confidence data
metadata (dict[str, Any]): Context metadata for logging and notifications
success (bool): Current success status to maintain across operations
Returns:
tuple[bool, int]: (overall_success, total_tags_written)
- overall_success: True if all configured tags were written successfully
- total_tags_written: Count of successfully written tags
"""
count = 0
if 'prediction_tags' in config:
@@ -204,17 +295,29 @@ class OPC(BaseActivity):
def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]:
"""
Processes the confidence of OPC server write operations and updates the DataFrame accordingly.
Process prediction confidence based on OPC write operation success.
If the write operation was not successful, sets the 'prediction_confidence' column in the DataFrame
to a predefined error confidence value and logs a debug message. Otherwise, logs a success message.
This method updates the prediction confidence values in the DataFrame
based on the success status of OPC server write operations. If any
write operations failed, it sets the confidence to a predefined error
value to indicate data quality issues.
The method implements a confidence degradation strategy:
- Success: Maintains original confidence values
- Failure: Sets confidence to error value for operational awareness
Args:
data (DataFrame): The DataFrame containing the data to be processed.
success (bool): Indicates whether the data was successfully written to the OPC servers.
data (DataFrame): DataFrame containing prediction and confidence data
success (bool): Overall success status of OPC write operations
metadata (dict[str, Any]): Context metadata for logging and notifications
Returns:
dict[Any, Any]: The processed data as a dictionary.
dict[Any, Any]: Processed data as a dictionary with updated confidence values
Note:
The error confidence value (OPC_WRITTING_ERROR_CONFIDENCE = 12) is
used to indicate that data was not successfully exported to OPC servers.
This allows downstream systems to handle data quality appropriately.
"""
if not success:
@@ -230,5 +333,24 @@ class OPC(BaseActivity):
return data.to_dict()
async def shutdown(self):
"""
Gracefully shutdown all OPC server connections and cleanup resources.
This method ensures proper cleanup of all active OPC server connections
by calling the disconnect method on each repository instance. It's
designed to be called during application shutdown to prevent resource
leaks and ensure clean termination.
The method performs the following cleanup operations:
1. Iterates through all active OPC repository connections
2. Calls disconnect() on each repository instance
3. Allows for graceful connection termination
4. Prevents resource leaks and connection hanging
Note:
This method should be called during application shutdown to ensure
proper cleanup. It handles all active connections regardless of
their current state and provides a clean shutdown experience.
"""
for opc in self.opc_repository.values():
await opc.disconnect()