From 7a776066ec8b6075f7bb94294b67d2869378b92c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 May 2026 16:43:20 -0300 Subject: [PATCH] SIENTIAPDE-1646 Update requirements and enhance OPC communication handling - Updated `requirements-local.txt` to use the latest versions of `sientia-dataops-library` (1.12.1) and `sientia-model-library` (0.10.0). - Improved documentation in `opc-communication.md` to clarify the transition from async to sync implementation and added details on error classification and reconnection behavior. - Refactored `init_opc` method in `opc.py` to enhance connection handling and logging, ensuring independent server initialization. - Enhanced validation and writing methods in `opc.py` to provide better feedback and error handling for OPC server operations. - Updated `opc_repository.py` to improve error payload construction and session management metrics. --- docs/opc-communication.md | 2 + laborious/activities/opc.py | 141 +++++++++++++++---- laborious/utils/repository/opc_repository.py | 23 ++- requirements-local.txt | 6 +- 4 files changed, 135 insertions(+), 37 deletions(-) diff --git a/docs/opc-communication.md b/docs/opc-communication.md index d272c61..cab045d 100644 --- a/docs/opc-communication.md +++ b/docs/opc-communication.md @@ -2,6 +2,8 @@ Laborious exports predictions to OPC UA servers through `OpcRepository` ([`laborious/utils/repository/opc_repository.py`](../laborious/utils/repository/opc_repository.py)) and the synchronous Temporal activity layer in [`laborious/activities/opc.py`](../laborious/activities/opc.py). The repository uses `asyncua.sync.Client` (asyncio on a background thread) so activities remain blocking without `async def`. +OPC reconnect, write error classification (`opc_error_kind`), and activity confidence/comment behavior are converted from the **async** implementation on `main` at `fcc8920a8be4` (`asyncua.Client` + `asyncio` reconnect task → `threading` reconnect thread). Re-convert with `scripts/convert_opc_async_to_sync.py` when `main` OPC files change. + Implementation plan for session/channel recovery on Tier-1 `Bad*` errors: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](../.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). ## Architecture diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 733ecb9..b9a512a 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -71,6 +71,7 @@ class OPC(SientiaMonitoring): 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 """ @@ -87,7 +88,7 @@ class OPC(SientiaMonitoring): self.opc_repository: dict[str, OpcRepository] = {} - def init_opc(self) -> None: + def init_opc(self): """ Initialize OPC server connections and establish communication channels. @@ -95,22 +96,36 @@ class OPC(SientiaMonitoring): 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.info('Initializing OPC servers...') for opc_id, server in self.opc_servers.items(): self.opc_repository[opc_id] = OpcRepository( opc_id=opc_id, - url=server['url'], server_name=server['server_name'], + url=server['url'], logger=self.logger, - notification_handler=self.notification_handler, - metrics_controller=self.metrics_controller, - reconnection_interval=server.get('reconnection_interval', 60), 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.get('reconnection_interval', 60), + metrics_controller=self.metrics_controller, ) is_connected, error_data = self.opc_repository[opc_id].connect() if not is_connected: @@ -129,8 +144,7 @@ class OPC(SientiaMonitoring): ) else: self.info( - f'OPC server {opc_id}:{server["server_name"]} connected successfully.', - None, + f'OPC server {opc_id}:{server["server_name"]} connected successfully.' ) def write_data( @@ -179,16 +193,25 @@ class OPC(SientiaMonitoring): def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool: """ - Validate that an OPC repository exists for the requested server identifier. + 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): OPC server identifier from workflow output config. - - metadata (dict[str, Any]): Workflow metadata used for logs/alerts. + server_id (str): Unique identifier for the OPC server to validate + metadata (dict[str, Any]): Context metadata for logging and notifications - Return: - bool: ``True`` when the server repository is available; ``False`` otherwise. + 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( @@ -266,10 +289,30 @@ class OPC(SientiaMonitoring): metadata: dict[str, Any], ) -> tuple[bool, dict[str, float | None], bool, str | None, bool]: """ - Write prediction and confidence values for one OPC server configuration. + Manage the writing of prediction and confidence data to OPC server tags. - Return: - tuple: success flag, per-tag response times, session_bad flags. + 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 """ response_times: dict[str, float | None] = {} session_bad_seen = False @@ -318,15 +361,21 @@ class OPC(SientiaMonitoring): self, input_data: dict[str, Any] ) -> tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]: """ - Execute OPC writes across all configured servers and collect per-tag metrics. + 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]): Payload containing workflow metadata, data - to write, and ``opc_output_config`` server/tag definitions. + - 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. - Return: - tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]: Updated - prediction payload dict and nested metrics per server/tag. """ metadata = input_data['metadata'] self.info('Writing data to OPC servers...', metadata) @@ -392,10 +441,29 @@ class OPC(SientiaMonitoring): reconnect_in_progress: bool = False, ) -> dict[Hashable, Any]: """ - Apply fallback confidence/comment values when OPC writes are not fully successful. + Process prediction confidence based on OPC write operation success. - Return: - dict[Hashable, Any]: Serialized dataframe dict with updated confidence/comments on failure. + 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: @@ -423,11 +491,26 @@ class OPC(SientiaMonitoring): return data.to_dict() - def close(self) -> None: - """ - Disconnect all tracked OPC repositories and clear in-memory references. + def close(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(): opc.disconnect() self.opc_repository.clear() diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 6a0b172..1c17020 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -4,7 +4,7 @@ Synchronous OPC UA client repository using asyncua ``sync`` API. ``asyncua.sync.Client`` runs the asyncio stack on a background thread so Temporal activities and other callers stay blocking while preserving the same session lifecycle, security policy, reconnect semantics, and write error classification -as the async ``origin/main`` implementation. +as the async ``main`` implementation at ``fcc8920a8be4`` (async → sync/thread conversion). """ import json @@ -169,7 +169,7 @@ class OpcRepository(SientiaMonitoring): Synchronous OPC UA repository for connect/disconnect and typed writes. Uses ``asyncua.sync.Client`` with the same session metrics, Tier-1 Bad* reconnect, - and structured write error payloads as the async repository on ``origin/main``. + and structured write error payloads as the async repository on ``main``. """ def __init__( @@ -194,10 +194,10 @@ class OpcRepository(SientiaMonitoring): self.private_key_path = private_key_path self.server_cert_path = server_cert_path self.reconnection_interval = reconnection_interval - self.last_reconnection_time: datetime | None = None + self.last_reconnection_time: None | datetime = None self.disconnection_interval = 10.0 self.notification_handler = notification_handler - self.client: Client | None = None + self.client: None | Client = None SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) @@ -214,6 +214,15 @@ class OpcRepository(SientiaMonitoring): self._allow_reconnect = True def _opc_debug_tags(self, session_id: str) -> dict[str, str]: + """ + Build Prometheus/log label tags for OPC session-scoped metrics. + + Args: + session_id (str): OPC UA session token string. + + Return: + dict[str, str]: Labels pod_id, server_name, runtime, opc_server_id, session_id. + """ return { 'pod_id': str(getattr(self, 'pod_id', 'unknown')), 'server_name': self.server_name, @@ -251,6 +260,12 @@ class OpcRepository(SientiaMonitoring): ).total_seconds() > self.reconnection_interval def _not_connected_error(self) -> dict[str, Any]: + """ + Build the standard error payload when validate_connection finds no open protocol. + + Return: + dict[str, Any]: Notification fields for OPC_CONNECTION_NOT_READY. + """ return { 'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}', 'message': f'OPC server {self.id} is not connected', diff --git a/requirements-local.txt b/requirements-local.txt index df0d122..73a9a47 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -3,10 +3,8 @@ psycopg2-binary sqlalchemy asyncua==1.0.6 redis -#git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.0 -/home/grezewave/Documents/projects/sientia/sientia-dataops-library -#git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.8.3 -/home/grezewave/Documents/projects/sientia/sientia-model-library +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 +git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0 prometheus-client botocore boto3