選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

1060 行
45KB

  1. using HealthMonitor.Common;
  2. using HealthMonitor.Common.helper;
  3. using HealthMonitor.Model.Config;
  4. using HealthMonitor.Model.Service.Mapper;
  5. using HealthMonitor.Service.Biz.db.Dto;
  6. using HealthMonitor.Util.Models;
  7. using Microsoft.EntityFrameworkCore.Metadata.Internal;
  8. using Microsoft.Extensions.Logging;
  9. using Microsoft.Extensions.Options;
  10. using Newtonsoft.Json;
  11. using Newtonsoft.Json.Linq;
  12. using SqlSugar;
  13. using SqlSugar.DbConvert;
  14. using SqlSugar.TDengine;
  15. using System;
  16. using System.Collections.Generic;
  17. using System.Data;
  18. using System.Linq;
  19. using System.Linq.Expressions;
  20. using System.Reflection;
  21. using System.Text;
  22. using System.Threading.Tasks;
  23. using System.Xml.Linq;
  24. using TDengineDriver;
  25. using TDengineDriver.Impl;
  26. using TDengineTMQ;
  27. using HealthMonitor.Service.Cache;
  28. namespace HealthMonitor.Service.Biz.db
  29. {
  30. public class TDengineService
  31. {
  32. private readonly ILogger<TDengineService> _logger;
  33. private readonly HttpHelper _httpHelper=default!;
  34. private readonly TDengineServiceConfig _configTDengineService;
  35. private readonly SqlSugarClient _clientSqlSugar;
  36. private readonly FhrPhrMapCacheManager _mgrFhrPhrMapCache;
  37. private readonly DeviceCacheManager _deviceCacheMgr;
  38. public TDengineService(ILogger<TDengineService> logger,
  39. IOptions<TDengineServiceConfig> configTDengineService,
  40. HttpHelper httpHelper,
  41. FhrPhrMapCacheManager fhrPhrMapCacheManager, DeviceCacheManager deviceCacheMgr
  42. )
  43. {
  44. _logger = logger;
  45. _configTDengineService = configTDengineService.Value;
  46. _httpHelper = httpHelper;
  47. _mgrFhrPhrMapCache = fhrPhrMapCacheManager;
  48. _deviceCacheMgr = deviceCacheMgr;
  49. _clientSqlSugar = new SqlSugarClient(new ConnectionConfig()
  50. {
  51. DbType = SqlSugar.DbType.TDengine,
  52. ConnectionString = $"Host={_configTDengineService.Host};Port={_configTDengineService.Port};Username={_configTDengineService.UserName};Password={_configTDengineService.Password};Database={_configTDengineService.DB};TsType=config_ns",
  53. IsAutoCloseConnection = true,
  54. AopEvents = new AopEvents
  55. {
  56. OnLogExecuting = (sql, p) =>
  57. {
  58. Console.WriteLine(SqlSugar.UtilMethods.GetNativeSql(sql, p));
  59. }
  60. },
  61. ConfigureExternalServices = new ConfigureExternalServices()
  62. {
  63. EntityService = (property, column) =>
  64. {
  65. if (column.SqlParameterDbType == null)
  66. {
  67. column.SqlParameterDbType = typeof(CommonPropertyConvert);
  68. }
  69. }
  70. },
  71. MoreSettings=new ConnMoreSettings()
  72. {
  73. PgSqlIsAutoToLower = false,
  74. PgSqlIsAutoToLowerCodeFirst = false,
  75. }
  76. });
  77. }
  78. public IntPtr Connection()
  79. {
  80. string host = _configTDengineService.Host;
  81. string user = _configTDengineService.UserName;
  82. string db = _configTDengineService.DB;
  83. short port = _configTDengineService.Port;
  84. string password = _configTDengineService.Password;
  85. //#if DEBUG
  86. // //string configDir = "C:/TDengine/cfg";
  87. // //TDengine.Options((int)TDengineInitOption.TSDB_OPTION_CONFIGDIR, configDir);
  88. // TDengine.Options((int)TDengineInitOption.TSDB_OPTION_TIMEZONE, "Asia/Beijing");
  89. //#endif
  90. IntPtr conn = TDengine.Connect(host, user, password, db, port);
  91. if (conn == IntPtr.Zero)
  92. {
  93. _logger.LogError($"连接 TDengine 失败....");
  94. }
  95. else
  96. {
  97. _logger.LogInformation($"连接 TDengine 成功....");
  98. }
  99. return conn;
  100. }
  101. public void ExecuteSQL(IntPtr conn, string sql)
  102. {
  103. IntPtr res = TDengine.Query(conn, sql);
  104. // Check if query success
  105. if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0))
  106. {
  107. Console.Write(sql + " failure, ");
  108. // Get error message while Res is a not null pointer.
  109. if (res != IntPtr.Zero)
  110. {
  111. Console.Write("reason:" + TDengine.Error(res));
  112. }
  113. }
  114. else
  115. {
  116. Console.Write(sql + " success, {0} rows affected", TDengine.AffectRows(res));
  117. //... do something with res ...
  118. // Important: need to free result to avoid memory leak.
  119. TDengine.FreeResult(res);
  120. }
  121. }
  122. public void ExecuteQuerySQL(IntPtr conn, string sql)
  123. {
  124. IntPtr res = TDengine.Query(conn, sql);
  125. // Check if query success
  126. if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0))
  127. {
  128. Console.Write(sql + " failure, ");
  129. // Get error message while Res is a not null pointer.
  130. if (res != IntPtr.Zero)
  131. {
  132. Console.Write("reason:" + TDengine.Error(res));
  133. }
  134. }
  135. else
  136. {
  137. Console.Write(sql + " success, {0} rows affected", TDengine.AffectRows(res));
  138. //... do something with res ...
  139. List<TDengineDriver.TDengineMeta> resMeta = LibTaos.GetMeta(res);
  140. List<object> resData = LibTaos.GetData(res);
  141. foreach (var meta in resMeta)
  142. {
  143. _logger.LogInformation("\t|{meta.name} {meta.TypeName()} ({meta.size})\t|", meta.name, meta.TypeName(), meta.size);
  144. }
  145. for (int i = 0; i < resData.Count; i++)
  146. {
  147. _logger.LogInformation($"|{resData[i].ToString()} \t");
  148. if (((i + 1) % resMeta.Count == 0))
  149. {
  150. _logger.LogInformation("");
  151. }
  152. }
  153. // Important: need to free result to avoid memory leak.
  154. TDengine.FreeResult(res);
  155. }
  156. }
  157. public void CheckRes(IntPtr conn, IntPtr res, String errorMsg)
  158. {
  159. if (TDengine.ErrorNo(res) != 0)
  160. {
  161. throw new Exception($"{errorMsg} since: {TDengine.Error(res)}");
  162. }
  163. }
  164. public void ExecuteInsertSQL(string sql)
  165. {
  166. var conn = Connection();
  167. try
  168. {
  169. //sql = "INSERT INTO d1001 USING meters TAGS('California.SanFrancisco', 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000) " +
  170. // "d1002 USING power.meters TAGS('California.SanFrancisco', 3) VALUES('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000) " +
  171. // "d1003 USING power.meters TAGS('California.LosAngeles', 2) VALUES('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000)('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000) " +
  172. // "d1004 USING power.meters TAGS('California.LosAngeles', 3) VALUES('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000)('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)";
  173. //#if DEBUG
  174. // //string configDir = "C:/TDengine/cfg";
  175. // //TDengine.Options((int)TDengineInitOption.TSDB_OPTION_CONFIGDIR, configDir);
  176. // TDengine.Options((int)TDengineInitOption.TSDB_OPTION_TIMEZONE, "Asia/Beijing");
  177. //#endif
  178. _logger.LogInformation($"Insert SQL: {sql}");
  179. IntPtr res = TDengine.Query(conn, sql);
  180. CheckRes(conn, res, "failed to insert data");
  181. int affectedRows = TDengine.AffectRows(res);
  182. _logger.LogInformation("affectedRows {affectedRows}" , affectedRows);
  183. TDengine.FreeResult(res);
  184. }
  185. finally
  186. {
  187. TDengine.Close(conn);
  188. }
  189. }
  190. #region TDengine.Connector async query
  191. public void QueryCallback(IntPtr param, IntPtr taosRes, int code)
  192. {
  193. if (code == 0 && taosRes != IntPtr.Zero)
  194. {
  195. FetchRawBlockAsyncCallback fetchRowAsyncCallback = new FetchRawBlockAsyncCallback(FetchRawBlockCallback);
  196. TDengine.FetchRawBlockAsync(taosRes, fetchRowAsyncCallback, param);
  197. }
  198. else
  199. {
  200. _logger.LogInformation("async query data failed, failed code {code}",code);
  201. }
  202. }
  203. // Iteratively call this interface until "numOfRows" is no greater than 0.
  204. public void FetchRawBlockCallback(IntPtr param, IntPtr taosRes, int numOfRows)
  205. {
  206. if (numOfRows > 0)
  207. {
  208. _logger.LogInformation("{numOfRows} rows async retrieved", numOfRows);
  209. IntPtr pdata = TDengine.GetRawBlock(taosRes);
  210. List<TDengineMeta> metaList = TDengine.FetchFields(taosRes);
  211. List<object> dataList = LibTaos.ReadRawBlock(pdata, metaList, numOfRows);
  212. for (int i = 0; i < metaList.Count; i++)
  213. {
  214. _logger.LogInformation("{0} {1}({2}) \t|", metaList[i].name, metaList[i].type, metaList[i].size);
  215. }
  216. _logger.LogInformation("");
  217. for (int i = 0; i < dataList.Count; i++)
  218. {
  219. if (i != 0 && i % metaList.Count == 0)
  220. {
  221. _logger.LogInformation("{dataList[i]}\t|", dataList[i]);
  222. }
  223. _logger.LogInformation("{dataList[i]}\t|", dataList[i]);
  224. }
  225. TDengine.FetchRawBlockAsync(taosRes, FetchRawBlockCallback, param);
  226. }
  227. else
  228. {
  229. if (numOfRows == 0)
  230. {
  231. _logger.LogInformation("async retrieve complete.");
  232. }
  233. else
  234. {
  235. _logger.LogInformation("FetchRawBlockCallback callback error, error code {numOfRows}", numOfRows);
  236. }
  237. TDengine.FreeResult(taosRes);
  238. }
  239. }
  240. public void ExecuteQueryAsync(string sql)
  241. {
  242. var conn = Connection();
  243. QueryAsyncCallback queryAsyncCallback = new QueryAsyncCallback(QueryCallback);
  244. TDengine.QueryAsync(conn, sql, queryAsyncCallback, IntPtr.Zero);
  245. }
  246. //public void ExecuteQuery(string sql)
  247. //{
  248. // var conn = Connection();
  249. // QueryAsyncCallback queryAsyncCallback = new QueryAsyncCallback(QueryCallback);
  250. // TDengine.QueryAsync(conn, sql, queryAsyncCallback, IntPtr.Zero);
  251. //}
  252. public Aggregate GetAggregateValue(string field, string tbName, string? condition)
  253. {
  254. List<int> data = new();
  255. var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  256. var conn = Connection();
  257. try
  258. {
  259. IntPtr res = TDengine.Query(conn, sql);
  260. // Check if query success
  261. if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0))
  262. {
  263. Console.Write(sql + " failure, ");
  264. // Get error message while Res is a not null pointer.
  265. if (res != IntPtr.Zero)
  266. {
  267. Console.Write("reason:" + TDengine.Error(res));
  268. }
  269. }
  270. else
  271. {
  272. Console.Write(sql + " success, {0} rows affected", TDengine.AffectRows(res));
  273. //... do something with res ...
  274. List<TDengineMeta> resMeta = LibTaos.GetMeta(res);
  275. List<object> resData = LibTaos.GetData(res);
  276. foreach (var meta in resMeta)
  277. {
  278. Console.Write($"\t|{meta.name} {meta.TypeName()} ({meta.size})\t|");
  279. }
  280. resData.ForEach(x => data.Add(SafeType.SafeInt(x)));
  281. // Important: need to free result to avoid memory leak.
  282. TDengine.FreeResult(res);
  283. }
  284. }
  285. finally
  286. {
  287. TDengine.Close(conn);
  288. }
  289. return new Aggregate
  290. {
  291. Max = data.Count.Equals(0) ? 0 : data[0],
  292. Min = data.Count.Equals(0) ? 0 : data[1],
  293. };
  294. }
  295. public int GetAvgExceptMaxMinValue(string field, string tbName, string? condition)
  296. {
  297. List<int> data = new();
  298. var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  299. var aggregate= GetAggregateValue(field, tbName, condition);
  300. var sqlAvg = $"SELECT AVG({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition} AND {field} < {aggregate.Max} and {field} > {aggregate.Min}";
  301. var conn = Connection();
  302. try
  303. {
  304. IntPtr res = TDengine.Query(conn, sqlAvg);
  305. // Check if query success
  306. if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0))
  307. {
  308. Console.Write(sqlAvg + " failure, ");
  309. // Get error message while Res is a not null pointer.
  310. if (res != IntPtr.Zero)
  311. {
  312. Console.Write("reason:" + TDengine.Error(res));
  313. }
  314. }
  315. else
  316. {
  317. Console.Write(sqlAvg + " success, {0} rows affected", TDengine.AffectRows(res));
  318. //... do something with res ...
  319. List<TDengineMeta> resMeta = LibTaos.GetMeta(res);
  320. List<object> resData = LibTaos.GetData(res);
  321. foreach (var meta in resMeta)
  322. {
  323. Console.Write($"\t|{meta.name} {meta.TypeName()} ({meta.size})\t|");
  324. }
  325. resData.ForEach(x => data.Add(SafeType.SafeInt(x)));
  326. // Important: need to free result to avoid memory leak.
  327. TDengine.FreeResult(res);
  328. }
  329. }
  330. finally
  331. {
  332. TDengine.Close(conn);
  333. }
  334. return data.Count.Equals(0) ? 0 : data[0];
  335. }
  336. #endregion
  337. #region RestAPI
  338. public async Task<string?> ExecuteQuerySQLRestResponse(string sql)
  339. {
  340. var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}";
  341. List<KeyValuePair<string, string>> headers = new()
  342. {
  343. new KeyValuePair<string, string>("Authorization", "Basic " + _configTDengineService.Token)
  344. };
  345. var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false);
  346. return result;
  347. }
  348. public async Task<string?> ExecuteSelectRestResponseAsync( string tbName, string condition="1", string field = "*")
  349. {
  350. var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}";
  351. var sql = $"SELECT {field} FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  352. List<KeyValuePair<string, string>> headers = new()
  353. {
  354. new KeyValuePair<string, string>("Authorization", "Basic " + _configTDengineService.Token)
  355. };
  356. _logger.LogInformation($"{nameof(ExecuteSelectRestResponseAsync)} --- SQL 语句执行 {sql}");
  357. var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false);
  358. return result;
  359. }
  360. public async Task<bool> GernalRestSql(string sql)
  361. {
  362. //"http://{server}:{port}/rest/sql/{db}"
  363. var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}";
  364. List<KeyValuePair<string, string>> headers = new()
  365. {
  366. new KeyValuePair<string, string>("Authorization", "Basic " + _configTDengineService.Token)
  367. };
  368. var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false);
  369. var res = JsonConvert.DeserializeObject<TDengineRestResBase>(result!);
  370. if (result != null)
  371. {
  372. if (res?.Code == 0)
  373. {
  374. _logger.LogInformation($"{nameof(GernalRestSql)},SQL 语句执行成功|{sql}");
  375. return true;
  376. }
  377. else
  378. {
  379. _logger.LogWarning($"{nameof(GernalRestSql)},SQL 语句执行失败||{sql}");
  380. return false;
  381. }
  382. }
  383. else
  384. {
  385. _logger.LogError($"{nameof(GernalRestSql)},TDengine 服务器IP:{_configTDengineService.Host} 错误,请联系运维人员");
  386. return false;
  387. }
  388. //return res.Code==0;
  389. }
  390. public async Task<string?> GernalRestSqlResTextAsync(string sql)
  391. {
  392. _logger.LogInformation($"执行 SQL: {nameof(GernalRestSqlResTextAsync)}--{sql}");
  393. var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}";
  394. List<KeyValuePair<string, string>> headers = new()
  395. {
  396. new KeyValuePair<string, string>("Authorization", "Basic " + _configTDengineService.Token)
  397. };
  398. var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false);
  399. return result;
  400. }
  401. /** 取消
  402. /// <summary>
  403. /// 最大值,最小值聚合
  404. /// </summary>
  405. /// <param name="field"></param>
  406. /// <param name="tbName"></param>
  407. /// <param name="condition"></param>
  408. /// <returns></returns>
  409. public async Task<Aggregate> GetAggregateValueAsync(string field,string tbName,string? condition)
  410. {
  411. var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  412. var result = await GernalRestSqlResTextAsync(sql);
  413. var res = JsonConvert.DeserializeObject<Aggregate>(result!);
  414. List<dynamic> data = res?.Data!;
  415. return new Aggregate
  416. {
  417. Max = data.Count.Equals(0) ? 0 : data[0][0],
  418. Min = data.Count.Equals(0) ? 0 : data[0][1],
  419. };
  420. }
  421. /// <summary>
  422. /// 去除最大值和最小值后的平均值
  423. /// </summary>
  424. /// <param name="field"></param>
  425. /// <param name="tbName"></param>
  426. /// <param name="condition"></param>
  427. /// <returns></returns>
  428. public async Task<int> GetAvgExceptMaxMinValueAsync(string field, string tbName, string? condition)
  429. {
  430. var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  431. var result = await GernalRestSqlResTextAsync(sql);
  432. var res = JsonConvert.DeserializeObject<TDengineRestResBase>(result!);
  433. _logger.LogInformation($"最大小值:{sql}");
  434. List<dynamic> data = res?.Data!;
  435. var sqlAvg = $"SELECT AVG({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition} AND {field} < { (data.Count.Equals(0)? 0: data[0][0]) } and {field} > {(data.Count.Equals(0) ? 0 : data[0][1])}";
  436. result = await GernalRestSqlResTextAsync(sqlAvg);
  437. _logger.LogInformation($"sqlAvg:{sqlAvg}");
  438. res = JsonConvert.DeserializeObject<TDengineRestResBase>(result!);
  439. data = res?.Data!;
  440. return data.Count.Equals(0)?0:(int)data[0][0];
  441. }
  442. /// <summary>
  443. /// 获取最后的记录
  444. /// </summary>
  445. /// <param name="tbName"></param>
  446. /// <param name="condition"></param>
  447. /// <returns></returns>
  448. public async Task<JArray?> GetLastAsync(string tbName, string? condition)
  449. {
  450. var sql = $"SELECT last_row(*) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  451. var result = await GernalRestSqlResTextAsync(sql);
  452. var res = JsonConvert.DeserializeObject<TDengineRestResBase>(result!);
  453. if(res?.Data?.Count==0) return null;
  454. List<dynamic> data = res?.Data!;
  455. return data[0] as JArray;
  456. }
  457. public async Task<int> GetCount(string tbName, string? condition)
  458. {
  459. var sql = $"SELECT count(ts) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}";
  460. var result = await GernalRestSqlResTextAsync(sql);
  461. var res = JsonConvert.DeserializeObject<TDengineRestResBase>(result!);
  462. List<dynamic> data = res?.Data!;
  463. return SafeType.SafeInt(data[0][0]);
  464. }
  465. */
  466. #endregion
  467. /// <summary>
  468. /// 平均值算法(去除最大值,最小值和大于标定值的平均值)
  469. /// </summary>
  470. /// <param name="numToRemove"></param>
  471. /// <param name="collection"></param>
  472. /// <param name="max"></param>
  473. /// <param name="min"></param>
  474. /// <returns></returns>
  475. public static decimal AverageAfterRemovingOneMinMaxRef(List<int> collection, int max, int min,int refValue)
  476. {
  477. collection.Remove(max);
  478. collection.Remove(min);
  479. collection.RemoveAll(_ => _ > refValue);
  480. if (collection.Count < 2)
  481. {
  482. throw new ArgumentException($"数据集{collection.ToArray()},去掉一个最大值 {max}和一个最小值{min},异常值(大于标定值{refValue}),后数据值不足");
  483. }
  484. return (decimal)collection.Average(x => x);
  485. //var values = ParseData.Select(valueSelector).ToList();
  486. //collection = values.Select(i => (int)i).ToArray();
  487. //if (values.Count <= 2)
  488. //{
  489. // throw new ArgumentException("Not enough elements to remove.");
  490. //}
  491. //// Remove the specified number of minimum and maximum values
  492. ////values.RemoveAll(_ => _ == values.Min());
  493. ////values.RemoveAll(_ => _ == values.Max());
  494. //max = (int)values.Max();
  495. //min = (int)values.Min();
  496. //values.Remove(max);
  497. //values.Remove(min);
  498. //// Remove values less than the specified threshold
  499. //values.RemoveAll(_ => _ > numToRemove);
  500. //// Calculate and return the average of the remaining values
  501. //return values.Average();
  502. }
  503. /// <summary>
  504. /// 去除最大值和最小值各一个(列表的头和尾),再去除异常值
  505. /// </summary>
  506. /// <param name="systolicRefValue"></param>
  507. /// <param name="hmBpParser"></param>
  508. /// <returns></returns>
  509. public decimal[] AverageAfterRemovingOneMinMaxRef(int systolicRefValue, ParseTDengineRestResponse<BloodPressureModel>? hmBpParser)
  510. {
  511. var sortedList = hmBpParser?.Select(i => i)
  512. .Where(i => i.IsDisplay.Equals(true))
  513. .OrderByDescending(i => i.SystolicValue)
  514. .ThenByDescending(i => i.DiastolicValue)
  515. .ToList();
  516. _logger.LogInformation($"计算时间段排列数据集:{JsonConvert.SerializeObject(sortedList)}");
  517. // 去除最大值和最小值各一个(列表的头和尾)
  518. var trimmedList = sortedList?
  519. .Skip(1)
  520. .Take(sortedList.Count - 2)
  521. .ToList();
  522. _logger.LogInformation($"计算去除最大值和最小值各一个数据集:{JsonConvert.SerializeObject(trimmedList)}");
  523. // 去除异常值
  524. var filteredList = trimmedList?.Where(bp => bp.SystolicValue < SafeType.SafeInt(systolicRefValue!)).ToList();
  525. _logger.LogInformation($"计算除异常值个数据集:{JsonConvert.SerializeObject(filteredList)}");
  526. if (filteredList?.Count < 2)
  527. {
  528. // throw new ArgumentException("数据不够不能计算");
  529. // 平均值为0,说明数据不足,不能计算增量值
  530. return new decimal[] { 0M, 0M };
  531. }
  532. var systolicAvg = filteredList?.Select(bp => bp.SystolicValue).Average();
  533. var diastolicAvg = filteredList?.Select(bp => bp.DiastolicValue).Average();
  534. return new decimal[] { (decimal)systolicAvg!, (decimal)diastolicAvg! };
  535. }
  536. /// <summary>
  537. /// 去除最大值和最小值各一个(列表的头和尾)
  538. /// </summary>
  539. /// <param name="systolicRefValue"></param>
  540. /// <param name="hmBpParser"></param>
  541. /// <returns></returns>
  542. public decimal[] AverageAfterRemovingOneMinMaxRef(ParseTDengineRestResponse<BloodPressureModel>? hmBpParser)
  543. {
  544. var sortedList = hmBpParser?.Select(i => i)
  545. .Where(i => i.IsDisplay.Equals(true))
  546. .OrderByDescending(i => i.SystolicValue)
  547. .ThenByDescending(i => i.DiastolicValue)
  548. .ToList();
  549. _logger.LogInformation($"计算时间段排列数据集:{JsonConvert.SerializeObject(sortedList)}");
  550. // 去除最大值和最小值各一个(列表的头和尾)
  551. var trimmedList = sortedList?
  552. .Skip(1)
  553. .Take(sortedList.Count - 2)
  554. .ToList();
  555. _logger.LogInformation($"计算去除最大值和最小值各一个数据集:{JsonConvert.SerializeObject(trimmedList)}");
  556. var filteredList = trimmedList?.ToList();
  557. _logger.LogInformation($"计算除异常值个数据集:{JsonConvert.SerializeObject(filteredList)}");
  558. if (filteredList?.Count < 2)
  559. {
  560. // throw new ArgumentException("数据不够不能计算");
  561. // 平均值为0,说明数据不足,不能计算增量值
  562. return new decimal[] { 0M, 0M };
  563. }
  564. var systolicAvg = filteredList?.Select(bp => bp.SystolicValue).Average();
  565. var diastolicAvg = filteredList?.Select(bp => bp.DiastolicValue).Average();
  566. return new decimal[] { (decimal)systolicAvg!, (decimal)diastolicAvg! };
  567. }
  568. #region SqlSugarClient
  569. /**
  570. public async Task InsertFetalHeartRateAsync()
  571. {
  572. var tableName = typeof(FetalHeartRateModel)
  573. .GetCustomAttribute<STableAttribute>()?
  574. .STableName;
  575. if (tableName == null)
  576. {
  577. throw new InvalidOperationException("STableAttribute not found on FetalHeartRateModel class.");
  578. }
  579. await _clientSqlSugar.Ado.ExecuteCommandAsync($"create table IF NOT EXISTS hm_fhr_00 using {tableName} tags('00')");
  580. _clientSqlSugar.Insertable(new FetalHeartRateModel()
  581. {
  582. Timestamp = DateTime.Now,
  583. CreateTime = DateTime.Now,
  584. FetalHeartRate = 90,
  585. FetalHeartRateId = Guid.NewGuid().ToString("D"),
  586. IsDisplay = false,
  587. Method = 1,
  588. PersonId = Guid.NewGuid().ToString("D"),
  589. MessageId = Guid.NewGuid().ToString("D"),
  590. SerialNumber = Guid.NewGuid().ToString("D"),
  591. DeviceKey = Guid.NewGuid().ToString("D"),
  592. SerialTailNumber = "00",
  593. LastUpdate = DateTime.Now,
  594. }).AS("hm_fhr_00").ExecuteCommand();
  595. }
  596. public Task<FetalHeartRateModel> GetFirst()
  597. {
  598. var tableName = typeof(FetalHeartRateModel)
  599. .GetCustomAttribute<STableAttribute>()?
  600. .STableName;
  601. var first = _clientSqlSugar
  602. .Queryable<FetalHeartRateModel>()
  603. .AS(tableName)
  604. .OrderByDescending(x => x.Timestamp).FirstAsync();
  605. return first;
  606. }
  607. */
  608. /// <summary>
  609. /// 插入记录
  610. /// </summary>
  611. /// <typeparam name="T"></typeparam>
  612. /// <param name="model"></param>
  613. /// <param name="tbName">子表名称</param>
  614. /// <returns></returns>
  615. /// <exception cref="InvalidOperationException"></exception>
  616. public async Task InsertAsync<T>(string tbName, T model)
  617. {
  618. var stbName = typeof(T)
  619. .GetCustomAttribute<STableAttribute>()?
  620. .STableName;
  621. if (stbName == null)
  622. {
  623. throw new InvalidOperationException($"STableAttribute not found on {nameof(T)} class.");
  624. }
  625. var tailNo = typeof(T).GetProperty("SerialTailNumber")?.GetValue(model)?.ToString();
  626. if (string.IsNullOrEmpty(tailNo))
  627. {
  628. throw new InvalidOperationException($"SerialNumberAttribute not found on {nameof(T)} class.");
  629. }
  630. var tbFullName = $"{tbName}_{tailNo}";
  631. await _clientSqlSugar.Ado.ExecuteCommandAsync($"create table IF NOT EXISTS {tbFullName} using {stbName} tags('{tailNo}')");
  632. //_clientSqlSugar.InsertableByDynamic(model).AS(tbFullName).ExecuteCommand();
  633. _clientSqlSugar.InsertableByObject(model).AS(tbFullName).ExecuteCommand();
  634. }
  635. public Task<T> GetLastAsync<T>() where T : class
  636. {
  637. var tableName = typeof(T)
  638. .GetCustomAttribute<STableAttribute>()?
  639. .STableName;
  640. // 创建一个表示 Timestamp 属性的表达式
  641. var parameter = Expression.Parameter(typeof(T), "x");
  642. var property = Expression.Property(parameter, "Timestamp");
  643. var lambda = Expression.Lambda<Func<T, object>>(Expression.Convert(property, typeof(object)), parameter);
  644. var first = _clientSqlSugar
  645. .Queryable<T>()
  646. .AS(tableName)
  647. .OrderByDescending(lambda).FirstAsync();
  648. return first;
  649. }
  650. public Task<T> GetLastAsync<T>(string serialNo) where T : class
  651. {
  652. var tableName = typeof(T)
  653. .GetCustomAttribute<STableAttribute>()?
  654. .STableName;
  655. // 创建表示 Timestamp 属性的表达式
  656. var parameter = Expression.Parameter(typeof(T), "x");
  657. var timestampProperty = Expression.Property(parameter, "Timestamp");
  658. var timestampLambda = Expression.Lambda<Func<T, object>>(Expression.Convert(timestampProperty, typeof(object)), parameter);
  659. // 创建表示 SerialNo 属性的表达式
  660. var serialNoProperty = Expression.Property(parameter, "SerialNumber");
  661. var serialNoConstant = Expression.Constant(serialNo);
  662. var equalExpression = Expression.Equal(serialNoProperty, serialNoConstant);
  663. var serialNoLambda = Expression.Lambda<Func<T, bool>>(equalExpression, parameter);
  664. var first = _clientSqlSugar
  665. .Queryable<T>()
  666. .AS(tableName)
  667. .Where(serialNoLambda)
  668. .OrderByDescending(timestampLambda).FirstAsync();
  669. return first;
  670. }
  671. public async Task<List<T>> GetBySerialNoAsync<T>(string serialNo) where T : class
  672. {
  673. var tableName = typeof(T)
  674. .GetCustomAttribute<STableAttribute>()?
  675. .STableName;
  676. // 创建表示 Timestamp 属性的表达式
  677. var parameter = Expression.Parameter(typeof(T), "x");
  678. var timestampProperty = Expression.Property(parameter, "Timestamp");
  679. var timestampLambda = Expression.Lambda<Func<T, object>>(Expression.Convert(timestampProperty, typeof(object)), parameter);
  680. // 创建表示 SerialNo 属性的表达式
  681. var serialNoProperty = Expression.Property(parameter, "SerialNumber");
  682. var serialNoConstant = Expression.Constant(serialNo);
  683. var equalExpression = Expression.Equal(serialNoProperty, serialNoConstant);
  684. var serialNoLambda = Expression.Lambda<Func<T, bool>>(equalExpression, parameter);
  685. var res = await _clientSqlSugar
  686. .Queryable<T>()
  687. .AS(tableName)
  688. .Where(serialNoLambda)
  689. .OrderByDescending(timestampLambda).ToListAsync();
  690. return res;
  691. }
  692. public async Task<List<T>> GetBySerialNoAsync<T>(string serialNo, int days) where T : class
  693. {
  694. var tableName = typeof(T)
  695. .GetCustomAttribute<STableAttribute>()?
  696. .STableName;
  697. // 创建表示 Timestamp 属性的表达式
  698. var parameter = Expression.Parameter(typeof(T), "x");
  699. var timestampProperty = Expression.Property(parameter, "Timestamp");
  700. var timestampLambda = Expression.Lambda<Func<T, object>>(Expression.Convert(timestampProperty, typeof(object)), parameter);
  701. // 创建表示 SerialNo 属性的表达式
  702. var serialNoProperty = Expression.Property(parameter, "SerialNumber");
  703. var serialNoConstant = Expression.Constant(serialNo);
  704. var equalExpression = Expression.Equal(serialNoProperty, serialNoConstant);
  705. // 创建表示 Timestamp 大于指定天数的表达式
  706. var daysAgo = DateTime.Now.AddDays(-days);
  707. var daysAgoConstant = Expression.Constant(daysAgo, typeof(DateTime));
  708. var greaterThanExpression = Expression.GreaterThan(timestampProperty, daysAgoConstant);
  709. // 合并 SerialNo 和 Timestamp 条件
  710. var combinedExpression = Expression.AndAlso(equalExpression, greaterThanExpression);
  711. var combinedLambda = Expression.Lambda<Func<T, bool>>(combinedExpression, parameter);
  712. var res = await _clientSqlSugar
  713. .Queryable<T>()
  714. .AS(tableName)
  715. .Where(combinedLambda)
  716. .OrderByDescending(timestampLambda).ToListAsync();
  717. return res;
  718. }
  719. #endregion
  720. #region 胎心算法
  721. /// <summary>
  722. /// 计算个人一般心率(最大值,最大值,最小值)
  723. /// </summary>
  724. /// <param name="serialNo"></param>
  725. /// <param name="days"></param>
  726. /// <param name="percentage"></param>
  727. /// <returns></returns>
  728. public async Task<PregnancyCommonHeartRateModel?> InitPregnancyCommonHeartRateModeAsync(string serialNo, int days = 7, int percentage = 90)
  729. {
  730. var tableName = typeof(PregnancyHeartRateModel)
  731. .GetCustomAttribute<STableAttribute>()?
  732. .STableName;
  733. var daysAgo = DateTime.Now.AddDays(-days);
  734. var collection = await _clientSqlSugar
  735. .Queryable<PregnancyHeartRateModel>()
  736. .AS(tableName)
  737. .Where(i => i.SerialNumber.Equals(serialNo))
  738. .Where(i => i.Timestamp > daysAgo)
  739. .OrderByDescending(i => i.Timestamp)
  740. .ToArrayAsync();
  741. var res = collection
  742. .Select(i => i.PregnancyHeartRate).ToList();
  743. // 心率数据量必须30个以上才进行计算
  744. if (res.Count < 30)
  745. {
  746. _logger.LogInformation($"{serialNo} 心率数据不足,无法计算其众数");
  747. return null;
  748. }
  749. #region 计算众数
  750. var mode = res.GroupBy(n => n)
  751. .OrderByDescending(g => g.Count())
  752. .First()
  753. .Key;
  754. Console.WriteLine("众数是: " + mode);
  755. // 如果有多个众数的情况
  756. var maxCount = res.GroupBy(n => n)
  757. .Max(g => g.Count());
  758. var modes = res.GroupBy(n => n)
  759. .Where(g => g.Count() == maxCount)
  760. .Select(g => g.Key)
  761. .ToList();
  762. // 多个众数,选择最接近平均数或中位数的众数
  763. if (modes.Count > 1)
  764. {
  765. // 计算平均值
  766. double average = res.Average();
  767. Console.WriteLine("平均值是: " + average);
  768. // 计算中位数
  769. double median;
  770. int count = res.Count;
  771. var sortedRes = res.OrderBy(n => n).ToList();
  772. if (count % 2 == 0)
  773. {
  774. // 偶数个元素,取中间两个数的平均值
  775. median = (sortedRes[count / 2 - 1] + sortedRes[count / 2]) / 2.0;
  776. }
  777. else
  778. {
  779. // 奇数个元素,取中间的数
  780. median = sortedRes[count / 2];
  781. }
  782. Console.WriteLine("中位数是: " + median);
  783. // 找出最接近平均值的众数
  784. //var closestToAverage = modes.OrderBy(m => Math.Abs(m - average)).First();
  785. //Console.WriteLine("最接近平均值的众数是: " + closestToAverage);
  786. // 找出最接近中位数的众数
  787. var closestToMedian = modes.OrderBy(m => Math.Abs(m - median)).First();
  788. Console.WriteLine("最接近中位数的众数是: " + closestToMedian);
  789. mode = closestToMedian;
  790. }
  791. #endregion
  792. // 计算需要的数量
  793. int requiredCount = (int)(res.Count * 0.8);
  794. // 从原始数据集中获取最接近众数的元素
  795. var closestToModeData = res.OrderBy(n => Math.Abs(n - mode))
  796. .Take(requiredCount)
  797. .ToList();
  798. // 输出新数据集
  799. Console.WriteLine("新数据集: " + string.Join(", ", closestToModeData));
  800. Console.WriteLine("新数据集的数量: " + closestToModeData.Count);
  801. var fhrMap = _mgrFhrPhrMapCache.GetHeartRatesMap();
  802. var watchConfig = await _deviceCacheMgr.GetGpsDeviceWatchConfigCacheObjectBySerialNoAsync(serialNo, "0067");
  803. if (watchConfig == null)
  804. {
  805. return null;
  806. }
  807. // long.TryParse(watchConfig["EDOC"]!.ToString(), out long edoc);
  808. // "EDOC": "1720860180652",当前时间 - (EDOC - 280) days =怀孕时间
  809. //edoc = edoc.ToString().Length == 10 ? edoc * 1000 : edoc;
  810. var edoc = DateTimeUtil.ToDateTime(watchConfig["EDOC"]!.ToString());
  811. int pregnancyWeek = (DateTime.Now - edoc.AddDays(-280)).Days / 7;
  812. _logger.LogInformation($"IMEI {serialNo},EDOC:{edoc},NOW:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")},SinceNOW:{edoc.AddDays(-280).ToString("yyyy-MM-dd HH:mm:ss")},怀孕周数 {pregnancyWeek}");
  813. float statMaxValueFprCoefficient = 0f;
  814. float statMinValueFprCoefficient = 0f;
  815. float StatModeAvgFprCoefficient = 0f;
  816. // 20-45周之间
  817. if (pregnancyWeek >= 12 && pregnancyWeek <= 45)
  818. {
  819. var map = fhrMap
  820. .Where(i =>
  821. i.PregnancyPeriod![0] <= pregnancyWeek &&
  822. i.PregnancyPeriod[1] >= pregnancyWeek &&
  823. i.PregnancyHeartRateRange![0] <= mode &&
  824. i.PregnancyHeartRateRange[1] >= mode)
  825. .FirstOrDefault();
  826. if (map != null)
  827. {
  828. statMaxValueFprCoefficient = (float)Math.Round((decimal)map.FetalHeartRateRange![1] / res.Max(), 3);
  829. statMinValueFprCoefficient = (float)Math.Round((decimal)map.FetalHeartRateRange[0] / res.Min(), 3);
  830. StatModeAvgFprCoefficient = (float)Math.Round((decimal)map.FetalHeartRateAverage / mode, 3);
  831. }
  832. }
  833. return new PregnancyCommonHeartRateModel()
  834. {
  835. Timestamp = DateTime.Now,
  836. PersonId = collection.First().DeviceKey,
  837. DeviceKey = collection.First().DeviceKey,
  838. SerialNumber = collection.First().SerialNumber,
  839. Mode = mode,
  840. Percentage = percentage,
  841. MaxValue = closestToModeData.Max(),
  842. MinValue = closestToModeData.Min(),
  843. OriginalMaxValue = res.Max(),
  844. OriginalMinValue = res.Min(),
  845. CreateTime = DateTime.Now,
  846. StatStartTime = collection.OrderBy(i => i.Timestamp).Select(i => i.Timestamp).First(),
  847. StatEndTime = collection.OrderBy(i => i.Timestamp).Select(i => i.Timestamp).Last(),
  848. StatMaxValueFprCoefficient = statMaxValueFprCoefficient,
  849. StatMinValueFprCoefficient = statMinValueFprCoefficient,
  850. StatModeAvgFprCoefficient = StatModeAvgFprCoefficient,
  851. Remark = string.Empty,
  852. SerialTailNumber = serialNo.Substring(serialNo.Length - 2)
  853. };
  854. }
  855. /// <summary>
  856. /// 获取孕妇心率众数
  857. /// </summary>
  858. /// <param name="serialNo"></param>
  859. /// <param name="days"></param>
  860. /// <returns></returns>
  861. //public async Task<int> GetPregnancyHeartRateModeAsync(string serialNo,int days=7)
  862. //{
  863. // var tableName = typeof(PregnancyHeartRateModel)
  864. // .GetCustomAttribute<STableAttribute>()?
  865. // .STableName;
  866. // var daysAgo = DateTime.Now.AddDays(-days);
  867. // var res = await _clientSqlSugar
  868. // .Queryable<PregnancyHeartRateModel>()
  869. // .AS(tableName)
  870. // .Where(i=>i.SerialNumber.Equals(serialNo))
  871. // .Where(i => i.Timestamp > daysAgo)
  872. // //.OrderByDescending(i => i.PregnancyHeartRate)
  873. // .Select(i =>i.PregnancyHeartRate)
  874. // .ToListAsync();
  875. // // 心率数据量必须30个以上才进行计算
  876. // if (res.Count < 30) return 0;
  877. // // 计算众数
  878. // var mode = res.GroupBy(n => n)
  879. // .OrderByDescending(g => g.Count())
  880. // .First()
  881. // .Key;
  882. // Console.WriteLine("众数是: " + mode);
  883. // // 如果有多个众数的情况
  884. // var maxCount = res.GroupBy(n => n)
  885. // .Max(g => g.Count());
  886. // var modes = res.GroupBy(n => n)
  887. // .Where(g => g.Count() == maxCount)
  888. // .Select(g => g.Key)
  889. // .ToList();
  890. // // 多个众数,选择最接近平均数或中位数的众数
  891. // if (modes.Count>1)
  892. // {
  893. // // 计算平均值
  894. // double average = res.Average();
  895. // Console.WriteLine("平均值是: " + average);
  896. // // 计算中位数
  897. // double median;
  898. // int count = res.Count;
  899. // var sortedRes = res.OrderBy(n => n).ToList();
  900. // if (count % 2 == 0)
  901. // {
  902. // // 偶数个元素,取中间两个数的平均值
  903. // median = (sortedRes[count / 2 - 1] + sortedRes[count / 2]) / 2.0;
  904. // }
  905. // else
  906. // {
  907. // // 奇数个元素,取中间的数
  908. // median = sortedRes[count / 2];
  909. // }
  910. // Console.WriteLine("中位数是: " + median);
  911. // // 找出最接近平均值的众数
  912. // //var closestToAverage = modes.OrderBy(m => Math.Abs(m - average)).First();
  913. // //Console.WriteLine("最接近平均值的众数是: " + closestToAverage);
  914. // // 找出最接近中位数的众数
  915. // var closestToMedian = modes.OrderBy(m => Math.Abs(m - median)).First();
  916. // Console.WriteLine("最接近中位数的众数是: " + closestToMedian);
  917. // mode = closestToMedian;
  918. // }
  919. // return mode;
  920. //}
  921. #endregion
  922. }
  923. }