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

SIENTIAPDE-1646

Browse Source 
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.
This commit is contained in:
vitor-aignosi
2026-05-19 16:43:20 -03:00
parent c919713075
commit 7a776066ec
4 changed files with 135 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
laborious/activities/opc.py
View File

@@ -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()

23
laborious/utils/repository/opc_repository.py
View File

@@ -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',