using JiShe.CollectBus.Common.BuildSendDatas; using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Common.Enums; using JiShe.CollectBus.Common.Extensions; using JiShe.CollectBus.Common.Helpers; using JiShe.CollectBus.Common.Models; using JiShe.CollectBus.Enums; using JiShe.CollectBus.IotSystems.Devices; 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.Models; using JiShe.CollectBus.Protocol.Contracts.SendData; using JiShe.CollectBus.Protocol.SendData; using Mapster; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using TouchSocket.Sockets; using Volo.Abp.Domain.Repositories; namespace JiShe.CollectBus.Protocol { public class T1882018ProtocolPlugin : T37612012ProtocolPlugin { private readonly ILogger _logger; private readonly IProducerService _producerService; private readonly IRepository _deviceRepository; private readonly ITcpService _tcpService; public readonly Dictionary T3761AFNHandlers; public readonly Dictionary T645ControlHandlers; /// /// Initializes a new instance of the class. /// /// The service provider. public T1882018ProtocolPlugin(IServiceProvider serviceProvider, ILogger logger, ITcpService tcpService) : base(serviceProvider, logger) { _logger = logger; //_logger = serviceProvider.GetRequiredService>(); _producerService = serviceProvider.GetRequiredService(); _deviceRepository = serviceProvider.GetRequiredService>(); _tcpService = tcpService; T3761AFNHandlers = Telemetry3761PacketBuilder.T3761AFNHandlers; T645ControlHandlers = Telemetry1882018PacketBuilder.T645ControlHandlers; } public sealed override ProtocolInfo Info => new(nameof(T1882018ProtocolPlugin), "376.1", "TCP", "376.1协议", "DTS1980"); public override async Task AnalyzeAsync(ITcpSessionClient client, string messageReceived, Action? sendAction = null) { TB3761? tB3761 = Analysis3761(messageReceived); if (tB3761 != null) { if (tB3761.AFN_FC?.AFN == (int)AFN.链路接口检测) { if (tB3761.A == null || tB3761.A.Code.IsNullOrWhiteSpace() || tB3761.A.A3?.D1_D7 == null || tB3761.SEQ?.PSEQ == null) { _logger.LogError($"解析AFN.链路接口检测报文失败,报文:{messageReceived},TB3761:{tB3761.Serialize()}"); } else { if (tB3761.DT?.Fn == (int)FN.登录) { // 登录回复 if (tB3761.SEQ.CON == (int)CON.需要对该帧进行确认) await LoginAsync(client, messageReceived, tB3761.A.Code, tB3761.A.A3?.D1_D7, tB3761.SEQ?.PSEQ); } else if (tB3761.DT?.Fn == (int)FN.心跳) { // 心跳回复 //心跳帧有两种情况: //1. 集中器先有登录帧,再有心跳帧 //2. 集中器没有登录帧,只有心跳帧 await HeartbeatAsync(client, messageReceived, tB3761.A.Code, tB3761.A.A3?.D1_D7, tB3761.SEQ?.PSEQ); } } } await OnTcpNormalReceived(client, tB3761); } return (tB3761 as T)!; } /// /// 正常帧处理,将不同的AFN进行分发 /// /// /// /// /// private async Task OnTcpNormalReceived(ITcpSessionClient tcpSessionClient, TB3761 tB3761) { //string topicName = string.Format(ProtocolConst.AFNTopicNameFormat, aFn); //todo 如何确定时标?目前集中器的采集频率,都是固定,数据上报的时候,根据当前时间,往后推测出应当采集的时间点作为时标。但是如果由于网络问题,数据一直没上报的情况改怎么计算? //await _producerBus.PublishAsync(ProtocolConst.SubscriberReceivedEventName, new MessageReceived //{ // ClientId = client.Id, // ClientIp = client.IP, // ClientPort = client.Port, // MessageHexString = messageHexString, // DeviceNo = deviceNo, // MessageId = NewId.NextGuid().ToString() //}); if (tB3761.AFN_FC.BaseHexMessage == null || tB3761.DT.BaseHexMessage == null || tB3761.BaseHexMessage.HexMessageString==null) { _logger.LogError("376.1协议解析AFN失败"); return; } // 登录心跳已做了处理,故需要忽略登录和心跳帧 if (tB3761.DT.Fn == (int)FN.登录 || tB3761.DT.Fn == (int)FN.心跳) return; //TODO:根据AFN进行分流推送到kafka string topicName = string.Format(ProtocolConst.AFNTopicNameFormat, tB3761.AFN_FC.AFN.ToString().PadLeft(2, '0')); MessageProtocolAnalysis messageReceivedAnalysis = new MessageProtocolAnalysis() { ClientId = tcpSessionClient.Id, ClientIp = tcpSessionClient.IP, ClientPort = tcpSessionClient.Port, MessageHexString = tB3761.BaseHexMessage.HexMessageString!, DeviceNo = tB3761.A.Code!, MessageId = Guid.NewGuid().ToString(), ReceivedTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), Data = tB3761 }; List topics = ProtocolConstExtensions.GetAllTopicNamesByReceived(); if (topics.Contains(topicName)) await _producerService.ProduceAsync(topicName, messageReceivedAnalysis); else { _logger.LogError($"不支持的上报kafka主题:{topicName}"); await _producerService.ProduceAsync(ProtocolConst.SubscriberReceivedEventName, messageReceivedAnalysis); } } /// /// 登录回复 /// /// /// /// /// /// public async Task LoginAsync(ITcpSessionClient client, string messageReceived, string code, int? msa, int? pseq) { string oldClientId = $"{client.Id}"; await client.ResetIdAsync(code); var deviceInfoList = await _deviceRepository.GetListAsync(a => a.Number == code); if (deviceInfoList != null && deviceInfoList.Count > 1) { //todo 推送集中器编号重复预警 _logger.LogError($"集中器编号:{code},存在多个集中器,请检查集中器编号是否重复"); return; } var entity = deviceInfoList?.FirstOrDefault(a => a.Number == code); if (entity == null) { await _deviceRepository.InsertAsync(new Device(code, oldClientId, DateTime.Now, DateTime.Now, DeviceStatus.Online)); } else { entity.UpdateByLoginAndHeartbeat(oldClientId); await _deviceRepository.UpdateAsync(entity); } var messageReceivedLoginEvent = new MessageReceivedLogin { ClientId = code, ClientIp = client.IP, ClientPort = client.Port, MessageHexString = messageReceived, DeviceNo = code, MessageId = Guid.NewGuid().ToString(), ReceivedTime=DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") }; await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent); var reqParam = new ReqParameter2 { AFN = AFN.确认或否认, FunCode = (int)CFromStationFunCode.链路数据, PRM = PRM.从动站报文, A = code, Seq = new Seq() { TpV = TpV.附加信息域中无时间标签, FIRFIN = FIRFIN.单帧, CON = CON.需要对该帧进行确认, PRSEQ = pseq!.Value }, MSA = msa!.Value, Pn = 0, Fn = 1 }; var bytes = Build3761SendData.BuildSendCommandBytes(reqParam); var issuedEventMessage = new IssuedEventMessage { ClientId = messageReceivedLoginEvent.ClientId, DeviceNo = messageReceivedLoginEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceivedLoginEvent.MessageId }; if (_tcpService.ClientExists(issuedEventMessage.ClientId)) { await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message); _logger.LogInformation($"集中器地址{issuedEventMessage.ClientId} 登录回复下发内容:{Convert.ToHexString(bytes)}"); await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, issuedEventMessage); } } /// /// 心跳帧解析 /// /// /// /// /// /// public async Task HeartbeatAsync(ITcpSessionClient client, string messageReceived, string code, int? msa, int? pseq) { string clientId = code; string oldClientId = $"{client.Id}"; var deviceInfoList = await _deviceRepository.GetListAsync(a => a.Number == code); if (deviceInfoList != null && deviceInfoList.Count > 1) { //todo 推送集中器编号重复预警 _logger.LogError($"集中器编号:{code},存在多个集中器,请检查集中器编号是否重复"); return; } var entity = deviceInfoList?.FirstOrDefault(a => a.Number == code); if (entity == null) //没有登录帧的设备,只有心跳帧 { await client.ResetIdAsync(clientId); await _deviceRepository.InsertAsync(new Device(code, oldClientId, DateTime.Now, DateTime.Now, DeviceStatus.Online)); } else { if (clientId != oldClientId) { entity.UpdateByLoginAndHeartbeat(oldClientId); } else { entity.UpdateByLoginAndHeartbeat(); } await _deviceRepository.UpdateAsync(entity); } var messageReceivedHeartbeatEvent = new MessageReceivedHeartbeat { ClientId = clientId, ClientIp = client.IP, ClientPort = client.Port, MessageHexString = messageReceived, DeviceNo = code, MessageId = Guid.NewGuid().ToString(), ReceivedTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") }; await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent); var reqParam = new ReqParameter2() { AFN = AFN.确认或否认, FunCode = (int)CFromStationFunCode.链路数据, PRM = PRM.从动站报文, A = code, Seq = new Seq() { TpV = TpV.附加信息域中无时间标签, FIRFIN = FIRFIN.单帧, CON = CON.不需要对该帧进行确认, PRSEQ = pseq!.Value, }, MSA = msa!.Value, Pn = 0, Fn = 1 }; var bytes = Build3761SendData.BuildSendCommandBytes(reqParam); IssuedEventMessage issuedEventMessage = new IssuedEventMessage { ClientId = messageReceivedHeartbeatEvent.ClientId, DeviceNo = messageReceivedHeartbeatEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceivedHeartbeatEvent.MessageId }; if (_tcpService.ClientExists(issuedEventMessage.ClientId)) { await _tcpService.SendAsync(issuedEventMessage.ClientId, bytes); _logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 心跳回复下发内容:{Convert.ToHexString(bytes)}"); await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, issuedEventMessage); } } /// /// 组装报文 /// /// 报文构建参数 /// public override async Task BuildAsync(ProtocolBuildRequest request) { if (request == null) { throw new Exception($"{nameof(T1882018ProtocolPlugin)} 报文构建失败,参数为空"); } var itemCodeArr = request.ItemCode.Split('_'); var aFNStr = itemCodeArr[0]; var aFN = (AFN)aFNStr.HexToDec(); var fn = int.Parse(itemCodeArr[1]); Telemetry3761PacketResponse builderResponse = null; List dataUnit = new List(); //数据转发场景 10H_F1_1CH if (aFNStr == "10" && request.SubProtocolRequest != null && string.IsNullOrWhiteSpace(request.SubProtocolRequest.ItemCode) == false) { var t645PacketHandlerName = $"C{request.SubProtocolRequest.ItemCode}_Send"; Telemetry1882018PacketResponse t645PacketResponse = null; if (T645ControlHandlers != null && T645ControlHandlers.TryGetValue(t645PacketHandlerName , out var t645PacketHandler)) { t645PacketResponse = t645PacketHandler(new Telemetry1882018PacketRequest() { MeterAddress = request.SubProtocolRequest.MeterAddress, Password = request.SubProtocolRequest.Password, ItemCode = request.SubProtocolRequest.ItemCode, }); } if (t645PacketResponse != null) { dataUnit = t645PacketResponse.Data; } } string afnMethonCode = $"AFN{aFNStr}_Fn_Send"; if (T3761AFNHandlers != null && T3761AFNHandlers.TryGetValue(afnMethonCode , out var handler)) { builderResponse = handler(new Telemetry3761PacketRequest() { FocusAddress = request.FocusAddress, Fn = fn, Pn = request.Pn, DataUnit = dataUnit, }); } if (builderResponse == null) { return new ProtocolBuildResponse(); } var result = builderResponse.Adapt(); result.IsSuccess = true; return await Task.FromResult(result); } #region 上行命令 //68 //32 00 //32 00 //68 //C9 1100'1001. 控制域C。 // D7=1, (终端发送)上行方向。 // D6=1, 此帧来自启动站。 // D5=0, (上行方向)要求访问位。表示终端无事件数据等待访问。 // D4=0, 保留 // D3~D0=9, 功能码。链路测试 //20 32 行政区划码 //90 26 终端地址 //00 主站地址和组地址标志。终端为单地址。 //3220 09 87 2 // 终端启动的发送帧的 MSA 应为 0, 其主站响应帧的 MSA 也应为 0. //02 应用层功能码。AFN=2, 链路接口检测 //70 0111'0000. 帧序列域。无时间标签、单帧、需要确认。 //00 00 信息点。DA1和DA2全为“0”时,表示终端信息点。 //01 00 信息类。F1, 登录。 //44 帧尾,包含用户区数据校验和 //16 帧结束标志 /// /// 解析上行命令 /// /// /// public CommandReulst? AnalysisCmd(string cmd) { CommandReulst? commandReulst = null; var hexStringList = cmd.StringToPairs(); if (hexStringList.Count < hearderLen) { return commandReulst; } //验证起始字符 if (!hexStringList[0].IsStartStr() || !hexStringList[5].IsStartStr()) { return commandReulst; } var lenHexStr = $"{hexStringList[2]}{hexStringList[1]}"; var lenBin = lenHexStr.HexToBin(); var len = lenBin.Remove(lenBin.Length - 2).BinToDec(); //验证长度 if (hexStringList.Count - 2 != hearderLen + len) return commandReulst; var userDataIndex = hearderLen; var c = hexStringList[userDataIndex];//控制域 1字节 userDataIndex += 1; var aHexList = hexStringList.Skip(userDataIndex).Take(5).ToList();//地址域 5字节 var a = AnalysisA(aHexList); var a3Bin = aHexList[4].HexToBin().PadLeft(8, '0'); var mSA = a3Bin.Substring(0, 7).BinToDec(); userDataIndex += 5; var aFN = (AFN)hexStringList[userDataIndex].HexToDec();//1字节 userDataIndex += 1; var seq = hexStringList[userDataIndex].HexToBin().PadLeft(8, '0'); var tpV = (TpV)Convert.ToInt32(seq.Substring(0, 1)); var fIRFIN = (FIRFIN)Convert.ToInt32(seq.Substring(1, 2)); var cON = (CON)Convert.ToInt32(seq.Substring(3, 1)); var prseqBin = seq.Substring(4, 4); userDataIndex += 1; // (DA2 - 1) * 8 + DA1 = pn var da1Bin = hexStringList[userDataIndex].HexToBin(); var da1 = da1Bin == "0" ? 0 : da1Bin.Length; userDataIndex += 1; var da2 = hexStringList[userDataIndex].HexToDec(); var pn = da2 == 0 ? 0 : (da2 - 1) * 8 + da1; userDataIndex += 1; //(DT2*8)+DT1=fn var dt1Bin = hexStringList[userDataIndex].HexToBin(); var dt1 = dt1Bin != "0" ? dt1Bin.Length : 0; userDataIndex += 1; var dt2 = hexStringList[userDataIndex].HexToDec(); var fn = dt2 * 8 + dt1; userDataIndex += 1; //数据单元 var datas = hexStringList.Skip(userDataIndex).Take(len + hearderLen - userDataIndex).ToList(); //EC //Tp commandReulst = new CommandReulst() { A = a, MSA = mSA, AFN = aFN, Seq = new Seq() { TpV = tpV, FIRFIN = fIRFIN, CON = cON, PRSEQ = prseqBin.BinToDec(), }, CmdLength = len, Pn = pn, Fn = fn, HexDatas = datas }; return commandReulst; } /// /// 解析地址 /// /// /// private string AnalysisA(List aHexList) { var a1 = aHexList[1] + aHexList[0]; var a2 = aHexList[3] + aHexList[2]; var a2Dec = a2.HexToDec(); var a3 = aHexList[4]; var a = $"{a1}{a2Dec.ToString().PadLeft(5, '0')}"; return a; } #endregion } }