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

Merge branch 'dev' into zhy_feat_cev_v8

Browse Source
This commit is contained in:
zenghongyao 2025-05-11 10:43:54 +08:00
commit acd0a5c155
12 changed files with 420 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/JiShe.CollectBus.Analyzers/ComplexTypeSourceAnalyzers.cs
View File

@ -573,7 +573,7 @@ namespace JiShe.CollectBus.IncrementalGenerator
$"new EntityMemberInfo(" +
$"\"{prop.Name}.{elementName}\", " +
$"typeof({elementType}), " +
$"\"{elementDeclaredName}\", " +
$"typeof({elementType}).Name, " +//$"\"{elementDeclaredName}\", " +
$"(e) => Get{prop.Name}_{elementName}(({entityType})e), " +
$"(e, v) => Set{prop.Name}_{elementName}(({entityType})e, ({elementType})v))");
}

22
modules/JiShe.CollectBus.IoTDB/Exceptions/IoTException.cs Normal file
View File

@ -0,0 +1,22 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDB.Exceptions
{
/// <summary>
/// IoTDB异常
/// </summary>
public class IoTException : Exception
{
public int ErrorCode { get; }
public IoTException(string message, int errorCode)
: base($"{message} (Code: {errorCode})")
{
ErrorCode = errorCode;
}
}
}

3
modules/JiShe.CollectBus.IoTDB/JiShe.CollectBus.IoTDB.csproj
View File

@ -14,7 +14,6 @@
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Analyzers.Shared\JiShe.CollectBus.Analyzers.Shared.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.Analyzers\JiShe.CollectBus.Analyzers.csproj"
OutputItemType="Analyzer" ReferenceOutputAssembly="false"/>
<ProjectReference Include="..\..\modules\JiShe.CollectBus.Analyzers\JiShe.CollectBus.Analyzers.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
</ItemGroup>
</Project>

23
modules/JiShe.CollectBus.IoTDB/Model/IoTEntity.cs
View File

@ -1,6 +1,7 @@
using JiShe.CollectBus.Common.Attributes;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.IoTDB.Attributes;
using Volo.Abp.Domain.Entities;
namespace JiShe.CollectBus.IoTDB.Model
{
@ -43,5 +44,27 @@ namespace JiShe.CollectBus.IoTDB.Model
/// 时标，也就是业务时间戳，单位毫秒,必须通过DateTimeOffset获取
/// </summary>
public long Timestamps { get; set; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
/// <summary>
/// 设备路径
/// </summary>
public virtual string DevicePath
{
get
{
return $"root.{SystemName.ToLower()}.`{ProjectId}`.`{DeviceType}`.{DataType}.`{DeviceId}`";
}
set
{
if (string.IsNullOrWhiteSpace(value))
{
DevicePath = $"root.{SystemName.ToLower()}.`{ProjectId}`.`{DeviceType}`.{DataType}.`{DeviceId}`";
}
else
{
DevicePath = value;
}
}
}
}
}

50
modules/JiShe.CollectBus.IoTDB/Provider/DeviceMetadata.cs
View File

@ -6,8 +6,18 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <summary>
/// 设备元数据
/// </summary>
public class DeviceMetadata
public sealed class DeviceMetadata
{
/// <summary>
/// 实体类名称
/// </summary>
public string EntityName { get; set; }
/// <summary>
/// 设备表名或树路径，如果实体没有添加TableNameOrTreePath,此处为空
/// </summary>
public string TableNameOrTreePath { get; set; }
/// <summary>
/// 实体类型枚举
/// </summary>
@ -31,6 +41,42 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <summary>
/// 值类型集合，用于构建Table的值类型，也就是dataTypes参数
/// </summary>
public List<TSDataType> DataTypes { get; } = new();
public List<TSDataType> DataTypes { get; set; } = new();
/// <summary>
/// 列处理信息集合
/// </summary>
public List<ColumnProcessor> Processors { get; } = new List<ColumnProcessor>();
}
/// <summary>
/// 列处理信息结构
/// </summary>
public struct ColumnProcessor
{
/// <summary>
/// 列名
/// </summary>
public string ColumnName;
/// <summary>
/// 值获取委托
/// </summary>
public Func<object, object> ValueGetter;
/// <summary>
/// 类型转换委托
/// </summary>
public Func<object, object> Converter;
/// <summary>
/// 是否单测点
/// </summary>
public bool IsSingleMeasuring;
/// <summary>
/// 单测点名称委托
/// </summary>
public Func<object, object> SingleMeasuringNameGetter;
}
}

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

@ -21,6 +21,11 @@ using Microsoft.Extensions.Logging;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Domain.Entities;
using JiShe.CollectBus.Analyzers.Shared;
using JiShe.CollectBus.IoTDB.Exceptions;
using System.Diagnostics.Metrics;
using Newtonsoft.Json.Linq;
using static System.Runtime.InteropServices.JavaScript.JSType;
using System.Text.RegularExpressions;
namespace JiShe.CollectBus.IoTDB.Provider
{
@ -67,8 +72,13 @@ namespace JiShe.CollectBus.IoTDB.Provider
var metadata = await GetMetadata<T>();
var tablet = BuildTablet(new[] { entity }, metadata);
if (tablet == null || tablet.Count <= 0)
{
_logger.LogError($"{nameof(InsertAsync)} IoTDB插入{typeof(T).Name}的数据时 tablet 为null");
return;
}
await CurrentSession.InsertAsync(tablet);
await CurrentSession.InsertAsync(tablet.First());
}
catch (Exception ex)
{
@ -94,7 +104,15 @@ namespace JiShe.CollectBus.IoTDB.Provider
foreach (var batch in batches)
{
var tablet = BuildTablet(batch, metadata);
await CurrentSession.InsertAsync(tablet);
if (tablet == null || tablet.Count <= 0)
{
_logger.LogError($"{nameof(InsertAsync)} IoTDB插入{typeof(T).Name}的数据时 tablet 为null");
return;
}
foreach (var item in tablet)
{
await CurrentSession.InsertAsync(item);
}
}
}
catch (Exception ex)
@ -122,7 +140,15 @@ namespace JiShe.CollectBus.IoTDB.Provider
foreach (var batch in batches)
{
var tablet = BuildTablet(batch, deviceMetadata);
await CurrentSession.InsertAsync(tablet);
if (tablet == null || tablet.Count <= 0)
{
_logger.LogError($"{nameof(InsertAsync)} IoTDB插入{typeof(T).Name}的数据时 tablet 为null");
return;
}
foreach (var item in tablet)
{
await CurrentSession.InsertAsync(item);
}
}
}
catch (Exception ex)
@ -180,22 +206,54 @@ namespace JiShe.CollectBus.IoTDB.Provider
var accessor = SourceEntityAccessorFactory.GetAccessor<T>();
var columns = CollectColumnMetadata<T>(accessor);
var metadata = BuildDeviceMetadata<T>(columns);
var tmpMetadata = BuildDeviceMetadata<T>(columns, accessor);
string tableNameOrTreePath = string.Empty;
var tableNameOrTreePathAttribute = typeof(T).GetCustomAttribute<TableNameOrTreePathAttribute>();
if (tableNameOrTreePathAttribute != null)
{
tableNameOrTreePath = tableNameOrTreePathAttribute.TableNameOrTreePath;
}
tmpMetadata.EntityName = accessor.EntityName;
tmpMetadata.EntityType = accessor.EntityType;
tmpMetadata.TableNameOrTreePath = tableNameOrTreePath;
var metaData = MetadataCache.AddOrUpdate(
typeof(T),
addValueFactory: t => metadata, // 如果键不存在，用此值添加
addValueFactory: t => tmpMetadata, // 如果键不存在，用此值添加
updateValueFactory: (t, existingValue) =>
{
var columns = CollectColumnMetadata(accessor);
var metadata = BuildDeviceMetadata<T>(columns);
var metadata = BuildDeviceMetadata(columns, accessor);
//对现有值 existingValue 进行修改，返回新值
string tableNameOrTreePath = string.Empty;
var tableNameOrTreePathAttribute = typeof(T).GetCustomAttribute<TableNameOrTreePathAttribute>();
if (tableNameOrTreePathAttribute != null)
{
tableNameOrTreePath = tableNameOrTreePathAttribute.TableNameOrTreePath;
}
existingValue.ColumnNames = metadata.ColumnNames;
existingValue.DataTypes = metadata.DataTypes;
return existingValue;
}
);
metadata.EntityType = accessor.EntityType;
//var metaData = MetadataCache.GetOrAdd(typeof(T), type =>
//{
// var columns = CollectColumnMetadata(accessor);
// var metadata = BuildDeviceMetadata(columns, accessor);
// string tableNameOrTreePath = string.Empty;
// var tableNameOrTreePathAttribute = typeof(T).GetCustomAttribute<TableNameOrTreePathAttribute>();
// if (tableNameOrTreePathAttribute != null)
// {
// tableNameOrTreePath = tableNameOrTreePathAttribute.TableNameOrTreePath;
// }
// metadata.EntityName = accessor.EntityName;
// metadata.EntityType = accessor.EntityType;
// metadata.TableNameOrTreePath = tableNameOrTreePath;
// return metadata;
//});
return await Task.FromResult(metaData);
}
@ -257,31 +315,26 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <param name="entities">表实体</param>
/// <param name="metadata">设备元数据</param></param>
/// <returns></returns>
private Tablet BuildTablet<T>(IEnumerable<T> entities, DeviceMetadata metadata) where T : IoTEntity
private List<Tablet> BuildTablet<T>(IEnumerable<T> entities, DeviceMetadata metadata) where T : IoTEntity
{
var timestamps = new List<long>();
var values = new List<List<object>>();
var devicePaths = new HashSet<string>();
List<string> tempColumnNames = new List<string>();
tempColumnNames.AddRange(metadata.ColumnNames);
var accessor = SourceEntityAccessorFactory.GetAccessor<T>();
var memberCache = new Dictionary<string, EntityMemberInfo>(); // 缓存优化查询
// 预构建成员缓存（Key: NameOrPath）
foreach (var member in accessor.MemberList)
var entitiyList = entities.ToList();
if (entitiyList == null || entitiyList.Count <= 0)
{
memberCache[member.NameOrPath] = member;
return null;
}
if (accessor.EntityType == null || metadata.EntityType == null)
//var accessor = SourceEntityAccessorFactory.GetAccessor<T>();
//var memberCache = BuildMemberCache(accessor);
if (metadata.EntityType == null)
{
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 {nameof(T)}的EntityType 没有指定，属于异常情况，-101");
}
if (metadata.EntityType != accessor.EntityType)
if (metadata.EntityType == EntityTypeEnum.Other)
{
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 {nameof(T)}的EntityType 和{nameof(DeviceMetadata)}的 EntityType 不一致，属于异常情况，-102");
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 {nameof(T)}的EntityType 不属于IoTDB数据模型实体，属于异常情况，-102");
}
if (metadata.EntityType == EntityTypeEnum.TreeModel && _runtimeContext.UseTableSessionPool == true)
@ -292,103 +345,90 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 table模型不能使用tree模型Session连接，属于异常情况，-104");
}
string tableNameOrTreePath = string.Empty;
var tableNameOrTreePathAttribute = typeof(T).GetCustomAttribute<TableNameOrTreePathAttribute>();
if (tableNameOrTreePathAttribute != null)
if (_runtimeContext.UseTableSessionPool)//表模型
{
tableNameOrTreePath = tableNameOrTreePathAttribute.TableNameOrTreePath;
}
foreach (var entity in entities)
{
timestamps.Add(entity.Timestamps);
var rowValues = new List<object>();
foreach (var measurement in tempColumnNames)
{
if (!memberCache.TryGetValue(measurement, out var member))
{
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构时{accessor.EntityName}没有找到{measurement}对应的member信息，-105");
}
var value = member.GetValue(entity);
// 特性查询优化
var attributes = member.CustomAttributes ?? Enumerable.Empty<Attribute>();
var singleMeasuringAttr = attributes.OfType<SingleMeasuringAttribute>().FirstOrDefault();
if (singleMeasuringAttr != null)//如果是单侧点
{
var tupleItemKey = $"{member.NameOrPath}.Item2";
if (!memberCache.TryGetValue(tupleItemKey, out var tupleMember))
{
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构时{accessor.EntityName} 没有找到{measurement}对应的member Item2 信息，-106");
}
value = tupleMember.GetValue(entity);
}
if (value != null)
{
var tempValue = member.DeclaredTypeName.ToUpper() switch
{
"DATETIME" => Convert.ToDateTime(value).GetDateTimeOffset().ToUnixTimeNanoseconds(),
_ => value
};
rowValues.Add(tempValue);
}
else
{
rowValues.Add(value);
}
}
values.Add(rowValues);
//如果指定了路径
if (!string.IsNullOrWhiteSpace(tableNameOrTreePath))
if (!string.IsNullOrWhiteSpace(metadata.TableNameOrTreePath))
{
devicePaths.Add(tableNameOrTreePath);
tableNameOrTreePath = metadata.TableNameOrTreePath;
}
else
{
if (!_runtimeContext.UseTableSessionPool)//树模型
tableNameOrTreePath = DevicePathBuilder.GetTableName<T>();
}
return new List<Tablet>() { BuildTablet(entitiyList, metadata, tableNameOrTreePath) };
}
else
{
//树模型的时候，实体的设备Id可能会不同，因此需要根据不同路径进行存储。
var tabletList = new List<Tablet>();
var groupEntities = entitiyList.GroupBy(d => d.DevicePath).ToList();
foreach (var group in groupEntities)
{
tabletList.Add(BuildTablet(group.ToList(), metadata, group.Key));
}
return tabletList;
}
}
private Tablet BuildTablet<T>(List<T> entities, DeviceMetadata metadata, string tableNameOrTreePath) where T : IoTEntity
{
// 预分配内存结构
var rowCount = entities.Count;
var timestamps = new long[rowCount];
var values = new object[rowCount][];
for (var i = 0; i < values.Length; i++)
{
values[i] = new object[metadata.ColumnNames.Count];
}
List<string> tempColumnNames = new List<string>();
tempColumnNames.AddRange(metadata.ColumnNames);
// 顺序处理数据（保证线程安全）
for (var row = 0; row < rowCount; row++)
{
var entity = entities[row];
timestamps[row] = entity.Timestamps;
for (int i = 0; i < metadata.ColumnNames.Count; i++)
{
var processor = metadata.Processors[i];
if (processor.IsSingleMeasuring)
{
devicePaths.Add(DevicePathBuilder.GetDevicePath(entity));
}
else
{
devicePaths.Add(DevicePathBuilder.GetTableName<T>());
tempColumnNames[i] = (string)processor.SingleMeasuringNameGetter(entity);
}
// 获取并转换值
values[row][i] = processor.ValueGetter(entity);
}
}
if (devicePaths.Count > 1)
{
throw new Exception($"{nameof(BuildTablet)} 构建Tablet《{typeof(T).Name}》时，批量插入的设备路径不一致。");
}
return _runtimeContext.UseTableSessionPool
? BuildTableSessionTablet(metadata, devicePaths.First(), values, timestamps)
: BuildSessionTablet(metadata, devicePaths.First(), values, timestamps);
? BuildTableSessionTablet(metadata, tableNameOrTreePath, tempColumnNames, values.Select(d => d.ToList()).ToList(), timestamps.ToList())
: BuildSessionTablet(metadata, tableNameOrTreePath, tempColumnNames, values.Select(d => d.ToList()).ToList(), timestamps.ToList());
}
/// <summary>
/// 构建tree模型的Tablet
/// </summary>
/// <param name="metadata">已解析的设备数据元数据</param>
/// <param name="devicePath">设备路径</param>
/// <param name="columns">数据列集合</param>
/// <param name="values">数据集合</param>
/// <param name="timestamps">时间戳集合</param>
/// <returns></returns>
private Tablet BuildSessionTablet(DeviceMetadata metadata, string devicePath, List<List<object>> values, List<long> timestamps)
private Tablet BuildSessionTablet(DeviceMetadata metadata, string devicePath, List<string> columns, List<List<object>> values, List<long> timestamps)
{
//todo 树模型需要去掉TAG类型和ATTRIBUTE类型的字段，只需要保留FIELD类型字段即可
return new Tablet(
devicePath,
metadata.ColumnNames,
columns,
metadata.DataTypes,
values,
timestamps
@ -400,14 +440,15 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// </summary>
/// <param name="metadata">已解析的设备数据元数据</param>
/// <param name="tableName">表名称</param>
/// <param name="columns">数据列集合</param>
/// <param name="values">数据集合</param>
/// <param name="timestamps">时间戳集合</param>
/// <returns></returns>
private Tablet BuildTableSessionTablet(DeviceMetadata metadata, string tableName, List<List<object>> values, List<long> timestamps)
private Tablet BuildTableSessionTablet(DeviceMetadata metadata, string tableName, List<string> columns, List<List<object>> values, List<long> timestamps)
{
var tablet = new Tablet(
tableName,
metadata.ColumnNames,
columns,
metadata.ColumnCategories,
metadata.DataTypes,
values,
@ -535,13 +576,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
var metadata = await GetMetadata<T>();
var accessor = SourceEntityAccessorFactory.GetAccessor<T>();
var memberCache = new Dictionary<string, EntityMemberInfo>(); // 缓存优化查询
// 预构建成员缓存（Key: NameOrPath）
foreach (var member in accessor.MemberList)
{
memberCache[member.NameOrPath] = member;
}
var memberCache = BuildMemberCache(accessor);
var columns = new List<string>() { "Timestamps" };
var dataTypes = new List<TSDataType>() { TSDataType.TIMESTAMP };
@ -596,13 +631,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
private List<ColumnInfo> CollectColumnMetadata<T>(ISourceEntityAccessor<T> accessor)
{
var columns = new List<ColumnInfo>();
var memberCache = new Dictionary<string, EntityMemberInfo>(); // 缓存优化查询
// 预构建成员缓存（Key: NameOrPath）
foreach (var member in accessor.MemberList)
{
memberCache[member.NameOrPath] = member;
}
var memberCache = BuildMemberCache(accessor);
foreach (var member in accessor.MemberList)
{
@ -625,15 +654,15 @@ namespace JiShe.CollectBus.IoTDB.Provider
ColumnInfo? column = null;
if (tagAttr != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.TAG, GetDataTypeFromTypeName(declaredTypeName), false);
column = new ColumnInfo(member.NameOrPath, ColumnCategory.TAG, GetDataTypeFromTypeName(declaredTypeName), false, member.DeclaredTypeName);
}
else if (attrColumn != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.ATTRIBUTE, GetDataTypeFromTypeName(declaredTypeName), false);
column = new ColumnInfo(member.NameOrPath, ColumnCategory.ATTRIBUTE, GetDataTypeFromTypeName(declaredTypeName), false, member.DeclaredTypeName);
}
else if (fieldColumn != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.FIELD, GetDataTypeFromTypeName(declaredTypeName), false);
column = new ColumnInfo(member.NameOrPath, ColumnCategory.FIELD, GetDataTypeFromTypeName(declaredTypeName), false, member.DeclaredTypeName);
}
// 单测模式处理
@ -644,7 +673,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
throw new Exception($"{nameof(CollectColumnMetadata)} {accessor.EntityName} {member.NameOrPath} 单侧点属性解析异常");
}
column = new ColumnInfo(member.NameOrPath, ColumnCategory.FIELD, GetDataTypeFromTypeName(tupleMember.DeclaredTypeName), true);
column = new ColumnInfo(member.NameOrPath, ColumnCategory.FIELD, GetDataTypeFromTypeName(tupleMember.DeclaredTypeName), true, tupleMember.DeclaredTypeName);
}
if (column.HasValue) columns.Add(column.Value);
@ -658,7 +687,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <param name="typeInfo">待解析的类</param>
/// <param name="columns">已处理好的数据列</param>
/// <returns></returns>
private DeviceMetadata BuildDeviceMetadata<T>(List<ColumnInfo> columns) where T : IoTEntity
private DeviceMetadata BuildDeviceMetadata<T>(List<ColumnInfo> columns, ISourceEntityAccessor<T> accessor) where T : IoTEntity
{
var metadata = new DeviceMetadata();
@ -677,9 +706,97 @@ namespace JiShe.CollectBus.IoTDB.Provider
ProcessCategory(groupedColumns, ColumnCategory.ATTRIBUTE, metadata);
ProcessCategory(groupedColumns, ColumnCategory.FIELD, metadata);
// 新增处理器初始化
foreach (var item in metadata.ColumnNames)
{
ColumnInfo column = columns.FirstOrDefault(d => d.Name == item);
var processor = new ColumnProcessor
{
ColumnName = column.Name,
IsSingleMeasuring = column.IsSingleMeasuring,
Converter = CreateConverter(column.DeclaredTypeName.ToUpper())
};
// 处理单测点
if (column.IsSingleMeasuring)
{
var item1Member = accessor.MemberList
.First(m => m.NameOrPath == $"{column.Name}.Item1");
processor.SingleMeasuringNameGetter = (obj) =>
{
// 获取原始值并转为字符串
object rawValue = item1Member.Getter(obj);
string value = rawValue?.ToString();
if (!string.IsNullOrEmpty(value))
{
// 规则1: 严格检查ASCII字母和数字（0-9, A-Z, a-z）
bool hasInvalidChars = value.Any(c =>
!((c >= 'A' && c <= 'Z') ||
(c >= 'a' && c <= 'z') ||
(c >= '0' && c <= '9')));
// 规则2: 首字符不能是数字
bool startsWithDigit = value[0] >= '0' && value[0] <= '9';
// 规则3: 全字符串不能都是数字
bool allDigits = value.All(c => c >= '0' && c <= '9');
// 按优先级抛出具体异常
if (hasInvalidChars)
{
throw new InvalidOperationException(
$"SingleMeasuring name '{value}' 包含非法字符，只允许字母和数字");
}
else if (startsWithDigit)
{
throw new InvalidOperationException(
$"SingleMeasuring name '{value}' 不能以数字开头");
}
else if (allDigits)
{
throw new InvalidOperationException(
$"SingleMeasuring name '{value}' 不能全为数字");
}
}
return value;
};
var item2Member = accessor.MemberList
.First(m => m.NameOrPath == $"{column.Name}.Item2");
processor.ValueGetter = (obj) => {
object rawValue = item2Member.Getter(obj);
return processor.Converter(rawValue);
};
}
else
{
// 获取对应的成员访问器
var member = accessor.MemberList.First(m => m.NameOrPath == column.Name);
processor.ValueGetter = (obj) => {
object rawValue = member.Getter(obj);
return processor.Converter(rawValue);
};
}
metadata.Processors.Add(processor);
}
return metadata;
}
private Func<object, object> CreateConverter(string declaredTypeName)
{
return declaredTypeName switch
{
"DATETIME" => value => ((DateTime)value).GetDateTimeOffset().ToUnixTimeNanoseconds(),
_ => value => value
};
}
/// <summary>
/// 处理不同列类型的逻辑
/// </summary>
@ -706,6 +823,11 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// </summary>
public string Name { get; }
/// <summary>
/// 声明的类型的名称
/// </summary>
public string DeclaredTypeName { get; }
/// <summary>
/// 是否是单测点
/// </summary>
@ -721,12 +843,13 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// </summary>
public TSDataType DataType { get; }
public ColumnInfo(string name, ColumnCategory category, TSDataType dataType, bool isSingleMeasuring)
public ColumnInfo(string name, ColumnCategory category, TSDataType dataType, bool isSingleMeasuring, string declaredTypeName)
{
Name = name;
Category = category;
DataType = dataType;
IsSingleMeasuring = isSingleMeasuring;
DeclaredTypeName = declaredTypeName;
}
}
@ -807,5 +930,23 @@ namespace JiShe.CollectBus.IoTDB.Provider
TSDataType.STRING => Convert.ToString(value),
_ => Convert.ToString(value)
};
/// <summary>
/// 缓存实体属性信息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="accessor"></param>
/// <returns></returns>
private Dictionary<string, EntityMemberInfo> BuildMemberCache<T>(ISourceEntityAccessor<T> accessor)
{
var cache = new Dictionary<string, EntityMemberInfo>(StringComparer.Ordinal);
foreach (var member in accessor.MemberList)
{
cache[member.NameOrPath] = member;
}
return cache;
}
private static readonly Regex _asciiAlphanumericRegex = new Regex(@"^[a-zA-Z0-9]*$", RegexOptions.Compiled);
}
}

14
services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs
View File

@ -140,13 +140,13 @@ namespace JiShe.CollectBus.DataChannels
// 批量写入数据库
await _dbProvider.BatchInsertAsync(metadata, records);
//// 限流推送Kafka
//await DeviceGroupBalanceControl.ProcessWithThrottleAsync(
// items: records,
// deviceIdSelector: data => data.DeviceId,
// processor: async (data, groupIndex) =>
// await KafkaProducerIssuedMessageAction(topicName, data, groupIndex)
//);
// 限流推送Kafka
await DeviceGroupBalanceControl.ProcessWithThrottleAsync(
items: records,
deviceIdSelector: data => data.DeviceId,
processor: async (data, groupIndex) =>
await KafkaProducerIssuedMessageAction(topicName, data, groupIndex)
);
}
catch (Exception ex)
{

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

@ -184,14 +184,14 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
time = DateTime.Now;
//System.Reflection.PropertyInfo;
//System.Reflection.FieldInfo
var meter = new TreeModelSingleMeasuringEntity<string>()
var meter = new TreeModelSingleMeasuringEntity<DateTime>()
{
SystemName = "energy",
DeviceId = "402440506",
DeviceType = "1",
ProjectId = "10059",
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
SingleMeasuring = (measuring, value)
SingleMeasuring = (measuring, time)
};
await _iotDBProvider.InsertAsync(meter);
}

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

@ -7,8 +7,9 @@ using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.DataChannels;
using JiShe.CollectBus.DataMigration.Options;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Model;
using JiShe.CollectBus.IoTDB.Options;
@ -17,8 +18,8 @@ using JiShe.CollectBus.IotSystems.Ammeters;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Interfaces;
using JiShe.CollectBus.Protocol.Models;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
@ -26,14 +27,6 @@ using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using JiShe.CollectBus.Protocol.Models;
using System.Threading.Channels;
using static IdentityModel.ClaimComparer;
using JiShe.CollectBus.DataChannels;
using JiShe.CollectBus.DataMigration.Options;
using static System.Runtime.InteropServices.JavaScript.JSType;
using static System.Formats.Asn1.AsnWriter;
using System.Threading;
namespace JiShe.CollectBus.ScheduledMeterReading
{
@ -329,7 +322,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
/// <returns></returns>
public virtual async Task InitAmmeterCacheData(string gatherCode = "")
{
return;
//return;
// 创建取消令牌源
//var cts = new CancellationTokenSource();
@ -744,7 +737,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
ItemCode = tempItem,
DataTimeMark = new Protocol.DataTimeMark()
{
Density = ammeterInfo.TimeDensity,//todo 转换成协议的值
Density = ammeterInfo.TimeDensity.GetDensity(),//转换成协议的值
Point = 1,
DataTime = timestamps,
}
@ -1354,7 +1347,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
}
List<MeterReadingTelemetryPacketInfo> taskList = new List<MeterReadingTelemetryPacketInfo>();
var metadata = await _dbProvider.GetMetadata<MeterReadingTelemetryPacketInfo>();
var itemCode = T37612012PacketItemCodeConst.AFN09HFN01H;
//var subItemCode = T6452007PacketItemCodeConst.C08;
@ -1643,36 +1635,43 @@ namespace JiShe.CollectBus.ScheduledMeterReading
/// <returns></returns>
protected MeterReadingTelemetryPacketInfo CreateAmmeterPacketInfo(AmmeterInfo ammeterInfo, long timestamps, ProtocolBuildResponse builderResponse, string itemCode, string subItemCode, DateTime pendingCopyReadTime, DateTime creationTime, TelemetryPacketTypeEnum packetType)
{
string taskMark = CommonHelper.GetTaskMark(builderResponse.AFn, builderResponse.Fn, ammeterInfo.MeteringCode, builderResponse.MSA, builderResponse.Seq);
return new MeterReadingTelemetryPacketInfo()
try
{
SystemName = SystemType,
ProjectId = $"{ammeterInfo.ProjectID}",
DeviceType = $"{MeterTypeEnum.Ammeter}",
DeviceId = $"{ammeterInfo.MeterId}",
Timestamps = timestamps,
DatabaseBusiID = ammeterInfo.DatabaseBusiID,
PendingCopyReadTime = pendingCopyReadTime,
CreationTime = creationTime,
MeterAddress = ammeterInfo.AmmerterAddress,
PacketType = (int)packetType,
AFN = builderResponse.AFn,
Fn = builderResponse.Fn,
Seq = builderResponse.Seq,
MSA = builderResponse.MSA,
FocusId = ammeterInfo.FocusId,
FocusAddress = ammeterInfo.FocusAddress,
ItemCode = itemCode,
SubItemCode = subItemCode,
TaskMark = taskMark,
IsSend = false,
ManualOrNot = false,
Pn = ammeterInfo.MeteringCode,
IssuedMessageId = GuidGenerator.Create().ToString(),
IssuedMessageHexString = Convert.ToHexString(builderResponse.Data),
IsReceived = false,
ScoreValue = $"{ammeterInfo.FocusAddress}.{taskMark}".Md5Fun(),
};
string taskMark = CommonHelper.GetTaskMark(builderResponse.AFn, builderResponse.Fn, ammeterInfo.MeteringCode, builderResponse.MSA, builderResponse.Seq);
return new MeterReadingTelemetryPacketInfo()
{
SystemName = SystemType,
ProjectId = $"{ammeterInfo.ProjectID}",
DeviceType = $"{MeterTypeEnum.Ammeter}",
DeviceId = $"{ammeterInfo.MeterId}",
Timestamps = timestamps,
DatabaseBusiID = ammeterInfo.DatabaseBusiID,
PendingCopyReadTime = pendingCopyReadTime,
CreationTime = creationTime,
MeterAddress = ammeterInfo.AmmerterAddress,
PacketType = (int)packetType,
AFN = builderResponse.AFn,
Fn = builderResponse.Fn,
Seq = builderResponse.Seq,
MSA = builderResponse.MSA,
FocusId = ammeterInfo.FocusId,
FocusAddress = ammeterInfo.FocusAddress,
ItemCode = itemCode,
SubItemCode = subItemCode,
TaskMark = taskMark,
IsSend = false,
ManualOrNot = false,
Pn = ammeterInfo.MeteringCode,
IssuedMessageId = Guid.NewGuid().ToString(),
IssuedMessageHexString = Convert.ToHexString(builderResponse.Data),
IsReceived = false,
ScoreValue = $"{ammeterInfo.FocusAddress}.{taskMark}".Md5Fun(),
};
}
catch (Exception ex)
{
throw ex;
}
}
#endregion

17
shared/JiShe.CollectBus.Common/Helpers/CommonHelper.cs
View File

@ -859,6 +859,21 @@ namespace JiShe.CollectBus.Common.Helpers
}
/// <summary>
/// 采集频率转换为集中器采集密度
/// </summary>
/// <param name="timeDensity"></param>
/// <returns></returns>
public static int GetDensity(this int timeDensity) =>
timeDensity switch
{
0 => 0,//无
1 => 255,//1分钟
5 => 245,//5分钟
15 => 1,//15分钟
30 => 2,//30分钟
60 => 3,//60分钟
_ => -1//采集项本身无密度位
};
}
}

1
web/JiShe.CollectBus.Host/Pages/Monitor.cshtml
View File

@ -16,7 +16,6 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet" />
<title>后端服务</title>
</head>
<body>

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

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