admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

优化协议插件定义，调整插件结构

Browse Source
This commit is contained in:
zenghongyao 2025-04-21 23:47:11 +08:00
parent f5f9434a97
commit 83efc6da00
13 changed files with 1105 additions and 276 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

264
protocols/JiShe.CollectBus.Protocol.Contracts/Abstracts/BaseProtocolPlugin.cs → protocols/JiShe.CollectBus.Protocol.Contracts/Abstracts/BaseProtocolPlugin_bak.cs
View File

@ -15,10 +15,10 @@ using JiShe.CollectBus.Common.Consts;
namespace JiShe.CollectBus.Protocol.Contracts.Abstracts namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
{ {
public abstract class BaseProtocolPlugin : IProtocolPlugin public abstract class BaseProtocolPlugin_bak //: IProtocolPlugin
{ {
private readonly IProducerService _producerService; private readonly IProducerService _producerService;
private readonly ILogger<BaseProtocolPlugin> _logger; private readonly ILogger<BaseProtocolPlugin_bak> _logger;
private readonly IRepository<ProtocolInfo, Guid> _protocolInfoRepository; private readonly IRepository<ProtocolInfo, Guid> _protocolInfoRepository;
//头部字节长度 //头部字节长度
@ -29,13 +29,13 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
public const string errorData = "EE"; public const string errorData = "EE";
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="BaseProtocolPlugin"/> class. /// Initializes a new instance of the <see cref="BaseProtocolPlugin_bak"/> class.
/// </summary> /// </summary>
/// <param name="serviceProvider">The service provider.</param> /// <param name="serviceProvider">The service provider.</param>
protected BaseProtocolPlugin(IServiceProvider serviceProvider) protected BaseProtocolPlugin_bak(IServiceProvider serviceProvider)
{ {
_logger = serviceProvider.GetRequiredService<ILogger<BaseProtocolPlugin>>(); _logger = serviceProvider.GetRequiredService<ILogger<BaseProtocolPlugin_bak>>();
_protocolInfoRepository = serviceProvider.GetRequiredService<IRepository<ProtocolInfo, Guid>>(); _protocolInfoRepository = serviceProvider.GetRequiredService<IRepository<ProtocolInfo, Guid>>();
_producerService = serviceProvider.GetRequiredService<IProducerService>(); _producerService = serviceProvider.GetRequiredService<IProducerService>();
} }
@ -603,149 +603,149 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
var time = Appendix.Appendix_A1(hexDatas.Take(6).ToList()); var time = Appendix.Appendix_A1(hexDatas.Take(6).ToList());
} }
/// <summary> ///// <summary>
/// 通用解析 ///// 通用解析
/// </summary> ///// </summary>
/// <param name="messageReceived"></param> ///// <param name="messageReceived"></param>
/// <param name="sendAction"></param> ///// <param name="sendAction"></param>
/// <returns></returns> ///// <returns></returns>
public virtual TB3761 AnalyzeReadingDataAsync(MessageReceived messageReceived, //public virtual TB3761 AnalyzeReadingDataAsync(MessageReceived messageReceived,
Action<byte[]>? sendAction = null) // Action<byte[]>? sendAction = null)
{ //{
var hexStringList = messageReceived.MessageHexString.StringToPairs(); // var hexStringList = messageReceived.MessageHexString.StringToPairs();
var afn = (AFN)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN); // var afn = (AFN)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
var fn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN); // var fn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
var tb3761 = QGDW3761Config.CommandList.FirstOrDefault(it => it.Afn == afn); // var tb3761 = QGDW3761Config.CommandList.FirstOrDefault(it => it.Afn == afn);
if (tb3761 == null) return null; // if (tb3761 == null) return null;
var tb3761Fn = tb3761.FnList.FirstOrDefault(it => it.Fn == fn); // var tb3761Fn = tb3761.FnList.FirstOrDefault(it => it.Fn == fn);
if (tb3761Fn == null) return null; // if (tb3761Fn == null) return null;
var analyzeValue = (List<string>)hexStringList.GetAnalyzeValue(CommandChunkEnum.Data); // var analyzeValue = (List<string>)hexStringList.GetAnalyzeValue(CommandChunkEnum.Data);
var m = 0; // var m = 0;
var rateNumberUpSort = -1; // var rateNumberUpSort = -1;
var rateNumberUp = tb3761Fn.UpList.FirstOrDefault(it => it.Name.Contains("费率数M")); // var rateNumberUp = tb3761Fn.UpList.FirstOrDefault(it => it.Name.Contains("费率数M"));
if (rateNumberUp != null) // if (rateNumberUp != null)
{ // {
var rateNumber = analyzeValue.Skip(rateNumberUp.DataIndex).Take(rateNumberUp.DataCount).FirstOrDefault(); // var rateNumber = analyzeValue.Skip(rateNumberUp.DataIndex).Take(rateNumberUp.DataCount).FirstOrDefault();
m = Convert.ToInt32(rateNumber); // m = Convert.ToInt32(rateNumber);
rateNumberUpSort = rateNumberUp.Sort; // rateNumberUpSort = rateNumberUp.Sort;
} // }
foreach (var up in tb3761Fn.UpList) // foreach (var up in tb3761Fn.UpList)
{ // {
var dataIndex = up.DataIndex; // var dataIndex = up.DataIndex;
if (dataIndex == 0 && up.Sort > rateNumberUpSort) // if (dataIndex == 0 && up.Sort > rateNumberUpSort)
{ // {
var sum1 = tb3761Fn.UpList.Where(it => it.Sort < up.Sort) // var sum1 = tb3761Fn.UpList.Where(it => it.Sort < up.Sort)
.Sum(it => it.DataCount); // .Sum(it => it.DataCount);
var sum2 = tb3761Fn.UpList.Where(it => it.Sort < up.Sort && it.Tb3761UpChildlList.Count > 0) // var sum2 = tb3761Fn.UpList.Where(it => it.Sort < up.Sort && it.Tb3761UpChildlList.Count > 0)
.Sum(it => it.Tb3761UpChildlList.Sum(c=> m * c.DataCount)); // .Sum(it => it.Tb3761UpChildlList.Sum(c=> m * c.DataCount));
dataIndex = sum1 + sum2; // dataIndex = sum1 + sum2;
} // }
var value = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, up.DataCount, up.DataType); // var value = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, up.DataCount, up.DataType);
if (value != null) // if (value != null)
{ // {
up.Value = value.ToString(); // up.Value = value.ToString();
} // }
if (up.Tb3761UpChildlList.Count > 0) //复费率根据费率数来解析 // if (up.Tb3761UpChildlList.Count > 0) //复费率根据费率数来解析
{ // {
var repeatCount = m; // var repeatCount = m;
foreach (var upChild in up.Tb3761UpChildlList) // foreach (var upChild in up.Tb3761UpChildlList)
{ // {
for (var j = 0; j < repeatCount; j++) // for (var j = 0; j < repeatCount; j++)
{ // {
var val = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, upChild.DataCount, upChild.DataType); // var val = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, upChild.DataCount, upChild.DataType);
if (val != null) // if (val != null)
{ // {
upChild.Name = string.Format(upChild.Name, j + 1); // upChild.Name = string.Format(upChild.Name, j + 1);
upChild.Value = val.ToString(); // upChild.Value = val.ToString();
} // }
dataIndex += upChild.DataCount; // dataIndex += upChild.DataCount;
} // }
} // }
} // }
} // }
return tb3761; // return tb3761;
} //}
/// <summary> ///// <summary>
/// 通用解析 日冻结曲线类 ///// 通用解析 日冻结曲线类
/// </summary> ///// </summary>
/// <param name="messageReceived"></param> ///// <param name="messageReceived"></param>
/// <param name="sendAction"></param> ///// <param name="sendAction"></param>
/// <returns></returns> ///// <returns></returns>
public virtual TB3761 AnalyzeReadingTdcDataAsync(MessageReceived messageReceived, //public virtual TB3761 AnalyzeReadingTdcDataAsync(MessageReceived messageReceived,
Action<byte[]>? sendAction = null) // Action<byte[]>? sendAction = null)
{ //{
var hexStringList = messageReceived.MessageHexString.StringToPairs(); // var hexStringList = messageReceived.MessageHexString.StringToPairs();
var afn = (AFN)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN); // var afn = (AFN)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
var fn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN); // var fn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
var tb3761 = QGDW3761Config.CommandTdcList.FirstOrDefault(it => it.Afn == afn); // var tb3761 = QGDW3761Config.CommandTdcList.FirstOrDefault(it => it.Afn == afn);
if (tb3761 == null) return null; // if (tb3761 == null) return null;
var tb3761Fn = tb3761.FnList.FirstOrDefault(it => it.Fn == fn); // var tb3761Fn = tb3761.FnList.FirstOrDefault(it => it.Fn == fn);
if (tb3761Fn == null) return null; // if (tb3761Fn == null) return null;
var analyzeValue = (List<string>)hexStringList.GetAnalyzeValue(CommandChunkEnum.Data); // var analyzeValue = (List<string>)hexStringList.GetAnalyzeValue(CommandChunkEnum.Data);
foreach (var up in tb3761Fn.UpList) // foreach (var up in tb3761Fn.UpList)
{ // {
var value = AnalyzeDataAccordingDataType(analyzeValue, up.DataIndex, up.DataCount, up.DataType); // var value = AnalyzeDataAccordingDataType(analyzeValue, up.DataIndex, up.DataCount, up.DataType);
if (value != null) // if (value != null)
{ // {
up.Value = value.ToString(); // up.Value = value.ToString();
if (up.Tb3761UpChildlList.Count > 0) // if (up.Tb3761UpChildlList.Count > 0)
{ // {
var dataIndex = up.DataIndex; // var dataIndex = up.DataIndex;
var repeatCount = (int)value; // var repeatCount = (int)value;
foreach (var upChild in up.Tb3761UpChildlList) // foreach (var upChild in up.Tb3761UpChildlList)
{ // {
for (var j = 0; j < repeatCount; j++) // for (var j = 0; j < repeatCount; j++)
{ // {
var val = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, upChild.DataCount, upChild.DataType); // var val = AnalyzeDataAccordingDataType(analyzeValue, dataIndex, upChild.DataCount, upChild.DataType);
if (val != null) // if (val != null)
{ // {
upChild.Value = val.ToString(); // upChild.Value = val.ToString();
upChild.Name = string.Format(upChild.Name, j + 1); // upChild.Name = string.Format(upChild.Name, j + 1);
} // }
dataIndex += upChild.DataCount; // dataIndex += upChild.DataCount;
} // }
} // }
} // }
} // }
} // }
return tb3761; // return tb3761;
//var freezeDensity = (FreezeDensity)Convert.ToInt32(hexDatas.Skip(5).Take(1)); // //var freezeDensity = (FreezeDensity)Convert.ToInt32(hexDatas.Skip(5).Take(1));
//var addMinute = 0; // //var addMinute = 0;
//switch (freezeDensity) // //switch (freezeDensity)
//{ // //{
// case FreezeDensity.No:break; // // case FreezeDensity.No:break;
// case FreezeDensity.Min15: // // case FreezeDensity.Min15:
// addMinute = 15; // // addMinute = 15;
// break; // // break;
// case FreezeDensity.Min30: // // case FreezeDensity.Min30:
// addMinute = 30; // // addMinute = 30;
// break; // // break;
// case FreezeDensity.Min60: // // case FreezeDensity.Min60:
// addMinute = 60; // // addMinute = 60;
// break; // // break;
// case FreezeDensity.Min5: break; // // case FreezeDensity.Min5: break;
// addMinute = 5; // // addMinute = 5;
// case FreezeDensity.Min1: // // case FreezeDensity.Min1:
// addMinute = 1; // // addMinute = 1;
// break; // // break;
// } // // }
} //}
private object? AnalyzeDataAccordingDataType(List<string> analyzeValue, int dataIndex,int dataCount,string dataType) private object? AnalyzeDataAccordingDataType(List<string> analyzeValue, int dataIndex,int dataCount,string dataType)
{ {

368
protocols/JiShe.CollectBus.Protocol.Contracts/Abstracts/ProtocolPlugin.cs Normal file
View File

@ -0,0 +1,368 @@
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.IotSystems.Protocols;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Protocol.Contracts.Models;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using TouchSocket.Sockets;
using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
{
public abstract class ProtocolPlugin:IProtocolPlugin
{
//头部字节长度
public const int hearderLen = 6;
public const int tPLen = 6;
public const string errorData = "EE";
private readonly ILogger _logger;
private readonly IRepository<ProtocolInfo, Guid> _protocolInfoRepository;
public ProtocolPlugin(IServiceProvider serviceProvider, ILogger logger)
{
_logger = logger;
_protocolInfoRepository = serviceProvider.GetRequiredService<IRepository<ProtocolInfo, Guid>>();
}
public abstract ProtocolInfo Info { get; }
public virtual async Task<ProtocolInfo> GetAsync() => await Task.FromResult(Info);
public virtual async Task AddAsync()
{
if (Info == null)
{
throw new ArgumentNullException(nameof(Info));
}
await _protocolInfoRepository.DeleteDirectAsync(a => a.Name == Info.Name);
await _protocolInfoRepository.InsertAsync(Info);
//await _protocolInfoCache.Get()
}
public abstract Task<T> AnalyzeAsync<T>(ITcpSessionClient client, string messageReceived, Action<T>? receivedAction = null) where T :class;
/// <summary>
/// 解析376.1帧
/// </summary>
/// <param name="messageReceived"></param>
/// <returns></returns>
public virtual TB3761? Analysis3761(string messageReceived)
{
try
{
var hexStringList = messageReceived.StringToPairs();
// 初步校验
if (hexStringList.Count < 6 || hexStringList.FirstOrDefault() != "68" || hexStringList.Skip(5).Take(1).FirstOrDefault() != "68" || hexStringList.Count < 18 || hexStringList.LastOrDefault() != "16")
{
_logger.LogError($"解析Analysis3761校验不通过,报文：{messageReceived}");
}
else
{
TB3761 tB3761 = new TB3761
{
C = Analysis_C(hexStringList),
A = Analysis_A(hexStringList),
AFN_FC = Analysis_AFN_FC(hexStringList),
SEQ = Analysis_SEQ(hexStringList),
UnitData = Analysis_UnitData(hexStringList),
DA = Analysis_DA(hexStringList),
DT = Analysis_DT(hexStringList)
};
return tB3761;
}
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis3761错误,报文：{messageReceived},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 控制域C解析
/// </summary>
/// <returns></returns>
public virtual C? Analysis_C(List<string> hexStringList)
{
try
{
if (hexStringList.Count > 6)
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(6, 1) // 控制域 1字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
string binStr = baseHexMessage.HexMessageString.HexToBin();
C c = new C
{
BaseHexMessage = baseHexMessage,
FC = binStr.Substring(binStr.Length - 4, 4).BinToDec(),
FCV = binStr.Substring(3, 1).BinToDec(),
FCB = binStr.Substring(2, 1).BinToDec(),
PRM = binStr.Substring(1, 1).BinToDec(),
DIR = binStr.Substring(0, 1).BinToDec()
};
return c;
}
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_C错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 地址域A解析
/// </summary>
/// <param name="hexStringList"></param>
/// <returns></returns>
public virtual A? Analysis_A(List<string> hexStringList)
{
try
{
if (hexStringList.Count > 7)
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(7, 5) // 地址域 5个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
A a = new A
{
BaseHexMessage = baseHexMessage,
A1 = baseHexMessage.HexMessageList.ListReverseToStr(0, 2),//.DataConvert(10);//行政区划码A1
A2 = baseHexMessage.HexMessageList.ListReverseToStr(2, 2).PadLeft(5, '0').HexToDec(),//终端地址A2
A3 = Analysis_A3(baseHexMessage.HexMessageList) //主站地址和组地址标志A3
};
a.Code = $"{a.A1.PadLeft(4, '0')}{a.A2.ToString().PadLeft(5, '0')}";
return a;
}
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_A错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 站地址和组地址标志A3
/// </summary>
/// <param name="hexAList">地址域A集合</param>
/// <returns></returns>
public virtual A3? Analysis_A3(List<string> hexAList)
{
try
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexAList.GetRange(4, 1) // 站地址和组地址标志A3 1个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
var binStr = baseHexMessage.HexMessageString.HexToBin();
A3 a3 = new A3
{
BaseHexMessage = baseHexMessage,
D0 = binStr.Substring(binStr.Length - 1, 1).BinToDec(),
D1_D7 = binStr.Substring(0, binStr.Length - 1).BinToDec()
};
return a3;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_A3错误,报文：{string.Join("", hexAList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// AFN_FC功能码
/// </summary>
/// <returns></returns>
public virtual AFN_FC? Analysis_AFN_FC(List<string> hexStringList)
{
try
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(12, 1) //AFN功能码 1个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
AFN_FC aFN_FC = new AFN_FC
{
BaseHexMessage = baseHexMessage,
AFN = baseHexMessage.HexMessageString.HexToDec(),
};
return aFN_FC;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_AFN_FC错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 解析帧序列域SEQ
/// </summary>
/// <returns></returns>
public virtual SEQ? Analysis_SEQ(List<string> hexStringList)
{
try
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(13, 1) //帧序列域 SEQ 1个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
var binStr = baseHexMessage.HexMessageString.HexToBin();
SEQ seq = new SEQ
{
PSEQ = binStr.Substring(binStr.Length - 4, 4).BinToDec(),
CON = binStr.Substring(3, 1).BinToDec(),
FIN = binStr.Substring(2, 1).BinToDec(),
FIR = binStr.Substring(1, 1).BinToDec(),
TpV = binStr.Substring(0, 1).BinToDec()
};
return seq;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_SEQ错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 数据单元标识及数据单元数据
/// </summary>
public virtual UnitData? Analysis_UnitData(List<string> hexStringList)
{
try
{
UnitData unitData = new UnitData
{
HexMessageList = hexStringList.GetRange(14, hexStringList.Count - 14 - 2) //总数字节数-固定长度报文头-控制域C-地址域A-校验和CS-结束字符（16H）
};
unitData.HexMessageString = string.Join("", unitData.HexMessageList);
if (unitData.HexMessageList.Count == 0)
return null;
return unitData;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_UnitData错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 信息点DA Pn
/// </summary>
/// <returns></returns>
public virtual DA? Analysis_DA(List<string> hexStringList)
{
try
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(14, 2) //信息点DA Pn 2个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
var da1 = baseHexMessage.HexMessageList[0];
var da2 = baseHexMessage.HexMessageList[1];
DA da = new DA()
{
BaseHexMessage = baseHexMessage,
Pn = CalculatePn(da1, da2)
};
return da;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_DA错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 信息类DT Fn
/// </summary>
/// <returns></returns>
public virtual DT? Analysis_DT(List<string> hexStringList)
{
try
{
BaseHexMessage baseHexMessage = new BaseHexMessage
{
HexMessageList = hexStringList.GetRange(16, 2) //信息类DT Fn 2个字节
};
baseHexMessage.HexMessageString = string.Join("", baseHexMessage.HexMessageList);
if (baseHexMessage.HexMessageList.Count == 0)
return null;
var dt1 = baseHexMessage.HexMessageList[0];
var dt2 = baseHexMessage.HexMessageList[1];
DT dt = new DT()
{
BaseHexMessage = baseHexMessage,
Fn = CalculateFn(dt1, dt2)
};
return dt;
}
catch (Exception ex)
{
_logger.LogError($"解析Analysis_DT错误,报文：{string.Join("", hexStringList)},异常：{ex.Message}");
}
return null;
}
/// <summary>
/// 计算Pn
/// </summary>
/// <param name="da1"></param>
/// <param name="da2"></param>
/// <returns></returns>
public int CalculatePn(string da1, string da2) => (da2.HexToDec() - 1) * 8 + (8 - da1.HexToBin().IndexOf(da1.Equals("00") ? "0" : "1"));
/// <summary>
/// 计算Fn
/// </summary>
/// <param name="dt1"></param>
/// <param name="dt2"></param>
/// <returns></returns>
public int CalculateFn(string dt1, string dt2) => dt2.HexToDec() * 8 + (8 - dt1.HexToBin().IndexOf("1"));
}
}

8
protocols/JiShe.CollectBus.Protocol.Contracts/Interfaces/IProtocolPlugin.cs
View File

@ -14,10 +14,12 @@ namespace JiShe.CollectBus.Protocol.Contracts.Interfaces
Task AddAsync(); Task AddAsync();
Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) where T : TB3761; Task<T> AnalyzeAsync<T>(ITcpSessionClient client, string messageReceived, Action<T>? sendAction = null) where T : class;
Task LoginAsync(MessageReceivedLogin messageReceived); TB3761? Analysis3761(string messageReceived);
Task HeartbeatAsync(MessageReceivedHeartbeat messageReceived); //Task LoginAsync(MessageReceivedLogin messageReceived);
//Task HeartbeatAsync(MessageReceivedHeartbeat messageReceived);
} }
} }

252
protocols/JiShe.CollectBus.Protocol.Contracts/Models/TB3761.cs
View File

@ -1,5 +1,4 @@
using JiShe.CollectBus.Common.Enums; using System;
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Text; using System.Text;
@ -7,49 +6,238 @@ using System.Threading.Tasks;
namespace JiShe.CollectBus.Protocol.Contracts.Models namespace JiShe.CollectBus.Protocol.Contracts.Models
{ {
/// <summary>
/// 解析3761报文
/// </summary>
public class TB3761 public class TB3761
{ {
public int Id { get; set; } /// <summary>
/// 控制域C
/// </summary>
public C? C { get; set; }
public AFN Afn { get; set; } /// <summary>
/// 地址域A
/// </summary>
public A? A { get; set; }
public List<TB3761FN> FnList { get; set; } /// <summary>
/// 帧序列域 SEQ
/// </summary>
public SEQ? SEQ { get; set; }
/// <summary>
/// 用户数据区
/// 功能码
/// </summary>
public AFN_FC? AFN_FC { get; set; }
/// <summary>
/// 用户数据区
/// 信息点DA Pn
/// </summary>
public DA? DA { get; set; }
/// <summary>
/// 用户数据区
/// 信息类DT Fn
/// </summary>
public DT? DT { get; set; }
/// <summary>
/// 数据单元标识和数据单元格式
/// </summary>
public UnitData? UnitData { get; set; }
}
#region
/// <summary>
/// 报文信息
/// </summary>
public class BaseHexMessage
{
/// <summary>
/// 报文
/// </summary>
public string? HexMessageString { get; set; }
/// <summary>
/// 报文数组
/// </summary>
public List<string>? HexMessageList { get; set; }
}
/// <summary>
/// 控制域C
/// </summary>
public class C
{
/// <summary>
/// 控制域C报文
/// </summary>
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 传输方向位D7 DIR=0，表示此帧报文是由主站发出的下行报文；DIR=1，表示此帧报文是由终端发出的上行报文。
/// </summary>
public int DIR { get; set; }
/// <summary>
/// D6启动标志位 0:表示此帧报文来自从动站(终端),1:表示此帧报文来自启动站(服务端)
/// </summary>
public int PRM { get; set; }
/// <summary>
/// D5下行：帧计数位(FCB)/上行(ACD)：要求访问位(终端有重要事件等待访问),
/// </summary>
public int FCB { get; set; }
/// <summary>
/// 下行：帧计数有效位(决定FCB位有效/无效)/上行：保留 D4
/// </summary>
public int FCV { get; set; }
/// <summary>
/// 功能码 D0-D3
/// </summary>
public int FC { get; set; }
} }
public class TB3761FN
{
public int Id { get; set; }
/// <summary>
/// 地址域A
/// </summary>
public class A
{
/// <summary>
/// 地址域报文
/// </summary>
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 集中器/终端编码
/// </summary>
public string? Code { get; set; }
/// <summary>
/// 行政区划码A1
/// </summary>
public string? A1 { get; set; }
/// <summary>
/// 终端地址A2
/// </summary>
public int A2 { get; set; }
/// <summary>
/// 站地址和组地址标志A3
/// </summary>
public A3? A3 { get; set; }
}
/// <summary>
/// 站地址和组地址标志A3
/// </summary>
public class A3
{
/// <summary>
/// 地址域A3报文
/// </summary>
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 终端组地址标志，D0=0即False 表示终端地址A2 为单地址
/// </summary>
public int D0 { get; set; }
/// <summary>
/// 主站地址 MSA D1～D7 组成 0～127
/// </summary>
public int D1_D7 { get; set; }
}
/// <summary>
/// 帧序列域 SEQ
/// </summary>
public class SEQ
{
/// <summary>
/// 帧序列域报文
/// </summary>
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 响应帧序号
/// </summary>
public int PSEQ { get; set; }
/// <summary>
/// CON为“1”，表示需要对该帧报文进行确认；置“0”，表示不需要对该帧报文进行确认。
/// </summary>
public int CON { get; set; }
/// <summary>
/// 末帧标志
/// </summary>
public int FIN { get; set; }
/// <summary>
/// 首帧标志
/// </summary>
public int FIR { get; set; }
/// <summary>
/// 帧时间标签有效位，TpV=0，表示在附加信息域中无时间标签Tp；TpV=1，表示在附加信息域中带有时间标签Tp
/// </summary>
public int TpV { get; set; }
}
/// <summary>
/// 用户数据区
/// 功能码
/// </summary>
public class AFN_FC
{
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 功能码
/// </summary>
public int AFN { get; set; }
}
/// <summary>
/// 用户数据区
/// 信息点DA Pn
/// </summary>
public class DA
{
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 信息点 DA Pn
/// </summary>
public int Pn { get; set; }
}
/// <summary>
/// 用户数据区
/// 信息类DT Fn
/// </summary>
public class DT
{
public BaseHexMessage? BaseHexMessage { get; set; }
/// <summary>
/// 信息类 DT Fn
/// </summary>
public int Fn { get; set; } public int Fn { get; set; }
public string Text { get; set; }
public List<TB3761UP> UpList { get; set; }
} }
public class TB3761UP
/// <summary>
/// 数据单元标识和数据单元格式
/// </summary>
public class UnitData: BaseHexMessage
{ {
public int Id { get; set; }
public string Name { get; set; }
public string? Value { get; set; }
public string DataType { get; set; }
public int DataIndex { get; set; }
//public int DataIndex { get; set; }
public int DataCount { get; set; }
//public int ParentId { get; set; }
public int Sort { get; set; }
public List<TB3761UP> Tb3761UpChildlList { get; set; }
} }
#endregion
} }

11
protocols/JiShe.CollectBus.Protocol.Test/TestProtocolPlugin.cs
View File

@ -4,22 +4,25 @@ using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.MessageReceiveds; using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.Protocols; using JiShe.CollectBus.IotSystems.Protocols;
using JiShe.CollectBus.Protocol.Contracts.Abstracts; using JiShe.CollectBus.Protocol.Contracts.Abstracts;
using Microsoft.Extensions.Logging;
using TouchSocket.Sockets;
namespace JiShe.CollectBus.Protocol.Test namespace JiShe.CollectBus.Protocol.Test
{ {
public class TestProtocolPlugin : BaseProtocolPlugin public class TestProtocolPlugin : ProtocolPlugin
{ {
private readonly ILogger<TestProtocolPlugin> _logger;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="TestProtocolPlugin"/> class. /// Initializes a new instance of the <see cref="TestProtocolPlugin"/> class.
/// </summary> /// </summary>
/// <param name="serviceProvider">The service provider.</param> /// <param name="serviceProvider">The service provider.</param>
public TestProtocolPlugin(IServiceProvider serviceProvider) : base(serviceProvider) public TestProtocolPlugin(IServiceProvider serviceProvider, ILogger<TestProtocolPlugin> logger) : base(serviceProvider, logger)
{ {
} }
public sealed override ProtocolInfo Info => new(nameof(TestProtocolPlugin), "Test", "TCP", "Test协议", "DTS1980-Test"); public sealed override ProtocolInfo Info => new(nameof(TestProtocolPlugin), "Test", "TCP", "Test协议", "DTS1980-Test");
public override async Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) public override Task<T> AnalyzeAsync<T>(ITcpSessionClient client, string messageReceived, Action<T>? receivedAction = null)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
@ -150,6 +153,8 @@ namespace JiShe.CollectBus.Protocol.Test
var a = $"{a1}{a2Dec.ToString().PadLeft(5, '0')}"; var a = $"{a1}{a2Dec.ToString().PadLeft(5, '0')}";
return a; return a;
} }
#endregion #endregion
} }
} }

5
protocols/JiShe.CollectBus.Protocol/JiSheCollectBusProtocolModule.cs
View File

@ -1,5 +1,8 @@
using JiShe.CollectBus.Protocol.Contracts.Interfaces; using JiShe.CollectBus.Protocol.Contracts.Abstracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using TouchSocket.Core;
using Volo.Abp; using Volo.Abp;
using Volo.Abp.Modularity; using Volo.Abp.Modularity;

241
protocols/JiShe.CollectBus.Protocol/StandardProtocolPlugin.cs
View File

@ -1,62 +1,245 @@
using JiShe.CollectBus.Common.Enums; using DeviceDetectorNET.Parser.Device;
using JiShe.CollectBus.Common.BuildSendDatas;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions; using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models; using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.Enums;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageReceiveds; using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.Protocols; using JiShe.CollectBus.IotSystems.Protocols;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Contracts.Abstracts; using JiShe.CollectBus.Protocol.Contracts.Abstracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Protocol.Contracts.Models; using JiShe.CollectBus.Protocol.Contracts.Models;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json.Linq; using Newtonsoft.Json.Linq;
using TouchSocket.Sockets;
using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.Protocol namespace JiShe.CollectBus.Protocol
{ {
public class StandardProtocolPlugin : BaseProtocolPlugin public class StandardProtocolPlugin : ProtocolPlugin
{ {
private readonly ILogger<StandardProtocolPlugin> _logger;
private readonly IProducerService _producerService;
private readonly IRepository<Device, Guid> _deviceRepository;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="StandardProtocolPlugin"/> class. /// Initializes a new instance of the <see cref="StandardProtocolPlugin"/> class.
/// </summary> /// </summary>
/// <param name="serviceProvider">The service provider.</param> /// <param name="serviceProvider">The service provider.</param>
public StandardProtocolPlugin(IServiceProvider serviceProvider) : base(serviceProvider) public StandardProtocolPlugin(IServiceProvider serviceProvider,ILogger<StandardProtocolPlugin> logger) : base(serviceProvider, logger)
{ {
_logger= logger;
//_logger = serviceProvider.GetRequiredService<ILogger<StandardProtocolPlugin>>();
_producerService = serviceProvider.GetRequiredService<IProducerService>();
_deviceRepository = serviceProvider.GetRequiredService<IRepository<Device, Guid>>();
} }
public sealed override ProtocolInfo Info => new(nameof(StandardProtocolPlugin), "376.1", "TCP", "376.1协议", "DTS1980"); public sealed override ProtocolInfo Info => new(nameof(StandardProtocolPlugin), "376.1", "TCP", "376.1协议", "DTS1980");
public override async Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) public override async Task<T> AnalyzeAsync<T>(ITcpSessionClient client, string messageReceived, Action<T>? sendAction = null)
{ {
var hexStringList = messageReceived.MessageHexString.StringToPairs(); TB3761? tB3761 = Analysis3761(messageReceived);
var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A); if (tB3761 != null)
var afn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
var fn = (int)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
T analyze = default;
switch ((AFN)afn)
{ {
case AFN.确认或否认: if (tB3761.AFN_FC?.AFN == (int)AFN.链路接口检测)
AnalyzeAnswerDataAsync(messageReceived, sendAction); {
break; if (tB3761.A == null || tB3761.A.Code.IsNullOrWhiteSpace() || tB3761.A.A3?.D1_D7 == null || tB3761.SEQ?.PSEQ == null)
case AFN.设置参数: break;
case AFN.查询参数: break;
case AFN.请求实时数据:
if (Enum.IsDefined(typeof(ATypeOfDataItems), fn))
{ {
analyze = (T?)AnalyzeReadingDataAsync(messageReceived, sendAction); _logger.LogError($"解析AFN.链路接口检测报文失败,报文：{messageReceived},TB3761:{tB3761.Serialize()}");
} }
break; else
case AFN.请求历史数据:
if (Enum.IsDefined(typeof(IIdataTypeItems), fn))
{ {
analyze = (T?)AnalyzeReadingTdcDataAsync(messageReceived, sendAction); if (tB3761.SEQ.CON == 1)
{
if (tB3761.DT?.Fn == 1)
{
// 登录回复
await LoginAsync(client, messageReceived, tB3761.A.Code, tB3761.A.A3?.D1_D7, tB3761.SEQ?.PSEQ);
}
else if (tB3761.DT?.Fn == 2)
{
// 心跳回复
await HeartbeatAsync(client, messageReceived, tB3761.A.Code, tB3761.A.A3?.D1_D7, tB3761.SEQ?.PSEQ);
}
}
} }
break;
case AFN.数据转发: }
AnalyzeTransparentForwardingAnswerAsync(messageReceived, sendAction);
break;
} }
return await Task.FromResult(analyze); return (tB3761 as T)!;
} }
/// <summary>
/// 登录回复
/// </summary>
/// <param name="client"></param>
/// <param name="code"></param>
/// <param name="msa"></param>
/// <param name="pseq"></param>
/// <returns></returns>
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()
};
//await _producerBus.PublishAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent);
await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent);
//await _producerBus.Publish( 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);
//await _producerBus.PublishAsync(ProtocolConst.SubscriberLoginIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, new IssuedEventMessage { ClientId = messageReceivedLoginEvent.ClientId, DeviceNo = messageReceivedLoginEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceivedLoginEvent.MessageId });
//await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
}
/// <summary>
/// 心跳帧解析
/// </summary>
/// <param name="client"></param>
/// <param name="code"></param>
/// <param name="msa"></param>
/// <param name="pseq"></param>
/// <returns></returns>
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()
};
//await _producerBus.PublishAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent);
await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent);
//await _producerBus.Publish(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);
//await _producerBus.PublishAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, new IssuedEventMessage { ClientId = messageReceivedHeartbeatEvent.ClientId, DeviceNo = messageReceivedHeartbeatEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceivedHeartbeatEvent.MessageId });
//await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
}
#region 上行命令 #region 上行命令
//68 //68

138
services/JiShe.CollectBus.Application/Plugins/TcpMonitor.cs
View File

@ -14,6 +14,10 @@ using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageReceiveds; using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.Kafka.Producer; using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Contracts; using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Contracts.Abstracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Protocol.Contracts.Models;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using TouchSocket.Core; using TouchSocket.Core;
using TouchSocket.Sockets; using TouchSocket.Sockets;
@ -21,6 +25,7 @@ using Volo.Abp.Caching;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
using Volo.Abp.Domain.Entities; using Volo.Abp.Domain.Entities;
using Volo.Abp.Domain.Repositories; using Volo.Abp.Domain.Repositories;
using static System.Formats.Asn1.AsnWriter;
using static FreeSql.Internal.GlobalFilter; using static FreeSql.Internal.GlobalFilter;
namespace JiShe.CollectBus.Plugins namespace JiShe.CollectBus.Plugins
@ -31,64 +36,83 @@ namespace JiShe.CollectBus.Plugins
private readonly ILogger<TcpMonitor> _logger; private readonly ILogger<TcpMonitor> _logger;
private readonly IRepository<Device, Guid> _deviceRepository; private readonly IRepository<Device, Guid> _deviceRepository;
private readonly IDistributedCache<AmmeterInfo> _ammeterInfoCache; private readonly IDistributedCache<AmmeterInfo> _ammeterInfoCache;
private readonly IServiceProvider _serviceProvider;
/// <summary> /// <summary>
/// ///
/// </summary> /// </summary>
/// <param name="producerService"></param> /// <param name="producerService"></param>
/// <param name="logger"></param> /// <param name="logger"></param>
/// <param name="deviceRepository"></param> /// <param name="deviceRepository"></param>
/// <param name="ammeterInfoCache"></param> /// <param name="ammeterInfoCache"></param>
/// <param name="serviceProvider"></param>
public TcpMonitor(IProducerService producerService, public TcpMonitor(IProducerService producerService,
ILogger<TcpMonitor> logger, ILogger<TcpMonitor> logger,
IRepository<Device, Guid> deviceRepository, IRepository<Device, Guid> deviceRepository,
IDistributedCache<AmmeterInfo> ammeterInfoCache) IDistributedCache<AmmeterInfo> ammeterInfoCache, IServiceProvider serviceProvider)
{ {
_producerService = producerService; _producerService = producerService;
_logger = logger; _logger = logger;
_deviceRepository = deviceRepository; _deviceRepository = deviceRepository;
_ammeterInfoCache = ammeterInfoCache; _ammeterInfoCache = ammeterInfoCache;
_serviceProvider= serviceProvider;
} }
public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e) public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e)
{ {
var messageHexString = Convert.ToHexString(e.ByteBlock.Span); var messageHexString = Convert.ToHexString(e.ByteBlock.Span);
var hexStringList = messageHexString.StringToPairs();
var aFn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
var fn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A);
if (aFn.HasValue && fn.HasValue && aTuple != null && !string.IsNullOrWhiteSpace(aTuple.Item1))
{
var tcpSessionClient = (ITcpSessionClient)client;
if ((AFN)aFn == AFN.链路接口检测) var tcpSessionClient = (ITcpSessionClient)client;
{ var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
switch (fn) if (protocolPlugin == null)
{
case 1:
await OnTcpLoginReceived(tcpSessionClient, messageHexString, aTuple.Item1);
break;
case 3:
//心跳帧有两种情况：
//1. 集中器先有登录帧，再有心跳帧
//2. 集中器没有登录帧，只有心跳帧
await OnTcpHeartbeatReceived(tcpSessionClient, messageHexString, aTuple.Item1);
break;
default:
_logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
break;
}
}
else
{
await OnTcpNormalReceived(tcpSessionClient, messageHexString, aTuple.Item1,aFn.ToString()!.PadLeft(2,'0'));
}
}
else
{ {
_logger.LogError($"指令初步解析失败，指令内容：{messageHexString}"); _logger.LogError("协议不存在！");
} }
protocolPlugin.Analysis3761(messageHexString);
TB3761? tB3761 = await protocolPlugin!.AnalyzeAsync<TB3761>(tcpSessionClient, messageHexString);
//var hexStringList = messageHexString.StringToPairs();
//var aFn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
//var fn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
//var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A);
//if (aFn.HasValue && fn.HasValue && aTuple != null && !string.IsNullOrWhiteSpace(aTuple.Item1))
//{
//var tcpSessionClient = (ITcpSessionClient)client;
//if ((AFN)aFn == AFN.链路接口检测)
//{
// switch (fn)
// {
// case 1:
// await OnTcpLoginReceived(tcpSessionClient, messageHexString, aTuple.Item1);
// break;
// case 3:
// //心跳帧有两种情况：
// //1. 集中器先有登录帧，再有心跳帧
// //2. 集中器没有登录帧，只有心跳帧
// await OnTcpHeartbeatReceived(tcpSessionClient, messageHexString, aTuple.Item1);
// break;
// default:
// _logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
// break;
// }
//}
//else
//{
//await OnTcpNormalReceived(tcpSessionClient, messageHexString); // , aTuple.Item1, aFn.ToString()!.PadLeft(2, '0')
//}
//}
//else
//{
// _logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
//}
await e.InvokeNext(); await e.InvokeNext();
} }
@ -236,8 +260,18 @@ namespace JiShe.CollectBus.Plugins
/// <param name="deviceNo"></param> /// <param name="deviceNo"></param>
/// <param name="aFn"></param> /// <param name="aFn"></param>
/// <returns></returns> /// <returns></returns>
private async Task OnTcpNormalReceived(ITcpSessionClient client, string messageHexString, string deviceNo,string aFn) private async Task OnTcpNormalReceived(ITcpSessionClient tcpSessionClient, string messageHexString) //, string deviceNo,string aFn
{ {
var _protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (_protocolPlugin == null)
{
_logger.LogError("协议不存在！");
}
TB3761? tB3761 = await _protocolPlugin!.AnalyzeAsync<TB3761>(tcpSessionClient, messageHexString);
//await _producerBus.Publish(new MessageReceived //await _producerBus.Publish(new MessageReceived
//{ //{
// ClientId = client.Id, // ClientId = client.Id,
@ -260,15 +294,19 @@ namespace JiShe.CollectBus.Plugins
// DeviceNo = deviceNo, // DeviceNo = deviceNo,
// MessageId = NewId.NextGuid().ToString() // MessageId = NewId.NextGuid().ToString()
//}); //});
await _producerService.ProduceAsync(ProtocolConst.SubscriberReceivedEventName, new MessageReceived
{ //TODO：根据AFN进行分流推送到kafka
ClientId = client.Id,
ClientIp = client.IP,
ClientPort = client.Port, //await _producerService.ProduceAsync(ProtocolConst.SubscriberReceivedEventName, new MessageReceived
MessageHexString = messageHexString, //{
DeviceNo = deviceNo, // ClientId = client.Id,
MessageId = Guid.NewGuid().ToString() // ClientIp = client.IP,
}); // ClientPort = client.Port,
// MessageHexString = messageHexString,
// DeviceNo = deviceNo,
// MessageId = Guid.NewGuid().ToString()
//});
} }
} }
} }

16
services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -151,16 +151,14 @@ namespace JiShe.CollectBus.Subscribers
} }
else else
{ {
//todo 会根据不同的协议进行解析，然后做业务处理 //todo 会根据不同的协议进行解析，然后做业务处理
TB3761 fN = await protocolPlugin.AnalyzeAsync<TB3761>(receivedMessage); TB3761 tB3761 = protocolPlugin.Analysis3761(receivedMessage.MessageHexString);
if(fN == null) if (tB3761 == null)
{ {
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}"); Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
return SubscribeAck.Success(); return SubscribeAck.Success();
} }
var tb3761FN = fN.FnList.FirstOrDefault(); if (tB3761.DT == null || tB3761.AFN_FC == null)
if (tb3761FN == null)
{ {
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}"); Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
return SubscribeAck.Success(); return SubscribeAck.Success();
@ -170,8 +168,8 @@ namespace JiShe.CollectBus.Subscribers
var entity = new MeterReadingRecords() var entity = new MeterReadingRecords()
{ {
ReceivedMessageHexString = receivedMessage.MessageHexString, ReceivedMessageHexString = receivedMessage.MessageHexString,
AFN = fN.Afn, AFN = (AFN)tB3761.AFN_FC.AFN,
Fn = tb3761FN.Fn, Fn = tB3761.DT.Fn,
Pn = 0, Pn = 0,
FocusAddress = "", FocusAddress = "",
MeterAddress = "", MeterAddress = "",
@ -208,7 +206,7 @@ namespace JiShe.CollectBus.Subscribers
} }
else else
{ {
await protocolPlugin.HeartbeatAsync(receivedHeartbeatMessage); //await protocolPlugin.HeartbeatAsync(receivedHeartbeatMessage);
await _messageReceivedHeartbeatEventRepository.InsertAsync(receivedHeartbeatMessage); await _messageReceivedHeartbeatEventRepository.InsertAsync(receivedHeartbeatMessage);
} }
} }
@ -228,7 +226,7 @@ namespace JiShe.CollectBus.Subscribers
} }
else else
{ {
await protocolPlugin.LoginAsync(receivedLoginMessage); //await protocolPlugin.LoginAsync(receivedLoginMessage);
await _messageReceivedLoginEventRepository.InsertAsync(receivedLoginMessage); await _messageReceivedLoginEventRepository.InsertAsync(receivedLoginMessage);
} }
} }

34
shared/JiShe.CollectBus.Common/BuildSendDatas/Build3761SendData.cs
View File

@ -375,22 +375,30 @@ namespace JiShe.CollectBus.Common.BuildSendDatas
#region 排序 #region 排序
var timeSets = timeSetDetails; var timeSets = timeSetDetails;
Dictionary<int, TimeSetDetail> dicTsDetails = new Dictionary<int, TimeSetDetail>(); // 旧
//Dictionary<int, TimeSetDetail> dicTsDetails = new Dictionary<int, TimeSetDetail>();
//foreach (var timeSet in timeSets)
//{
// int firstMonty = timeSet.Months[0];
// if (!dicTsDetails.Keys.Contains(firstMonty))
// dicTsDetails.Add(firstMonty, timeSet);
//}
//var sortKeys = dicTsDetails.Keys.OrderBy(n => n).ToList();
//List<TimeSetDetail> orderTsDetails = new List<TimeSetDetail>();
//foreach (var key in sortKeys)
//{
// orderTsDetails.Add(dicTsDetails[key]);
//}
//timeSetDetails = orderTsDetails;
// 新
foreach (var timeSet in timeSets) foreach (var timeSet in timeSets)
{ {
int firstMonty = timeSet.Months[0]; timeSet.Months = timeSet.Months.OrderBy(m => m).ToArray();
if (!dicTsDetails.Keys.Contains(firstMonty))
dicTsDetails.Add(firstMonty, timeSet);
} }
timeSetDetails = timeSets;
var sortKeys = dicTsDetails.Keys.OrderBy(n => n).ToList();
List<TimeSetDetail> orderTsDetails = new List<TimeSetDetail>();
foreach (var key in sortKeys)
{
orderTsDetails.Add(dicTsDetails[key]);
}
timeSetDetails = orderTsDetails;
#endregion #endregion

37
shared/JiShe.CollectBus.Common/Extensions/StringExtensions.cs
View File

@ -1128,6 +1128,21 @@ namespace JiShe.CollectBus.Common.Extensions
return string.Join(" ", strArr.Reverse()); return string.Join(" ", strArr.Reverse());
} }
/// <summary>
/// 高低位反转，并转换成字符串
/// </summary>
/// <param name="list">16进制字符集合</param>
/// <param name="index">16进制字符集合开始索引</param>
/// <param name="count">取多少个数据</param>
/// <returns></returns>
public static string ListReverseToStr(this List<string> list, int index, int count)
{
var addrList = list.GetRange(index, count);
addrList.Reverse();//高低位反转
return string.Join("", addrList);
}
/// <summary> /// <summary>
/// 二进制转十六进制 /// 二进制转十六进制
/// </summary> /// </summary>
@ -1163,6 +1178,18 @@ namespace JiShe.CollectBus.Common.Extensions
return decimalNumber; return decimalNumber;
} }
///// <summary>
///// 十六进制转二进制
///// </summary>
///// <param name="hexString"></param>
///// <returns></returns>
//public static string HexToBin(this string hexString)
//{
// var binaryValue = Convert.ToString(Convert.ToInt32(hexString, 16), 2);
// return binaryValue;
//}
/// <summary> /// <summary>
/// 十六进制转二进制 /// 十六进制转二进制
/// </summary> /// </summary>
@ -1170,8 +1197,14 @@ namespace JiShe.CollectBus.Common.Extensions
/// <returns></returns> /// <returns></returns>
public static string HexToBin(this string hexString) public static string HexToBin(this string hexString)
{ {
var binaryValue = Convert.ToString(Convert.ToInt32(hexString, 16), 2); string result = string.Empty;
return binaryValue; foreach (char c in hexString)
{
int v = Convert.ToInt32(c.ToString(), 16);
int v2 = int.Parse(Convert.ToString(v, 2));
result += string.Format("{0:d4}", v2);
}
return result;
} }
/// <summary> /// <summary>

5
web/JiShe.CollectBus.Host/JiShe.CollectBus.Host.csproj
View File

@ -62,7 +62,10 @@
<ItemGroup> <ItemGroup>
<None Update="Plugins\JiShe.CollectBus.Protocol.dll"> <None Update="Plugins\JiShe.CollectBus.Protocol.dll">
<CopyToOutputDirectory>Always</CopyToOutputDirectory> <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</None>
<None Update="Plugins\JiShe.CollectBus.Protocol.Test.dll">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</None> </None>
</ItemGroup> </ItemGroup>

2
web/JiShe.CollectBus.Host/appsettings.json
View File

@ -84,7 +84,7 @@
"SaslPassword": "lixiao1980", "SaslPassword": "lixiao1980",
"KafkaReplicationFactor": 3, "KafkaReplicationFactor": 3,
"NumPartitions": 30, "NumPartitions": 30,
"ServerTagName": "JiSheCollectBus2" "ServerTagName": "JiSheCollectBus99"
}, },
"IoTDBOptions": { "IoTDBOptions": {
"UserName": "root", "UserName": "root",