Compare commits
6 Commits
02f2a2cafc
...
309f5c37d2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
309f5c37d2 | ||
|
|
ca1a23fd33 | ||
|
|
3c6a04f947 | ||
|
|
38283db1b2 | ||
|
|
cb6cfea4dd | ||
|
|
cc59b32742 |
@ -0,0 +1,19 @@
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Attribute
|
||||
{
|
||||
/// <summary>
|
||||
/// IoTDB实体类型特性
|
||||
/// </summary>
|
||||
[AttributeUsage(AttributeTargets.Class)]
|
||||
public class EntityTypeAttribute : System.Attribute
|
||||
{
|
||||
public EntityTypeEnum EntityType { get; }
|
||||
|
||||
|
||||
public EntityTypeAttribute(EntityTypeEnum entityType)
|
||||
{
|
||||
EntityType = entityType;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,7 +1,7 @@
|
||||
namespace JiShe.CollectBus.IoTDB.Attribute
|
||||
{
|
||||
/// <summary>
|
||||
/// 用于标识当前实体为单侧点模式,单侧点模式只有一个Filed标识字段,类型是Tuple<string,object>,Item1=>测点名称,Item2=>测点值,泛型
|
||||
/// 用于标识当前实体为单侧点模式,单侧点模式只有一个Filed标识字段,类型是Tuple<string,T>,Item1=>测点名称,Item2=>测点值,泛型
|
||||
/// </summary>
|
||||
[AttributeUsage(AttributeTargets.Property)]
|
||||
public class SingleMeasuringAttribute : System.Attribute
|
||||
|
||||
@ -0,0 +1,18 @@
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Attribute
|
||||
{
|
||||
/// <summary>
|
||||
/// IoTDB实体存储路径或表名称,一般用于已经明确的存储路径或表名称,例如日志存储
|
||||
/// </summary>
|
||||
[AttributeUsage(AttributeTargets.Class)]
|
||||
public class TableNameOrTreePathAttribute : System.Attribute
|
||||
{
|
||||
public string TableNameOrTreePath { get; }
|
||||
|
||||
public TableNameOrTreePathAttribute(string tableNameOrTreePath)
|
||||
{
|
||||
TableNameOrTreePath = tableNameOrTreePath;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -17,7 +17,7 @@ namespace JiShe.CollectBus.IoTDB.Context
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 是否使用表模型存储, 默认false,使用tree模型存储
|
||||
/// 存储模型切换标识,是否使用table模型存储, 默认为false,标识tree模型存储
|
||||
/// </summary>
|
||||
public bool UseTableSessionPool { get; set; }
|
||||
|
||||
|
||||
24
modules/JiShe.CollectBus.IoTDB/EnumInfo/EntityTypeEnum.cs
Normal file
24
modules/JiShe.CollectBus.IoTDB/EnumInfo/EntityTypeEnum.cs
Normal file
@ -0,0 +1,24 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Enums
|
||||
{
|
||||
/// <summary>
|
||||
/// IoTDB实体类型枚举
|
||||
/// </summary>
|
||||
public enum EntityTypeEnum
|
||||
{
|
||||
/// <summary>
|
||||
/// 树模型
|
||||
/// </summary>
|
||||
TreeModel = 1,
|
||||
|
||||
/// <summary>
|
||||
/// 表模型
|
||||
/// </summary>
|
||||
TableModel = 2,
|
||||
}
|
||||
}
|
||||
@ -1,6 +1,6 @@
|
||||
using JiShe.CollectBus.Common.Models;
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
using JiShe.CollectBus.IoTDB.Options;
|
||||
using JiShe.CollectBus.IoTDB.Provider;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Interface
|
||||
{
|
||||
|
||||
@ -6,7 +6,6 @@
|
||||
<Nullable>enable</Nullable>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<!--<PackageReference Include="Apache.IoTDB" Version="1.3.3.1" />-->
|
||||
<PackageReference Include="Apache.IoTDB" Version="2.0.2" />
|
||||
<PackageReference Include="Volo.Abp" Version="8.3.3" />
|
||||
</ItemGroup>
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Provider
|
||||
namespace JiShe.CollectBus.IoTDB.Model
|
||||
{
|
||||
/// <summary>
|
||||
/// IoT实体基类
|
||||
/// IoT实体基类,此类适用于多个数据测点记录场景,单个测点请使用子类 SingleMeasuring
|
||||
/// </summary>
|
||||
public abstract class IoTEntity
|
||||
{
|
||||
@ -11,29 +11,29 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// 系统名称
|
||||
/// </summary>
|
||||
[TAGColumn]
|
||||
public string SystemName { get; set; }
|
||||
public required string SystemName { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 项目编码
|
||||
/// </summary>
|
||||
[TAGColumn]
|
||||
public string ProjectCode { get; set; }
|
||||
public required string ProjectCode { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 设备类型集中器、电表、水表、流量计、传感器等
|
||||
/// </summary>
|
||||
[TAGColumn]
|
||||
public string DeviceType { get; set; }
|
||||
public required string DeviceType { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 设备ID
|
||||
/// </summary>
|
||||
[TAGColumn]
|
||||
public string DeviceId { get; set; }
|
||||
public required string DeviceId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 当前时间戳,单位毫秒
|
||||
/// 当前时间戳,单位毫秒,必须通过DateTimeOffset获取
|
||||
/// </summary>
|
||||
public long Timestamps { get; set; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
|
||||
public required long Timestamps { get; set; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,24 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
using JiShe.CollectBus.IoTDB.Provider;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Model
|
||||
{
|
||||
/// <summary>
|
||||
/// Table模型单项数据实体
|
||||
/// </summary>
|
||||
[EntityType(EntityTypeEnum.TableModel)]
|
||||
public class TableModelSingleMeasuringEntity<T> : IoTEntity
|
||||
{
|
||||
/// <summary>
|
||||
/// 单项数据键值对
|
||||
/// </summary>
|
||||
[SingleMeasuring(nameof(SingleColumn))]
|
||||
public required Tuple<string, T> SingleColumn { get; set; }
|
||||
}
|
||||
}
|
||||
@ -4,19 +4,21 @@ using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
using JiShe.CollectBus.IoTDB.Provider;
|
||||
|
||||
namespace JiShe.CollectBus.IotSystems.AFNEntity
|
||||
namespace JiShe.CollectBus.IoTDB.Model
|
||||
{
|
||||
/// <summary>
|
||||
/// AFN单项数据实体
|
||||
/// Tree模型单项数据实体
|
||||
/// </summary>
|
||||
public class SingleMeasuringAFNDataEntity<T> : IoTEntity
|
||||
[EntityType(EntityTypeEnum.TreeModel)]
|
||||
public class TreeModelSingleMeasuringEntity<T> : IoTEntity
|
||||
{
|
||||
/// <summary>
|
||||
/// 单项数据对象
|
||||
/// 单项数据键值对
|
||||
/// </summary>
|
||||
[SingleMeasuring(nameof(SingleMeasuring))]
|
||||
public Tuple<string, T> SingleMeasuring { get; set; }
|
||||
public required Tuple<string, T> SingleMeasuring { get; set; }
|
||||
}
|
||||
}
|
||||
@ -1,4 +1,5 @@
|
||||
using Apache.IoTDB;
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Provider
|
||||
{
|
||||
@ -7,6 +8,11 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// </summary>
|
||||
public class DeviceMetadata
|
||||
{
|
||||
/// <summary>
|
||||
/// IoTDB实体类型枚举
|
||||
/// </summary>
|
||||
public EntityTypeEnum EntityType { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 是否有单测量值
|
||||
/// </summary>
|
||||
|
||||
@ -1,4 +1,6 @@
|
||||
namespace JiShe.CollectBus.IoTDB.Provider
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Provider
|
||||
{
|
||||
/// <summary>
|
||||
/// 设备路径构建器
|
||||
@ -13,7 +15,7 @@
|
||||
/// <returns></returns>
|
||||
public static string GetDevicePath<T>(T entity) where T : IoTEntity
|
||||
{
|
||||
return $"root.{entity.SystemName.ToLower()}.`{entity.ProjectCode}`.`{entity.DeviceId}`";
|
||||
return $"root.{entity.SystemName.ToLower()}.`{entity.ProjectCode}`.`{entity.DeviceType}`.`{entity.DeviceId}`";
|
||||
}
|
||||
|
||||
|
||||
@ -28,6 +30,17 @@
|
||||
var type = typeof(T);
|
||||
return $"{type.Name.ToLower()}";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取表名称,用作单侧点表模型特殊处理。
|
||||
/// </summary>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <param name="entity"></param>
|
||||
/// <returns></returns>
|
||||
public static string GetDeviceTableName<T>(T entity) where T : IoTEntity
|
||||
{
|
||||
return $"{entity.SystemName.ToLower()}.`{entity.ProjectCode}`.`{entity.DeviceType}`.`{entity.DeviceId}`";
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Reflection;
|
||||
using System.Text;
|
||||
using Apache.IoTDB;
|
||||
@ -7,9 +8,11 @@ using JiShe.CollectBus.Common.Models;
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
using JiShe.CollectBus.IoTDB.Context;
|
||||
using JiShe.CollectBus.IoTDB.Interface;
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
using JiShe.CollectBus.IoTDB.Options;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Volo.Abp.DependencyInjection;
|
||||
using Volo.Abp.Domain.Entities;
|
||||
|
||||
namespace JiShe.CollectBus.IoTDB.Provider
|
||||
{
|
||||
@ -56,12 +59,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
var tablet = BuildTablet(new[] { entity }, metadata);
|
||||
|
||||
await CurrentSession.InsertAsync(tablet);
|
||||
|
||||
//int result = await _currentSession.InsertAsync(tablet);
|
||||
//if (result <= 0)
|
||||
//{
|
||||
// _logger.LogError($"{typeof(T).Name}插入数据没有成功");
|
||||
//}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@ -80,11 +77,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
{
|
||||
var tablet = BuildTablet(batch, metadata);
|
||||
await CurrentSession.InsertAsync(tablet);
|
||||
//var result = await _currentSession.InsertAsync(tablet);
|
||||
//if (result <= 0)
|
||||
//{
|
||||
// _logger.LogWarning($"{typeof(T).Name} 批量插入数据第{batch}批次没有成功,共{batches}批次。");
|
||||
//}
|
||||
}
|
||||
}
|
||||
|
||||
@ -146,10 +138,39 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
List<string> tempColumnNames = new List<string>();
|
||||
tempColumnNames.AddRange(metadata.ColumnNames);
|
||||
|
||||
var entityTypeAttribute = typeof(T).GetCustomAttribute<EntityTypeAttribute>();
|
||||
|
||||
if (entityTypeAttribute == null)
|
||||
{
|
||||
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 {nameof(T)}的EntityType 没有指定,属于异常情况,-101");
|
||||
}
|
||||
|
||||
if (metadata.EntityType != entityTypeAttribute.EntityType)
|
||||
{
|
||||
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 {nameof(T)}的EntityType 和{nameof(DeviceMetadata)}的 EntityType 不一致,属于异常情况,-102");
|
||||
}
|
||||
|
||||
if (metadata.EntityType == Enums.EntityTypeEnum.TreeModel && _runtimeContext.UseTableSessionPool == true)
|
||||
{
|
||||
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 tree模型不能使用table模型Session连接,属于异常情况,-103");
|
||||
}
|
||||
else if (metadata.EntityType == Enums.EntityTypeEnum.TableModel && _runtimeContext.UseTableSessionPool == false)
|
||||
{
|
||||
throw new ArgumentException($"{nameof(BuildTablet)} 构建存储结构{nameof(Tablet)}时 table模型不能使用tree模型Session连接,属于异常情况,-104");
|
||||
}
|
||||
|
||||
string tableNameOrTreePath = string.Empty;
|
||||
var tableNameOrTreePathAttribute = typeof(T).GetCustomAttribute<TableNameOrTreePathAttribute>();
|
||||
if (tableNameOrTreePathAttribute != null)
|
||||
{
|
||||
tableNameOrTreePath = tableNameOrTreePathAttribute.TableNameOrTreePath;
|
||||
}
|
||||
|
||||
foreach (var entity in entities)
|
||||
{
|
||||
timestamps.Add(entity.Timestamps);
|
||||
var rowValues = new List<object>();
|
||||
|
||||
foreach (var measurement in tempColumnNames)
|
||||
{
|
||||
|
||||
@ -160,7 +181,9 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
}
|
||||
|
||||
var value = propertyInfo.GetValue(entity);
|
||||
if (propertyInfo.IsDefined(typeof(SingleMeasuringAttribute), false) && value != null)//表示当前对象是单测点模式
|
||||
if (propertyInfo.IsDefined(typeof(SingleMeasuringAttribute), false) && metadata.IsSingleMeasuring == true)//表示当前对象是单测点模式
|
||||
{
|
||||
if (value != null)
|
||||
{
|
||||
Type tupleType = value.GetType();
|
||||
Type[] tupleArgs = tupleType.GetGenericArguments();
|
||||
@ -176,27 +199,35 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
metadata.ColumnNames[indexOf] = (string)item1!;
|
||||
|
||||
rowValues.Add(item2);
|
||||
|
||||
}
|
||||
else
|
||||
{
|
||||
if (value != null)
|
||||
rowValues.Add(null);
|
||||
}
|
||||
|
||||
//同时如果是单测点模式,且是table模型存储,路径只能通过DevicePathBuilder.GetDeviceTableName(entity)获取
|
||||
if (_runtimeContext.UseTableSessionPool)
|
||||
{
|
||||
tableNameOrTreePath = DevicePathBuilder.GetDeviceTableName(entity);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
|
||||
rowValues.Add(value);
|
||||
}
|
||||
else
|
||||
{
|
||||
//填充默认数据值
|
||||
DataTypeDefaultValueMap.TryGetValue(propertyInfo.PropertyType.Name, out object defaultValue);
|
||||
|
||||
rowValues.Add(defaultValue);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
values.Add(rowValues);
|
||||
|
||||
//如果指定了路径
|
||||
if (!string.IsNullOrWhiteSpace(tableNameOrTreePath))
|
||||
{
|
||||
devicePaths.Add(tableNameOrTreePath);
|
||||
}
|
||||
else
|
||||
{
|
||||
if (!_runtimeContext.UseTableSessionPool)//树模型
|
||||
{
|
||||
devicePaths.Add(DevicePathBuilder.GetDevicePath(entity));
|
||||
@ -207,6 +238,8 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if (devicePaths.Count > 1)
|
||||
{
|
||||
throw new Exception($"{nameof(BuildTablet)} 构建Tablet《{typeof(T).Name}》时,批量插入的设备路径不一致。");
|
||||
@ -220,14 +253,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// <summary>
|
||||
/// 构建tree模型的Tablet
|
||||
/// </summary>
|
||||
/// <param name="metadata"></param>
|
||||
/// <param name="devicePath"></param>
|
||||
/// <param name="values"></param>
|
||||
/// <param name="timestamps"></param>
|
||||
/// <param name="metadata">已解析的设备数据元数据</param>
|
||||
/// <param name="devicePath">设备路径</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)
|
||||
{
|
||||
//todo 树模型需要去掉TAG类型和ATTRIBUTE类型的字段,只需要保留FIELD类型字段即可
|
||||
|
||||
return new Tablet(
|
||||
devicePath,
|
||||
metadata.ColumnNames,
|
||||
@ -240,16 +275,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// <summary>
|
||||
/// 构建表模型的Tablet
|
||||
/// </summary>
|
||||
/// <param name="metadata"></param>
|
||||
/// <param name="devicePath"></param>
|
||||
/// <param name="values"></param>
|
||||
/// <param name="timestamps"></param>
|
||||
/// <param name="metadata">已解析的设备数据元数据</param>
|
||||
/// <param name="tableName">表名称</param>
|
||||
/// <param name="values">数据集合</param>
|
||||
/// <param name="timestamps">时间戳集合</param>
|
||||
/// <returns></returns>
|
||||
private Tablet BuildTableSessionTablet(DeviceMetadata metadata, string devicePath,
|
||||
private Tablet BuildTableSessionTablet(DeviceMetadata metadata, string tableName,
|
||||
List<List<object>> values, List<long> timestamps)
|
||||
{
|
||||
var tablet = new Tablet(
|
||||
devicePath,
|
||||
tableName,
|
||||
metadata.ColumnNames,
|
||||
metadata.ColumnCategories,
|
||||
metadata.DataTypes,
|
||||
@ -399,34 +434,21 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// <returns></returns>
|
||||
private DeviceMetadata GetMetadata<T>() where T : IoTEntity
|
||||
{
|
||||
|
||||
var columns = CollectColumnMetadata(typeof(T));
|
||||
var metadata = BuildDeviceMetadata(columns);
|
||||
|
||||
var metadata = BuildDeviceMetadata<T>(columns);
|
||||
return MetadataCache.AddOrUpdate(
|
||||
typeof(T),
|
||||
addValueFactory: t => metadata, // 如果键不存在,用此值添加
|
||||
updateValueFactory: (t, existingValue) =>
|
||||
{
|
||||
var columns = CollectColumnMetadata(t);
|
||||
var metadata = BuildDeviceMetadata(columns);
|
||||
var metadata = BuildDeviceMetadata<T>(columns);
|
||||
|
||||
//对现有值 existingValue 进行修改,返回新值
|
||||
existingValue.ColumnNames = metadata.ColumnNames;
|
||||
return existingValue;
|
||||
}
|
||||
);
|
||||
|
||||
//return _metadataCache.GetOrAdd(typeof(T), type =>
|
||||
//{
|
||||
// var columns = CollectColumnMetadata(type);
|
||||
// var metadata = BuildDeviceMetadata(columns);
|
||||
// //if (metadata.IsSingleMeasuring)
|
||||
// //{
|
||||
// // _metadataCache.Remove(typeof(T));
|
||||
// //}
|
||||
// return metadata;
|
||||
//});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@ -440,21 +462,36 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
|
||||
foreach (var prop in type.GetProperties())
|
||||
{
|
||||
|
||||
string typeName = string.Empty;
|
||||
|
||||
Type declaredType = prop.PropertyType;
|
||||
// 处理可空类型
|
||||
if (declaredType.IsGenericType && declaredType.GetGenericTypeDefinition() == typeof(Nullable<>))
|
||||
{
|
||||
Type underlyingType = Nullable.GetUnderlyingType(declaredType);
|
||||
typeName = underlyingType.Name;
|
||||
}
|
||||
else
|
||||
{
|
||||
typeName = declaredType.Name;
|
||||
}
|
||||
|
||||
//先获取Tag标签和属性标签
|
||||
ColumnInfo? column = prop.GetCustomAttribute<TAGColumnAttribute>() is not null ? new ColumnInfo(
|
||||
name: prop.Name,
|
||||
category: ColumnCategory.TAG,
|
||||
dataType: GetDataTypeFromTypeName(prop.PropertyType.Name),
|
||||
dataType: GetDataTypeFromTypeName(typeName),
|
||||
false
|
||||
) : prop.GetCustomAttribute<ATTRIBUTEColumnAttribute>() is not null ? new ColumnInfo(
|
||||
prop.Name,
|
||||
ColumnCategory.ATTRIBUTE,
|
||||
GetDataTypeFromTypeName(prop.PropertyType.Name),
|
||||
GetDataTypeFromTypeName(typeName),
|
||||
false
|
||||
) : prop.GetCustomAttribute<FIELDColumnAttribute>() is not null ? new ColumnInfo(
|
||||
prop.Name,
|
||||
ColumnCategory.FIELD,
|
||||
GetDataTypeFromTypeName(prop.PropertyType.Name),
|
||||
GetDataTypeFromTypeName(typeName),
|
||||
false)
|
||||
: null;
|
||||
|
||||
@ -490,9 +527,10 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
/// <summary>
|
||||
/// 构建设备元数据
|
||||
/// </summary>
|
||||
/// <param name="columns"></param>
|
||||
/// <param name="typeInfo">待解析的类</param>
|
||||
/// <param name="columns">已处理好的数据列</param>
|
||||
/// <returns></returns>
|
||||
private DeviceMetadata BuildDeviceMetadata(List<ColumnInfo> columns)
|
||||
private DeviceMetadata BuildDeviceMetadata<T>(List<ColumnInfo> columns) where T : IoTEntity
|
||||
{
|
||||
var metadata = new DeviceMetadata();
|
||||
|
||||
@ -511,6 +549,15 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
ProcessCategory(groupedColumns, ColumnCategory.ATTRIBUTE, metadata);
|
||||
ProcessCategory(groupedColumns, ColumnCategory.FIELD, metadata);
|
||||
|
||||
var entityTypeAttribute = typeof(T).GetCustomAttribute<EntityTypeAttribute>();
|
||||
|
||||
if (entityTypeAttribute == null)
|
||||
{
|
||||
throw new ArgumentException($"{nameof(BuildDeviceMetadata)} 构建设备元数据时 {nameof(IoTEntity)} 的EntityType 没有指定,属于异常情况,-101");
|
||||
}
|
||||
|
||||
metadata.EntityType = entityTypeAttribute.EntityType;
|
||||
|
||||
return metadata;
|
||||
}
|
||||
|
||||
|
||||
@ -56,7 +56,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
var result = await _sessionPool.InsertAlignedTabletAsync(tablet);
|
||||
if (result != 0)
|
||||
{
|
||||
throw new Exception($"{nameof(TableSessionPoolAdapter)} ");
|
||||
throw new Exception($"{nameof(SessionPoolAdapter)} Tree模型数据入库没有成功,返回结果为:{result}");
|
||||
}
|
||||
|
||||
return result;
|
||||
|
||||
@ -54,7 +54,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
|
||||
var result = await _sessionPool.InsertAsync(tablet);
|
||||
if (result != 0)
|
||||
{
|
||||
throw new Exception($"{nameof(TableSessionPoolAdapter)} ");
|
||||
throw new Exception($"{nameof(TableSessionPoolAdapter)} table模型数据入库没有成功,返回结果为:{result}");
|
||||
}
|
||||
|
||||
return result;
|
||||
|
||||
@ -42,6 +42,8 @@ namespace JiShe.CollectBus.RedisDataCache
|
||||
Instance = _freeRedisProvider.Instance;
|
||||
}
|
||||
|
||||
//todo 单个数据查询
|
||||
|
||||
/// <summary>
|
||||
/// 单个添加数据
|
||||
/// </summary>
|
||||
|
||||
@ -1,33 +1,28 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
using Apache.IoTDB.DataStructure;
|
||||
using Apache.IoTDB;
|
||||
using Confluent.Kafka;
|
||||
using JiShe.CollectBus.Ammeters;
|
||||
using JiShe.CollectBus.FreeSql;
|
||||
using JiShe.CollectBus.IotSystems.PrepayModel;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using JiShe.CollectBus.IotSystems.AFNEntity;
|
||||
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using JiShe.CollectBus.Common.Consts;
|
||||
using JiShe.CollectBus.Common.Enums;
|
||||
using System.Diagnostics.Metrics;
|
||||
using JiShe.CollectBus.Common.DeviceBalanceControl;
|
||||
using JiShe.CollectBus.Kafka.Attributes;
|
||||
using System.Text.Json;
|
||||
using JiShe.CollectBus.Ammeters;
|
||||
using JiShe.CollectBus.Application.Contracts;
|
||||
using JiShe.CollectBus.Common.Models;
|
||||
using System.Diagnostics;
|
||||
using JiShe.CollectBus.Common.Consts;
|
||||
using JiShe.CollectBus.Common.DeviceBalanceControl;
|
||||
using JiShe.CollectBus.Common.Enums;
|
||||
using JiShe.CollectBus.Common.Extensions;
|
||||
using JiShe.CollectBus.FreeSql;
|
||||
using JiShe.CollectBus.IoTDB.Context;
|
||||
using JiShe.CollectBus.IoTDB.Interface;
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
using JiShe.CollectBus.IoTDB.Options;
|
||||
using JiShe.CollectBus.IoTDB.Provider;
|
||||
using JiShe.CollectBus.IotSystems.PrepayModel;
|
||||
using JiShe.CollectBus.Kafka.Attributes;
|
||||
using JiShe.CollectBus.Kafka.Internal;
|
||||
using JiShe.CollectBus.Common.Extensions;
|
||||
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace JiShe.CollectBus.Samples;
|
||||
|
||||
@ -52,23 +47,13 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
/// <summary>
|
||||
/// 测试 UseSessionPool
|
||||
/// </summary>
|
||||
/// <param name="timestamps"></param>
|
||||
/// <param name="testTime"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task UseSessionPool(long timestamps)
|
||||
public async Task UseSessionPool(DateTime testTime)
|
||||
{
|
||||
string? messageHexString = null;
|
||||
if (timestamps == 0)
|
||||
{
|
||||
timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
|
||||
_logger.LogError($"timestamps_{timestamps}");
|
||||
}
|
||||
else
|
||||
{
|
||||
messageHexString = messageHexString + timestamps;
|
||||
}
|
||||
|
||||
ElectricityMeter meter = new ElectricityMeter()
|
||||
ElectricityMeterTreeModel meter = new ElectricityMeterTreeModel()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
@ -77,8 +62,8 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
MeterModel = "DDZY-1980",
|
||||
ProjectCode = "10059",
|
||||
Voltage = 10,
|
||||
IssuedMessageHexString = messageHexString,
|
||||
Timestamps = timestamps,
|
||||
IssuedMessageHexString = "messageHexString",
|
||||
Timestamps = testTime.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
@ -88,9 +73,10 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task UseTableSessionPool()
|
||||
public async Task UseTableSessionPool(DateTime time)
|
||||
{
|
||||
ElectricityMeter meter2 = new ElectricityMeter()
|
||||
var testTime = time;
|
||||
ElectricityMeterTreeModel meter2 = new ElectricityMeterTreeModel()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
@ -99,7 +85,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
MeterModel = "DDZY-1980",
|
||||
ProjectCode = "10059",
|
||||
Voltage = 10,
|
||||
Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
Timestamps = testTime.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
};
|
||||
|
||||
await _iotDBProvider.InsertAsync(meter2);
|
||||
@ -115,9 +101,135 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
MeterModel = "DDZY-1980",
|
||||
ProjectCode = "10059",
|
||||
Voltage = 10,
|
||||
Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
Timestamps = testTime.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试Session切换3
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task UseTableSessionPool3(DateTime time)
|
||||
{
|
||||
var testTime = time;
|
||||
ElectricityMeterTreeModel meter2 = new ElectricityMeterTreeModel()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
Current = 10,
|
||||
MeterModel = "DDZY-1980",
|
||||
ProjectCode = "10059",
|
||||
Voltage = 10,
|
||||
IssuedMessageHexString = "dsdfsfd",
|
||||
Timestamps = testTime.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
|
||||
};
|
||||
|
||||
await _iotDBProvider.InsertAsync(meter2);
|
||||
|
||||
_dbContext.UseTableSessionPool = true;
|
||||
|
||||
|
||||
ElectricityMeter meter3 = new ElectricityMeter()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
Current = 10,
|
||||
MeterModel = "DDZY-1980",
|
||||
ProjectCode = "10059",
|
||||
Voltage = 10,
|
||||
Currentd = 22,
|
||||
IssuedMessageHexString = "dsdfsfd",
|
||||
Timestamps = testTime.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter3);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试树模型单个测点数据项
|
||||
/// </summary>
|
||||
/// <param name="measuring"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task TestTreeModelSingleMeasuringEntity(string measuring, string value, DateTime time)
|
||||
{
|
||||
var meter = new TreeModelSingleMeasuringEntity<string>()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
ProjectCode = "10059",
|
||||
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
SingleMeasuring = new Tuple<string, string>(measuring, value)
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试树模型单个测点数据项2
|
||||
/// </summary>
|
||||
/// <param name="measuring"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task TestTreeModelSingleMeasuringEntity2(string measuring, int value, DateTime time)
|
||||
{
|
||||
var meter = new TreeModelSingleMeasuringEntity<int>()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
ProjectCode = "10059",
|
||||
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
SingleMeasuring = new Tuple<string, int>(measuring, value)
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试表模型单个测点数据项
|
||||
/// </summary>
|
||||
/// <param name="measuring"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task TestTableModelSingleMeasuringEntity(string measuring, string value, DateTime time)
|
||||
{
|
||||
var meter = new TableModelSingleMeasuringEntity<string>()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
ProjectCode = "10059",
|
||||
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
SingleColumn = new Tuple<string, string>(measuring, value)
|
||||
};
|
||||
_dbContext.UseTableSessionPool = true;
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试表模型单个测点数据项2
|
||||
/// </summary>
|
||||
/// <param name="measuring"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task TestTableModelSingleMeasuringEntity2(string measuring, int value, DateTime time)
|
||||
{
|
||||
var meter = new TableModelSingleMeasuringEntity<int>()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
ProjectCode = "10059",
|
||||
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
|
||||
SingleColumn = new Tuple<string, int>(measuring, value)
|
||||
};
|
||||
_dbContext.UseTableSessionPool = true;
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@ -172,26 +284,6 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 测试单个测点数据项
|
||||
/// </summary>
|
||||
/// <param name="measuring"></param>
|
||||
/// <returns></returns>
|
||||
[HttpGet]
|
||||
public async Task TestSingleMeasuringAFNData(string measuring, string value)
|
||||
{
|
||||
var meter = new SingleMeasuringAFNDataEntity<string>()
|
||||
{
|
||||
SystemName = "energy",
|
||||
DeviceId = "402440506",
|
||||
DeviceType = "Ammeter",
|
||||
ProjectCode = "10059",
|
||||
Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
SingleMeasuring = new Tuple<string, string>(measuring, value)
|
||||
};
|
||||
await _iotDBProvider.InsertAsync(meter);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 测试Redis批量读取10万条数据性能
|
||||
/// </summary>
|
||||
|
||||
@ -12,7 +12,6 @@ using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using JiShe.CollectBus.Common.Helpers;
|
||||
using JiShe.CollectBus.IotSystems.AFNEntity;
|
||||
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using JiShe.CollectBus.Cassandra;
|
||||
|
||||
@ -4,10 +4,12 @@ using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
using JiShe.CollectBus.IoTDB.Provider;
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
|
||||
namespace JiShe.CollectBus.Ammeters
|
||||
{
|
||||
[EntityType(EntityTypeEnum.TableModel)]
|
||||
public class ElectricityMeter : IoTEntity
|
||||
{
|
||||
[ATTRIBUTEColumn]
|
||||
@ -33,5 +35,8 @@ namespace JiShe.CollectBus.Ammeters
|
||||
|
||||
[FIELDColumn]
|
||||
public double Power => Voltage * Current;
|
||||
|
||||
[FIELDColumn]
|
||||
public double? Currentd { get; set; }
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,37 @@
|
||||
using JiShe.CollectBus.IoTDB.Attribute;
|
||||
using JiShe.CollectBus.IoTDB.Enums;
|
||||
using JiShe.CollectBus.IoTDB.Model;
|
||||
|
||||
namespace JiShe.CollectBus.Ammeters
|
||||
{
|
||||
[EntityType(EntityTypeEnum.TreeModel)]
|
||||
public class ElectricityMeterTreeModel : IoTEntity
|
||||
{
|
||||
[ATTRIBUTEColumn]
|
||||
public string MeterModel { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 下发消息内容
|
||||
/// </summary>
|
||||
[FIELDColumn]
|
||||
public string IssuedMessageHexString { get; set; }
|
||||
|
||||
///// <summary>
|
||||
///// 下发消息Id
|
||||
///// </summary>
|
||||
//[FIELDColumn]
|
||||
//public string IssuedMessageId { get; set; }
|
||||
|
||||
[FIELDColumn]
|
||||
public double Voltage { get; set; }
|
||||
|
||||
[FIELDColumn]
|
||||
public double Current { get; set; }
|
||||
|
||||
[FIELDColumn]
|
||||
public double Power => Voltage * Current;
|
||||
|
||||
[FIELDColumn]
|
||||
public double? Currentd { get; set; }
|
||||
}
|
||||
}
|
||||
@ -233,5 +233,22 @@ namespace JiShe.CollectBus.Common.Extensions
|
||||
.AddHours(hours)
|
||||
.AddMinutes(minutes);
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 将 DateTime 时间转换为 DateTimeOffset 时间
|
||||
/// </summary>
|
||||
/// <param name="rawDateTime"></param>
|
||||
/// <returns></returns>
|
||||
public static DateTimeOffset GetDateTimeOffset(this DateTime rawDateTime)
|
||||
{
|
||||
//确保 Kind 为 Local(如果是 Unspecified)
|
||||
DateTime localDateTime = rawDateTime.Kind == DateTimeKind.Unspecified
|
||||
? DateTime.SpecifyKind(rawDateTime, DateTimeKind.Local)
|
||||
: rawDateTime;
|
||||
|
||||
// 转换为 DateTimeOffset(自动应用本地时区偏移)
|
||||
return new DateTimeOffset(localDateTime);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user