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

SIENTIAPDE-1243: Remove OPC server integration and add code quality tools.

Browse Source 
This commit removes the OPC server integration from the Model Manager, including related activities, repositories, metrics, and configuration. It also adds code quality tools such as Ruff (linting/formatting), mypy (type checking), and Bandit (security analysis) along with a validation script and CI/CD integration for automated code validation. The README has been updated to reflect these changes.
This commit is contained in:
Bruno Domingues
2025-10-01 16:25:08 -03:00
parent bc4d98f78d
commit b102f79087
24 changed files with 453 additions and 1810 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
model_manager/activities/activities.py
View File

@@ -6,28 +6,25 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.observability.logger import Logger
from model_manager.activities.mlflow import MLFlow
from model_manager.activities.gates import Gates
from model_manager.activities.opc import OPC
from typing import Any
class Activities(Postgres, MLFlow, Gates, OPC):
class Activities(Postgres, MLFlow, Gates):
"""
Main activities orchestrator for the Model Manager 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.
MLFlow model interactions, and data quality validation.
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
"""
@@ -35,7 +32,6 @@ class Activities(Postgres, MLFlow, Gates, OPC):
def __init__(self,
postgres_config: dict[str, Any],
mlflow_config: dict[str, Any],
opc_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler):
"""
@@ -49,8 +45,6 @@ class Activities(Postgres, MLFlow, Gates, OPC):
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
@@ -78,22 +72,15 @@ class Activities(Postgres, MLFlow, Gates, OPC):
Gates.__init__(self, logger=logger,
notification_handler=notification_handler)
OPC.__init__(self,
opc_servers=opc_config,
logger=logger,
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)

356
model_manager/activities/opc.py
View File

@@ -1,356 +0,0 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.logger import Logger
from model_manager.utils.repository.opc_repository import OpcRepository
from typing import Any
import traceback
from pandas import DataFrame
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):
self.logger = logger
self.notification_handler = notification_handler
self.opc_servers = opc_servers
BaseActivity.__init__(
self, logger, notification_handler, set_error_counter=True)
self.opc_repository: dict[str, OpcRepository] = {}
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(
id=server['id'],
url=server['url'],
logger=self.logger,
server_uri=server['server_uri'],
cert_path=server['cert_path'],
private_key_path=server['private_key_path'],
server_cert_path=server['server_cert_path'],
notification_handler=self.notification_handler,
reconnection_interval=server['reconnection_interval'],
pod_id=self.pod_id
)
is_connected, error_data = await self.opc_repository[id].connect()
if not is_connected:
self.send_notification(
metadata={
'model_id': '-',
'model_name': '-',
'workflow_name': '-',
'schedule_name': 'INITIALIZATION'
},
notification_id=error_data['notification_id'],
message=error_data['message'],
block=error_data['block'],
level=error_data.get('level', NotificationLevel.ERROR),
attachment_content=error_data.get(
'attachment_content', None)
)
else:
self.logger.info(
f"OPC server {id} connected successfully.")
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 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.
- tag (str): The tag to write to.
- data (Any): The data to write.
- data_type (str): The data type.
- tag_type (str): The tag type.
Returns:
- bool: True if the data was written successfully, False otherwise.
"""
try:
is_success, error_data = await self.opc_repository[server_id].write_data(
tag, data, data_type, self.logger, metadata)
if not is_success:
self.send_notification(
metadata=metadata,
notification_id=error_data['notification_id'],
message=error_data['message'],
block=error_data['block'],
level=error_data.get('level', NotificationLevel.ERROR),
attachment_content=error_data.get(
'attachment_content', None)
)
return False
return True
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR",
message=f"Error writing data to OPC server: {e}",
block="write_opc_data",
level=NotificationLevel.ERROR,
attachment_content=trace
)
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(
metadata=metadata,
notification_id="OPC_SERVER_NOT_FOUND",
message=message,
block="write_opc_data",
level=NotificationLevel.ERROR,
attachment_content=f"OPC servers: {list(self.opc_repository.keys())}"
)
return False
return True
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:
for tag, tag_config in config['prediction_tags'].items():
local_success = await self.write_data(
server_id=server_id,
tag=tag,
data=data.head(1)['prediction'].values[0],
data_type=tag_config['data_type'],
tag_type='prediction',
metadata=metadata
)
if local_success:
self.info(
f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata)
count += 1
success = success and local_success
if 'confidence_tags' in config:
for tag, tag_config in config['confidence_tags'].items():
local_success = await self.write_data(
server_id=server_id,
tag=tag,
data=data.head(1)['prediction_confidence'].values[0],
data_type=tag_config['data_type'],
tag_type='confidence',
metadata=metadata
)
if local_success:
self.info(
f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata)
count += 1
success = success and local_success
return success, count
@activity.defn(name='write_opc_data')
async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]:
"""
Write prediction and confidence data to OPC servers. The two writing
operations are optional and independent of each other.
Args:
- input_data(dict[str, Any]): The input data. Contains the following keys:
- data(dict[str, Any]): The dataframe that contains the data to write
to the OPC servers.
- opc_output_config(dict[str, Any]): The OPC writing configuration.
The keys are the OPC server names and the values contain:
- prediction_tags(dict[str, Any]): The tags to write to the OPC servers.
- confidence_tags(dict[str, Any]): The tags to write to the OPC servers.
Returns:
- dict[Any, Any]: The data that was written to the OPC servers.
"""
metadata = input_data['metadata']
self.info("Writing data to OPC servers...", metadata)
data = DataFrame(input_data['data'])
opc_output_config = input_data['opc_output_config']
self.info(f"Data to write: {data.size} rows", metadata)
success = True
for server_id, config in opc_output_config.items():
if not self.validate_server(server_id, metadata):
success = False
continue
local_success, local_count = await self.manage_output_tags(
server_id, config, data, metadata, success)
success = success and local_success
self.info(
f"Process completed for OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata)
return self.process_confidence(data, success, metadata)
def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]:
"""
Process prediction confidence based on OPC write operation success.
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): 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]: 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:
data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE
self.debug(
f"Some data could not be written to OPC servers, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.",
metadata
)
else:
self.debug("Data written to OPC servers successfully.", metadata)
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()

17
model_manager/metrics.py
View File

@@ -13,14 +13,13 @@ Key Metric Categories:
- Application Health: Overall system status and availability
- Prediction Operations: Count and performance of prediction operations
- Data Quality: Confidence levels and validation results
- Export Operations: Database and OPC export performance
- Export Operations: Database export performance
- Response Times: Performance monitoring for various operations
Metric Labels:
- pod_id: Kubernetes pod identifier for multi-instance deployments
- model_name: Name of the ML model being used
- pipeline_name: Name of the prediction pipeline
- opc_server_id: Identifier for OPC server operations
"""
from prometheus_client import Gauge, Counter, Histogram
@@ -56,17 +55,3 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram(
CORE_LABELS,
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
)
# OPC export metrics
PREDICTION_OPC_WRITING_COUNT = Counter(
"model_manager_prediction_opc_writing_count",
"Number of predictions written to the OPC server",
[*CORE_LABELS, "opc_server_id"],
)
PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR = Histogram(
"model_manager_prediction_opc_writing_response_time_monitor",
"Current response time of each prediction written to the OPC server",
[*CORE_LABELS, "opc_server_id"],
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
)

39
model_manager/utils/connectors_config.py
View File

@@ -59,45 +59,6 @@ def build_mlflow_config() -> Dict[str, Any]:
}
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_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
"""
opc_raw = getenv('OPC_CONFIG', None)
if opc_raw:
return json.loads(opc_raw)
return {
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'))
}
}
def build_mongodb_config() -> Dict[str, Any]:
"""
Build MongoDB configuration from environment variables.

359
model_manager/utils/repository/opc_repository.py
View File

@@ -1,359 +0,0 @@
import asyncio
import traceback
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from asyncua import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from asyncua.ua import DataValue, Variant, VariantType, DateTime
from regex import F
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from model_manager import metrics
data_type_map = {
'float': {
'converter': float,
'opc_type': VariantType.Float,
},
'double': {
'converter': float,
'opc_type': VariantType.Double,
},
'int': {
'converter': int,
'opc_type': VariantType.Int32,
},
'bool': {
'converter': bool,
'opc_type': VariantType.Boolean,
},
'str': {
'converter': str,
'opc_type': VariantType.String,
}
}
class OpcRepository():
def __init__(self, id: str, url: str, logger: Logger,
notification_handler: NotificationHandler,
reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None,
private_key_path: str = None, server_cert_path: str = None, pod_id: str = None):
self.url = url
self.id = id
self.server_uri = server_uri
self.cert_path = cert_path
self.private_key_path = private_key_path
self.server_cert_path = server_cert_path
self.logger = logger
self.error_count = 0
self.reconnection_interval = reconnection_interval
self.last_reconnection_time = None
self.notification_handler = notification_handler
self.client = None
self.pod_id = pod_id
self.metadata = {
'model_name': '-',
'model_id': '-',
'workflow_name': 'opc_repository',
'schedule_name': '-'
}
async def set_security(self):
"""
Configures the security settings for the OPC UA client.
This method sets up the security policy, certificates, and timeouts
required for establishing a secure connection with the OPC UA server.
Raises:
ValueError: If either the certificate path or private key path is not provided.
Attributes:
- cert_path (str): Path to the client's certificate file.
- private_key_path (str): Path to the client's private key file.
- server_cert_path (str, optional): Path to the server's certificate file.
- server_uri (str): The URI of the server to be used as the application URI.
- client (opcua.Client): The OPC UA client instance.
- logger (logging.Logger): Logger instance for logging information.
Security Settings:
- Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms
- Session Timeout: 10,000,000 ms
"""
if not all([self.cert_path, self.private_key_path]):
raise ValueError(
"Certificate and private key paths must be provided for secure connection.")
cert = Path(self.cert_path)
private_key = Path(self.private_key_path)
server_cert = Path(
self.server_cert_path) if self.server_cert_path else None
self.client.application_uri = self.server_uri
self.logger.custom_info('Setting security...', self.metadata)
await self.client.set_security(
SecurityPolicyBasic256,
certificate=str(cert),
private_key=str(private_key),
server_certificate=str(server_cert)
)
self.client.secure_channel_timeout = 10000000
self.client.session_timeout = 10000000
async def connect(self) -> tuple[bool, dict[str, Any]]:
"""
Establishes a connection to the OPC server.
This method initializes the OPC client using the provided URL and
sets up security if a certificate path is specified. It then
attempts to connect to the server and logs the connection status.
Raises:
Exception: If the connection to the OPC server fails.
"""
self.client = Client(self.url)
if self.cert_path:
await self.set_security()
self.logger.custom_info(
f'Starting connection to OPC server {self.id}...', self.metadata)
return await self.try_connect()
async def try_connect(self) -> tuple[bool, dict[str, Any]]:
"""
Attempt to establish connection to the OPC server.
This method performs the actual connection attempt to the OPC server
and handles connection failures with comprehensive error reporting.
It updates reconnection timing and provides detailed error information
for operational monitoring and debugging.
Returns:
tuple[bool, dict[str, Any]]: Connection result
- bool: True if connection successful, False otherwise
- dict: Error information if connection failed
"""
try:
self.last_reconnection_time = datetime.now()
await self.client.connect()
return True, {}
except Exception as e:
trace = traceback.format_exc()
self.logger.custom_error(trace, self.metadata)
return False, {
"notification_id": f"OPC_CONNECTION_ERROR_{self.id}",
"message": f"Failed to connect to OPC server: {e}",
"block": "opc_repository",
"level": NotificationLevel.ERROR,
"attachment_content": trace
}
async def disconnect(self):
"""
Gracefully disconnect from the OPC server.
This method safely terminates the connection to the OPC server
and cleans up client resources. It handles disconnection errors
gracefully and ensures proper resource cleanup.
"""
if self.client is None:
return
try:
await self.client.disconnect()
self.logger.custom_info(
'Disconnected from OPC server', self.metadata)
except Exception as e:
self.logger.custom_error(
f"Failed to disconnect from OPC server: {e}", self.metadata)
self.client = None
async def validate_connection(self) -> tuple[bool, dict[str, Any]]:
"""
Validate and maintain OPC server connection health.
This method performs comprehensive connection validation and
implements automatic reconnection logic for production reliability.
It handles various connection states and implements intelligent
reconnection strategies with error counting and timing controls.
Connection Validation:
1. Checks client existence and connection state
2. Implements error counting with automatic disconnection
3. Enforces reconnection timing windows
4. Provides detailed error reporting and notifications
Reconnection Strategy:
- Error Count Threshold: Disconnects after 5 consecutive errors
- Reconnection Window: Enforces minimum intervals between attempts
- Automatic Recovery: Attempts reconnection when conditions allow
- State Monitoring: Continuously monitors connection health
Args:
None
Returns:
tuple[bool, dict[str, Any]]: Connection validation result
- bool: True if connection is healthy, False otherwise
- dict: Error information if validation fails
"""
if self.client is None:
return await self.connect()
if self.error_count > 5:
self.logger.custom_warning(
f"OPC server {self.id} will be disconnected due to multiple errors", self.metadata)
try:
await self.disconnect()
except Exception as e:
trace = traceback.format_exc()
self.logger.custom_error(
f"Failed to disconnect from OPC server: {e}", self.metadata)
self.logger.custom_error(trace, self.metadata)
self.logger.custom_info(
f"Attempting to reconnect to OPC server {self.id}...", self.metadata)
return await self.connect()
# Check if client is connected using asyncua's connection state
try:
if self.client.uaclient.protocol is None or self.client.uaclient.protocol.state == "closed":
# OPC server is not connected
self.logger.custom_error(
f"OPC server {self.id} is not connected", self.metadata)
if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds(
) > self.reconnection_interval:
await self.disconnect()
self.logger.custom_info(
f"Trying to reconnect to OPC server {self.id}...", self.metadata)
return await self.connect()
return False, {
"notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}",
"message": f"OPC server {self.id} is not connected, waiting for next reconnection window...",
"block": "opc_repository",
"level": NotificationLevel.WARNING
}
return True, {}
except Exception as e:
trace = traceback.format_exc()
message = f"Failed to validate connection to OPC server: {e}"
self.logger.custom_error(message, self.metadata)
return False, {
"notification_id": f"OPC_CONNECTION_CHECK_ERROR_{self.id}",
"message": message,
"block": "opc_repository",
"level": NotificationLevel.ERROR,
"attachment_content": trace
}
async def write_data(self, node: str, value: Any, data_type: str,
logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
"""
Write data to OPC server with comprehensive validation and monitoring.
This method provides secure and reliable data writing to OPC servers
with automatic connection validation, data type conversion, and
comprehensive error handling. It implements performance monitoring
and metrics collection for operational visibility.
Data Writing Process:
1. Connection validation and automatic reconnection
2. Node validation and error handling
3. Data type conversion and validation
4. OPC data writing with timestamp
5. Performance metrics collection
6. Error handling and notification
Args:
node (str): OPC node identifier to write data to
value (Any): Data value to write to the OPC node
data_type (str): Data type for OPC conversion
logger (Logger): Logger instance for operation logging
metadata (dict[str, Any]): Context metadata for logging and metrics
Returns:
tuple[bool, dict[str, Any]]: Write operation result
- bool: True if write successful, False otherwise
- dict: Error information if write failed
"""
is_connected, error = await self.validate_connection()
if not is_connected:
return False, error
start_time = time.time()
try:
node_obj = self.client.get_node(node)
except Exception as e:
trace = traceback.format_exc()
logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
self.error_count += 1
return False, {
"notification_id": f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
"message": f"Failed to get node from OPC server: {e} | metadata: {metadata}",
"block": "opc_repository",
"level": NotificationLevel.ERROR,
"attachment_content": trace
}
if data_type not in data_type_map:
return False, {
"notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
"message": f"Unsupported data type: {data_type} | metadata: {metadata}",
"block": "opc_repository",
"level": NotificationLevel.ERROR
}
data = data_type_map[data_type]['converter'](value)
logger.custom_info(
f'Writing {data} - {type(data)} to {node}', metadata)
now = datetime.now()
ua_data = DataValue(
Variant(data, data_type_map[data_type]['opc_type']),
SourceTimestamp=DateTime(
now.year,
now.month,
now.day,
now.hour,
now.minute,
now.second,
now.microsecond
)
)
try:
await node_obj.write_value(ua_data)
metrics.PREDICTION_OPC_WRITING_COUNT.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
opc_server_id=self.id
).inc()
end_time = time.time()
response_time = end_time - start_time
metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
opc_server_id=self.id
).observe(response_time)
except Exception as e:
trace = traceback.format_exc()
logger.custom_error(trace, metadata)
self.error_count += 1
return False, {
"notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}",
"message": f"Failed to write data to OPC server: {e} | metadata: {metadata}",
"block": "opc_repository",
"level": NotificationLevel.ERROR,
"attachment_content": trace
}
self.error_count = 0
return True, {}

12
model_manager/worker/worker.py
View File

@@ -42,7 +42,6 @@ with workflow.unsafe.imports_passed_through():
from model_manager.utils.connectors_config import (
build_postgres_config,
build_mlflow_config,
build_opc_config,
build_mongodb_config
)
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -63,9 +62,8 @@ async def main():
2. Starts Prometheus metrics server
3. Initializes notification handler
4. Creates and configures activities
5. Initializes OPC connections
6. Starts Temporal client and workers
7. Manages worker lifecycle and graceful shutdown
5. Starts Temporal client and workers
6. Manages worker lifecycle and graceful shutdown
The function runs indefinitely until interrupted or an error occurs.
On error, it performs cleanup and exits with a non-zero status code.
@@ -105,14 +103,10 @@ async def main():
activities = Activities(
postgres_config=build_postgres_config(),
mlflow_config=build_mlflow_config(),
opc_config=build_opc_config(),
logger=logger,
notification_handler=notification_handler
)
logger.custom_info('Initializing OPC...', metadata)
await activities.init_opc()
logger.custom_info(
f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
@@ -167,8 +161,6 @@ async def main():
activities.format_prediction,
activities.format_default_prediction,
activities.get_last_timestamp,
# OPC
activities.write_opc_data,
# Postgres
activities.load_custom_query,
activities.repeat_last_prediction,

4
model_manager/workflows/predictions_batch.py
View File

@@ -58,7 +58,7 @@ class PredictionsBatch():
- mlflow_predict_filters (dict, optional): MLFlow prediction filters
- model_retention (int, optional): Model retention period in minutes
- path_priority (list[str]): Decision path priority configuration
- opc_output_config (dict, optional): OPC server export configuration
- datetime_columns (list[str], optional): Columns to treat as datetime
Returns:
@@ -115,7 +115,7 @@ class PredictionsBatch():
}),
'model_config': input_data.get('model_config', {}),
'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']),
'opc_output_config': input_data.get('opc_output_config', {}),
'prediction_store_policy': input_data.get(
'prediction_store_policy', 'lts:1')
}

24
model_manager/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -14,9 +14,9 @@ class FormatAndExportPrediction():
Data formatting and export workflow for prediction results.
This workflow handles the final stages of the prediction pipeline, including
data formatting, database persistence, OPC server export, and metrics recording.
data formatting, database persistence, and metrics recording.
It implements flexible formatting based on prediction quality and provides
comprehensive export capabilities to multiple destinations.
comprehensive export capabilities.
The workflow supports two main prediction paths:
1. Normal Prediction: Formats and exports successful prediction results
@@ -24,7 +24,6 @@ class FormatAndExportPrediction():
Export Destinations:
- PostgreSQL Database: Persistent storage with timestamp conversion
- OPC Servers: Real-time industrial system integration
- Prometheus Metrics: Performance monitoring and operational visibility
"""
@@ -36,14 +35,12 @@ class FormatAndExportPrediction():
This method orchestrates the complete data export process by:
1. Determining the appropriate formatting strategy based on path_flag
2. Formatting prediction data according to quality and requirements
3. Exporting data to OPC servers for real-time industrial access
4. Persisting data to PostgreSQL database with comprehensive metadata
5. Recording performance metrics for operational monitoring
3. Persisting data to PostgreSQL database with comprehensive metadata
4. Recording performance metrics for operational monitoring
The method implements flexible formatting strategies:
- Normal predictions: Full data formatting with confidence scores
- Error predictions: Default formatting with error indicators
- Comprehensive export: Multi-destination data distribution
Args:
input_data: Complete configuration for the export workflow
@@ -58,7 +55,6 @@ class FormatAndExportPrediction():
- comment (str): Operational comment or error description
- schema (str): Database schema for data storage
- table_name (str): Target table for data persistence
- opc_output_config (dict[str, Any]): OPC server export configuration
- prediction_store_policy (str, optional): Data retention policy
Returns:
@@ -100,18 +96,6 @@ class FormatAndExportPrediction():
start_to_close_timeout=timedelta(seconds=60)
)
# write to opc
prediction = await workflow.execute_activity_method(
Activities.write_opc_data,
{
**metadata,
'opc_output_config': input_data['opc_output_config'],
'data': prediction
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60)
)
# write to postgres
await workflow.execute_activity_method(
Activities.export_data_to_postgres,

4
model_manager/workflows/sub_workflows/prediction_process.py
View File

@@ -64,7 +64,7 @@ class PredictionProcess():
- mlflow_predict_filters (dict): MLFlow prediction filters
- model_retention (int): Model retention period in minutes
- path_priority (list[str]): Decision path priority configuration
- opc_output_config (dict): OPC server export configuration
Returns:
None: The workflow completes successfully when export workflow finishes
@@ -210,7 +210,6 @@ class PredictionProcess():
'model_id': model_id,
'model_name': model_name,
'model_config': model_config,
'opc_output_config': input_data['opc_output_config'],
'schema': input_data['schema'],
'table_name': input_data['table_name'],
'comment': comment,
@@ -287,7 +286,6 @@ class PredictionProcess():
'schema': schema,
'table_name': table_name,
'comment': comment,
'opc_output_config': input_data['opc_output_config'],
'prediction_store_policy': input_data['prediction_store_policy']
}
)