diff --git a/HealthMonitor.Service/Resolver/BloodpressResolver.cs b/HealthMonitor.Service/Resolver/BloodpressResolver.cs index dcf1f9d..6a58305 100644 --- a/HealthMonitor.Service/Resolver/BloodpressResolver.cs +++ b/HealthMonitor.Service/Resolver/BloodpressResolver.cs @@ -73,101 +73,101 @@ namespace HealthMonitor.Service.Resolver Console.WriteLine($"ExecuteMessageAsync" + messageId); var bp = _msgData.Value!; - //#region 获取个人信息 + #region 获取个人信息 var person = await _personCacheMgr.GetDeviceGpsPersonCacheBySerialNoAsync(bp.MessageId, bp.Serialno).ConfigureAwait(false); Console.WriteLine(person?.Person.PersonName); - // 验证这个信息是否存在 - //if (person == null || person?.Person.BornDate == null) - //{ - // Console.WriteLine("验证这个信息是否存在"); - // return; - //} - //// 验证年龄是否在范围 （2 - 120） - //var age = SafeType.SafeInt(DateTime.Today.Year - person?.Person.BornDate!.Value.Year!); - //if (age < 1 || age > 120) - //{ - // Console.WriteLine("验证年龄是否在范围 （2 - 120）"); - //} - //#endregion - - //var gender = person?.Person.Gender == true ? 1 : 2; - //var isHypertension = SafeType.SafeBool(person?.Person.Ishypertension!); - //var height = SafeType.SafeDouble(person?.Person.Height!); - //var weight = SafeType.SafeDouble(person?.Person.Weight!); - - //#region 计算增量值 - //var bpRef = await _bpRefValCacheManager.GetBloodPressReferenceValueAsync(age, gender, isHypertension); - - //var systolicRefValue = bpRef?.Systolic;//? - //var diastolicRefValue = bpRef?.Diastolic;//? - //int duration = 30; - //// 获取历史数据 - ////DateTime now = DateTime.Now; - //DateTime now = (DateTime)bp.LastUpdate!; //测试 - //DateTime startTime = now.AddDays(-duration); - //DateTime endTime = now; - - //// - //var systolicAggregate = await _serviceTDengine.GetAggregateValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'"); - //var diastolicAggregate = await _serviceTDengine.GetAggregateValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'"); - - //// 最大值 - //var systolicMax = systolicAggregate.Max; - //var diastolicMax = diastolicAggregate.Max; - //// 最小值 - //var systolicMin = systolicAggregate.Min; - //var diastolicMin = diastolicAggregate.Min; - - - //// 计算去除最大值和最小值和异常值的平均值 - //var systolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and systolic_value < {systolicRefValue} "); - //var diastolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and diastolic_value < {diastolicRefValue}"); - - //// 偏移参数 - //var avgOffset = 0.25M; - - //var systolicAvgOffset = avgOffset; - //var diastolicAvgOffset = avgOffset; - - - //// 增量值=（标定值-平均值）* 0.25 - //var systolicInc = systolicAvg.Equals(0M) ? 0 : (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!; - //var diastolicInc = diastolicAvg.Equals(0M) ? 0 : (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!; - - //#endregion - - //#region 插入BP增量值 hm_bloodpress_stats_inc - //var sql = $"INSERT INTO health_monitor.hm_bloodpress_stats_inc VALUES(" + - // $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," + - // $"'{bp.BloodPressId}'," + - // $"'{bp.MessageId}'," + - // $"'{bp.Serialno}'," + - // $"{bp.SystolicValue}," + - // $"{systolicRefValue}," + - // $"{systolicAvg}," + - // $"{systolicMax}," + - // $"{systolicMin}," + - // $"{systolicAvgOffset}," + - // $"{systolicInc}," + - // $"{bp.DiastolicValue}," + - // $"{diastolicRefValue}," + - // $"{diastolicAvg}," + - // $"{diastolicMax}," + - // $"{diastolicMin}," + - // $"{diastolicAvgOffset}," + - // $"{diastolicInc}," + - // $"{gender}," + - // $"{age}," + - // $"{height}," + - // $"{weight}," + - // $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," + - // $"{duration}," + - // $"'{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," + - // $"'{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," + - // $"'{string.Empty}')"; - //var res = await _serviceTDengine.GernalRestSql(sql); - //#endregion + //验证这个信息是否存在 + if (person == null || person?.Person.BornDate == null) + { + Console.WriteLine("验证这个信息是否存在"); + return; + } + // 验证年龄是否在范围 （2 - 120） + var age = SafeType.SafeInt(DateTime.Today.Year - person?.Person.BornDate!.Value.Year!); + if (age < 1 || age > 120) + { + Console.WriteLine("验证年龄是否在范围 （2 - 120）"); + } +#endregion + + var gender = person?.Person.Gender == true ? 1 : 2; + var isHypertension = SafeType.SafeBool(person?.Person.Ishypertension!); + var height = SafeType.SafeDouble(person?.Person.Height!); + var weight = SafeType.SafeDouble(person?.Person.Weight!); + + #region 计算增量值 + var bpRef = await _bpRefValCacheManager.GetBloodPressReferenceValueAsync(age, gender, isHypertension); + + var systolicRefValue = bpRef?.Systolic;//? + var diastolicRefValue = bpRef?.Diastolic;//? + int duration = 30; + // 获取历史数据 + //DateTime now = DateTime.Now; + DateTime now = (DateTime)bp.LastUpdate!; //测试 + DateTime startTime = now.AddDays(-duration); + DateTime endTime = now; + + // + var systolicAggregate = await _serviceTDengine.GetAggregateValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'"); + var diastolicAggregate = await _serviceTDengine.GetAggregateValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'"); + + // 最大值 + var systolicMax = systolicAggregate.Max; + var diastolicMax = diastolicAggregate.Max; + // 最小值 + var systolicMin = systolicAggregate.Min; + var diastolicMin = diastolicAggregate.Min; + + + // 计算去除最大值和最小值和异常值的平均值 + var systolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("systolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and systolic_value < {systolicRefValue} "); + var diastolicAvg = await _serviceTDengine.GetAvgExceptMaxMinValue("diastolic_value", "hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}' and diastolic_value < {diastolicRefValue}"); + + // 偏移参数 + var avgOffset = 0.25M; + + var systolicAvgOffset = avgOffset; + var diastolicAvgOffset = avgOffset; + + + // 增量值=（标定值-平均值）* 0.25 + var systolicInc = systolicAvg.Equals(0M) ? 0 : (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!; + var diastolicInc = diastolicAvg.Equals(0M) ? 0 : (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!; + + #endregion + + #region 插入BP增量值 hm_bloodpress_stats_inc + var sql = $"INSERT INTO health_monitor.hm_bloodpress_stats_inc VALUES(" + + $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," + + $"'{bp.BloodPressId}'," + + $"'{bp.MessageId}'," + + $"'{bp.Serialno}'," + + $"{bp.SystolicValue}," + + $"{systolicRefValue}," + + $"{systolicAvg}," + + $"{systolicMax}," + + $"{systolicMin}," + + $"{systolicAvgOffset}," + + $"{systolicInc}," + + $"{bp.DiastolicValue}," + + $"{diastolicRefValue}," + + $"{diastolicAvg}," + + $"{diastolicMax}," + + $"{diastolicMin}," + + $"{diastolicAvgOffset}," + + $"{diastolicInc}," + + $"{gender}," + + $"{age}," + + $"{height}," + + $"{weight}," + + $"'{bp.LastUpdate:yyyy-MM-ddTHH:mm:ss.fffZ}'," + + $"{duration}," + + $"'{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," + + $"'{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}'," + + $"'{string.Empty}')"; + var res = await _serviceTDengine.GernalRestSql(sql); + #endregion } diff --git a/HealthMonitor.Service/Sub/MsgManager.cs b/HealthMonitor.Service/Sub/MsgManager.cs index 54a038b..c41c5ee 100644 --- a/HealthMonitor.Service/Sub/MsgManager.cs +++ b/HealthMonitor.Service/Sub/MsgManager.cs @@ -2,6 +2,7 @@ using HealthMonitor.Service.Resolver.Interface; using Microsoft.Extensions.Logging; using System; +using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; @@ -15,13 +16,15 @@ namespace HealthMonitor.Service.Sub public class MsgManager { + private readonly MsgQueueManager _msgQueueManager; private readonly BloodpressResolver _resolver; private readonly ILogger _logger; private PackageMsgModel _msg = default!; - public MsgManager(ILogger logger, BloodpressResolver resolver) + public MsgManager(ILogger logger, BloodpressResolver resolver, MsgQueueManager msgQueueManager) { _logger = logger; _resolver = resolver; + _msgQueueManager = msgQueueManager; } public void EnqueueMsg(PackageMsgModel msg) @@ -30,18 +33,38 @@ namespace HealthMonitor.Service.Sub Console.WriteLine($"GetMsgResolver-{_msg.MessageId}"); } - public IResolver? GetMsgResolver() { - if (_msg == null) + + //if (_msg == null) + //{ + // return null; + //} + //Console.WriteLine($"GetMsgResolver-{_msg.MessageId}"); + //_msgQueueManager.TryDequeue(out object obj); + _msgQueueManager.TryDequeue(out var msg); + if (msg == null) { return null; } - Console.WriteLine($"GetMsgResolver-{_msg.MessageId}"); - _resolver.SetResolveInfo(_msg); - _msg = default!; + Console.WriteLine($"GetMsgResolver-{msg.MessageId}"); + _resolver.SetResolveInfo(msg); + return _resolver; } + //public IReso + + //public IResolver? GetMsgResolver() + //{ + // if (_msg == null) + // { + // return null; + // } + // Console.WriteLine($"GetMsgResolver-{_msg.MessageId}"); + // _resolver.SetResolveInfo(_msg); + // _msg = default!; + // return _resolver; + //} //public IResolver? GetMsgResolver(PackageMsgModel msg) //{ // if (_msg == null) diff --git a/HealthMonitor.Service/Sub/MsgQueueManager.cs b/HealthMonitor.Service/Sub/MsgQueueManager.cs index 9df1a0e..689d440 100644 --- a/HealthMonitor.Service/Sub/MsgQueueManager.cs +++ b/HealthMonitor.Service/Sub/MsgQueueManager.cs @@ -1,4 +1,6 @@ -using System; +using HealthMonitor.Service.Resolver; +using Microsoft.Extensions.Logging; +using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; @@ -9,6 +11,10 @@ namespace HealthMonitor.Service.Sub { public class MsgQueueManager: ConcurrentQueue { - + private readonly ILogger _logger; + public MsgQueueManager(ILogger logger) + { + _logger = logger; + } } } diff --git a/HealthMonitor.Service/Sub/TDengineDataSubcribe.cs b/HealthMonitor.Service/Sub/TDengineDataSubcribe.cs index 1d5e086..9e09281 100644 --- a/HealthMonitor.Service/Sub/TDengineDataSubcribe.cs +++ b/HealthMonitor.Service/Sub/TDengineDataSubcribe.cs @@ -25,6 +25,7 @@ namespace HealthMonitor.Service.Sub private IConsumer _consumer = default!; private IntPtr _conn = default!; private readonly MsgManager _msgManager; + private readonly MsgQueueManager _msgQueueManager; private readonly TDengineService _serviceTDengine; private readonly PersonCacheManager _personCacheMgr; private readonly BloodPressReferenceValueCacheManager _bpRefValCacheManager; @@ -39,6 +40,7 @@ namespace HealthMonitor.Service.Sub BloodPressReferenceValueCacheManager bpRefValCacheManager, IResolverFactory resolverFactory, MsgManager msgManager, + MsgQueueManager msgQueueManager, ILogger logger ) { @@ -48,6 +50,7 @@ namespace HealthMonitor.Service.Sub _logger = logger; _resolverFactory = resolverFactory; _msgManager = msgManager; + _msgQueueManager = msgQueueManager; _conn = GetConnection(); } public void BeginListen(CancellationToken stoppingToken) @@ -206,8 +209,8 @@ namespace HealthMonitor.Service.Sub Console.WriteLine(msg!.MessageId); if (msg == null) return; // ConcurrentQueue messageQueue = new ConcurrentQueue(); - _msgManager.EnqueueMsg(msg!); - + //_msgManager.EnqueueMsg(msg!); + _msgQueueManager.Enqueue(msg); } diff --git a/HealthMonitor.WebApi/Program.cs b/HealthMonitor.WebApi/Program.cs index 2f46e24..3e7dc06 100644 --- a/HealthMonitor.WebApi/Program.cs +++ b/HealthMonitor.WebApi/Program.cs @@ -166,6 +166,7 @@ namespace HealthMonitor.WebApi #endregion #region Worker + builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton();