1. 订阅函数变化更新 新增订阅源数据比对更新

2. S7读取修改

README.md

@@ -311,3 +311,7 @@ while(true)
 1. 自定义订阅修改
 2. 修改对象比对
 3. 版本更新
+
+#### 2023-10-17
+1. 订阅函数变化更新 新增订阅源数据比对更新
+2. S7读取修改

src/YSAI.DAQ/YSAI.Core/interface/IBaseAbstract.cs

@@ -1,4 +1,5 @@
 using YSAI.Core.data;
+using YSAI.Core.@interface;
 
 namespace YSAI.Core.@interface
 {
@@ -13,7 +14,7 @@ namespace YSAI.Core.@interface
         /// </summary>
         /// <param name="sender">自身对象</param>
         /// <param name="e">事件结果</param>
-        protected virtual void OnEventHandler(object? sender, EventResult e)
+        protected void OnEventHandler(object? sender, EventResult e)
         {
             OnEvent?.Invoke(sender, e);
         }

src/YSAI.DAQ/YSAI.Core/subscribe/SubscribeOperate.cs

@@ -3,6 +3,7 @@ using System.Runtime.CompilerServices;
 using YSAI.Core.data;
 using YSAI.Core.@enum;
 using YSAI.Core.@interface;
+using YSAI.Core.subscribe.core;
 using YSAI.Unility;
 
 namespace YSAI.Core.subscription
@@ -27,7 +28,7 @@ namespace YSAI.Core.subscription
         /// <summary>
         /// 单例模式
         /// </summary>
-        /// <param name="OpcUaData.Basics">基础数据</param>
+        /// <param name="basics">基础数据</param>
         /// <returns></returns>
         public static SubscribeOperate Instance(SubscribeData.Basics basics)
         {
@@ -63,6 +64,7 @@ namespace YSAI.Core.subscription
         {
             //设置参数
             this.basics = basics;
+            subscribeService.OnEvent += SubscribeService_OnEvent;
         }
 
         /// <summary>
@@ -389,12 +391,23 @@ namespace YSAI.Core.subscription
         }
 
         /// <summary>
-        /// 数据缓存
+        /// 订阅服务
         /// </summary>
-        ConcurrentDictionary<string, AddressValue> DataCache = new ConcurrentDictionary<string, AddressValue>();
-
-
-
+        SubscribeService<AddressValue> subscribeService = SubscribeService<AddressValue>.Instance();
+        /// <summary>
+        /// 订阅服务事件
+        /// </summary>
+        /// <param name="sender"></param>
+        /// <param name="e"></param>
+        /// <exception cref="NotImplementedException"></exception>
+        private void SubscribeService_OnEvent(object? sender, EventResult e)
+        {
+            SubscribeSource<AddressValue>? source = e.RData as SubscribeSource<AddressValue>;
+            if (source != null)
+            {
+                OnEventHandler(this, new EventResult(true, "变化数据", new ConcurrentDictionary<string, AddressValue> { [source.Source.AddressName] = source.Source }, RType: ResultType.KeyValue));
+            }
+        }
         /// <summary>
         /// 任务处理
         /// </summary>
@@ -413,20 +426,7 @@ namespace YSAI.Core.subscription
                     {
                         if (queueData != null && !token.IsCancellationRequested)
                         {
-                            if (DataCache.ContainsKey(queueData.AddressName))
-                            {
-                                if (!DataCache[queueData.AddressName].Equals(queueData))
-                                {
-                                    //抛出差异数据
-                                    OnEventHandler(this, new EventResult(true, "变化数据", new ConcurrentDictionary<string, AddressValue> { [queueData.AddressName] = queueData }, RType: ResultType.KeyValue));
-                                }
-                            }
-                            else
-                            {
-                                //抛出差异数据
-                                OnEventHandler(this, new EventResult(true, "变化数据", new ConcurrentDictionary<string, AddressValue> { [queueData.AddressName] = queueData }, RType: ResultType.KeyValue));
-                            }
-                            DataCachePool.AddOrUpdate(queueData.AddressName, queueData, (k, v) => queueData);
+                            subscribeService.SetAsync(queueData.AddressName, queueData);
                         }
                     }
                     //队列里面的数据处理完休息一下

src/YSAI.DAQ/YSAI.Core/subscribe/core/SubscribeService.cs

@@ -0,0 +1,137 @@
+using System.Collections.Concurrent;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using YSAI.Core.reflection;
+using YSAI.Core.subscription;
+using YSAI.Log;
+
+namespace YSAI.Core.subscribe.core
+{
+    /// <summary>
+    /// 订阅服务数据层，与调用者共存亡
+    /// 
+    /// 确保SN 是唯一，不然会导致数据冲突
+    /// </summary>
+    public class SubscribeService<T> : IBaseAbstract
+    {
+        protected override string TAG => "SubscribeService";
+        /// <summary>
+        /// 锁
+        /// </summary>
+        private static readonly object Lock = new object();
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        private static SubscribeService<T>? ThisObjList; 
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static SubscribeService<T> Instance()
+        {
+            if (ThisObjList == null)
+            {
+                lock (Lock)
+                {
+                    if (ThisObjList == null)
+                    {
+                        ThisObjList = new SubscribeService<T>();
+                    }
+                }
+            }
+            return ThisObjList;
+        }
+        /// <summary>
+        /// 数据源
+        /// </summary>
+        private ConcurrentDictionary<string, SubscribeSource<T>> Source = new ConcurrentDictionary<string, SubscribeSource<T>>();
+        /// <summary>
+        /// 订阅事件
+        /// </summary>
+        public event EventHandler<EventResult> OnEvent;
+        /// <summary>
+        /// 事件传递
+        /// </summary>
+        /// <param name="sender">源</param>
+        /// <param name="e">事件结果</param>
+        private void OnEventHandler(object? sender, EventResult e)
+        {
+            OnEvent?.Invoke(sender, e);
+        }
+        /// <summary>
+        /// 释放
+        /// </summary>
+        public void Dispose()
+        {
+            foreach (var item in Source)
+            {
+                item.Value.Dispose();
+            }
+            Source.Clear();
+            GC.SuppressFinalize(this);
+            GC.Collect();
+        }
+        /// <summary>
+        /// 设置数据
+        /// </summary>
+        /// <param name="SN">唯一标识符</param>
+        /// <returns>设置的啥类型，返回啥类型</returns>
+        public OperateResult Get(string SN)
+        {
+            if (ThisObjList == null)
+            {
+                throw new Exception("please use singleton mode");
+            }
+            Depart("Get");
+            if (Source.ContainsKey(SN))
+            {
+               return Source[SN].Get();
+            }
+            return Break("Get", false,$"({SN}) Instance does not exist");
+        }
+        /// <summary>
+        /// 设置数据异步
+        /// </summary>
+        /// <param name="SN">唯一标识符</param>
+        /// <returns>设置的啥类型，返回啥类型</returns>
+        public Task<OperateResult> GetAsync(string SN)
+        {
+            return Task.Run(() => Get(SN));
+        }
+
+        /// <summary>
+        /// 设置数据
+        /// </summary>
+        /// <typeparam name="T">对象</typeparam>
+        /// <param name="SN">唯一标识符</param>
+        /// <param name="Data">数据</param>
+        /// <returns>统一结果</returns>
+        public OperateResult Set(string SN, T Data)
+        {
+            if (ThisObjList == null)
+            {
+                throw new Exception("please use singleton mode");
+            }
+            Depart("Set");
+            if (!Source.ContainsKey(SN))
+            {
+                SubscribeSource<T> core = new SubscribeSource<T>(SN);
+                core.OnEvent += OnEventHandler;
+                Source.AddOrUpdate(SN, core, (k, v) => core);
+            }
+            Source[SN].Set(Data);
+            return Break("Set", true);
+        }
+        /// <summary>
+        /// 设置数据异步
+        /// </summary>
+        /// <typeparam name="T">对象</typeparam>
+        /// <param name="SN">唯一标识符</param>
+        /// <param name="Data">数据</param>
+        /// <returns>统一结果</returns>
+        public Task<OperateResult> SetAsync(string SN, T Data)
+        {
+            return Task.Run(() => Set(SN, Data));
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Core/subscribe/core/SubscribeSource.cs

@@ -0,0 +1,111 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using YSAI.Core.reflection;
+using YSAI.Unility;
+
+namespace YSAI.Core.subscribe.core
+{
+    /// <summary>
+    /// 订阅数据库
+    /// </summary>
+    public class SubscribeSource<T>: IBaseAbstract, IDisposable
+    {
+        /// <summary>
+        /// 构造函数传入SN
+        /// </summary>
+        /// <param name="sN"></param>
+        public SubscribeSource(string sN)
+        {
+            SN = sN;
+        }
+        /// <summary>
+        /// 订阅事件
+        /// </summary>
+        public event EventHandler<EventResult> OnEvent;
+        protected virtual string TAG => "SubscribeSource";
+        /// <summary>
+        /// 唯一标识符
+        /// </summary>
+        public string SN { get; set; }
+        /// <summary>
+        /// 数据源
+        /// </summary>
+        public T Source { get; set; }
+        /// <summary>
+        /// 更新时间
+        /// </summary>
+        public DateTime UpdateTime { get; set; } = DateTime.Now;
+        /// <summary>
+        /// 设置数据
+        /// </summary>
+        /// <typeparam name="T">对象</typeparam>
+        /// <param name="Data">数据</param>
+        /// <returns>统一结果</returns>
+        public OperateResult Set(T Data)
+        {
+            Depart("Set");
+            UpdateTime = DateTime.Now;
+            if (!Data.Comparer(Source, new string[] { "Time" }).result)
+            {
+                Source = Data;
+                OnEventHandler(this, new EventResult(true, "Data Update", this, @enum.ResultType.Object));
+            }
+            return Break("Set", true);
+        }
+
+        /// <summary>
+        /// 设置数据异步
+        /// </summary>
+        /// <typeparam name="T">对象</typeparam>
+        /// <param name="Data">数据</param>
+        /// <returns>统一结果</returns>
+        public Task<OperateResult> SetAsync(T Data)
+        {
+            return Task.Run(()=> Set(Data));
+        }
+
+
+        /// <summary>
+        /// 获取数据
+        /// </summary>
+        /// <returns>设置的啥类型，返回啥类型</returns>
+        public OperateResult Get()
+        {
+            return Break(Depart("Get"), true, "Get Succeed", this, @enum.ResultType.Object);
+        }
+
+        /// <summary>
+        /// 获取数据
+        /// </summary>
+        /// <returns>设置的啥类型，返回啥类型</returns>
+        public Task<OperateResult> GetAsync()
+        {
+            return Task.Run(() => Get());
+        }
+
+
+        /// <summary>
+        /// 事件传递
+        /// </summary>
+        /// <param name="sender">源</param>
+        /// <param name="e">事件结果</param>
+        private void OnEventHandler(object? sender, EventResult e)
+        {
+            OnEvent?.Invoke(sender, e);
+        }
+        /// <summary>
+        /// 释放
+        /// </summary>
+        public void Dispose()
+        {
+            GC.SuppressFinalize(this);
+            GC.Collect();
+        }
+    }
+}

src/YSAI.DAQ/YSAI.S7/YSAI.S7.csproj

@@ -14,11 +14,11 @@
 
 	<ItemGroup>
     <PackageReference Include="S7netplus" Version="0.20.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.54" />
+    <!--<PackageReference Include="YSAI.Core" Version="1.0.0.54" />-->
   </ItemGroup>
 
-	<!--<ItemGroup>
+	<ItemGroup>
 		<ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
-	</ItemGroup>-->
+	</ItemGroup>
 	
 </Project>

src/YSAI.DAQ/YSAI.S7/client/S7ClientOperate.cs

@@ -189,26 +189,110 @@ namespace YSAI.S7.client
                         {
                             //获取地址数据项
                             DataItem dataItem = DataItem.FromAddress(item.AddressName);
+                            //switch (dataItem.VarType)
+                            //{
+                            //    case VarType.Bit:   //bool 类型
+                            //        Value = ((bool)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.Byte:  //字节数据
+                            //        Value = ((byte)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.Word:  //二进制
+                            //        Value = ((ushort)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.DWord:  //二进制
+                            //        Value = ((uint)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.Int:  //16位 int 在C# 中指 uint
+                            //        Value = ((uint)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.DInt:  //32位 int 在C# 中指 int
+                            //        Value = ((int)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.Real:  //浮点型  Float
+                            //        Value = ((uint)PlcS7.ReadAsync(item.AddressName).Result).ConvertToFloat().ToString();
+                            //        break;
+                            //    case VarType.LReal:  // 浮点型  double
+                            //        Value = ((double)PlcS7.ReadAsync(item.AddressName).Result).ToDouble().ToString();
+                            //        break;
+                            //    case VarType.String:   //字符串
+                            //    case VarType.S7String:  //字符串
+                            //    case VarType.S7WString:  //字符串
+                            //        byte[]? head = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, 2);
+                            //        byte[]? bytes = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr + 2, head[1]);
+                            //        Value = Encoding.ASCII.GetString(bytes).TrimEnd(new char[] { '\0' });
+                            //        break;
+                            //    case VarType.Timer:  //时间
+                            //        Value = ((double)PlcS7.ReadAsync(item.AddressName).Result).ToDouble().ToString();
+                            //        break;
+                            //    case VarType.Counter:  //计算器？
+                            //        Value = ((ushort)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.DateTime:  //时间类型
+                            //        Value = ((System.DateTime)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //    case VarType.DateTimeLong:   //时间类型
+                            //        Value = ((System.DateTime)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                            //        break;
+                            //}
+
                             switch (item.AddressDataType)
                             {
                                 case Core.@enum.DataType.Bool:
-                                    Value = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.Count).ValToBinString().BinStringToByte().Value.SelectBit(0).ToString();
+                                    Value = ((bool)PlcS7.ReadAsync(item.AddressName).Result).ToString();
                                     break;
                                 case Core.@enum.DataType.String:
                                     byte[]? head = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, 2);
                                     byte[]? bytes = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr + 2, head[1]);
                                     Value = Encoding.ASCII.GetString(bytes).TrimEnd(new char[] { '\0' });
                                     break;
+                                case Core.@enum.DataType.Byte:
+                                    Value = ((byte)PlcS7.ReadAsync(item.AddressName).Result).ToString();
+                                    break;
+                                case Core.@enum.DataType.Double:
+                                    Value = ((ushort)PlcS7.ReadAsync(item.AddressName).Result).ToDouble().ToString();
+                                    break;
                                 case Core.@enum.DataType.Float:
-                                    Value = ((uint)PlcS7?.Read(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.VarType, dataItem.Count)).ConvertToFloat().ToString();
+                                    Value = ((uint)PlcS7.ReadAsync(item.AddressName).Result).ConvertToFloat().ToString();
                                     break;
                                 case Core.@enum.DataType.Int:
-                                    Value = ((uint)PlcS7?.Read(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.VarType, dataItem.Count)).ConvertToInt().ToString();
+                                    Value = ((uint)PlcS7.ReadAsync(item.AddressName).Result).ConvertToInt().ToString();
+                                    break;
+                                case Core.@enum.DataType.Uint:
+                                    Value = ((int)PlcS7.ReadAsync(item.AddressName).Result).ConvertToUInt().ToString();
                                     break;
                                 default:
                                     Value = $"不支持{item.AddressDataType}类型读取";
                                     break;
                             }
+
+
+                            if (string.IsNullOrEmpty(Value))
+                            {
+                                Value = "null";
+                            }
+                            //switch (item.AddressDataType)
+                            //{
+                            //    case Core.@enum.DataType.Bool:
+                            //        Value = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.Count).ValToBinString().BinStringToByte().Value.SelectBit(0).ToString();
+                            //        break;
+                            //    case Core.@enum.DataType.String:
+                            //        byte[]? head = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, 2);
+                            //        byte[]? bytes = PlcS7?.ReadBytes(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr + 2, head[1]);
+                            //        Value = Encoding.ASCII.GetString(bytes).TrimEnd(new char[] { '\0' });
+                            //        break;
+                            //    case Core.@enum.DataType.Byte:
+                            //        break;
+                            //    case Core.@enum.DataType.Float:
+                            //        Value = ((uint)PlcS7?.Read(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.VarType, dataItem.Count)).ConvertToFloat().ToString();
+                            //        break;
+                            //    case Core.@enum.DataType.Int:
+                            //        Value = ((uint)PlcS7?.Read(dataItem.DataType, dataItem.DB, dataItem.StartByteAdr, dataItem.VarType, dataItem.Count)).ConvertToInt().ToString();
+                            //        break;
+                            //    default:
+                            //        Value = $"不支持{item.AddressDataType}类型读取";
+                            //        break;
+                            //}
                         }
 
                         //数据处理

src/YSAI.DAQ/YSAI.Test.All/Program.cs

@@ -1,7 +1,33 @@
-using S7.Net;
+
+
+//using YSAI.Core.subscribe.core;
+//using YSAI.Unility;
+
+//SubscribeService<double> subscribeService = SubscribeService<double>.Instance();
+//subscribeService.OnEvent += SubscribeService_OnEvent;
+
+
+//while (true)
+//{
+//    for (int i = 0; i < 100000; i++)
+//    {
+//        subscribeService.SetAsync($"{i}号点位",i);
+//    }
+//    Thread.Sleep(10000);
+//}
+
+
+
+//void SubscribeService_OnEvent(object? sender, YSAI.Core.data.EventResult e)
+//{
+//    Console.WriteLine(e.ToJson().JsonFormatting());
+//}
+
+using S7.Net;
 using System.Collections.Concurrent;
 using YSAI.Core.data;
 using YSAI.Core.@enum;
+using YSAI.Core.subscribe.core;
 using YSAI.Log;
 using YSAI.S7.client;
 using YSAI.Unility;
@@ -14,10 +40,10 @@ S7ClientOperate s7ClientOperate = S7ClientOperate.Instance(new S7ClientData.Basi
     Rack = 0,
     Slot = 1,
     S7CpuType = CpuType.S71200,
-    ChangeOut = true,
-    HandleInterval = 1,
+    ChangeOut = false,
+    HandleInterval = 0,
     SN = Guid.NewGuid().ToString()
-}); 
+});
 //打开
 OperateResult operateResult = s7ClientOperate.On();
 Console.WriteLine(operateResult.Message);
@@ -30,7 +56,7 @@ address.AddressArray = new List<AddressDetails>();
 address.AddressArray.Add(new AddressDetails()
 {
     AddressName = "DB71.DBD4",
-    AddressDataType = YSAI.Core.@enum.DataType.String,
+    AddressDataType = YSAI.Core.@enum.DataType.Float,
     AddressType = AddressType.Reality
 });
 
@@ -52,17 +78,20 @@ s7ClientOperate.Subscribe(address);
 
 while (true)
 {
-    string value = Console.ReadLine();
+    float value = float.Parse(Console.ReadLine());
     //写入
-    ConcurrentDictionary<string, string> keyValuePairs = new ConcurrentDictionary<string, string>();
+    ConcurrentDictionary<string, float> keyValuePairs = new ConcurrentDictionary<string, float>();
     keyValuePairs.TryAdd("DB71.DBD4", value);
     operateResult = s7ClientOperate.Write(keyValuePairs);
     Console.WriteLine(operateResult.Message);
+
+    SubscribeService<AddressValue> subscribeService = SubscribeService<AddressValue>.Instance();
+    Console.WriteLine();
 }
 
 void S7ClientOperate_OnEvent(object? sender, EventResult e)
 {
-    Console.WriteLine(e.ToJson());
+    Console.WriteLine(e.ToJson().JsonFormatting());
 }
 
 

src/YSAI.DAQ/YSAI.Test.Console/Program.cs

@@ -1,13 +1,13 @@
 
 
-using YSAI.Opc.ua.client;
+//using YSAI.Opc.ua.client;
 
-OpcUaClientOperate opcUaClientOperate = OpcUaClientOperate.Instance(new OpcUaClientData.Basics
-{
-    ServerUrl= "opc.tcp://127.0.0.1:8866/Opc.Ua.Service"
-});
+//OpcUaClientOperate opcUaClientOperate = OpcUaClientOperate.Instance(new OpcUaClientData.Basics
+//{
+//    ServerUrl= "opc.tcp://127.0.0.1:8866/Opc.Ua.Service"
+//});
 
-Console.WriteLine(opcUaClientOperate.On().Message); 
+//Console.WriteLine(opcUaClientOperate.On().Message); 
 
 
 //using System.Runtime.InteropServices;