HealthMonitor/HealthMonitor.Service/Resolver/BloodpressResolver.cs


using HealthMonitor.Common;
using HealthMonitor.Common.helper;
using HealthMonitor.Service.Biz.db;
using HealthMonitor.Service.Cache;
using HealthMonitor.Service.Etcd;
using HealthMonitor.Service.Resolver.Interface;
using HealthMonitor.Service.Sub;
using HealthMonitor.Service.Sub.Topic.Model;
using Microsoft.EntityFrameworkCore.Metadata;
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.GpsCard;
using TelpoDataService.Util;
using TelpoDataService.Util.Entities.GpsLocationHistory;
using HealthMonitor.Service.Biz;
using HealthMonitor.Model.Service;

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 HttpHelper _httpHelper = default!;

        private readonly GpsCardAccessorClient<GpsPerson> _gpsPersonApiClient;
        private readonly IotWebApiService _serviceIotWebApi;

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

        private readonly EtcdService _serviceEtcd;

        public BloodpressResolver(
            TDengineService serviceDengine,
            BloodPressReferenceValueCacheManager bpRefValCacheManager,
            PersonCacheManager personCacheMgr, HttpHelper httpHelper,
            GpsCardAccessorClient<GpsPerson> gpsPersonApiClient,
            IotWebApiService iotWebApiService,
            EtcdService serviceEtcd,

            ILogger<BloodpressResolver> logger) 
        {
            _httpHelper = httpHelper;
            _serviceTDengine = serviceDengine;
            _bpRefValCacheManager = bpRefValCacheManager;
            _gpsPersonApiClient = gpsPersonApiClient;
            _serviceIotWebApi = iotWebApiService;
            _logger = logger;
            _personCacheMgr = personCacheMgr;
            _serviceEtcd = serviceEtcd;

        }

        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 lastPush = await _serviceTDengine.GetLastAsync("stb_hm_bp_push_ref_inc_value", $"serialno='{bp.Serialno}' order by ts desc");
            //var ts = last?[0];
            var systolicRefValue = bpRef?.Systolic;//?
            var diastolicRefValue = bpRef?.Diastolic;//?
            long duration = 7 * 24 * 3600 * 1000;

            int systolicInc;
            int diastolicInc;
            string sql = string.Empty;

            var remarkFlag = false;

            // 曾经有下发记录
            if (lastPush?.Count!=0)
            {
                // 重置设备，取正常值标定值
                if (
                   (int)lastPush![2] == 0
                   && (int)lastPush![3] == 0
                   && (int)lastPush![4] == 0
                   && (int)lastPush![5] == 0
                   )
                {
                     systolicRefValue = bpRef?.Systolic;//?
                     diastolicRefValue = bpRef?.Diastolic;//?
                }
                // 取最后一条下推的标定值
                else
                {
                    systolicRefValue = (int)lastPush?[2]!;
                    diastolicRefValue = (int)lastPush?[3]!;
                }

                if (DateTime.TryParse(lastPush?[0]!.ToString(), out DateTime newTs))
                {
                    duration = SafeType.SafeInt64(((DateTime)bp.LastUpdate! - newTs).TotalMilliseconds);
                }

            }

            
            TimeSpan ts= TimeSpan.FromMilliseconds(duration);    
            // 获取历史数据
            ////DateTime now = DateTime.Now;
            //DateTime now = (DateTime)bp.LastUpdate!; //测试
            //DateTime startTime = now.AddDays(-duration);
            //DateTime endTime = now;

            DateTime endTime = (DateTime)bp.LastUpdate!; //测试
            DateTime startTime = endTime-ts;


            // 如果hm_bp_config_manual_calibration存在数据,使用最新数据

            //
            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 avgOffset = 0.25M;
            var systolicAvgOffset = avgOffset;
            var diastolicAvgOffset = avgOffset;


            // 计算去除最大值和最小值和异常值的平均值
            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不保存");
                systolicAvg = bp.SystolicValue;
                diastolicAvg = bp.DiastolicValue;
                systolicInc = (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)! > 0? (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!:0;
                diastolicInc = (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)! >0? (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!:0;
               
                // 初始化 remark
                remarkFlag = await _serviceIotWebApi.UpdatePersonRemarksAsync(bp.Serialno, (int)systolicRefValue!, (int)diastolicRefValue!, systolicInc, diastolicInc).ConfigureAwait(false);
                if (remarkFlag)
                {
                    // 下推
                    BloodPressCalibrationConfigModel bpIncData = new()
                    {
                        Imei = bp.Serialno,
                        SystolicRefValue = (int)systolicRefValue!,  //收缩压标定值，值为0 表示不生效
                        DiastolicRefValue = (int)diastolicRefValue!, //舒张压标定值，值为0表示不生效
                        SystolicIncValue = systolicInc, //收缩压显示增量，值为0 表示不生效
                        DiastolicIncValue = diastolicInc   //舒张压显示增量，值为0 表示不生效
                    };
                    // 下发 IOT 增量值
                    var flagIot = await _serviceIotWebApi.SetBloodPressCalibrationConfigAsync(bpIncData).ConfigureAwait(false);
                    if (flagIot)
                    {
                        #region 保存下推记录 stb_hm_bp_push_ref_inc_value
                        sql = $"INSERT INTO health_monitor.hm_bp_push_ref_inc_value_{bp.Serialno.Substring(bp.Serialno.Length - 2)} " +
                                $"USING health_monitor.stb_hm_bp_push_ref_inc_value " +
                                $"TAGS ('{bp.Serialno.Substring(bp.Serialno.Length - 2)}')  " +
                                $"VALUES(" +
                                    $"'{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}'," +
                                    $"'{bp.Serialno}'," +
                                    $"{systolicRefValue}," +
                                    $"{diastolicRefValue}," +
                                    $"{systolicInc}," +
                                    $"{diastolicInc}," +
                                    $"{true})";

                        _serviceTDengine.ExecuteInsertSQL(sql);
                        #endregion
                    }
                }

                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}");

            // 增量值=（标定值-平均值）* 0.25
             systolicInc = systolicAvg.Equals(0M) ? 0 : (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!;
             diastolicInc = diastolicAvg.Equals(0M) ? 0 : (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!;
             
            #endregion

            #region 插入BP增量值 hm_bloodpress_stats_inc
            // 自动建表
            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
            // 初始化 remark
            remarkFlag = await _serviceIotWebApi.UpdatePersonRemarksAsync(bp.Serialno, (int)systolicRefValue!, (int)diastolicRefValue!, systolicInc, diastolicInc).ConfigureAwait(false);
            if (remarkFlag)
            {
                // 下推
                BloodPressCalibrationConfigModel bpIncData = new()
                {

                    Imei = bp.Serialno,
                    SystolicRefValue = (int)systolicRefValue!,  //收缩压标定值，值为0 表示不生效
                    DiastolicRefValue = (int)diastolicRefValue!, //舒张压标定值，值为0表示不生效
                    SystolicIncValue = systolicInc, //收缩压显示增量，值为0 表示不生效
                    DiastolicIncValue = diastolicInc   //舒张压显示增量，值为0 表示不生效
                };
                // 下发 IOT 增量值
                var flagIot = await _serviceIotWebApi.SetBloodPressCalibrationConfigAsync(bpIncData).ConfigureAwait(false);
                if (flagIot)
                {
                    #region 保存下推记录 stb_hm_bp_push_ref_inc_value
                    sql = $"INSERT INTO health_monitor.hm_bp_push_ref_inc_value_{bp.Serialno.Substring(bp.Serialno.Length - 2)} " +
                            $"USING health_monitor.stb_hm_bp_push_ref_inc_value " +
                            $"TAGS ('{bp.Serialno.Substring(bp.Serialno.Length - 2)}')  " +
                            $"VALUES(" +
                                $"'{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}'," +
                                $"'{bp.Serialno}'," +
                                $"{systolicRefValue}," +
                                $"{diastolicRefValue}," +
                                $"{systolicInc}," +
                                $"{diastolicInc}," +
                                $"{true})";

                    _serviceTDengine.ExecuteInsertSQL(sql);
                    #endregion
                }
            }

            // 注册定时下发事件
            // 获取当前时间
            //DateTime sNow = DateTime.Now;

            //// 计算距离明天12点的时间间隔
            //TimeSpan timeUntil = new DateTime(sNow.Year, sNow.Month, sNow.Day, 12, 0, 0) - sNow;

            //// 如果当前时间已经超过了12点，将等待到明天
            //if (timeUntil < TimeSpan.Zero)
            //{
            //    timeUntil = timeUntil.Add(TimeSpan.FromHours(24));
            //}
            //var data = new
            //{
            //    imei = bp.Serialno,
            //    systolicCalibrationValue = systolicRefValue,  //收缩压标定值，值为0 表示不生效
            //    diastolicCalibrationValue = diastolicRefValue, //舒张压标定值，值为0表示不生效
            //    systolicIncValue = systolicInc, //收缩压显示增量，值为0 表示不生效
            //    diastolicIncValue = diastolicInc   //舒张压显示增量，值为0 表示不生效
            //};

            //var url = $"http://id.ssjlai.com/webapi/api/Command/SetBloodPressCalibrationConfig";
            //List<KeyValuePair<string, string>> headers = new()
            //{
            //        new KeyValuePair<string, string>("AuthKey", "key1")
            //};

            //var data = new
            //{
            //    imei = bp.Serialno,
            //    systolicCalibrationValue = systolicRefValue,  //收缩压标定值，值为0 表示不生效
            //    diastolicCalibrationValue = diastolicRefValue, //舒张压标定值，值为0表示不生效
            //    systolicIncValue = systolicInc, //收缩压显示增量，值为0 表示不生效
            //    diastolicIncValue = diastolicInc   //舒张压显示增量，值为0 表示不生效
            //};


            //var result = JsonConvert.SerializeObject(data);
            //var result = bp.Serialno;

            var key = $"health_moniter/schedule_push/imei/{bp.Serialno}";
            var schedule_push = await _serviceEtcd.GetValAsync(key).ConfigureAwait(false);
            if (string.IsNullOrWhiteSpace(schedule_push))
            {
                // 注册首次下推
#if DEBUG
                //  await _serviceEtcd.PutValAsync(key, result, 60*1, false).ConfigureAwait(false);

                var interval = 0;
                // 获取当前时间
                DateTime now = DateTime.Now;

                // 计算距离下一个$interval天后的8点的时间间隔
                DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, now.Hour, now.Minute+1, 58).AddDays(interval);
                TimeSpan timeUntilNextRun = nextRunTime - now;

                // 如果当前时间已经超过了8点，将等待到明天后的8点
                if (timeUntilNextRun < TimeSpan.Zero)
                {
                    timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromMinutes(1));
                    nextRunTime += timeUntilNextRun;
                }
                var ttl = (long)timeUntilNextRun.TotalSeconds;
                var data = new
                {
                    imei = bp.Serialno,
                    create_time = now.ToString("yyyy-MM-dd HH:mm:ss"),
                    ttl,
                    next_run_time= nextRunTime.ToString("yyyy-MM-dd HH:mm:ss")
                };
                var result=JsonConvert.SerializeObject(data);
                await _serviceEtcd.PutValAsync(key, result, ttl, false).ConfigureAwait(false);

#else
                //DateTime sNow = DateTime.Now;

                //// 计算距离19:59:55点的时间间隔
                //TimeSpan timeUntil = new DateTime(sNow.Year, sNow.Month, sNow.Day, 19, 59, 55) - sNow;

                //// 如果当前时间已经超过了12点，将等待到明天
                //if (timeUntil < TimeSpan.Zero)
                //{
                //    timeUntil = timeUntil.Add(TimeSpan.FromHours(24));
                //}
                //var ttl = (long)timeUntil.TotalSeconds;


                var interval = 0;
                // 获取当前时间
                DateTime now = DateTime.Now;

                // 计算距离下一个$interval天后的8点的时间间隔
                DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, 19, 59, 58).AddDays(interval);
                TimeSpan timeUntilNextRun = nextRunTime - now;

                // 如果当前时间已经超过了8点，将等待到明天后的8点
                if (timeUntilNextRun < TimeSpan.Zero)
                {
                    timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromDays(1));
                     nextRunTime += timeUntilNextRun;
                }

                var ttl =(long)timeUntilNextRun.TotalSeconds;
                var data = new
                {
                    imei = bp.Serialno,
                    create_time = now.ToString("yyyy-MM-dd HH:mm:ss"),
                    ttl,
                    next_run_time = nextRunTime.ToString("yyyy-MM-dd HH:mm:ss")
                };
                var result = JsonConvert.SerializeObject(data);
                await _serviceEtcd.PutValAsync(key, result,ttl, false).ConfigureAwait(false);
#endif

            }


            //// 生效
            //if (DateTime.Now.Hour == 1)
            //{
            //    var data = new
            //    {
            //        imei = bp.Serialno,
            //        systolicCalibrationValue = systolicRefValue,  //收缩压标定值，值为0 表示不生效
            //        diastolicCalibrationValue = diastolicRefValue, //舒张压标定值，值为0表示不生效
            //        systolicIncValue = systolicInc, //收缩压显示增量，值为0 表示不生效
            //        diastolicIncValue = diastolicInc   //舒张压显示增量，值为0 表示不生效
            //    };
            //    var result = await _httpHelper.HttpToPostAsync(url, data, headers).ConfigureAwait(false);
            //    _logger.LogInformation($"将 {JsonConvert.SerializeObject(data)}发送到 {url} 并且返回 {JsonConvert.SerializeObject(result)}");
            //}
            //// 不生效
            //else if (DateTime.Now.Hour == 3)
            //{
            //    var data = new
            //    {
            //        imei = bp.Serialno,
            //        systolicCalibrationValue = 0,  //收缩压标定值，值为0 表示不生效
            //        diastolicCalibrationValue = 0, //舒张压标定值，值为0表示不生效
            //        systolicIncValue = 0, //收缩压显示增量，值为0 表示不生效
            //        diastolicIncValue = 0   //舒张压显示增量，值为0 表示不生效
            //    };
            //    var result = await _httpHelper.HttpToPostAsync(url, data, headers).ConfigureAwait(false);
            //    _logger.LogInformation($"将 {JsonConvert.SerializeObject(data)}发送到 {url} 并且返回 {JsonConvert.SerializeObject(result)}");
            //}


        }


    }
}