3ee076214c
Implement server validation and output tag management in OPC class - Added a new method to validate the existence of OPC servers before writing data, improving error handling. - Introduced a method to manage writing of prediction and confidence tags, streamlining the data writing process. - Refactored the write_opc_data method to utilize the new validation and management methods for better code organization and clarity.
219 lines
8.8 KiB
Python
219 lines
8.8 KiB
Python
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.temporal.utils.logger import Logger
|
|
from laborious.utils.repository.opc_repository import OpcRepository
|
|
from typing import Any
|
|
import traceback
|
|
from pandas import DataFrame
|
|
|
|
OPC_WRITTING_ERROR_CONFIDENCE = 12
|
|
|
|
|
|
class OPC(BaseActivity):
|
|
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] = {}
|
|
for id, server in 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 = 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)
|
|
)
|
|
|
|
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.
|
|
|
|
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 = 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:
|
|
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], success: bool) -> bool:
|
|
if 'prediction_tags' in config:
|
|
for tag, tag_config in config['prediction_tags'].items():
|
|
local_success = 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)
|
|
success = success and local_success
|
|
|
|
if 'confidence_tags' in config:
|
|
for tag, tag_config in config['confidence_tags'].items():
|
|
local_success = 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)
|
|
success = success and local_success
|
|
|
|
return success
|
|
|
|
@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.debug("Writing data to OPC servers...", metadata)
|
|
data = DataFrame(input_data['data'])
|
|
opc_output_config = input_data['opc_output_config']
|
|
self.debug(data, metadata)
|
|
|
|
success = True
|
|
|
|
for server_id, config in opc_output_config.items():
|
|
|
|
if not self.validate_server(server_id, metadata):
|
|
success = False
|
|
continue
|
|
|
|
success = success and self.manage_output_tags(
|
|
server_id, config, data, metadata, success)
|
|
|
|
return self.process_confidence(data, success, metadata)
|
|
|
|
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.
|
|
|
|
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.
|
|
|
|
Args:
|
|
data (DataFrame): The DataFrame containing the data to be processed.
|
|
success (bool): Indicates whether the data was successfully written to the OPC servers.
|
|
|
|
Returns:
|
|
dict[Any, Any]: The processed data as a dictionary.
|
|
"""
|
|
|
|
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()
|
|
|
|
def shutdown(self):
|
|
for opc in self.opc_repository.values():
|
|
opc.disconnect()
|