9b4333a83e
Update requirements and refactor datetime handling in OpcManager for timezone support - Commented out the specific version of sientia-dataops-library in requirements.txt and added a local path. - Updated GITHUB_BRANCH in values.yaml for consistency with current development. - Refactored source_timestamp handling in OpcManager to include timezone information. - Adjusted timestamp formatting in test cases to reflect the new datetime format with timezone.
343 lines
15 KiB
Python
343 lines
15 KiB
Python
import json
|
|
from pathlib import Path
|
|
from asyncua.crypto.security_policies import SecurityPolicyBasic256
|
|
from asyncua.sync import Client
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.temporal.activities.base import BaseActivity
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.temporal.constants import OPC_TIMEZONE, DATETIME_FORMAT_WITH_TZ
|
|
from ingestor.managers.data_manager import DataManager
|
|
import ingestor.metrics as metrics
|
|
|
|
|
|
class OpcManager(BaseActivity):
|
|
def __init__(self, name: str, url: str, data_manager: DataManager, logger: Logger,
|
|
server_uri: str, notification_handler: NotificationHandler, metadata: dict,
|
|
cert_path: str = None, private_key_path: str = None, server_cert_path: str = None):
|
|
self.url = url
|
|
self.name = name
|
|
self.server_uri = server_uri
|
|
self.data_queue = {}
|
|
self.non_receive_count = 0
|
|
self.client = None
|
|
self.cert_path = cert_path
|
|
self.private_key_path = private_key_path
|
|
self.server_cert_path = server_cert_path
|
|
self.nodes = {}
|
|
self.subscriptions = {}
|
|
self.data_manager = data_manager
|
|
self.metadata = metadata
|
|
|
|
BaseActivity.__init__(self, logger=logger,
|
|
notification_handler=notification_handler,
|
|
set_error_counter=True)
|
|
|
|
metrics.OPC_CONNECTION_STATUS.labels(
|
|
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0)
|
|
metrics.OPC_TAGS_SUBSCRIBED.labels(
|
|
pod_id=self.pod_id, server_name=self.name).set(0)
|
|
|
|
def __str__(self):
|
|
return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
|
|
f"nodes={self.nodes}, subscriptions={self.subscriptions}"
|
|
|
|
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.info('Setting security...')
|
|
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
|
|
|
|
def connect(self):
|
|
"""
|
|
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.
|
|
"""
|
|
|
|
metrics.OPC_CONNECTIONS_TOTAL.labels(
|
|
pod_id=self.pod_id, server_name=self.name).inc()
|
|
try:
|
|
self.client = Client(self.url)
|
|
if self.cert_path:
|
|
self.set_security()
|
|
self.logger.info(f'Starting connection to {self.name}...')
|
|
self.client.connect()
|
|
metrics.OPC_CONNECTION_STATUS.labels(
|
|
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(1)
|
|
self.logger.info(f'Connection to {self.name} successful.')
|
|
except Exception as e:
|
|
metrics.OPC_CONNECTION_STATUS.labels(
|
|
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0)
|
|
metrics.OPC_CONNECTIONS_FAILED.labels(
|
|
pod_id=self.pod_id, server_name=self.name).inc()
|
|
self.logger.error(f"Failed to connect to {self.name}: {e}")
|
|
raise
|
|
|
|
def create_subscription(self, name: str, period: int = 500):
|
|
"""
|
|
Creates a subscription with the specified monitoring period.
|
|
This method establishes a subscription to monitor data changes or events
|
|
from the OPC UA server. If the client is not connected, an exception is raised.
|
|
Args:
|
|
period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
|
|
Raises:
|
|
ValueError: If the client is not connected.
|
|
Side Effects:
|
|
- Sets the `self.period` attribute to the specified or default period.
|
|
- Creates a subscription and assigns it to `self.subscription`.
|
|
- Logs the creation of the subscription.
|
|
"""
|
|
|
|
if not self.client:
|
|
raise ValueError("Client not connected. Call connect first.")
|
|
try:
|
|
p = period if period is not None else 500
|
|
self.subscriptions[name] = self.client.create_subscription(p, self)
|
|
self.logger.info(f'Subscription {name} created on {self.name}.')
|
|
metrics.OPC_SUBSCRIPTIONS_CREATED.labels(
|
|
pod_id=self.pod_id, server_name=self.name, slot_name=name).inc()
|
|
except Exception as e:
|
|
self.logger.error(
|
|
f"Failed to create subscription {name} on {self.name}: {e}")
|
|
raise
|
|
|
|
def subscribe(self, subscription: str, nodes: dict, collect_period: int):
|
|
"""
|
|
Subscribes to a set of OPC UA nodes for data change notifications.
|
|
This method adds the specified nodes to the subscription and configures
|
|
their data collection rules based on the provided collection period and
|
|
node-specific frequency.
|
|
Args:
|
|
nodes (dict): A dictionary where keys are node identifiers (e.g., node
|
|
IDs or paths) and values are configurations for each node. Each
|
|
configuration must include a 'frequency' key indicating the
|
|
frequency of data collection in Hz.
|
|
collect_period (int): The data collection period in seconds.
|
|
Raises:
|
|
ValueError: If the subscription has not been created by calling
|
|
`create_subscription` prior to this method.
|
|
"""
|
|
|
|
if not self.subscriptions.get(subscription):
|
|
raise ValueError(
|
|
"Subscription not created. Call create_subscription first.")
|
|
|
|
self.logger.info(f"Subscribing to {subscription} on {self.name}...")
|
|
self.logger.info(f"Subscribing to nodes: {nodes}")
|
|
addr_nodes = [self.client.get_node(
|
|
n) for n in nodes]
|
|
self.logger.debug(f"Addr nodes: {addr_nodes}")
|
|
self.nodes.update(nodes)
|
|
self.logger.debug(f"Nodes: {self.nodes}")
|
|
metrics.OPC_TAGS_SUBSCRIBED.labels(
|
|
pod_id=self.pod_id, server_name=self.name).set(len(self.nodes))
|
|
self.collect_period = collect_period
|
|
|
|
for node, config in self.nodes.items():
|
|
self.nodes[node]['cycle_rule'] = {
|
|
'cycle_increment': collect_period*1000/float(config['frequency']),
|
|
'cycle_count': 0
|
|
}
|
|
|
|
self.subscriptions[subscription].subscribe_data_change(addr_nodes)
|
|
|
|
def unsubscribe(self, subscription: str):
|
|
"""
|
|
Unsubscribes from a given subscription.
|
|
Args:
|
|
subscription (str): The name of the subscription to unsubscribe from.
|
|
Logs:
|
|
- A warning if the specified subscription does not exist.
|
|
- An info message upon successful unsubscription.
|
|
Behavior:
|
|
- If the subscription exists, it is deleted and removed from the
|
|
subscriptions dictionary.
|
|
- If the subscription does not exist, no action is taken.
|
|
"""
|
|
|
|
if not self.subscriptions.get(subscription):
|
|
self.logger.warning(
|
|
f"Subscription '{subscription}' not found. Cannot unsubscribe.")
|
|
return
|
|
self.subscriptions[subscription].delete()
|
|
del self.subscriptions[subscription]
|
|
self.logger.info(f"Unsubscribed from {subscription}.")
|
|
|
|
def __del__(self):
|
|
self.disconnect()
|
|
|
|
def disconnect(self):
|
|
"""
|
|
Disconnects from the OPC UA server.
|
|
This method handles the disconnection process by deleting the subscription
|
|
and disconnecting the client from the OPC UA server. It logs the disconnection
|
|
process and handles any exceptions that may occur during cleanup.
|
|
Raises:
|
|
Exception: If an error occurs while deleting the subscription or disconnecting
|
|
from the OPC UA server, it logs the error details.
|
|
"""
|
|
|
|
self.logger.warning('Disconnecting from OPC server')
|
|
if self.client is None:
|
|
self.logger.warning("Client already disconnected.")
|
|
return
|
|
try:
|
|
_a = [self.subscriptions[sub].delete()
|
|
for sub in self.subscriptions]
|
|
self.logger.warning("Deleted all subscriptions.")
|
|
except Exception as sub_error:
|
|
self.logger.error(f"Failed to clean up subscription: {sub_error}")
|
|
|
|
try:
|
|
self.client.disconnect()
|
|
except Exception as conn_error:
|
|
self.logger.error(
|
|
f"Failed to disconnect from OPC UA server: {conn_error}")
|
|
finally:
|
|
del self.client
|
|
self.client = None
|
|
metrics.OPC_CONNECTION_STATUS.labels(
|
|
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0)
|
|
metrics.OPC_TAGS_SUBSCRIBED.labels(
|
|
pod_id=self.pod_id, server_name=self.name).set(0)
|
|
self.logger.warning("Disconnected from OPC UA server.")
|
|
|
|
def datachange_notification(self, node, _val, data):
|
|
"""
|
|
Handles data change notifications for monitored OPC UA nodes.
|
|
This method is triggered when a monitored node's value changes. It processes
|
|
the notification, updates internal state, and publishes the data to the
|
|
appropriate topics.
|
|
Args:
|
|
node (NodeId): The OPC UA node that triggered the data change notification.
|
|
_val (Any): The new value of the node (unused in this implementation).
|
|
data (DataChangeNotification): The data change notification object containing
|
|
details about the change.
|
|
Behavior:
|
|
- Extracts the value and source timestamp from the monitored item.
|
|
- Resets the cycle count for the node's cycle rule.
|
|
- Resets the non-receive count.
|
|
- Constructs a data dictionary containing the tag, tag name, timestamp, and value.
|
|
- Publishes the data to all topics associated with the node.
|
|
"""
|
|
|
|
# get data value
|
|
monitored_item = data.monitored_item
|
|
value = monitored_item.Value.Value.Value
|
|
# source_timestamp
|
|
source_timestamp = monitored_item.Value.SourceTimestamp.replace(
|
|
tzinfo=OPC_TIMEZONE)
|
|
tag = str(node)
|
|
|
|
self.logger.debug(
|
|
f"Data change notification received for tag:"
|
|
f"{tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles")
|
|
|
|
self.nodes[tag]['cycle_rule']['cycle_count'] = 0
|
|
self.non_receive_count = 0
|
|
metrics.OPC_CYCLES_WITHOUT_DATA.labels(
|
|
pod_id=self.pod_id, server_name=self.name).set(0)
|
|
|
|
data = {
|
|
'tag': tag,
|
|
'name': self.nodes[str(node)]['tag_name'],
|
|
'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
|
|
'value': value
|
|
}
|
|
|
|
_a = [self.data_manager.publish(e, data)
|
|
for e in self.nodes[tag]['topics']]
|
|
|
|
def check_cycles(self):
|
|
"""
|
|
Checks the cycle counts for all monitored nodes and sends
|
|
notifications if thresholds are exceeded.
|
|
|
|
This method iterates through all monitored nodes and updates their cycle counts based on
|
|
configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
|
|
a warning notification.
|
|
|
|
"""
|
|
for node, config in self.nodes.items():
|
|
self.nodes[node]['cycle_rule']['cycle_count'] += config[
|
|
'cycle_rule']['cycle_increment']
|
|
if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
|
|
name = config['tag_name']
|
|
cycles = self.nodes[node]['cycle_rule']['cycle_count']
|
|
self.send_notification(
|
|
metadata=self.metadata,
|
|
notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
|
|
message=f'{cycles} cycles without receive from {node}:{name}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.WARNING
|
|
)
|
|
|
|
def check_opc_listenning(self) -> bool:
|
|
"""
|
|
Checks the OPC connection and triggers notifications if the connection is lost.
|
|
Returns:
|
|
bool: True if the connection is lost, False otherwise.
|
|
"""
|
|
|
|
self.non_receive_count += 1
|
|
metrics.OPC_CYCLES_WITHOUT_DATA.labels(
|
|
pod_id=self.pod_id, server_name=self.name).set(self.non_receive_count)
|
|
if self.non_receive_count >= 5:
|
|
self.send_notification(
|
|
metadata=self.metadata,
|
|
notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
|
|
message=f'{self.non_receive_count} cycles without '
|
|
f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.ERROR
|
|
)
|
|
if self.non_receive_count >= 15:
|
|
metrics.OPC_RECONNECTIONS_TOTAL.labels(
|
|
pod_id=self.pod_id, server_name=self.name).inc()
|
|
self.send_notification(
|
|
metadata=self.metadata,
|
|
notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
|
|
message=f'Retrying to connect to server {self.name}',
|
|
block="opc_manager",
|
|
level=NotificationLevel.ERROR
|
|
)
|
|
return True
|
|
return False
|