e22b0bc7c3
SIENTIAPDE-1646 Update dependencies and refactor OPC integration - Replaced `opcua` with `asyncua` in `requirements-local.txt` and `requirements.txt` to utilize the async capabilities. - Updated `values.yaml` to change the GitHub branch from `feature/SIENTIAPDE-1646` to `release/SIENTIAPDE-1646`. - Refactored `opc.py` and `opc_repository.py` to accommodate the new `asyncua` library, ensuring compatibility with the synchronous API. - Adjusted tests in `test_opc_repository.py` to reflect changes in the client implementation and maintain functionality.
338 lines
13 KiB
Python
338 lines
13 KiB
Python
from temporalio import activity, workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
import traceback
|
|
from collections.abc import Hashable
|
|
from typing import Any
|
|
|
|
from pandas import DataFrame
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
|
|
|
from laborious.utils.repository.opc_repository import OpcRepository
|
|
|
|
OPC_WRITTING_ERROR_CONFIDENCE = 12
|
|
|
|
|
|
class OPC(SientiaMonitoring):
|
|
"""
|
|
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,
|
|
metrics_controller: MetricsController,
|
|
):
|
|
self.logger = logger
|
|
self.notification_handler = notification_handler
|
|
self.opc_servers = opc_servers
|
|
|
|
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
|
|
|
|
self.opc_repository: dict[str, OpcRepository] = {}
|
|
|
|
def init_opc(self) -> None:
|
|
"""
|
|
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.
|
|
"""
|
|
|
|
self.logger.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'],
|
|
logger=self.logger,
|
|
notification_handler=self.notification_handler,
|
|
metrics_controller=self.metrics_controller,
|
|
server_uri=server['server_uri'],
|
|
cert_path=server['cert_path'],
|
|
private_key_path=server['private_key_path'],
|
|
server_cert_path=server['server_cert_path'],
|
|
)
|
|
ok, err = self.opc_repository[opc_id].connect()
|
|
if not ok:
|
|
self.send_notification(
|
|
metadata={
|
|
'model_id': '-',
|
|
'model_name': '-',
|
|
'workflow_name': '-',
|
|
'schedule_name': 'INITIALIZATION',
|
|
},
|
|
notification_id=f'OPC_CONNECTION_ERROR_{server.get("id", opc_id)}',
|
|
message=err.get('message', 'Failed to connect to OPC server'),
|
|
block='opc_repository',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=err.get('attachment_content', traceback.format_exc()),
|
|
)
|
|
else:
|
|
self.logger.info(
|
|
f'OPC server {opc_id}:{server["server_name"]} connected successfully.'
|
|
)
|
|
|
|
def write_data(
|
|
self,
|
|
server_id: str,
|
|
tag: str,
|
|
data: Any,
|
|
data_type: str,
|
|
tag_type: str,
|
|
metadata: dict[str, Any],
|
|
) -> float | None:
|
|
"""
|
|
Write data to a specific OPC server tag with comprehensive error handling.
|
|
|
|
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:
|
|
- float | None: Response time in seconds if successful, None otherwise.
|
|
"""
|
|
|
|
try:
|
|
is_success, info_data = 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=info_data['notification_id'],
|
|
message=info_data['message'],
|
|
block=info_data['block'],
|
|
level=info_data.get('level', NotificationLevel.ERROR),
|
|
attachment_content=info_data.get('attachment_content', None),
|
|
)
|
|
return None
|
|
return info_data['response_time']
|
|
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
|
|
|
|
def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
|
|
"""
|
|
Validate that an OPC repository exists for the requested server identifier.
|
|
|
|
This guard prevents write attempts against unknown/uninitialized servers.
|
|
When the server is missing, it emits an error notification with the list
|
|
of available repositories to help operators diagnose configuration drift.
|
|
|
|
Args:
|
|
- server_id (str): OPC server identifier from workflow output config.
|
|
- metadata (dict[str, Any]): Workflow metadata used for logs/alerts.
|
|
|
|
Return:
|
|
bool: ``True`` when the server repository is available; ``False`` otherwise.
|
|
"""
|
|
|
|
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
|
|
|
|
def manage_output_tags(
|
|
self,
|
|
server_id: str,
|
|
config: dict[str, Any],
|
|
data: DataFrame,
|
|
metadata: dict[str, Any],
|
|
) -> tuple[bool, dict[str, float | None]]:
|
|
"""
|
|
Write prediction and confidence values for one OPC server configuration.
|
|
|
|
The method iterates through optional ``prediction_tags`` and
|
|
``confidence_tags``, performs synchronous writes for each tag, collects
|
|
per-tag response times, and returns an aggregate success flag
|
|
(all tags successful) with a metrics-friendly response map.
|
|
|
|
Args:
|
|
- server_id (str): Target OPC server id.
|
|
- config (dict[str, Any]): Server output configuration containing optional
|
|
``prediction_tags`` and ``confidence_tags`` sections.
|
|
- data (DataFrame): Prediction dataframe used as source values.
|
|
- metadata (dict[str, Any]): Workflow metadata for logging/notifications.
|
|
|
|
Return:
|
|
tuple[bool, dict[str, float | None]]: Global success flag and response-time
|
|
map per tag (``None`` for failed writes).
|
|
"""
|
|
|
|
response_times: dict[str, float | None] = {}
|
|
|
|
if 'prediction_tags' in config:
|
|
for tag, tag_config in config['prediction_tags'].items():
|
|
response_time = 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 response_time is not None:
|
|
self.info(
|
|
f'Prediction data written to OPC server {server_id} for tag {tag}.',
|
|
metadata,
|
|
)
|
|
response_times[tag] = response_time
|
|
|
|
if 'confidence_tags' in config:
|
|
for tag, tag_config in config['confidence_tags'].items():
|
|
response_time = 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 response_time is not None:
|
|
self.info(
|
|
f'Confidence data written to OPC server {server_id} for tag {tag}.',
|
|
metadata,
|
|
)
|
|
response_times[tag] = response_time
|
|
|
|
success = None not in response_times.values()
|
|
|
|
return success, response_times
|
|
|
|
@activity.defn(name='write_opc_data')
|
|
def write_opc_data(
|
|
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.
|
|
|
|
For each server in ``opc_output_config``, this activity validates server
|
|
availability, writes enabled prediction/confidence tags, accumulates
|
|
response-time metrics, and then normalizes confidence/comments in the
|
|
returned prediction payload when at least one write fails.
|
|
|
|
Args:
|
|
- input_data (dict[str, Any]): Payload containing workflow metadata, data
|
|
to write, and ``opc_output_config`` server/tag definitions.
|
|
|
|
Return:
|
|
tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]: Updated
|
|
prediction payload dict and nested metrics
|
|
``{server_id: {tag_name: response_time_or_none}}``.
|
|
"""
|
|
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
|
|
|
|
metrics: dict[str, dict[str, float | None]] = {}
|
|
|
|
for server_id, config in opc_output_config.items():
|
|
if not self.validate_server(server_id, metadata):
|
|
success = False
|
|
continue
|
|
|
|
local_success, local_response_times = self.manage_output_tags(
|
|
server_id, config, data, metadata
|
|
)
|
|
metrics[server_id] = local_response_times
|
|
local_count = len(local_response_times)
|
|
success = success and local_success
|
|
|
|
n_pred = len(config.get('prediction_tags') or {})
|
|
n_conf = len(config.get('confidence_tags') or {})
|
|
self.info(
|
|
f'Process completed for OPC server {server_id}: {local_count} of {n_pred} prediction tags and {n_conf} confidence tags',
|
|
metadata,
|
|
)
|
|
|
|
return self.process_confidence(data, success, metadata), metrics
|
|
|
|
def process_confidence(
|
|
self, data: DataFrame, success: bool, metadata: dict[str, Any]
|
|
) -> dict[Hashable, Any]:
|
|
"""
|
|
Apply fallback confidence/comment values when OPC writes are not fully successful.
|
|
|
|
Args:
|
|
- data (DataFrame): Prediction dataframe to be returned to downstream steps.
|
|
- success (bool): Aggregate write status across all attempted OPC tags.
|
|
- metadata (dict[str, Any]): Workflow metadata used for debug logs.
|
|
|
|
Return:
|
|
dict[Hashable, Any]: Serialized dataframe dict with original values on success,
|
|
or downgraded confidence/comment fields on failure.
|
|
"""
|
|
|
|
message = 'Some data could not be written to OPC servers'
|
|
|
|
if not success:
|
|
data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE
|
|
data['comments'] = message
|
|
self.debug(
|
|
f'{message}, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.',
|
|
metadata,
|
|
)
|
|
|
|
else:
|
|
self.debug('Data written to OPC servers successfully.', metadata)
|
|
|
|
return data.to_dict()
|
|
|
|
def close(self) -> None:
|
|
"""
|
|
Disconnect all tracked OPC repositories and clear in-memory references.
|
|
|
|
This method should be called during worker shutdown to ensure every
|
|
synchronous OPC session is explicitly closed before process exit.
|
|
"""
|
|
|
|
for opc in self.opc_repository.values():
|
|
opc.disconnect()
|
|
self.opc_repository.clear()
|