Python操作TDengine并进行简单风机预测应用

Python操作TDengine并进行简单风机预测应用

一、安装

Linux端安装Python Connector

1、确保Linux环境下安装了Python环境，我的python版本Python3.6.0。

2、从https://github.com/taosdata/TDengine下载源码，解压后放到服务器上创建的TDengine目录下，在源代码的src/connector/python文件夹下可以找到linux和windows两个版本的安装包。进入linux目录，通过pip命令来安装taos。

pip install python3/

3、执行命令python进入交互式界面，输入import taos报错。

4、出现上面错误的原因是：libtaos.so文件在/usr/lib目录下，而centos默认不会找/usr/lib下的文件。

解决方法：在环境变量/etc/profile中配置libtaos.so路径。

编辑配置文件：vim /etc/profile
添加配置：export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/lib
使修改生效：source /etc/profile

5、再次进入python交互式界面，导入taos模块成功。

python接收MQTT数据

Windows端安装Python Connector：

1、利用pip安装TDengine源代码中src\connector\python\windows目录下的python3对应的taos模块。

2、从官网https://www.taosdata.com/cn/getting-started/中下载TDengine Windows客户端tdengine-windows-client-x64-1.6.1.6.exe。

3、双击exe可执行文件，安装客户端，默认下一步到安装完成即可。

4、通过代码，查看windows是否可以访问TDengine

#!/usr/bin/env python
# coding=utf-8
#######################################################################
#    > File Name: 
#    > Author: cuiyufei
#    > Mail: XXX@qq.com
#    > Created Time: 2019年8月22日
#######################################################################
import taos
import pandas as pdif __name__ == '__main__':  #连接taos数据库conn = taos.connect(host='10.1.131.14', user='root', password='taosdata', database='log')# 通过获取到的数据库连接conn下的cursor()方法来创建游标cursor = conn.cursor()# Create a database named db#try:#    cursor.execute('create database test1')#except Exception as err:#    conn.close()#    raise(err)sql = "SELECT * FROM log"# read_sql 方法返回的数据类型是DataFramedataframe = pd.read_sql(sql, con=conn)print dataframe 

出现如下图错误。

最后发现，是由于客户端版本和服务端版本不一致导致的。

taos -h you IP

windows端：

Linux端：

taos -V

升级Linux端的taos版本

systemctl stop taosd
rpm -qa | grep -i tdengine
rpm -e tdengine-1.6.1.5-3.el7.x86_64rpm -ivh tdengine-1.6.1.6-3.el7.x86_64.rpm 
systemctl start taosd
systemctl status taosd

升级后，连接成功。

二、python操作TDengine

官网python的相关例子https://github.com/taosdata/TDengine/blob/master/tests/examples/python/read_example.py

#!/usr/bin/env python
# coding=utf-8
#######################################################################
#    > File Name: 
#    > Author: cuiyufei
#    > Mail: XXX@qq.com
#    > Created Time: 2019年8月22日
#######################################################################
import taos
import sys
import datetime
import randomdef exitProgram(conn):conn.close()sys.exit()if __name__ == '__main__':start_time = datetime.datetime(2019, 7, 1)time_interval = datetime.timedelta(seconds=60)# Connect to TDengine server.# # parameters:# @host     : TDengine server IP address # @user     : Username used to connect to TDengine server# @password : Password # @database : Database to use when connecting to TDengine server# @config   : Configuration directoryconn = taos.connect(host="your IP", user="root", password="taosdata", config="/etc/taos")# Generate a cursor object to run SQL commandsc1 = conn.cursor()# Create a database named dbtry:c1.execute('create database db')except Exception as err:conn.close()raise(err)# use databasetry:c1.execute('use db')except Exception as err:conn.close()raise(err)# create tabletry:c1.execute('create table if not exists t (ts timestamp, a int, b float, c binary(20))')except Exception as err:conn.close()raise(err)# insert data for i in range(10000):try:c1.execute("insert into t values ('%s', %d, %f, '%s')" % (start_time, random.randint(1,10), random.randint(1,10)/10.0, 'hello'))except Exception as err:conn.close()raise(err)start_time += time_interval# query data and return data in the form of listtry:c1.execute('select * from db.t')except Exception as err:conn.close()raise(err)# Column names are in c1.description listcols = c1.description# Use fetchall to fetch data in a listdata = c1.fetchall()try:c1.execute('select * from db.t')except Exception as err:conn.close()raise(err)# Use iterator to go through the retreived datafor col in c1:print(col)conn.close()

三、接收mqtt数据并存入TDengine

从mqtt接收数据，通过模型进行数据预测，然后把数据写入TDengine

#!/usr/bin/env python
# coding=utf-8
#######################################################################
#    > File Name: 
#    > Author: cuiyufei
#    > Mail: XXX@qq.com
#    > Created Time: 2019年8月22日
#######################################################################
import paho.mqtt.client as mqtt
import datetime
from sklearn.externals import joblib
from influxdb import InfluxDBClient
import taos
import random# INFLUXDB
TAOS_HOST = "your IP"
TAOS_USER = "root"
TAOS_PASSWORD = "taosdata"
TASO_DB = 'your database'# MQTT
MQTT_HOST = "your IP"
MQTT_PORT = PORT
MQTT_TOPIC = 'your topic'def insert_influxdb(msg):current_time = datetime.datetime.utcnow()#.isoformat()print(msg.topic+" " + ":" + str(msg.payload))feats = ['wind_speed', 'generator_speed', 'power', 'wind_direction','wind_direction_mean', 'yaw_position', 'yaw_speed', 'pitch1_angle','pitch2_angle', 'pitch3_angle', 'pitch1_speed', 'pitch2_speed','pitch3_speed', 'pitch1_moto_tmp', 'pitch2_moto_tmp', 'pitch3_moto_tmp','acc_x', 'acc_y', 'environment_tmp', 'int_tmp','pitch1_ng5_tmp', 'pitch2_ng5_tmp', 'pitch3_ng5_tmp', 'pitch1_ng5_DC','pitch2_ng5_DC', 'pitch3_ng5_DC', 'group']feats_value = [float(value) for value in (eval(msg.payload)[1:])]# 风机预测model = './model/gbm.pkl'clf = joblib.load(model)sub_preds = clf.predict_proba([feats_value], num_iteration=clf.best_iteration_)[:, 1]print(sub_preds[0])conn = taos.connect(host=TAOS_HOST, user=TAOS_USER, password=TAOS_PASSWORD, database=TASO_DB)cursor = conn.cursor()try:#cursor.execute("insert into t values ('%s', %d, %.14f, '%s')" % (current_time, feats_value[0], feats_value[1], 'hello'))cursor.execute("insert into wind_driven_generator values ('%s', %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f, %f)" %(current_time, feats_value[0], feats_value[1], feats_value[2], feats_value[3], feats_value[4], feats_value[5], feats_value[6], feats_value[7], feats_value[8], feats_value[9], feats_value[10], feats_value[11], feats_value[12], feats_value[13], feats_value[14], feats_value[15], feats_value[16], feats_value[17], feats_value[18], feats_value[19], feats_value[20], feats_value[21], feats_value[22], feats_value[23], feats_value[24], feats_value[25], feats_value[26], sub_preds[0]))except Exception as err:conn.close()raise(err)conn.close()
if __name__ == '__main__':#influxdb_client = InfluxDBClient(host=INFLUXDB_HOST, port=INFLUXDB_PORT, username=INFLUXDB_USER, password=INFLUXDB_PASSWORD, database=INFLUXDB_DB)mqtt_client = mqtt.Client()mqtt_client.on_connect = lambda self, mosq, obj, rc: self.subscribe(MQTT_TOPIC)mqtt_client.on_message = lambda client, userdata, msg: insert_influxdb(msg)mqtt_client.connect(MQTT_HOST, MQTT_PORT, 60)try:mqtt_client.loop_forever()except KeyboardInterrupt:print("stop...")

参考文章

Python操作TDengine

https://blog.csdn.net/jinwang3526/article/details/81537297

https://blog.csdn.net/qq_28877125/article/details/78325003

涛思官网