协议池应用02

protocols/JiShe.CollectBus.Protocol.Contracts/Interfaces/IProtocolPlugin.cs

@@ -25,7 +25,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Interfaces
         /// <typeparam name="T">是否需要转发的扩展协议入参对象</typeparam>
         /// <param name="afnFnCode">映射读取执行方法的Code，例如10_1，表示10H_F1</param>
         /// <returns></returns>
-        Task<BuildResponse> BuildAsync(BuildRequest request);
+        Task<ProtocolBuildResponse> BuildAsync(ProtocolBuildRequest request);
 
         //Task LoginAsync(MessageReceivedLogin messageReceived);
 

protocols/JiShe.CollectBus.Protocol.Contracts/Models/ProtocolBuildRequest.cs

@@ -1,9 +1,11 @@
-namespace JiShe.CollectBus.Protocol.Contracts.SendData
+using JiShe.CollectBus.Protocol.Contracts.Models;
+
+namespace JiShe.CollectBus.Protocol.Contracts.SendData
 {
     /// <summary>
     /// 报文构建参数
     /// </summary>
-    public class BuildRequest
+    public class ProtocolBuildRequest
     {
         /// <summary>
         /// 集中器地址
@@ -21,8 +23,8 @@
         public required string ItemCode { get; set; }
 
         /// <summary>
-        /// 集中器转发协议构建组合功能码
+        /// 集中器转发协议构建构建参数
         /// </summary>
-        public string SubItemCode { get; set; } 
+        public SubProtocolBuildRequest SubProtocolRequest { get; set; } 
     }
 }

protocols/JiShe.CollectBus.Protocol.Contracts/Models/ProtocolBuildResponse.cs

@@ -3,7 +3,7 @@
     /// <summary>
     /// 报文构建返回结果
     /// </summary>
-    public class BuildResponse
+    public class ProtocolBuildResponse
     {
         /// <summary>
         /// 是否成功

protocols/JiShe.CollectBus.Protocol.Contracts/Models/SubProtocolBuildRequest.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Protocol.Contracts.Models
+{
+    /// <summary>
+    /// 子协议构建参数
+    /// </summary>
+    public class SubProtocolBuildRequest
+    {
+        /// <summary>
+        /// 表地址
+        /// </summary>
+        public required string MeterAddress { get; set; }
+
+        /// <summary>
+        /// 密码
+        /// </summary>
+        public required string Password { get; set; }
+
+        /// <summary>
+        /// 操作码
+        /// </summary>
+        public required string ItemCode { get; set; }
+    }
+}

protocols/JiShe.CollectBus.Protocol/SendData/Telemetry645PacketBuilder.cs

@@ -77,7 +77,7 @@ namespace JiShe.CollectBus.Protocol.SendData
         /// </summary>
         /// <param name="request"></param>
         /// <returns></returns>
-        public static Telemetry645PacketResponse C1C_02_Send(Telemetry645PacketRequest request)
+        public static Telemetry645PacketResponse C1C_03_Send(Telemetry645PacketRequest request)
         {
             var itemCodeArr = request.ItemCode.Split('_');
             var c_data = itemCodeArr[0];

protocols/JiShe.CollectBus.Protocol/StandardProtocolPlugin.cs

@@ -1,6 +1,4 @@
-using Azure.Core;
+using JiShe.CollectBus.Common.BuildSendDatas;
-using DeviceDetectorNET.Parser.Device;
-using JiShe.CollectBus.Common.BuildSendDatas;
 using JiShe.CollectBus.Common.Consts;
 using JiShe.CollectBus.Common.Enums;
 using JiShe.CollectBus.Common.Extensions;
@@ -12,14 +10,12 @@ using JiShe.CollectBus.IotSystems.MessageReceiveds;
 using JiShe.CollectBus.IotSystems.Protocols;
 using JiShe.CollectBus.Kafka.Producer;
 using JiShe.CollectBus.Protocol.Contracts.Abstracts;
-using JiShe.CollectBus.Protocol.Contracts.Interfaces;
 using JiShe.CollectBus.Protocol.Contracts.Models;
 using JiShe.CollectBus.Protocol.Contracts.SendData;
 using JiShe.CollectBus.Protocol.SendData;
 using Mapster;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
-using Newtonsoft.Json.Linq;
 using TouchSocket.Sockets;
 using Volo.Abp.Domain.Repositories;
 
@@ -276,12 +272,14 @@ namespace JiShe.CollectBus.Protocol
         /// <summary>
         /// 组装报文
         /// </summary>
-        /// <typeparam name="T"></typeparam>
+        /// <param name="request">报文构建参数</param>
-        /// <param name="entity">设备数据实体</param>
-        /// <param name="afnFnCode">映射读取执行方法的Code，例如10_1，表示 10H_F1_00000,10H_F1_00001，统一英文下划线分隔</param>
         /// <returns></returns>
-        public override async Task<BuildResponse> BuildAsync(BuildRequest request)
+        public override async Task<ProtocolBuildResponse> BuildAsync(ProtocolBuildRequest request)
         {
+            if (request == null)
+            {
+                throw new Exception($"{nameof(StandardProtocolPlugin)} 报文构建失败，参数为空");
+            }
             var itemCodeArr = request.ItemCode.Split('_');
             var aFNStr = itemCodeArr[0];
             var aFN = (AFN)aFNStr.HexToDec();
@@ -289,21 +287,20 @@ namespace JiShe.CollectBus.Protocol
             Telemetry3761PacketResponse builderResponse = null;
 
             List<string> dataUnit = new List<string>();
-            //10H_F1_1CH
+            //数据转发场景 10H_F1_1CH
-            if (aFNStr == "10" && string.IsNullOrWhiteSpace(request.SubItemCode) == false)
+            if (aFNStr == "10" && request.SubProtocolRequest != null && string.IsNullOrWhiteSpace(request.SubProtocolRequest.ItemCode) == false)
             {
-
+                var t645PacketHandlerName = $"C{request.SubProtocolRequest.ItemCode}_Send";
-                var subItem = $"C{request.SubItemCode}_Send";
                 Telemetry645PacketResponse t645PacketResponse = null;
 
-                if (T645ControlHandlers != null && T645ControlHandlers.TryGetValue(subItem
+                if (T645ControlHandlers != null && T645ControlHandlers.TryGetValue(t645PacketHandlerName
-                  , out var cchandler))
+                  , out var t645PacketHandler))
                 {
-                    t645PacketResponse = cchandler(new Telemetry645PacketRequest()
+                    t645PacketResponse = t645PacketHandler(new Telemetry645PacketRequest()
                     {
-                        MeterAddress = "",
+                        MeterAddress = request.SubProtocolRequest.MeterAddress,
-                        Password = "",
+                        Password = request.SubProtocolRequest.Password,
-                        ItemCode = "",
+                        ItemCode = request.SubProtocolRequest.ItemCode,
                     });
                 }
 
@@ -328,10 +325,10 @@ namespace JiShe.CollectBus.Protocol
 
             if (builderResponse == null)
             {
-                return new BuildResponse();
+                return new ProtocolBuildResponse();
             }
 
-            var result = builderResponse.Adapt<BuildResponse>();
+            var result = builderResponse.Adapt<ProtocolBuildResponse>();
             result.IsSuccess = true;
 
             return await Task.FromResult(result);

services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs

@@ -20,6 +20,7 @@ using JiShe.CollectBus.IotSystems.Watermeter;
 using JiShe.CollectBus.Kafka.Internal;
 using JiShe.CollectBus.Kafka.Producer;
 using JiShe.CollectBus.Protocol.Contracts.Interfaces;
+using JiShe.CollectBus.Protocol.Contracts.Models;
 using JiShe.CollectBus.Protocol.Contracts.SendData;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -619,7 +620,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 //var fn = int.Parse(itemCodeArr[1]);
 
                 //TODO:特殊表 
-                BuildResponse builderResponse = await protocolPlugin.BuildAsync(new BuildRequest()
+                ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest()
                 {
                     FocusAddress = ammeterInfo.FocusAddress,
                     Pn = ammeterInfo.MeteringCode,
@@ -898,7 +899,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 tempCodes = new List<string>() { "10_1" };
             }
 
-            //根据电表型号获取协议插件
+            //根据表型号获取协议插件
             var protocolPlugin = await _protocolService.GetProtocolServiceAsync(watermeter.Code);
             if (protocolPlugin == null)
             {
@@ -944,11 +945,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 //    continue;
                 //}
 
-                BuildResponse builderResponse = await protocolPlugin.BuildAsync(new BuildRequest()
+                ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest()
                 {
                     FocusAddress = watermeter.FocusAddress,
                     Pn = watermeter.MeteringCode,
                     ItemCode = tempItem,
+                    SubProtocolRequest = new SubProtocolBuildRequest()
+                    {
+                        MeterAddress = watermeter.MeterAddress,
+                        Password = watermeter.Password,
+                        ItemCode = tempItem,
+                    }
                 });
                 if (builderResponse == null || builderResponse.Data.Length <= 0)
                 {

services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs

@@ -16,6 +16,7 @@ using JiShe.CollectBus.IotSystems.Watermeter;
 using JiShe.CollectBus.Kafka.Internal;
 using JiShe.CollectBus.Kafka.Producer;
 using JiShe.CollectBus.Protocol.Contracts.Interfaces;
+using JiShe.CollectBus.Protocol.Contracts.Models;
 using JiShe.CollectBus.Protocol.Contracts.SendData;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.Extensions.DependencyInjection;
@@ -226,16 +227,22 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                     //获取对应的缓存电表信息
                     var ammeterInfo = ammeterInfos.First();
                     bool tripStateResult = false;
+                    string subItemCode = string.Empty;
                     if (settingInfo.TripType.Equals("on"))
                     {
                         ammeterInfo.TripState = 0;
                         tripStateResult = true;
+                        subItemCode = T645PacketItemCodeConst.C1C01C;
+                        if (ammeterInfo.TypeName != 1)
+                        {
+                            subItemCode = T645PacketItemCodeConst.C1C01B;
+                        }
                     }
                     else if (settingInfo.TripType.Equals("off"))
                     {
                         ammeterInfo.TripState = 1;
                         tripStateResult = false;
-
+                        subItemCode = T645PacketItemCodeConst.C1C01A;
                     }
                     else
                     {
@@ -253,11 +260,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                         return;
                     }
 
-                     BuildResponse builderResponse = await protocolPlugin.BuildAsync(new BuildRequest()
+                    ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest()
                     {
                         FocusAddress = ammeterInfo.FocusAddress,
                         Pn = ammeterInfo.MeteringCode,
                         ItemCode = temCode,
+                        SubProtocolRequest = new SubProtocolBuildRequest()
+                        {
+                            MeterAddress = ammeterInfo.AmmerterAddress,
+                            Password = ammeterInfo.Password,
+                            ItemCode = subItemCode,
+                        }
                     });
                      
                     string taskMark = CommonHelper.GetTaskMark(builderResponse.AFn, builderResponse.Fn, ammeterInfo.MeteringCode, builderResponse.MSA,builderResponse.Seq);

services/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs

@@ -82,6 +82,7 @@ namespace JiShe.CollectBus.Subscribers
         [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName)]
         public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo receivedMessage)
         {
+            //todo 如果是时段自动阀控，需要检查当前的时间，如果时间在自动阀控时间段内，则发送自动阀控报文，否则不发送，尤其是消息队列阻塞或者延时过长的时候。以免造成生产事故。
             _logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
             return await SendMessagesAsync(receivedMessage);
         }

shared/JiShe.CollectBus.Common/Consts/T645PacketItemCodeConst.cs

@@ -0,0 +1,59 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Consts
+{
+    /// <summary>
+    /// T645报文项编码
+    /// </summary>
+    public class T645PacketItemCodeConst
+    {
+        #region 跳合闸、报警、保电
+        /// <summary>
+        /// 跳闸
+        /// </summary>
+        public const string C1C01A = "1C_1A";
+
+        /// <summary>
+        /// 单相合闸
+        /// </summary>
+        public const string C1C01B = "1C_1B";
+
+        /// <summary>
+        /// 三相合闸
+        /// </summary>
+        public const string C1C01C = "1C_1C";
+
+        /// <summary>
+        /// 触发报警
+        /// </summary>
+        public const string C1C02A = "1C_2A";
+
+        /// <summary>
+        /// 报警解除
+        /// </summary>
+        public const string C1C02B = "1C_2B";
+
+        /// <summary>
+        /// 保电开始
+        /// </summary>
+        public const string C1C03A = "1C_3A";
+
+        /// <summary>
+        /// 保电结束
+        /// </summary>
+        public const string C1C03B = "1C_3B";
+        #endregion
+
+        #region 广播校时
+
+        /// <summary>
+        /// 广播校时
+        /// </summary>
+        public const string C08 = "08";
+        #endregion
+    }
+}