HealthMonitor/HealthMonitor.Service/Resolver/BloodpressResolver.cs

using HealthMonitor.Common;
using HealthMonitor.Service.Biz.db;
using HealthMonitor.Service.Cache;
using HealthMonitor.Service.Resolver.Interface;
using HealthMonitor.Service.Sub;
using HealthMonitor.Service.Sub.Topic.Model;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Data.Common;
using System.Linq;
using System.Text;
using System.Text.Json.Serialization;
using System.Threading.Tasks;
using TDengineTMQ;
using TelpoDataService.Util.Entities.GpsLocationHistory;

namespace HealthMonitor.Service.Resolver
{
    
    public class BloodpressResolver: IResolver
    {
        private readonly ILogger<BloodpressResolver> _logger;
        private readonly PersonCacheManager _personCacheMgr;
        private readonly TDengineService _serviceTDengine;
        private readonly BloodPressReferenceValueCacheManager _bpRefValCacheManager;

        private readonly AsyncLocal<string> _messageId = new();
        private readonly AsyncLocal<HisGpsBloodPress> _msgData = new();

        public BloodpressResolver(
            TDengineService serviceDengine,
            BloodPressReferenceValueCacheManager bpRefValCacheManager,
            PersonCacheManager personCacheMgr,
            ILogger<BloodpressResolver> logger) 
        {
            _serviceTDengine = serviceDengine;
            _bpRefValCacheManager = bpRefValCacheManager;
            _logger = logger;
            _personCacheMgr = personCacheMgr;
        }

        public void SetResolveInfo(PackageMsgModel msg)
        {
            var topicHmBloodPress = JsonConvert.DeserializeObject<TopicHmBloodPress>(msg.DetailData.ToString()!);
            _messageId.Value = msg.MessageId;
            _msgData.Value = new HisGpsBloodPress()
            {
                BloodPressId = topicHmBloodPress!.BloodPressId,
                MessageId = topicHmBloodPress!.MessageId,
                Serialno= topicHmBloodPress!.Serialno,
                SystolicValue = topicHmBloodPress!.SystolicValue,
                DiastolicValue= topicHmBloodPress!.DiastolicValue,
                LastUpdate= DateTimeUtil.GetDateTimeFromUnixTimeMilliseconds(SafeType.SafeInt64(topicHmBloodPress.LastUpdate) / 1000000),
                CreateTime= DateTimeUtil.GetDateTimeFromUnixTimeMilliseconds(SafeType.SafeInt64(topicHmBloodPress.CreateTime) / 1000000),
                Method= topicHmBloodPress!.Method,
                IsDisplay=topicHmBloodPress!.IsDisplay ? 1 : 0
            };
        }

        public override string ToString()
        {
            return $"{nameof(BloodpressResolver)}[{_messageId.Value}]";
        }

        public async Task ExecuteMessageAsync()
        {
            var messageId = _messageId.Value;
            var bp = _msgData.Value!;

            #region 获取个人信息

            var person = await _personCacheMgr.GetDeviceGpsPersonCacheBySerialNoAsync(bp.MessageId, bp.Serialno).ConfigureAwait(false);
            //验证这个信息是否存在
            if (person == null || person?.Person.BornDate == null)
            {
                _logger.LogWarning("验证个人信息，找不到个人信息，跳过此消息");
                return;
            }
            // 验证年龄是否在范围 （2 - 120）
            var age = SafeType.SafeInt(DateTime.Today.Year - person?.Person.BornDate!.Value.Year!);
            if (age < 1 || age > 120)
            {
                _logger.LogWarning("验证年龄，不在范围 （2 - 120）岁，跳过此消息");
                return;
            }
            #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 = 2;
            // 获取历史数据
            //DateTime now = DateTime.Now;
            DateTime now = (DateTime)bp.LastUpdate!; //测试
            DateTime startTime = now.AddDays(-duration);
            DateTime endTime = now;

            //
            var systolicAggregate = await _serviceTDengine.GetAggregateValueAsync("systolic_value", "stb_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.GetAggregateValueAsync("diastolic_value", "stb_hm_bloodpress", $"ts>='{startTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and ts <='{endTime:yyyy-MM-ddTHH:mm:ss.fffZ}' and serialno='{bp.Serialno}'");
            //var systolicAggregate =  _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 =  _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.GetAvgExceptMaxMinValueAsync("systolic_value", "stb_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.GetAvgExceptMaxMinValueAsync("diastolic_value", "stb_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}");

            if (systolicAvg.Equals(0)|| diastolicAvg.Equals(0))
            {
                _logger.LogWarning("平均值为0不保存");
                return;
            }


            //var systolicAvg =  _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 =  _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_bp_stats_inc_{bp.Serialno.Substring(bp.Serialno.Length - 2)} " +
                $"USING health_monitor.stb_hm_bloodpress_stats_inc " +
                $"TAGS ('{bp.Serialno.Substring(bp.Serialno.Length - 2)}')  " +
                $"VALUES(" +
                    $"'{bp.LastUpdate:yyyy-MM-dd HH:mm:ss.fff}'," +
                    $"'{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-dd HH:mm:ss.fff}'," +
                    $"{duration}," +
                    $"'{startTime:yyyy-MM-dd HH:mm:ss.fff}'," +
                    $"'{endTime:yyyy-MM-dd HH:mm:ss.fff}'," +
                    $"'{string.Empty}'," +
                    $"{isHypertension})";

            _serviceTDengine.ExecuteInsertSQL(sql);
            #endregion
        }


    }
}