Display_D_HP/TeamAAS-VM/Core/PLCs/OPCuaClientPLC.cs

using Opc.Ua;
using Opc.Ua.Client;
using OpcUaHelper;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using TeamAAS_VP.Enums;
using TeamAAS_VP.Interfaces;
using TeamAAS_VP.Models;

namespace TeamAAS_VP.Core.PLCs
{
    /// <summary>
    /// OPC UA 节点订阅模式。
    /// </summary>
    public enum OpcUaSubscriptionMode
    {
        /// <summary>
        /// 使用 OPC UA 原生订阅机制，由服务端主动推送节点变更。
        /// </summary>
        Native,

        /// <summary>
        /// 使用客户端轮询方式定时读取节点值，并在检测到变化时触发通知。
        /// </summary>
        CustomPolling
    }

    /// <summary>
    /// OPC UA 订阅配置选项，用于控制订阅模式、轮询间隔以及首次扫描通知行为。
    /// </summary>
    public class OpcUaSubscriptionOptions
    {
        /// <summary>
        /// 获取或设置订阅模式。
        /// 默认为 <see cref="OpcUaSubscriptionMode.Native"/>。
        /// </summary>
        public OpcUaSubscriptionMode Mode { get; set; } = OpcUaSubscriptionMode.Native;

        /// <summary>
        /// 获取或设置轮询间隔，单位为毫秒。
        /// 仅在 <see cref="Mode"/> 为 <see cref="OpcUaSubscriptionMode.CustomPolling"/> 时生效。
        /// 默认值为 50。
        /// </summary>
        public int PollingInterval { get; set; } = 50;

        /// <summary>
        /// 获取或设置首次扫描时是否立即通知当前值。
        /// 仅在自定义轮询订阅模式下生效。
        /// </summary>
        public bool NotifyOnFirstScan { get; set; }
    }

    public class OpcUaClientPLC : IPlc
    {
        private readonly object _pollingSubscriptionLock = new object();
        private readonly Dictionary<string, PollingSubscriptionContext> _pollingSubscriptions = new Dictionary<string, PollingSubscriptionContext>();

        public Guid Id{ get; private set; }

        public CommunicationType CommunicationType { get; private set; }

        public string EndpointUrl { get; private set; }

        public int Index { get; set; } 

        public string Name { get; private set; }

        public OpcUaClient OpcUaClient { get; private set; }

        public bool IsConnected { get { return OpcUaClient.Connected; } }

        /// <summary>
        /// 节点头
        /// </summary>
        public string NodeHeader { get; private set; }

        /// <summary>
        /// 获取或设置默认的节点订阅模式。
        /// 在未显式传入订阅选项时，`SubscribeNodes` 将使用该模式决定采用 OPC UA 原生订阅还是客户端轮询订阅。
        /// 默认值为 <see cref="OpcUaSubscriptionMode.Native"/>。
        /// </summary>
        public OpcUaSubscriptionMode DefaultSubscriptionMode { get; set; } = OpcUaSubscriptionMode.Native;

        /// <summary>
        /// 获取或设置默认的订阅轮询间隔，单位为毫秒。
        /// 仅当默认订阅模式或实际订阅选项使用自定义轮询时生效。
        /// 默认值为 `100`。
        /// </summary>
        public int DefaultSubscriptionPollingInterval { get; set; } = 50;

        /// <summary>
        /// 获取或设置在默认轮询订阅模式下，首次扫描节点时是否立即触发一次数据变更通知。
        /// </summary>
        public bool DefaultNotifyOnFirstScan { get; set; }

        /// <summary>
        /// PLC 连接状态变化事件。
        /// 当 OPC UA 客户端完成连接、开始重连或重连完成时触发，第二个参数表示当前连接状态。
        /// </summary>
        public event Action<object, bool> ConnectChangedEvent;

        /// <summary>
        /// 轮询订阅上下文。
        /// 用于保存某个自定义轮询订阅任务的节点集合、上次读取值、回调处理器以及取消控制信息。
        /// </summary>
        private class PollingSubscriptionContext
        {
            /// <summary>
            /// 订阅唯一标识，用于区分不同的订阅任务。
            /// </summary>
            public string Key { get; set; }

            /// <summary>
            /// 当前订阅需要轮询的节点集合。
            /// </summary>
            public List<string> NodeIds { get; set; }

            /// <summary>
            /// 记录每个节点上一次读取到的值，用于比较是否发生变化。
            /// </summary>
            public Dictionary<string, object> LastValues { get; } = new Dictionary<string, object>();

            /// <summary>
            /// 节点值发生变化时执行的回调委托。
            /// </summary>
            public Action<(string key, string nodeId, object value)> DataChangeHandler { get; set; }

            /// <summary>
            /// 用于控制轮询任务取消的令牌源。
            /// </summary>
            public CancellationTokenSource CancellationTokenSource { get; set; }

            /// <summary>
            /// 轮询周期，单位为毫秒。
            /// </summary>
            public int PollingInterval { get; set; }

            /// <summary>
            /// 指示首次扫描时是否立即上报当前节点值。
            /// </summary>
            public bool NotifyOnFirstScan { get; set; }
        }

        public OpcUaClientPLC(Guid id,int index, string name, string endpointUrl,string nodeHeader, CommunicationType communicationType)
        {
            Id = id;
            Index = index;
            Name = name;
            EndpointUrl = endpointUrl;
            NodeHeader = nodeHeader;
            CommunicationType = communicationType;
            OpcUaClient = new OpcUaClient();
            // 初始化PLC连接
            OpcUaClient.UserIdentity = new UserIdentity(new AnonymousIdentityToken());
            // connect to server, this is a sample
            try
            {
                OpcUaClient.ConnectComplete += OpcUaClient_ConnectComplete;
                OpcUaClient.ReconnectStarting += OpcUaClient_ReconnectStarting;
                OpcUaClient.ReconnectComplete += OpcUaClient_ReconnectComplete;
                OpcUaClient.OpcStatusChange += OpcUaClient_OpcStatusChange;
            }
            catch (Exception)
            {
                return;
            }
        }

        public OpcUaClientPLC(PlcInfo plcInfo)
        {
            Id = plcInfo.Id;
            Index=plcInfo.Index;
            Name = plcInfo.Name;
            NodeHeader=plcInfo.NodeHeader;
            EndpointUrl = plcInfo.EndpointUrl;
            CommunicationType = plcInfo.CommunicationType;
            OpcUaClient = new OpcUaClient();
            // 初始化PLC连接
            OpcUaClient.UserIdentity = new UserIdentity(new AnonymousIdentityToken());
            // connect to server, this is a sample
            try
            {
                OpcUaClient.ConnectComplete += OpcUaClient_ConnectComplete;
                OpcUaClient.ReconnectStarting += OpcUaClient_ReconnectStarting;
                OpcUaClient.ReconnectComplete += OpcUaClient_ReconnectComplete;
                OpcUaClient.OpcStatusChange += OpcUaClient_OpcStatusChange;
            }
            catch (Exception)
            {
                return;
            }
        }

        public void Connect()
        {
            OpcUaClient.ConnectServer(EndpointUrl);
        }

        public Task ConnectAsync()
        {
            return OpcUaClient.ConnectServer(EndpointUrl);
        }

        public void Disconnect()
        {
            OpcUaClient.Disconnect();
        }

        public void Dispose()
        {
            StopAllPollingSubscriptions();
            OpcUaClient.Disconnect();
            OpcUaClient.ConnectComplete -= OpcUaClient_ConnectComplete;
            OpcUaClient.ReconnectStarting -= OpcUaClient_ReconnectStarting;
            OpcUaClient.ReconnectComplete -= OpcUaClient_ReconnectComplete;
            OpcUaClient.OpcStatusChange -= OpcUaClient_OpcStatusChange;
        }

        /// <summary>
        /// 读取多个节点的值
        /// </summary>
        /// <param name="nodeIds"></param>
        /// <returns></returns>
        public Dictionary<string, object> ReadNodes(string[] nodeIds)
        {
            var result = new Dictionary<string, object>();
            var readNodeIds = nodeIds.Select(s =>
            {
                if (s.StartsWith(NodeHeader))
                {
                    return s;
                }
                else
                {
                    return NodeHeader + s;
                }
            }).ToArray();
            List<NodeId> readNodeIdList = new List<NodeId>();
            foreach (var readNodeId in readNodeIds)
            {
                readNodeIdList.Add(new NodeId(readNodeId));
            }
            var values = OpcUaClient.ReadNodes(readNodeIdList.ToArray());
            for (int i = 0; i < nodeIds.Length; i++)
            {
                result[nodeIds[i]] = values[i].Value;
            }
            return result;
        }

        /// <summary>
        /// 异步读取多个节点的值
        /// </summary>
        /// <param name="nodeIds"></param>
        /// <returns></returns>
        public async Task<Dictionary<string, object>> ReadNodesAsync(string[] nodeIds)
        {
            var result = new Dictionary<string, object>();
            var readNodeIds = nodeIds.Select(s =>
            {
                if (s.StartsWith(NodeHeader))
                {
                    return s;
                }
                else
                {
                    return NodeHeader + s;
                }
            }).ToArray();
            List<NodeId> readNodeIdList = new List<NodeId>();
            foreach (var readNodeId in readNodeIds)
            {
                readNodeIdList.Add(new NodeId(readNodeId));
            }
            var values = await OpcUaClient.ReadNodesAsync(readNodeIdList.ToArray());
            for (int i = 0; i < nodeIds.Length; i++)
            {
                result[nodeIds[i]] = values[i].Value;
            }
            return result;
        }

        /// <summary>
        /// 读取单个节点的值
        /// </summary>
        /// <param name="nodeId"></param>
        /// <returns></returns>
        public object ReadNode(string nodeId)
        {
            string readNodeId = nodeId;
            if (!nodeId.StartsWith(NodeHeader))
            {
                readNodeId = NodeHeader + nodeId;
            }

            var value = OpcUaClient.ReadNode(new NodeId(readNodeId));
            return value.Value;
        }

        /// <summary>
        /// 异步读取单个节点的值
        /// </summary>
        /// <param name="nodeId"></param>
        /// <returns></returns>
        public Task<object> ReadNodeAsync(string nodeId)
        {
            return Task.Run(() =>
            {
                return ReadNode(nodeId);
            });
        }

        /// <summary>
        /// 读取单个节点的值
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="nodeId"></param>
        /// <returns></returns>
        public T ReadNode<T>(string nodeId)
        {
            string readNodeId = nodeId;
            if (!nodeId.StartsWith(NodeHeader))
            {
                readNodeId = NodeHeader + nodeId;
            }
            return OpcUaClient.ReadNode<T>(readNodeId);
        }

        /// <summary>
        /// 异步读取单个节点的值
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="nodeId"></param>
        /// <returns></returns>
        public async Task<T> ReadNodeAsync<T>(string nodeId)
        {
            string readNodeId = nodeId;
            if (!nodeId.StartsWith(NodeHeader))
            {
                readNodeId = NodeHeader + nodeId;
            }
            return await OpcUaClient.ReadNodeAsync<T>(readNodeId);
        }

        /// <summary>
        /// 写入单个节点的值
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="nodeId"></param>
        /// <param name="value"></param>
        /// <returns></returns>
        public bool WriteNode<T>(string nodeId, T value)
        {
            string writeNodeId = nodeId;
            if (!nodeId.StartsWith(NodeHeader))
            {
                writeNodeId = NodeHeader + nodeId;
            }
            return OpcUaClient.WriteNode<T>(writeNodeId, value);
        }

        /// <summary>
        /// 异步写入单个节点的值
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="nodeId"></param>
        /// <param name="value"></param>
        /// <returns></returns>
        public async Task<bool> WriteNodeAsync<T>(string nodeId, T value)
        {
            string writeNodeId = nodeId;
            if (!nodeId.StartsWith(NodeHeader))
            {
                writeNodeId = NodeHeader + nodeId;
            }
            return await OpcUaClient.WriteNodeAsync<T>(writeNodeId, value);
        }

        /// <summary>
        /// 写入多个节点
        /// </summary>
        /// <param name="nodeValues"></param>
        /// <returns></returns>
        public bool WriteNodes(Dictionary<string, object> nodeValues)
        {
            var writeNodeValues = new Dictionary<string, object>();
            foreach (var kvp in nodeValues)
            {
                string writeNodeId = kvp.Key;
                if (!kvp.Key.StartsWith(NodeHeader))
                {
                    writeNodeId = NodeHeader + kvp.Key;
                }
                writeNodeValues[writeNodeId] = kvp.Value;
            }
            return OpcUaClient.WriteNodes(writeNodeValues.Keys.ToArray(), writeNodeValues.Values.ToArray());
        }

        /// <summary>
        /// 写入多个节点异步
        /// </summary>
        /// <param name="nodeValues"></param>
        /// <returns></returns>
        public Task<bool> WriteNodesAsync(Dictionary<string, object> nodeValues)
        {
            return Task.Run(() =>
            {
                return WriteNodes(nodeValues);
            });
        }

        /// <summary>
        /// 使用当前实例的默认订阅配置订阅多个节点。
        /// </summary>
        /// <param name="key">订阅唯一标识，用于区分和管理同一客户端上的不同订阅任务。</param>
        /// <param name="nodeIds">需要订阅的节点标识集合，可传入带或不带 <see cref="NodeHeader"/> 前缀的节点名。</param>
        /// <param name="dataChangeHandler">节点值变化时触发的回调，返回订阅键、节点标识和值。</param>
        public void SubscribeNodes(string key, List<string> nodeIds, Action<(string key,string nodeId,object value)> dataChangeHandler)
        {
            SubscribeNodes(key, nodeIds, dataChangeHandler, new OpcUaSubscriptionOptions
            {
                Mode = DefaultSubscriptionMode,
                PollingInterval = DefaultSubscriptionPollingInterval,
                NotifyOnFirstScan = DefaultNotifyOnFirstScan
            });
        }

        /// <summary>
        /// 根据传入的OpcUaSubscriptionMode选项订阅多个节点
        /// </summary>
        /// <param name="key"></param>
        /// <param name="nodeIds"></param>
        /// <param name="dataChangeHandler"></param>
        /// <param name="mode"></param>
        public void SubscribeNodes(string key, List<string> nodeIds, Action<(string key, string nodeId, object value)> dataChangeHandler, OpcUaSubscriptionMode mode)
        {
            var options = new OpcUaSubscriptionOptions
            {
                Mode = mode,
                PollingInterval = DefaultSubscriptionPollingInterval,
                NotifyOnFirstScan = DefaultNotifyOnFirstScan
            };
            SubscribeNodes(key, nodeIds, dataChangeHandler, options);
        }

        /// <summary>
        /// 根据指定订阅选项订阅多个节点。
        /// </summary>
        /// <remarks>
        /// 该方法会先停止同一 <paramref name="key"/> 对应的轮询订阅，再根据 <paramref name="options"/> 中的模式
        /// 选择使用 OPC UA 原生订阅或客户端轮询订阅。节点列表会自动去重并忽略空白项。
        /// </remarks>
        /// <param name="key">订阅唯一标识，不能为空或仅包含空白字符。</param>
        /// <param name="nodeIds">待订阅的节点集合。</param>
        /// <param name="dataChangeHandler">节点值变化通知回调。</param>
        /// <param name="options">订阅配置；为 <c>null</c> 时使用默认配置。</param>
        /// <exception cref="ArgumentException"><paramref name="key"/> 为空或仅包含空白字符时抛出。</exception>
        /// <exception cref="ArgumentNullException"><paramref name="nodeIds"/> 或 <paramref name="dataChangeHandler"/> 为 <c>null</c> 时抛出。</exception>
        public void SubscribeNodes(string key, List<string> nodeIds, Action<(string key, string nodeId, object value)> dataChangeHandler, OpcUaSubscriptionOptions options)
        {
            if (string.IsNullOrWhiteSpace(key))
            {
                throw new ArgumentException("订阅Key不能为空。", nameof(key));
            }
            if (nodeIds == null)
            {
                throw new ArgumentNullException(nameof(nodeIds));
            }
            if (dataChangeHandler == null)
            {
                throw new ArgumentNullException(nameof(dataChangeHandler));
            }

            var subscriptionOptions = options ?? new OpcUaSubscriptionOptions();
            var normalizedNodeIds = nodeIds.Where(s => !string.IsNullOrWhiteSpace(s)).Distinct().ToList();
            StopPollingSubscription(key);

            if (subscriptionOptions.Mode == OpcUaSubscriptionMode.CustomPolling)
            {
                StartPollingSubscription(key, normalizedNodeIds, dataChangeHandler, subscriptionOptions);
                return;
            }

            SubscribeNodesByOpc(key, normalizedNodeIds, dataChangeHandler);
        }

        /// <summary>
        /// 使用 OPC UA 原生订阅机制订阅节点。
        /// </summary>
        /// <remarks>
        /// 订阅前会自动补齐节点前缀；收到服务端推送后，会移除返回节点中的 <see cref="NodeHeader"/> 前缀，
        /// 再将变化值通过回调透出给业务层。
        /// </remarks>
        /// <param name="key">订阅唯一标识。</param>
        /// <param name="nodeIds">节点集合。</param>
        /// <param name="dataChangeHandler">节点值变化回调。</param>
        private void SubscribeNodesByOpc(string key, List<string> nodeIds, Action<(string key, string nodeId, object value)> dataChangeHandler)
        {
            OpcUaClient.AddSubscription(key, nodeIds.Select(s =>
            {
                return NormalizeNodeId(s);
            }).ToArray(), (key1, monitoredItem, args) =>
            {
                MonitoredItemNotification notification = args.NotificationValue as MonitoredItemNotification;
                if (notification == null)
                {
                    return;
                }
                string nodeId = monitoredItem.StartNodeId.ToString();
                nodeId = TrimNodeHeader(nodeId);
                // 触发数据变化事件
                dataChangeHandler?.Invoke((key1, nodeId, notification.Value.WrappedValue.Value));
            });
        }

        /// <summary>
        /// 启动自定义轮询订阅任务。
        /// </summary>
        /// <remarks>
        /// 该方法会创建轮询上下文并登记到内部订阅字典中，然后在后台任务中执行轮询循环。
        /// 轮询间隔最小限制为 10 毫秒，以避免过于频繁的读取请求。
        /// </remarks>
        /// <param name="key">订阅唯一标识。</param>
        /// <param name="nodeIds">需要轮询的节点集合。</param>
        /// <param name="dataChangeHandler">节点值变化回调。</param>
        /// <param name="options">轮询订阅选项。</param>
        private void StartPollingSubscription(string key, List<string> nodeIds, Action<(string key, string nodeId, object value)> dataChangeHandler, OpcUaSubscriptionOptions options)
        {
            var context = new PollingSubscriptionContext
            {
                Key = key,
                NodeIds = nodeIds,
                DataChangeHandler = dataChangeHandler,
                CancellationTokenSource = new CancellationTokenSource(),
                PollingInterval = Math.Max(10, options.PollingInterval),
                NotifyOnFirstScan = options.NotifyOnFirstScan
            };

            lock (_pollingSubscriptionLock)
            {
                _pollingSubscriptions[key] = context;
            }

            Task.Run(() => PollingSubscriptionLoopAsync(context));
        }

        /// <summary>
        /// 执行轮询订阅的后台循环。
        /// </summary>
        /// <remarks>
        /// 当客户端已连接且存在待轮询节点时，循环会批量读取节点值，并与上次缓存值进行比较：
        /// 首次读取时根据配置决定是否立即通知，后续仅在值发生变化时触发回调。
        /// 当取消令牌触发或发生取消异常时，循环退出。
        /// </remarks>
        /// <param name="context">轮询订阅上下文。</param>
        /// <returns>表示异步轮询操作的任务。</returns>
        private async Task PollingSubscriptionLoopAsync(PollingSubscriptionContext context)
        {
            while (!context.CancellationTokenSource.IsCancellationRequested)
            {
                try
                {
                    if (IsConnected && context.NodeIds.Count > 0)
                    {
                        var values = await ReadNodesAsync(context.NodeIds.ToArray()).ConfigureAwait(false);
                        foreach (var nodeId in context.NodeIds)
                        {
                            object currentValue;
                            if (!values.TryGetValue(nodeId, out currentValue))
                            {
                                continue;
                            }

                            object lastValue;
                            if (!context.LastValues.TryGetValue(nodeId, out lastValue))
                            {
                                context.LastValues[nodeId] = currentValue;
                                if (context.NotifyOnFirstScan)
                                {
                                    context.DataChangeHandler?.Invoke((context.Key, nodeId, currentValue));
                                }
                                continue;
                            }

                            if (!Utils.IsEqual(lastValue, currentValue))
                            {
                                context.LastValues[nodeId] = currentValue;
                                context.DataChangeHandler?.Invoke((context.Key, nodeId, currentValue));
                            }
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception)
                {
                }

                try
                {
                    await Task.Delay(context.PollingInterval, context.CancellationTokenSource.Token).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
            }
        }

        /// <summary>
        /// 停止指定键对应的轮询订阅。
        /// </summary>
        /// <param name="key">要停止的订阅唯一标识。</param>
        private void StopPollingSubscription(string key)
        {
            PollingSubscriptionContext context = null;
            lock (_pollingSubscriptionLock)
            {
                if (_pollingSubscriptions.TryGetValue(key, out context))
                {
                    _pollingSubscriptions.Remove(key);
                }
            }

            context?.CancellationTokenSource.Cancel();
            context?.CancellationTokenSource.Dispose();
        }

        /// <summary>
        /// 停止并清理当前实例中的所有轮询订阅。
        /// </summary>
        private void StopAllPollingSubscriptions()
        {
            List<PollingSubscriptionContext> contexts;
            lock (_pollingSubscriptionLock)
            {
                contexts = _pollingSubscriptions.Values.ToList();
                _pollingSubscriptions.Clear();
            }

            foreach (var context in contexts)
            {
                context.CancellationTokenSource.Cancel();
                context.CancellationTokenSource.Dispose();
            }
        }

        /// <summary>
        /// 规范化节点标识，确保其包含节点头前缀。
        /// </summary>
        /// <param name="nodeId">原始节点标识。</param>
        /// <returns>包含 <see cref="NodeHeader"/> 前缀的完整节点标识。</returns>
        private string NormalizeNodeId(string nodeId)
        {
            if (nodeId.StartsWith(NodeHeader))
            {
                return nodeId;
            }

            return NodeHeader + nodeId;
        }

        /// <summary>
        /// 去除节点标识中的节点头前缀。
        /// </summary>
        /// <param name="nodeId">完整节点标识。</param>
        /// <returns>移除 <see cref="NodeHeader"/> 前缀后的节点标识；若原值不包含此前缀则直接返回原值。</returns>
        private string TrimNodeHeader(string nodeId)
        {
            if (nodeId.StartsWith(NodeHeader))
            {
                return nodeId.Substring(NodeHeader.Length);
            }

            return nodeId;
        }
        #region OPC事件
        private void OpcUaClient_OpcStatusChange(object sender, OpcUaStatusEventArgs e)
        {
            //LogHelper.WriteLogInfo($"OPC状态发生改变：Error：{e.Error},Time：{e.Time},Text：{e.Text}");
        }

        private void OpcUaClient_ReconnectComplete(object sender, EventArgs e)
        {
            ConnectChangedEvent?.Invoke(this, OpcUaClient.Connected);
        }

        private void OpcUaClient_ReconnectStarting(object sender, EventArgs e)
        {
            ConnectChangedEvent?.Invoke(this, OpcUaClient.Connected);
        }

        private void OpcUaClient_ConnectComplete(object sender, EventArgs e)
        {
            ConnectChangedEvent?.Invoke(this, OpcUaClient.Connected);
        }
        #endregion
    }
}