using HealthMonitor.Common.helper; using HealthMonitor.Model.Config; using HealthMonitor.Service.Biz.db.Dto; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using TDengineDriver; using TDengineDriver.Impl; namespace HealthMonitor.Service.Biz.db { public class TDengineService { private readonly ILogger _logger; private readonly HttpHelper _httpHelper=default!; private readonly TDengineServiceConfig _configTDengineService; public TDengineService(ILogger logger, IOptions configTDengineService, HttpHelper httpHelper ) { _logger = logger; _configTDengineService = configTDengineService.Value; _httpHelper = httpHelper; } public IntPtr Connection() { string host = _configTDengineService.Host; string user = _configTDengineService.UserName; string db = _configTDengineService.DB; short port = _configTDengineService.Port; string password = _configTDengineService.Password; IntPtr conn = TDengine.Connect(host, user, password, db, port); //string host = "172.16.255.180"; //short port = 6030; //string user = "root"; //string password = "taosdata"; //string db = "health_monitor"; //IntPtr conn = TDengine.Connect(host, user, password, db, port); // Check if get connection success if (conn == IntPtr.Zero) { Console.WriteLine("Connect to TDengine failed"); } else { Console.WriteLine("Connect to TDengine success"); } return conn; } public void ExecuteSQL(IntPtr conn, string sql) { IntPtr res = TDengine.Query(conn, sql); // Check if query success if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0)) { Console.Write(sql + " failure, "); // Get error message while Res is a not null pointer. if (res != IntPtr.Zero) { Console.Write("reason:" + TDengine.Error(res)); } } else { Console.Write(sql + " success, {0} rows affected", TDengine.AffectRows(res)); //... do something with res ... // Important: need to free result to avoid memory leak. TDengine.FreeResult(res); } } public void ExecuteQuerySQL(IntPtr conn, string sql) { IntPtr res = TDengine.Query(conn, sql); // Check if query success if ((res == IntPtr.Zero) || (TDengine.ErrorNo(res) != 0)) { Console.Write(sql + " failure, "); // Get error message while Res is a not null pointer. if (res != IntPtr.Zero) { Console.Write("reason:" + TDengine.Error(res)); } } else { Console.Write(sql + " success, {0} rows affected", TDengine.AffectRows(res)); //... do something with res ... List resMeta = LibTaos.GetMeta(res); List resData = LibTaos.GetData(res); foreach (var meta in resMeta) { Console.Write($"\t|{meta.name} {meta.TypeName()} ({meta.size})\t|"); } for (int i = 0; i < resData.Count; i++) { Console.Write($"|{resData[i].ToString()} \t"); //Console.WriteLine("{0},{1},{2}", i, resMeta.Count, (i) % resMeta.Count); if (((i + 1) % resMeta.Count == 0)) { Console.WriteLine(""); } } // Important: need to free result to avoid memory leak. TDengine.FreeResult(res); } } public void CheckRes(IntPtr conn, IntPtr res, String errorMsg) { if (TDengine.ErrorNo(res) != 0) { throw new Exception($"{errorMsg} since: {TDengine.Error(res)}"); } } public void ExecuteInsertSQL(IntPtr conn, string sql) { try { //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) " + // "d1002 USING power.meters TAGS('California.SanFrancisco', 3) VALUES('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000) " + // "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) " + // "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)"; IntPtr res = TDengine.Query(conn, sql); CheckRes(conn, res, "failed to insert data"); int affectedRows = TDengine.AffectRows(res); Console.WriteLine("affectedRows " + affectedRows); TDengine.FreeResult(res); } finally { TDengine.Close(conn); } } public async Task GernalRestSql(string sql) { //"http://{server}:{port}/rest/sql/{db}" var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}"; List> headers = new() { new KeyValuePair("Authorization", "Basic " + _configTDengineService.Token) }; var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false); var res = JsonConvert.DeserializeObject(result!); if (result != null) { if (res?.Code == 0) { _logger.LogInformation($"{nameof(GernalRestSql)},SQL 语句执行成功|{sql}"); return true; } else { _logger.LogWarning($"{nameof(GernalRestSql)},SQL 语句执行失败||{sql}"); return false; } } else { _logger.LogError($"{nameof(GernalRestSql)},TDengine 服务器IP:{_configTDengineService.Host} 错误,请联系运维人员"); return false; } //return res.Code==0; } public async Task GernalRestSqlResText(string sql) { var url = $"http://{_configTDengineService.Host}:{_configTDengineService.RestPort}/rest/sql/{_configTDengineService.DB}"; List> headers = new() { new KeyValuePair("Authorization", "Basic " + _configTDengineService.Token) }; var result = await _httpHelper.HttpToPostAsync(url, sql, headers).ConfigureAwait(false); return result; } public async Task GetAggregateValue(string field,string tbName,string? condition) { var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}"; var result = await GernalRestSqlResText(sql); var res = JsonConvert.DeserializeObject(result!); List data = res?.Data!; return new Aggregate { Max = data.Count.Equals(0) ? 0 : data[0][0], Min = data.Count.Equals(0) ? 0 : data[0][1], }; } public async Task GetAvgExceptMaxMinValue(string field, string tbName, string? condition) { var sql = $"SELECT MAX({field}), MIN({field}) FROM {_configTDengineService.DB}.{tbName} WHERE {condition}"; var result = await GernalRestSqlResText(sql); var res = JsonConvert.DeserializeObject(result!); List data = res?.Data!; 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])}"; result = await GernalRestSqlResText(sqlAvg); res = JsonConvert.DeserializeObject(result!); data = res?.Data!; return data.Count.Equals(0)?0:(int)data[0][0]; } } }