|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661 |
- using dotnet_etcd;
- using Etcdserverpb;
- using Google.Protobuf.WellKnownTypes;
- using HealthMonitor.Common;
- using HealthMonitor.Common.helper;
- using HealthMonitor.Core.Common.Extensions;
- using HealthMonitor.Model.Config;
- using HealthMonitor.Model.Service;
- using HealthMonitor.Model.Service.Mapper;
- using HealthMonitor.Service.Biz;
- using HealthMonitor.Service.Biz.db;
- using HealthMonitor.Service.Etcd;
- using HealthMonitor.Service.Sub;
- using Microsoft.AspNetCore.Mvc.RazorPages;
- using Microsoft.EntityFrameworkCore.Metadata.Internal;
- using Microsoft.Extensions.Options;
- using Newtonsoft.Json;
- using Newtonsoft.Json.Linq;
- using System.Reflection;
- using System.Threading.Channels;
- using TDengineDriver;
- using TDengineTMQ;
-
- namespace HealthMonitor.WebApi
- {
- public class Worker : BackgroundService
- {
- private readonly ILogger<Worker> _logger;
- private readonly IServiceProvider _services;
- private readonly TDengineDataSubcribe _tdEngineDataSubcribe;
- private readonly PackageProcess _processor;
- private readonly TDengineService _serviceTDengine;
- private readonly EtcdService _serviceEtcd;
- private readonly HttpHelper _httpHelper = default!;
- private readonly IotWebApiService _serviceIotWebApi;
- private readonly BoodPressResolverConfig _configBoodPressResolver;
-
- private CancellationTokenSource _tokenSource=default!;
-
- public Worker(ILogger<Worker> logger, IServiceProvider services, IotWebApiService iotWebApiService, IOptions<BoodPressResolverConfig> optionBoodPressResolver, PackageProcess processor,TDengineDataSubcribe tdEngineDataSubcribe, TDengineService serviceDengine, HttpHelper httpHelper, EtcdService serviceEtcd)
- {
- _logger = logger;
- _tdEngineDataSubcribe = tdEngineDataSubcribe;
- _services = services;
- _serviceIotWebApi = iotWebApiService;
- _processor = processor;
- _serviceEtcd = serviceEtcd;
- _serviceTDengine = serviceDengine;
- _httpHelper = httpHelper;
- _configBoodPressResolver = optionBoodPressResolver.Value;
- }
-
- public override Task StartAsync(CancellationToken cancellationToken)
- {
- _logger.LogInformation("------StartAsync");
- _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-
-
- return base.StartAsync(cancellationToken);
- }
-
- public override Task StopAsync(CancellationToken cancellationToken)
- {
- _logger.LogInformation("------StopAsync");
- _tokenSource.Cancel();
-
-
- return base.StopAsync(cancellationToken);
- }
-
- protected override Task ExecuteAsync(CancellationToken stoppingToken)
- {
-
- TaskFactory factory = new(_tokenSource.Token);
- factory.StartNew(async () =>
- {
- if (_tokenSource.IsCancellationRequested)
- _logger.LogWarning("Worker exit");
-
- _logger.LogInformation("------ResolveAsync");
- while (!_tokenSource.IsCancellationRequested)
- {
-
- await _processor.ResolveAsync().ConfigureAwait(false);
-
- }
-
- }, TaskCreationOptions.LongRunning);
-
- factory.StartNew(() =>
- {
- _logger.LogInformation("------_tdEngineDataSubcribe");
- while (!_tokenSource.IsCancellationRequested)
- {
-
- _tdEngineDataSubcribe.BeginListen(_tokenSource.Token);
- }
- }, TaskCreationOptions.LongRunning);
-
- Task.Run(() =>
- _serviceEtcd.WacthKeysWithPrefixResponseAsync($"health_moniter/schedule_push", WatchEvents)
- , stoppingToken);
-
-
-
-
-
-
-
-
-
-
-
-
- return Task.Delay(1000, _tokenSource.Token);
-
- }
-
- private void WatchEvents(WatchEvent[] response)
- {
- foreach (WatchEvent e1 in response)
- {
-
-
- switch (e1.Type.ToString())
- {
-
-
-
-
- case "Delete":
-
- Console.WriteLine($"--- {e1.Key}:{e1.Value}:{e1.Type}");
- break;
- }
- }
- }
-
-
- private void WatchEvents(WatchResponse response)
- {
-
- response.Events.ToList().ForEach(async e =>
- {
-
- switch (e.Type.ToString())
- {
- case "Put":
-
- Console.BackgroundColor = ConsoleColor.Blue;
- Console.WriteLine($"--- {e.Type}--{e.Kv.Key.ToStringUtf8()}--{e.Kv.Value.ToStringUtf8()}---{DateTime.Now}");
- Console.BackgroundColor = ConsoleColor.Black;
- break;
-
- case "Delete":
-
- Console.BackgroundColor = ConsoleColor.Green;
- Console.WriteLine($"--- {e.Type}--{e.Kv.Key.ToStringUtf8()}--{e.Kv.Value.ToStringUtf8()}---{DateTime.Now}");
-
-
- var key = e.Kv.Key.ToStringUtf8();
- var imeiDel = key.Split('/')[3];
- var schedule_push = await _serviceEtcd.GetValAsync(key).ConfigureAwait(false);
- if (string.IsNullOrWhiteSpace(schedule_push))
- {
- int systolicInc;
- int diastolicInc;
-
- int systolicRefValue;
- int diastolicRefValue;
-
- int systolicAvg;
- int diastolicAvg;
-
- int systolicMax = 0;
- int diastolicMax = 0;
-
- int systolicMin = 0;
- int diastolicMin = 0;
-
-
- var avgOffset = 0.25M;
- var systolicAvgOffset = avgOffset;
- var diastolicAvgOffset = avgOffset;
-
-
- int lastPushSystolicInc = 0;
- int lastPushDiastolicInc = 0;
-
-
- var startTime = DateTime.Now;
-
- #region 统计定时下发增量值
-
-
-
-
- var condition = $"serialno='{imeiDel}' order by last_update desc";
- var field = "last_row(*)";
- var lastHmBpResponse = await _serviceTDengine.ExecuteSelectRestResponseAsync("stb_hm_bloodpress", condition, field);
- var lastHmBpParser = JsonConvert.DeserializeObject<ParseTDengineRestResponse<BloodPressureModel>>(lastHmBpResponse!);
- var lastHmBp = lastHmBpParser?.Select().FirstOrDefault();
-
-
-
-
-
-
-
- if (lastHmBp?.Timestamp.AddDays(7) > DateTime.Now)
- {
-
- condition= $"serialno='{imeiDel}' order by ts desc";
- var lastPushResponse = await _serviceTDengine.ExecuteSelectRestResponseAsync("stb_hm_bp_push_ref_inc_value", condition, field);
- if (lastPushResponse == null)
- {
- _logger.LogInformation($"{imeiDel}--没有下发记录");
- break;
- }
- var lastPushParser = JsonConvert.DeserializeObject<ParseTDengineRestResponse<BloodPressurePushRefIncModel>>(lastPushResponse);
- var lastPush = lastPushParser!.Select().FirstOrDefault();
-
- systolicRefValue = lastPush!.SystolicRefValue;
- diastolicRefValue = lastPush!.DiastolicRefValue;
-
- condition = $"ts between '{lastPush?.Timestamp:yyyy-MM-dd HH:mm:ss.fff}' and '{startTime:yyyy-MM-dd HH:mm:ss.fff}' " +
- $"and serialno='{imeiDel}' " +
- $"and is_display = true";
- var hmBpResponse = await _serviceTDengine.ExecuteSelectRestResponseAsync("stb_hm_bloodpress", condition);
- var hmBpParser = JsonConvert.DeserializeObject<ParseTDengineRestResponse<BloodPressureModel>>(hmBpResponse!);
- var hmBp = hmBpParser?.Select();
- if (hmBp?.ToList().Count < 2)
- {
- _logger.LogInformation($"{imeiDel} -- {nameof(Worker)} --{nameof(WatchEvents)} -- 统计定时下发,计算增量值的数据条目不足");
- break;
- }
-
-
- systolicMax = (int)hmBpParser?.Select(i => i.SystolicValue).Max()!;
- diastolicMax = (int)hmBpParser?.Select(i => i.DiastolicValue).Max()!;
-
- systolicMin = (int)hmBpParser?.Select(i => i.SystolicValue).Min()!;
- diastolicMin = (int)hmBpParser?.Select(i => i.DiastolicValue).Min()!;
-
- systolicAvg = (int)(hmBpParser?.AverageAfterRemovingOneMinMaxRef(i => i.SystolicValue, SafeType.SafeInt(systolicRefValue!)))!;
- diastolicAvg = (int)(hmBpParser?.AverageAfterRemovingOneMinMaxRef(i => i.DiastolicValue, SafeType.SafeInt(diastolicRefValue!)))!;
-
-
- var currentSystolicInc = (int)((systolicRefValue - systolicAvg) * systolicAvgOffset)!;
- var currentDiastolicInc = (int)((diastolicRefValue - diastolicAvg) * diastolicAvgOffset)!;
-
-
- systolicInc = currentSystolicInc + lastPushSystolicInc;
- diastolicInc = currentDiastolicInc + lastPushDiastolicInc;
-
- _logger.LogInformation($"{nameof(Worker)} 开启血压标定值下发: {_configBoodPressResolver.EnableBPRefPush}");
- if (_configBoodPressResolver.EnableBPRefPush)
-
- {
- BloodPressCalibrationConfigModel bpIncData = new()
- {
-
- Imei = imeiDel,
- SystolicRefValue = SafeType.SafeInt(((int)systolicRefValue!)),
- DiastolicRefValue = SafeType.SafeInt(((int)diastolicRefValue!)),
- SystolicIncValue = SafeType.SafeInt(((int)systolicInc!)),
- DiastolicIncValue = SafeType.SafeInt(((int)diastolicInc!))
- };
- var pushedBP = await _serviceIotWebApi.SetBloodPressCalibrationConfigAsync(bpIncData).ConfigureAwait(false);
- if (pushedBP)
- {
-
- #region 保存下推记录 stb_hm_bp_push_ref_inc_value
- var sql = $"INSERT INTO health_monitor.hm_bp_push_ref_inc_value_{imeiDel.Substring(imeiDel.Length - 2)} " +
- $"USING health_monitor.stb_hm_bp_push_ref_inc_value " +
- $"TAGS ('{imeiDel.Substring(imeiDel.Length - 2)}') " +
- $"VALUES(" +
- $"'{startTime:yyyy-MM-dd HH:mm:ss.fff}'," +
- $"'{imeiDel}'," +
- $"{bpIncData.SystolicRefValue}," +
- $"{bpIncData.DiastolicRefValue}," +
- $"{bpIncData.SystolicIncValue}," +
- $"{bpIncData.DiastolicIncValue}," +
- $"{false}," +
- $"{systolicAvg}," +
- $"{diastolicAvg}," +
- $"{systolicAvgOffset}," +
- $"{diastolicAvgOffset}," +
- $"'{lastPush?.Timestamp:yyyy-MM-dd HH:mm:ss.fff}'," +
- $"'{startTime:yyyy-MM-dd HH:mm:ss.fff}'" +
- $")";
- _serviceTDengine.ExecuteInsertSQL(sql);
- #endregion
-
- #region 注册定时下发
-
- var endTime = DateTime.Now;
-
- #if DEBUG
-
-
-
-
-
- var interval = 0;
-
- DateTime now = DateTime.Now;
-
-
- DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, now.Hour, now.Minute + 2, 0).AddDays(interval);
- TimeSpan timeUntilNextRun = nextRunTime - now;
-
-
- if (timeUntilNextRun < TimeSpan.Zero)
- {
- timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromMinutes(1));
- nextRunTime += timeUntilNextRun;
- }
-
-
- long ttl = (long)((timeUntilNextRun.TotalMilliseconds - (endTime - startTime).TotalMilliseconds) / 1000);
- var data = new
- {
- imei = imeiDel,
- 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
-
- var interval = 1;
-
- DateTime now = DateTime.Now;
-
-
- DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, 20, 0, 0).AddDays(interval);
- TimeSpan timeUntilNextRun = nextRunTime - now;
-
-
- if (timeUntilNextRun < TimeSpan.Zero)
- {
- timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromDays(1));
- nextRunTime += timeUntilNextRun;
- }
-
-
- long ttl = (long)((timeUntilNextRun.TotalMilliseconds-(endTime-startTime).TotalMilliseconds)/1000);
- var data = new
- {
- imei = imeiDel,
- 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
- #endregion
-
- }
- else
- {
- _logger.LogInformation($"错误响应,没有下推数据");
- }
- }
- }
- else
- {
- _logger.LogInformation($"向{imeiDel}统计数据已经失效");
- }
- #endregion
-
-
- }
-
- break;
- }
-
- });
- }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- }
- }
|