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

修复IoTDB驱动decimal类型解析赋值异常的问题

Browse Source
This commit is contained in:
ChenYi 2025-05-15 23:59:10 +08:00
parent 52207a92fa
commit 6d95430b63
11 changed files with 190 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/JiShe.CollectBus.IoTDB/Context/IoTDBRuntimeContext.cs
View File

@ -7,7 +7,7 @@ namespace JiShe.CollectBus.IoTDB.Context
/// <summary>
/// IoTDB SessionPool 运行时上下文
/// </summary>
public class IoTDBRuntimeContext: IScopedDependency
public class IoTDBRuntimeContext: IScopedDependency//ITransientDependency
{
private readonly bool _defaultValue;

2
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBProvider.cs
View File

@ -16,6 +16,8 @@ namespace JiShe.CollectBus.IoTDB.Interface
///// <param name="useTableSession">是否使用表模型</param>
//void SwitchSessionPool(bool useTableSession);
IIoTDbProvider GetSessionPool(bool sessionpolType);
/// <summary>
/// 插入数据
/// </summary>

25
modules/JiShe.CollectBus.IoTDB/Provider/IoTDBProvider.cs
View File

@ -42,6 +42,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
private IIoTDbSessionPool CurrentSession =>
_sessionFactory.GetSessionPool(_runtimeContext.UseTableSessionPool);
//private IIoTDbSessionPool CurrentSession { get; set; }
public IIoTDbProvider GetSessionPool(bool sessionpolType)
{
//CurrentSession = _sessionFactory.GetSessionPool(sessionpolType);
return this;
}
/// <summary>
/// IoTDbProvider
/// </summary>
@ -96,6 +106,11 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
try
{
if (entities == null || entities.Count() <= 0)
{
_logger.LogError($"{nameof(BatchInsertAsync)} 参数异常，-101");
return;
}
var metadata = await GetMetadata<T>();
var batchSize = 1000;
@ -321,9 +336,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
return null;
}
//var accessor = SourceEntityAccessorFactory.GetAccessor<T>();
//var memberCache = BuildMemberCache(accessor);
if (metadata.EntityType == null)
{
@ -795,7 +807,8 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
return declaredTypeName switch
{
"DATETIME" => value => value != null ? ((DateTime)value).GetDateTimeOffset().ToUnixTimeNanoseconds() : null,
"DATETIME" => value => value != null ? Convert.ToDateTime(value).GetDateTimeOffset().ToUnixTimeNanoseconds() : null,
"DECIMAL" => value => value != null ? Convert.ToDouble( value) : null,
_ => value => value
};
}
@ -897,7 +910,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
["DATETIME"] = TSDataType.TIMESTAMP,
["DATE"] = TSDataType.DATE,
["BLOB"] = TSDataType.BLOB,
["DECIMAL"] = TSDataType.STRING,
["DECIMAL"] = TSDataType.DOUBLE,
["STRING"] = TSDataType.STRING
};
@ -933,7 +946,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
TSDataType.BOOLEAN => Convert.ToBoolean(value),
TSDataType.INT32 => Convert.ToInt32(value),
TSDataType.INT64 => Convert.ToInt64(value),
TSDataType.FLOAT => Convert.ToDouble(value),
TSDataType.FLOAT => Convert.ToSingle(value),
TSDataType.DOUBLE => Convert.ToDouble(value),
TSDataType.TEXT => Convert.ToString(value),
TSDataType.NONE => null,

23
protocols/JiShe.CollectBus.Protocol.T6452007/SendData/Telemetry6452007PacketBuilder.cs
View File

@ -99,5 +99,28 @@ namespace JiShe.CollectBus.Protocol.T6452007.SendData
return new Telemetry6452007PacketResponse() { Data = dataList };
}
#endregion
#region 读数据
/// <summary>
/// 变量数据标识编码处理
/// </summary>
/// <param name="request"></param>
/// <returns></returns>
public static Telemetry6452007PacketResponse C11_02_Send(Telemetry6452007PacketRequest request)
{
var itemCodeArr = request.ItemCode.Split('_');//11_02_80_00_02
var c_data = itemCodeArr[0];
var DI3 = itemCodeArr[1];
var DI2 = itemCodeArr[2];
var DI1 = itemCodeArr[3];
var DI0 = itemCodeArr[4];
var dataUnit = new List<string>() { DI3, DI2, DI1, DI0 };
var dataList = Build645SendData.Build645SendCommand(request.MeterAddress, c_data, dataUnit);
return new Telemetry6452007PacketResponse() { Data = dataList };
}
#endregion
}
}

10
services/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -55,8 +55,8 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
_dbContext = dbContext;
_logger = logger;
_redisDataCacheService = redisDataCacheService;
_producerService =producerService;
_tcpService=tcpService;
_producerService = producerService;
_tcpService = tcpService;
}
/// <summary>
@ -91,6 +91,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
//TableModelSingleMeasuringEntityExtension
//TableModelSingleMeasuringEntityAccessor.GetSystemName(meter);
//ElectricityMeterAccessor
await _iotDBProvider.GetSessionPool(true).InsertAsync(meter);
await _iotDBProvider.InsertAsync(meter);
}
@ -211,14 +212,15 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
time = DateTime.Now;
//System.Reflection.PropertyInfo;
//System.Reflection.FieldInfo
var meter = new TreeModelSingleMeasuringEntity<DateTime>()
//TreeModelSingleMeasuringEntityAccessor
var meter = new TreeModelSingleMeasuringEntity<decimal?>()
{
SystemName = "energy",
DeviceId = "402440506",
DeviceType = "1",
ProjectId = "10059",
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
SingleMeasuring = (measuring, time)
SingleMeasuring = (measuring, 34.534m)
};
await _iotDBProvider.InsertAsync(meter);
}

64
services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -337,34 +337,34 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_ = _dataChannelManage.ScheduledMeterTaskReadingAsync(DataChannelManage.TaskDataChannel.Reader);
// //此处代码不要删除
//#if DEBUG
// var redisCacheDeviceInfoHashKeyTemp = $"CollectBus:Energy:JiSheCollectBus2:DeviceInfo";
// //此处代码不要删除
//#if DEBUG
// var redisCacheDeviceInfoHashKeyTemp = $"CollectBus:Energy:JiSheCollectBus2:DeviceInfo";
// var timer1 = Stopwatch.StartNew();
// Dictionary<string, List<DeviceInfo>> keyValuePairsTemps = FreeRedisProvider.Instance.HGetAll<List<DeviceInfo>>(redisCacheDeviceInfoHashKeyTemp);
// List<DeviceInfo> meterInfos = new List<DeviceInfo>();
// List<string> focusAddressDataLista = new List<string>();
// foreach (var item in keyValuePairsTemps)
// {
// foreach (var subItem in item.Value)
// {
// if (subItem.MeterType == MeterTypeEnum.Ammeter && subItem.TimeDensity == 15)
// {
// meterInfos.Add(subItem);
// focusAddressDataLista.Add(subItem.MeterId.ToString());
// }
// }
// }
// var timer1 = Stopwatch.StartNew();
// Dictionary<string, List<DeviceInfo>> keyValuePairsTemps = FreeRedisProvider.Instance.HGetAll<List<DeviceInfo>>(redisCacheDeviceInfoHashKeyTemp);
// List<DeviceInfo> meterInfos = new List<DeviceInfo>();
// List<string> focusAddressDataLista = new List<string>();
// foreach (var item in keyValuePairsTemps)
// {
// foreach (var subItem in item.Value)
// {
// if (subItem.MeterType == MeterTypeEnum.Ammeter && subItem.TimeDensity == 15)
// {
// meterInfos.Add(subItem);
// focusAddressDataLista.Add(subItem.MeterId.ToString());
// }
// }
// }
// timer1.Stop();
// _logger.LogError($"电表初始化读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒");
// DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
// return;
//#else
// var meterInfos = await GetAmmeterInfoList(gatherCode);
//#endif
// timer1.Stop();
// _logger.LogError($"电表初始化读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒");
// DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
// return;
//#else
// var meterInfos = await GetAmmeterInfoList(gatherCode);
//#endif
var meterInfos = await GetAmmeterInfoList(gatherCode);
if (meterInfos == null || meterInfos.Count <= 0)
{
@ -728,17 +728,27 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//var aFN = (AFN)aFNStr.HexToDec();
//var fn = int.Parse(itemCodeArr[1]);
var itemCodeInfo = T37612012PacketItemCodeConst.MappingItemCodeTo645SubCodeRelationship(tempItem);
//TODO:特殊表
ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest()
{
FocusAddress = ammeterInfo.FocusAddress,
Pn = ammeterInfo.MeteringCode,
ItemCode = tempItem,
ItemCode = itemCodeInfo.Item1,
DataTimeMark = new Protocol.DataTimeMark()
{
Density = ammeterInfo.TimeDensity.GetFocusDensity(),//转换成协议的值
Point = 1,
DataTime = timestamps,
},
SubProtocolRequest = new SubProtocolBuildRequest()
{
MeterAddress = ammeterInfo.MeterAddress,
Password = ammeterInfo.Password,
MeteringPort = ammeterInfo.MeteringPort,
Baudrate = ammeterInfo.Baudrate,
ItemCode = itemCodeInfo.Item2, //10_97 => 11_02_80_00_02
}
});
if (builderResponse == null || builderResponse.Data.Length <= 0)
@ -1106,7 +1116,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
List<string> deviceIds = new List<string>();//用于处理Kafka主题分区数据的分发和处理。
//根据采集频率分组，获得采集频率分组
var meterInfoGroupByTimeDensity = meterInfos.Select(d=>d.TimeDensity).GroupBy(d => d);
var meterInfoGroupByTimeDensity = meterInfos.Select(d => d.TimeDensity).GroupBy(d => d);
var currentTime = DateTime.Now;
if (_applicationOptions.FirstCollectionTime.HasValue == false)
{

88
services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -222,7 +222,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
#if DEBUG
sql = $@"{sql} and c.Address in('542410000504','442405000040','442405000039','402410040506')";
// sql = $@"{sql} and c.Address in('542410000504','442405000040','442405000039','402410040506')";
sql = $@"{sql} and c.Address in('402410040506')";
#endif
if (!string.IsNullOrWhiteSpace(gatherCode))
@ -294,35 +295,49 @@ namespace JiShe.CollectBus.ScheduledMeterReading
if (settingInfos == null || settingInfos.Count <= 0)
{
settingInfos = new List<AmmeterAutoValveControlSetting>();
settingInfos.Add(new AmmeterAutoValveControlSetting()
{
MeterType = MeterTypeEnum.Ammeter,
AmmerterAddress = "442405000040",
FocusAddress = "442400040",
FocusId = 57683,
ProjectID = 1,
TripType = "on",
TripTime = $"{DateTime.Now:HH:mm}",
MeterId = 78971,
LoopType = "EachDay",
EachDayWithout = "周六,周日",
TimeDensity = 15,
});
settingInfos.Add(new AmmeterAutoValveControlSetting()
{
MeterType = MeterTypeEnum.Ammeter,
AmmerterAddress = "442405000039",
FocusAddress = "442400039",
FocusId = 57684,
ProjectID = 1,
TripType = "on",
TripTime = $"{DateTime.Now:HH:mm}",
MeterId = 78972,
LoopType = "EachDay",
EachDayWithout = "周六,周日",
TimeDensity = 15,
});
//settingInfos.Add(new AmmeterAutoValveControlSetting()
//{
// MeterType = MeterTypeEnum.Ammeter,
// AmmerterAddress = "442405000040",
// FocusAddress = "442400040",
// FocusId = 57683,
// ProjectID = 1,
// TripType = "on",
// TripTime = $"{DateTime.Now:HH:mm}",
// MeterId = 78971,
// LoopType = "EachDay",
// EachDayWithout = "周六,周日",
// TimeDensity = 15,
//});
//settingInfos.Add(new AmmeterAutoValveControlSetting()
//{
// MeterType = MeterTypeEnum.Ammeter,
// AmmerterAddress = "442405000039",
// FocusAddress = "442400039",
// FocusId = 57684,
// ProjectID = 1,
// TripType = "on",
// TripTime = $"{DateTime.Now:HH:mm}",
// MeterId = 78972,
// LoopType = "EachDay",
// EachDayWithout = "周六,周日",
// TimeDensity = 15,
//});
//settingInfos.Add(new AmmeterAutoValveControlSetting()
//{
// MeterType = MeterTypeEnum.Ammeter,
// AmmerterAddress = "542410000504",
// FocusAddress = "542400504",
// FocusId = 57686,
// ProjectID = 1,
// TripType = "on",
// TripTime = $"{DateTime.Now:HH:mm}",
// MeterId = 78974,
// LoopType = "EachDay",
// EachDayWithout = "周六,周日",
// TimeDensity = 15,
//});
settingInfos.Add(new AmmeterAutoValveControlSetting()
{
@ -339,21 +354,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
TimeDensity = 15,
});
settingInfos.Add(new AmmeterAutoValveControlSetting()
{
MeterType = MeterTypeEnum.Ammeter,
AmmerterAddress = "542410000504",
FocusAddress = "542400504",
FocusId = 57686,
ProjectID = 1,
TripType = "on",
TripTime = $"{DateTime.Now:HH:mm}",
MeterId = 78974,
LoopType = "EachDay",
EachDayWithout = "周六,周日",
TimeDensity = 15,
});
FreeRedisProvider.Instance.Set(redisCacheDeviceSettingInfoHashKey, settingInfos);
}

20
shared/JiShe.CollectBus.Common/Consts/T37612012PacketItemCodeConst.cs
View File

@ -19,6 +19,12 @@ namespace JiShe.CollectBus.Common.Consts
/// </summary>
public const string AFN10HFN01H = $"10_01";
/// <summary>
/// 电网频率
/// </summary>
public const string AFN10HFN97H = $"10_97";
/// <summary>
/// 读取终端信息
/// </summary>
@ -181,5 +187,19 @@ namespace JiShe.CollectBus.Common.Consts
}
#endregion
/// <summary>
/// 特殊645编码关系映射
/// </summary>
/// <param name="itemCode"></param>
/// <returns></returns>
public static (string,string) MappingItemCodeTo645SubCodeRelationship(string itemCode)
{
return itemCode switch
{
AFN10HFN97H => (AFN10HFN01H,T6452007PacketItemCodeConst.C1102800002),
_=> (itemCode,""),
};
}
}
}

10
shared/JiShe.CollectBus.Common/Consts/T6452007PacketItemCodeConst.cs
View File

@ -56,6 +56,16 @@ namespace JiShe.CollectBus.Common.Consts
/// </summary>
public const string C08 = "08";
#endregion
#region 读数据
/// <summary>
/// 电网频率
/// </summary>
public const string C1102800002 = "11_02_80_00_02";
#endregion
#endregion
}
}

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

@ -141,7 +141,7 @@
}
},
"ServerApplicationOptions": {
"ServerTagName": "JiSheCollectBus99",
"ServerTagName": "JiSheCollectBus8",
"SystemType": "Energy",
"FirstCollectionTime": "2025-04-28 15:07:00",
"AutomaticVerificationTime": "16:07:00",