using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Data;
using InfluxData.Net.Common;
using InfluxData.Net.InfluxDb;
namespace Mesnac.Basic
{
///
/// 时序数据库InfluxDb辅助类
///
public class InfluxDbHelper
{
#region 字段定义
private int _errCount = 0; //失败异常的次数
private string influxDbPath = @"D:\influxdb-1.4.3-1"; //时序数据库InfluxDb的根目录
private string endpointUri = "http://localhost:8086"; //时序数据库服务器的访问Uri
private string username = "root"; //时序数据库服务器用户名
private string password = "root"; //时序数据服务器密码
private bool _isStart = false; //时序数据库服务器是否启动
private InfluxDbClient _client = null; //时序数据库客户端对象
private readonly string backViewDbName = "MesnacBackView"; //回放数据库的名称
private readonly string backViewTableName = "Monitor"; //回放数据表名称
private readonly string curveDbName = "MesnacCurve"; //曲线数据库的名称
private readonly string mixCurveTableName = "MixCurveData"; //密炼曲线数据表名称
private readonly string millCurveTableName = "MillCurveData"; //开炼曲线数据表名称
private readonly string backViewRetentionPolicyName = "BackViewDataSaveDays"; //回放数据库保留策略名称
private int backViewDataSaveDays = 30; //回放数据保留的天数
#endregion
#region 单例实现
private static InfluxDbHelper _instance = null;
private InfluxDbHelper()
{
this.influxDbPath = AppConfigHelper.GetAppSettingValue("InfluxDb_Path", @"D:\influxdb-1.4.3-1");
this.endpointUri = AppConfigHelper.GetAppSettingValue("InfluxDb_EndPointUri", "http://localhost:8086");
this.username = AppConfigHelper.GetAppSettingValue("InfluxDb_UserName", "root");
this.password = AppConfigHelper.GetAppSettingValue("InfluxDb_Password", "root");
this.backViewDataSaveDays = AppConfigHelper.GetAppSettingValue("BackViewDataSaveDays", 30);
//this.StopDbServer();
//System.Threading.Thread.Sleep(10);
//this.StartDbServer();
//System.Threading.Thread.Sleep(50);
if (!this.IsStartDbServer)
{
this.StartDbServer();
}
}
public static InfluxDbHelper Instance
{
get
{
lock (String.Empty)
{
if (_instance == null)
{
_instance = new InfluxDbHelper();
}
return _instance;
}
}
}
#endregion
#region 属性定义
///
/// 回放数据库名称
///
public string BackViewDbName
{
get
{
return this.backViewDbName;
}
}
///
/// 返回数据保存策略名称
///
public string BackViewRetentionPolicyName
{
get
{
return this.backViewRetentionPolicyName;
}
}
///
/// 回放数据表名称
///
public string BackViewTableName
{
get
{
return this.backViewTableName;
}
}
///
/// 密炼曲线数据表名称
///
public string MixCurveTableName
{
get
{
return this.mixCurveTableName;
}
}
///
/// 开炼曲线数据表名称
///
public string MillCurveTableName
{
get
{
return this.millCurveTableName;
}
}
///
/// 服务是否启动
///
public bool StartState
{
get
{
return this._isStart;
}
}
///
/// InfluxDb客户端对象,用于执行查询
///
public InfluxDbClient Client
{
get
{
if (_client == null)
{
if (!this.StartState)
{
this.StartDbServer();
}
if (this.StartState)
{
_client = new InfluxDbClient(this.endpointUri, this.username, this.password, InfluxData.Net.Common.Enums.InfluxDbVersion.Latest);
}
}
return _client;
}
}
#endregion
#region 方法定义
#region 判断InfluxDbServer是否已启动
///
/// 判断InfluxDbServer是否已启动
///
public bool IsStartDbServer
{
get
{
this._isStart = WinCmdHelper.CheckProcessExists("influxd", String.Empty);
if (this._isStart)
{
this.CreateBackViewDb(); //创建回放数据库
this.CreateBackViewRetentionPolicy(); //创建回放数据保留策略
this.CreateCurveDb();
}
return this._isStart;
}
}
#endregion
#region 启动InfluxDbServer
///
/// 启动InfluxDbServer
///
public void StartDbServer()
{
System.Threading.Thread t = new System.Threading.Thread(new System.Threading.ThreadStart(this.DoStartDbServerTask));
t.Start();
this._isStart = true;
this.CreateBackViewDb(); //创建回放数据库
this.CreateBackViewRetentionPolicy(); //创建回放数据保留策略
this.CreateCurveDb(); //创建曲线数据库
}
private void DoStartDbServerTask()
{
string cmd = String.Format("{0} -config {1}", System.IO.Path.Combine(this.influxDbPath, "influxd.exe"), System.IO.Path.Combine(this.influxDbPath, "influxdb.conf"));
bool result = Mesnac.Basic.WinCmdHelper.ExcuteDosCommand(cmd, false, false);
if (result)
{
this._isStart = true;
ICSharpCode.Core.LoggingService.Info("The InfluxDb Server start success!");
}
else
{
this._isStart = false;
ICSharpCode.Core.LoggingService.Error("The InfluxDb Server Start failure!");
}
}
#endregion
#region 停止InfluxDbServer
///
/// 停止InfluxDbServer
///
public void StopDbServer()
{
bool result = WinCmdHelper.KillProcessExists("influxd", String.Empty);
if (result)
{
this._isStart = false;
ICSharpCode.Core.LoggingService.Info("The InfluxDb Server Stop success!");
}
else
{
ICSharpCode.Core.LoggingService.Error("The InfluxDb Server Stop failure!");
}
}
#endregion
#region 创建回放数据库
///
/// 创建回放数据库
///
public void CreateBackViewDb()
{
try
{
if (!this.StartState)
{
return;
}
Task task = this.Client.Database.CreateDatabaseAsync(this.backViewDbName);
ICSharpCode.Core.LoggingService.Info("Create BackView Db Result success = " + task.Result.Success);
ICSharpCode.Core.LoggingService.Info("Create BackView Db Result Body = " + task.Result.Body);
ICSharpCode.Core.LoggingService.Info("Create BackView Db Result StatusCode =" + task.Result.StatusCode);
}
catch (Exception ex)
{
this._errCount++;
ICSharpCode.Core.LoggingService.Error("创建回放数据库异常:" + ex.Message, ex);
}
}
#endregion
#region 创建曲线数据库
///
/// 创建曲线数据库
///
public void CreateCurveDb()
{
try
{
if (!this.StartState)
{
return;
}
Task task = this.Client.Database.CreateDatabaseAsync(this.curveDbName);
ICSharpCode.Core.LoggingService.Info("Create Curve Db Result success = " + task.Result.Success);
ICSharpCode.Core.LoggingService.Info("Create Curve Db Result Body = " + task.Result.Body);
ICSharpCode.Core.LoggingService.Info("Create Curve Db Result StatusCode =" + task.Result.StatusCode);
}
catch (Exception ex)
{
this._errCount++;
ICSharpCode.Core.LoggingService.Error("创建曲线数据库异常:" + ex.Message, ex);
}
}
#endregion
#region 创建回放数据库的保留策略
///
/// 创建回放数据库的保留策略
///
public void CreateBackViewRetentionPolicy()
{
try
{
if (!this.StartState)
{
return;
}
string duration = String.Format("{0}d", this.backViewDataSaveDays);
Task task = this.Client.Retention.CreateRetentionPolicyAsync(this.backViewDbName, this.backViewRetentionPolicyName, duration, 1);
ICSharpCode.Core.LoggingService.Info("Create BackView Retention Policy Result success = " + task.Result.Success);
ICSharpCode.Core.LoggingService.Info("Create BackView Retention Policy Result Body = " + task.Result.Body);
ICSharpCode.Core.LoggingService.Info("Create BackView Retention Policy Result StatusCode =" + task.Result.StatusCode);
}
catch (Exception ex)
{
this._errCount++;
ICSharpCode.Core.LoggingService.Error("创建回放数据库的保留策略异常:" + ex.Message, ex);
}
}
#endregion
#region 写数据至回放数据库
///
/// 写数据至回放数据库
///
/// 要写入的数据
public void WriteDataToBackViewDb(Dictionary data)
{
try
{
if (!this.StartState)
{
return;
}
//ICSharpCode.Core.LoggingService.Info("field count = " + data.Count);
InfluxData.Net.InfluxDb.Models.Point point = new InfluxData.Net.InfluxDb.Models.Point();
point.Name = this.backViewTableName; //表名
//point.Tags = new Dictionary() { { "EquipCode", "01001" } };
point.Fields = data;
point.Timestamp = DateTime.Now;
Task task = this.Client.Client.WriteAsync(point, this.backViewDbName, this.backViewRetentionPolicyName, InfluxData.Net.Common.Constants.TimeUnit.Seconds);
if (task != null && task.Result != null && !task.Result.Success)
{
ICSharpCode.Core.LoggingService.Warn("Write Data to backView db result success =" + task.Result.Success);
ICSharpCode.Core.LoggingService.Warn("Write Data to backView db result Body =" + task.Result.Body);
ICSharpCode.Core.LoggingService.Warn("Write Data to backView db result StatusCode =" + task.Result.StatusCode);
}
}
catch (Exception ex)
{
this._errCount++;
ICSharpCode.Core.LoggingService.Error("保存回放数据异常:" + ex.Message, ex);
}
}
#endregion
#region 写数据至曲线数据库
///
/// 写数据至曲线数据库
///
/// 机台编号
/// 曲线数据
public void WriteDataToCurveDb(string equipCode, Dictionary data)
{
try
{
if (!this.StartState)
{
return;
}
InfluxData.Net.InfluxDb.Models.Point point = new InfluxData.Net.InfluxDb.Models.Point();
point.Name = this.mixCurveTableName; //表名
point.Tags = new Dictionary() { { "EquipCode", equipCode } };
point.Fields = data;
point.Timestamp = DateTime.Now;
Task task = this.Client.Client.WriteAsync(point, this.curveDbName, null, InfluxData.Net.Common.Constants.TimeUnit.Seconds);
if (task != null && task.Result != null && !task.Result.Success)
{
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result success =" + task.Result.Success);
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result Body =" + task.Result.Body);
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result StatusCode =" + task.Result.StatusCode);
}
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("保存曲线数据异常:" + ex.Message, ex);
}
}
///
/// 写曲线数据至曲线数据库表
///
/// 要写入的表名
/// 机台编号
/// 设备编号
/// 曲线数据
public void WriteDataToCurveDb(string tableName, string equipCode, string unitNum, Dictionary data, DateTime? time)
{
try
{
if (!this.StartState)
{
return;
}
InfluxData.Net.InfluxDb.Models.Point point = new InfluxData.Net.InfluxDb.Models.Point();
point.Name = tableName; //表名
point.Tags = new Dictionary() { { "EquipCode", equipCode }, { "UnitNum", unitNum } };
point.Fields = data;
if (time == null)
{
point.Timestamp = DateTime.Now;
}
else
{
point.Timestamp = time;
}
Task task = this.Client.Client.WriteAsync(point, this.curveDbName, null, InfluxData.Net.Common.Constants.TimeUnit.Seconds);
if (task != null && task.Result != null && !task.Result.Success)
{
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result success =" + task.Result.Success);
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result Body =" + task.Result.Body);
ICSharpCode.Core.LoggingService.Warn("Write Data to curve db result StatusCode =" + task.Result.StatusCode);
}
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("保存曲线数据异常:" + ex.Message, ex);
}
}
#endregion
#region 从曲线数据库获取数据
///
/// 从曲线数据库获取数据
///
/// 机台号
/// 字段列表
/// 开始时间
/// 结束时间
/// 返回对应的曲线数据
public Dictionary> GetDataFromCurveDb(string equipCode, List columnList, DateTime begin, DateTime end)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1} where EquipCode = '{2}' and time >= '{3:yyyy-MM-dd HH:mm:ss}' and time <= '{4:yyyy-MM-dd HH:mm:ss}'";
query = String.Format(query, String.Join(",", columnList.ToArray()), this.mixCurveTableName, equipCode, begin, end);
Task> task = this.Client.Client.QueryAsync(query, this.curveDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从曲线数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
///
/// 从曲线数据库获取数据
///
/// 要读取的表名
/// 机台编号
/// 设备编号
/// 要查询的字段列表
/// 开始时间
/// 结束时间
/// 返回对应的曲线数据
public Dictionary> GetDataFromCurveDb(string tableName, string equipCode, string unitNum, List columnList, DateTime begin, DateTime end)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1} where EquipCode = '{2}' and UnitNum = '{3}' and time >= '{4:yyyy-MM-dd HH:mm:ss}' and time <= '{5:yyyy-MM-dd HH:mm:ss}'";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, tableName, equipCode, unitNum, begin, end);
Task> task = this.Client.Client.QueryAsync(query, this.curveDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从曲线数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
///
/// 从曲线数据库获取数据
///
/// 字段列表
/// 开始时间
/// 结束时间
/// 返回对应的曲线数据
public Dictionary> GetDataFromCurveDb(List columnList, DateTime begin, DateTime end)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1} where time >= '{2:yyyy-MM-dd HH:mm:ss}' and time <= '{3:yyyy-MM-dd HH:mm:ss}'";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, this.mixCurveTableName, begin, end);
Task> task = this.Client.Client.QueryAsync(query, this.curveDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从曲线数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
///
/// 从曲线数据库获取数据
///
/// 要读取的表名
/// 设备编号
/// 字段列表
/// 开始时间
/// 结束时间
/// 返回对应的曲线数据
public Dictionary> GetDataFromCurveDb(string tableName, string unitNum, List columnList, DateTime begin, DateTime end)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1} where UnitNum = '{2}' and time >= '{3:yyyy-MM-dd HH:mm:ss}' and time <= '{4:yyyy-MM-dd HH:mm:ss}'";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, tableName, unitNum, begin, end);
Task> task = this.Client.Client.QueryAsync(query, this.curveDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从曲线数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
#endregion
#region 把数据集合转化为DataTable
///
/// 把数据集合转化为DataTable
///
/// 要转化的数据集合
/// 返回转化后的DataTable对象
public static DataTable ConvertDicToDataTable(Dictionary> data)
{
DataTable dt = new DataTable();
if (data != null && data.Count > 0)
{
string lastKey = String.Empty;
foreach(string key in data.Keys)
{
lastKey = key;
dt.Columns.Add(key, typeof(object));
}
for (int i = 0; i < data[lastKey].Count; i++)
{
DataRow row = dt.NewRow();
foreach(string key in data.Keys)
{
if (i < data[key].Count)
{
row[key] = data[key][i];
}
else
{
row[key] = 0;
}
}
dt.Rows.Add(row);
}
}
return dt;
}
#endregion
#region 从回放数据库获取数据
///
/// 从回放数据库获取数据
///
/// 回放数据表名称
/// 字段列表
/// 回放时间点
/// 返回对应时间点的回放数据
public Dictionary> GetDataFromBackViewDb(string tableName, List columnList, DateTime time)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1}.{2} where time = '{3:yyyy-MM-dd HH:mm:ss}'";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, this.backViewRetentionPolicyName, tableName, time);
Task> task = this.Client.Client.QueryAsync(query, this.BackViewDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从回放数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
///
/// 从回放数据库获取数据
///
/// 回放时间点
/// 返回对应回放时间点的回放数据
public Dictionary> GetDataFromBackViewDb(DateTime time)
{
try
{
if (!this.StartState)
{
return null;
}
Dictionary> result = new Dictionary>();
string query = "select {0} from {1}.{2} where time = '{3:yyyy-MM-dd HH:mm:ss}'";
string fields = "*";
query = String.Format(query, fields, this.backViewRetentionPolicyName, this.backViewTableName, time);
Task> task = this.Client.Client.QueryAsync(query, this.BackViewDbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
for (int i = 0; i < serie.Columns.Count; i++)
{
result.Add(serie.Columns[i], serie.Values[i]);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从回放数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
#endregion
#region 从InfluxDb数据库获取数据
///
/// 从InfluxDb数据库获取数据
///
/// 数据库名称
/// 表名称
/// 字段列表
/// 时间条件yyyy-MM-dd HH:mm:ss
/// 限制返回结果的最大条数
/// 返回符合条件的n条数据
public List> GetData(string dbName, string tableName, List columnList, DateTime time, int limit)
{
try
{
if (!this.StartState)
{
return null;
}
List> result = new List>();
string query = "select {0} from {1} where time > '{2:yyyy-MM-dd HH:mm:ss}' limit {3}";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, tableName, time, limit);
Task> task = this.Client.Client.QueryAsync(query, dbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
Dictionary data = null;
for (int i = 0; i < limit; i++)
{
data = new Dictionary();
for (int colIndex = 0; colIndex < serie.Columns.Count; colIndex++)
{
data.Add(serie.Columns[colIndex], serie.Values[i][colIndex]);
}
result.Add(data);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error("从InfluxDb数据库获取数据异常:" + ex.Message, ex);
}
return null;
}
///
/// 从InfluxDb数据库获取数据
///
/// 数据库名称
/// 数据保存策略
/// 数据表
/// 字段列表
/// 时间条件yyyy-MM-dd HH:mm:ss
/// 限制返回结果的最大条数
/// 返回符合条件的n条数据
public List> GetData(string dbName, string retentionPolicyName, string tableName, List columnList, DateTime time, int limit)
{
try
{
if (!this.StartState)
{
return null;
}
List> result = new List>();
string query = "select {0} from {1}.{2} where time > '{3:yyyy-MM-dd HH:mm:ss}' limit {4}";
string fields = "*";
if (columnList != null && columnList.Count > 0)
{
fields = String.Join(",", columnList.ToArray());
}
query = String.Format(query, fields, retentionPolicyName, tableName, time, limit);
Task> task = this.Client.Client.QueryAsync(query, dbName, null, null);
if (task != null && task.Result != null)
{
InfluxData.Net.InfluxDb.Models.Responses.Serie serie = task.Result.FirstOrDefault();
if (serie != null)
{
Dictionary data = null;
for (int i = 0; i < limit; i++)
{
data = new Dictionary();
for (int colIndex = 0; colIndex < serie.Columns.Count; colIndex++)
{
data.Add(serie.Columns[colIndex], serie.Values[i][colIndex]);
}
result.Add(data);
}
}
}
return result;
}
catch (Exception ex)
{
ICSharpCode.Core.LoggingService.Error(String.Format("从InfluxDb数据库{0}.{1}.{2}中获取数据异常: {3}", dbName, retentionPolicyName, tableName, ex.Message), ex);
}
return null;
}
#endregion
#endregion
}
}