using JiShe.IoT.BusinessSystemAggregation.Dto;
using JiShe.IoT.DeviceAggregation;
using JiShe.ServicePro;
using JiShe.ServicePro.ApacheIoTDB.Provider.Model;
using JiShe.ServicePro.ApacheIoTDB.Provider.Options;
using JiShe.ServicePro.Core;
using JiShe.ServicePro.DataChannelManages;
using JiShe.ServicePro.DeviceManagement.DeviceInfos;
using JiShe.ServicePro.DeviceManagement.ThingModels;
using JiShe.ServicePro.Dto;
using JiShe.ServicePro.Encrypt;
using JiShe.ServicePro.Enums;
using JiShe.ServicePro.FreeRedisProvider;
using JiShe.ServicePro.IoTDBManagement.DataChannels;
using JiShe.ServicePro.IoTDBManagement.TreeModels;
using Mapster;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Collections.Generic;
using Volo.Abp;
using Volo.Abp.Auditing;
namespace JiShe.IoT.BusinessSystemAggregation
{
///
/// 业务系统聚合服务
///
[DisableAuditing]
public class BusinessSystemAggregationService(IOptions options, IReliableRedisPubSubService redisPubSubService, IDeviceAppService deviceAppService, IIoTDBDataChannelManageService ioTDBDataChannelManageService, IOptions _ioTDBOptions, ITreeModelService treeModelService, IDeviceAggregationService deviceAggregationService, ILogger _logger, IIoTPlatformThingModelInfoAppService platformThingModelInfoAppService) : IoTAppService, IBusinessSystemAggregationService
{
ServerApplicationOptions serverOptions = options.Value;
IoTDBOptions ioTDBOptions = _ioTDBOptions.Value;
///
/// 接收业务系统指令信息
///
[AllowAnonymous]
public async Task ReceiveCommandInfoAsync(OpenApiRequest input)
{
try
{
var handleResult = HandleOpenApiRequest(input, serverOptions);
if (handleResult.Success == false)
{
return handleResult;
}
var messageBody = handleResult.Data;
if (messageBody == null || messageBody.Commands == null || messageBody.Commands.Count <= 0)
{
return HttpDataResultExtensions.Failed("设备指令不能为空", -103, ResponeResultEnum.Fail);
}
//限定来源类型必须为业务系统
if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem)
{
return HttpDataResultExtensions.Failed("设备指令来源类型错误,业务系统传固定值2", -104, ResponeResultEnum.Fail);
}
var deviceInfo = await deviceAppService.FindByDeviceAddressAsync(messageBody.DeviceAddress);
if (deviceInfo == null)
{
return HttpDataResultExtensions.Failed("设备不存在", -1041, ResponeResultEnum.Fail);
}
var packetTaskInfo = GetDeviceTelemetryPacketTaskInfo(ioTDBOptions, input, deviceInfo.Adapt(), messageBody.Commands.Serialize());
//将指令存储IoTDB数据库和Redis发布通道
if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
{
//获取设备对应的平台端物模型信息,校验前端传入的属性标识集合是否存在不合法的属性标识符
var platformThingModelInfo = await platformThingModelInfoAppService.FindByPlatformProductIdAsync(new IdInput() { Id = deviceInfo.IoTPlatformProductId });
if (platformThingModelInfo == null)
{
throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}的平台端物模型信息不存在。");
}
foreach (var item in messageBody.Commands)
{
var tempPlatformThingModelInfo = platformThingModelInfo.Where(d => d.IoTPlatformRawFieldName == item.Key).FirstOrDefault();
if (tempPlatformThingModelInfo == null)
{
throw new UserFriendlyException($"业务系统推送指令时设备设备{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。");
}
//排除升级指令
if (tempPlatformThingModelInfo.StandardFieldName.ToLowerInvariant() == ThingModelFixedTypeConst.FIRMWARE_UPGRADE.ToLowerInvariant())
{
throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是升级指令操作,被禁止。");
}
if (deviceInfo.IsNeedConfigDevicMdoel && deviceInfo.DeviceThingModelDataId.HasValue && item.Key.ToLowerInvariant() == ThingModelFixedTypeConst.SpecialCommand.ToLowerInvariant())
{
throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是特殊指令操作,被禁止。");
}
}
//数据写入遥测任务数据存储通道
await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETCommandIssuedEventName, packetTaskInfo));
await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.OneNETCommandIssuedEventName, input);
}
else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing)
{
await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName, input);
}
else
{
return HttpDataResultExtensions.Failed("指令处理失败,当前设备平台类型异常", -105);
}
return HttpDataResultExtensions.Success("指令下发成功");
}
catch (Exception ex)
{
return HttpDataResultExtensions.Failed($"指令处理失败,发送异常:{ex.Message}", -106);
}
}
///
/// 业务系统批量查询设备数据,Msg 字段为 BatchQueryDeviceDataInfoInput 实体
///
///
///
[AllowAnonymous]
public async Task>> BatchQueryDeviceDataInfoAsync(OpenApiRequest input)
{
try
{
var handleResult = HandleOpenApiRequest(input, serverOptions);
if (handleResult.Success == false)
{
return HttpDataResultExtensions.Failed>(null, handleResult.Message, handleResult.LocationCode);
}
var messageBody = handleResult.Data;
if (messageBody.DeviceAddresses == null || messageBody.DeviceAddresses.Count <= 0)
{
return HttpDataResultExtensions.Failed>(null, "设备地址不能为空", -103);
}
// Lua脚本
string luaScript = @"
local hashKey = KEYS[1]
local fieldKeys = ARGV
return redis.call('HMGET', hashKey, unpack(fieldKeys))";
//执行脚本
var result = await FreeRedisProvider.Instance.EvalAsync
(
luaScript,
new[] { RedisConst.CacheAllDeviceInfoHashKey },
messageBody.DeviceAddresses.ToArray()
);
List deviceCacheInfos = new List();
// 处理返回结果
if (result is object[] values)
{
foreach (var value in values)
{
var tempFocusInfo = ServiceProJsonSerializer.Deserialize(value as string);
deviceCacheInfos.Add(tempFocusInfo);
}
}
List queryResult = new List();
foreach (var deviceAddress in messageBody.DeviceAddresses)
{
var deviceCacheInfo = deviceCacheInfos.FirstOrDefault(x => x.DeviceAddress == deviceAddress);
if (deviceCacheInfo == null)
{
_logger.LogError($"{nameof(BatchQueryDeviceDataInfoAsync)} 业务系统批量查询设备数据,设备地址:{deviceAddress}未找到设备地址缓存信息,消息体为:{input.Serialize()}");
continue;
}
var pageResult = await treeModelService.OpenRequestDeviceDataInfoPageAsync(new DeviceTreeModelDataInfoInput()
{
DeviceAddress = deviceAddress,
DeviceType = messageBody.DeviceType,
IoTDataType = messageBody.IoTDataType,
IsNeedPaging = false,
StartCreationTime = messageBody.BeginTime,
EndCreationTime = messageBody.EndTime
});
//todo 根据业务系统时间间隔要求进行过滤
if (pageResult.Items != null && pageResult.Items.Count > 0)
{
queryResult.AddRange(pageResult.Items);
}
}
return HttpDataResultExtensions.Success(queryResult, "查询成功");
}
catch (Exception ex)
{
return HttpDataResultExtensions.Failed>(null, $"查询设备数据失败,发送异常:{ex.Message}", -106);
}
}
///
/// 业务系统查询单个设备数据,Msg 字段为 QueryDeviceDataInfoInput 实体
///
///
///
[AllowAnonymous]
public async Task>> QueryDeviceDataInfoAsync(OpenApiRequest input)
{
try
{
var handleResult = HandleOpenApiRequest(input, serverOptions);
if (handleResult.Success == false)
{
return HttpDataResultExtensions.Failed>(null, handleResult.Message, handleResult.LocationCode);
}
var messageBody = handleResult.Data;
if (string.IsNullOrWhiteSpace(messageBody.DeviceAddress))
{
return HttpDataResultExtensions.Failed>(null, "设备地址不能为空", -103);
}
// Lua脚本
string luaScript = @"
local hashKey = KEYS[1]
local fieldKeys = ARGV
return redis.call('HMGET', hashKey, unpack(fieldKeys))";
//执行脚本
var result = await FreeRedisProvider.Instance.EvalAsync
(
luaScript,
new[] { RedisConst.CacheAllDeviceInfoHashKey },
new List() { messageBody.DeviceAddress }.ToArray()
);
List deviceCacheInfos = new List();
// 处理返回结果
if (result is object[] values)
{
foreach (var value in values)
{
var tempFocusInfo = ServiceProJsonSerializer.Deserialize(value as string);
deviceCacheInfos.Add(tempFocusInfo);
}
}
List queryResult = new List();
var deviceCacheInfo = deviceCacheInfos.FirstOrDefault(x => x.DeviceAddress == messageBody.DeviceAddress);
if (deviceCacheInfo == null)
{
_logger.LogError($"{nameof(BatchQueryDeviceDataInfoAsync)} 业务系统单个查询设备数据,设备地址:{messageBody.DeviceAddress}未找到设备地址缓存信息,消息体为:{input.Serialize()}");
}
var pageResult = await treeModelService.OpenRequestDeviceDataInfoPageAsync(new DeviceTreeModelDataInfoInput()
{
DeviceAddress = messageBody.DeviceAddress,
DeviceType = messageBody.DeviceType,
IoTDataType = messageBody.IoTDataType,
IsNeedPaging = false,
StartCreationTime = messageBody.BeginTime,
EndCreationTime = messageBody.EndTime,
SubDeviceAddress = messageBody.SubDeviceAddress,
});
//todo 根据业务系统时间间隔要求进行过滤
if (pageResult.Items != null && pageResult.Items.Count > 0)
{
queryResult.AddRange(pageResult.Items);
}
return HttpDataResultExtensions.Success(queryResult, "查询成功");
}
catch (Exception ex)
{
return HttpDataResultExtensions.Failed>(null, $"查询设备数据失败,发送异常:{ex.Message}", -106);
}
}
///
/// 业务系统批量新增设备数据,Msg 字段为 BatchCreateDeviceInfoInput 实体
///
///
///
[AllowAnonymous]
public async Task BatchCreateDeviceInfoAsync(OpenApiRequest input)
{
try
{
var handleResult = HandleOpenApiRequest(input, serverOptions);
if (handleResult.Success == false)
{
return HttpDataResultExtensions.Failed(handleResult.Message, handleResult.LocationCode);
}
var messageBody = handleResult.Data;
if (messageBody.DeviceAddresses == null || messageBody.DeviceAddresses.Count <= 0)
{
return HttpDataResultExtensions.Failed("设备地址不能为空", -101);
}
var createResult = await deviceAggregationService.BatchCreateDeviceBusinessSystemAsync(new BatchCreateDeviceAggregationInput()
{
DeviceSourceTypeEnum = messageBody.DeviceSourceType,
IoTPlatform = messageBody.IoTPlatform,
IoTPlatformProductId = messageBody.IoTPlatformProductId,
AddressList = messageBody.DeviceAddresses
});
if (createResult == false)
{
_logger.LogError($"{nameof(BatchCreateDeviceInfoAsync)} 业务系统批量新增设备数据没有成功,消息体为:{input.Serialize()}");
return HttpDataResultExtensions.Failed("新增设备失败", -102);
}
return HttpDataResultExtensions.Success("新增设备成功");
}
catch (Exception ex)
{
return HttpDataResultExtensions.Failed($"查询设备数据失败,发送异常:{ex.Message}", -106);
}
}
}
}