|
@@ -1,6 +1,7 @@
|
|
|
import datetime
|
|
|
import json
|
|
|
import os
|
|
|
+import time
|
|
|
from MyUtils.ZillionUtil import ZillionUtil
|
|
|
from MyUtils.MysqlUtils import MysqlUtils
|
|
|
from dateutil.relativedelta import relativedelta
|
|
@@ -17,7 +18,14 @@ def get_data_time(hbase_database, hbase_table, building, from_time, to_time):
|
|
|
Criteria = {
|
|
|
"building": building,
|
|
|
"energyModelSign": building + "EM01",
|
|
|
- "energyModelNodeSign": "EI1001",
|
|
|
+ "energyModelNodeSign": {
|
|
|
+ "in":[
|
|
|
+ "EI10102010101001",
|
|
|
+ "EI10102010102001",
|
|
|
+ "EI1010202001",
|
|
|
+ "EI1001"
|
|
|
+ ]
|
|
|
+ },
|
|
|
"data_time": {
|
|
|
"$gte": from_time,
|
|
|
"$lt": to_time
|
|
@@ -25,14 +33,14 @@ def get_data_time(hbase_database, hbase_table, building, from_time, to_time):
|
|
|
}
|
|
|
datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
|
|
|
return datas
|
|
|
-# mysql插入数据
|
|
|
-def insert_mysql(sqls,building,my_table):
|
|
|
- print("%s,开始往mysql插入%s的%s数据..."%(datetime_now(),building,my_table))
|
|
|
- for i in range(0, len(sqls), 1000):
|
|
|
- sqlranges = sqls[i:i + 1000]
|
|
|
- sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
|
|
|
- MysqlUtil.update(sqlranges)
|
|
|
- print("%s,mysql数据%s,项目%s插入成功,合计%s条..." % (datetime_now(),my_table,building,len(sqls)))
|
|
|
+# # mysql插入数据
|
|
|
+# def insert_mysql(sqls,building,my_table):
|
|
|
+# print("%s,开始往mysql插入%s的%s数据..."%(datetime_now(),building,my_table))
|
|
|
+# for i in range(0, len(sqls), 1000):
|
|
|
+# sqlranges = sqls[i:i + 1000]
|
|
|
+# sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
|
|
|
+# MysqlUtil.update(sqlranges)
|
|
|
+# print("%s,mysql数据%s,项目%s插入成功,合计%s条..." % (datetime_now(),my_table,building,len(sqls)))
|
|
|
|
|
|
|
|
|
# with open("config.json", "r") as f:
|
|
@@ -54,34 +62,63 @@ mysql = {
|
|
|
"passwd": os.getenv("passwd"),
|
|
|
"database": os.getenv("database")
|
|
|
}
|
|
|
+my_database = os.getenv("database")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+while True:
|
|
|
+ time_now = time.strftime("%H:%M:%S", time.localtime()) # 刷新
|
|
|
+ set_time = "05:00:00"
|
|
|
+ if time_now == set_time:#此处设置每天定时的时间
|
|
|
+ today = datetime.date.today().strftime("%Y%m%d")
|
|
|
+ yesterday = (datetime.date.today()-datetime.timedelta(days=1)).strftime("%Y%m%d")
|
|
|
+
|
|
|
+ # #连接hbase
|
|
|
+ zillionUtil = ZillionUtil(url)
|
|
|
+ # #连接hbase
|
|
|
+ MysqlUtil = MysqlUtils(**mysql)
|
|
|
+ datas = get_data_time(hbase_database, "data_energydata_1d", building, yesterday, today)
|
|
|
+ if datas:
|
|
|
+ project_id = "Pj" + building
|
|
|
+ # 删除上月数据
|
|
|
+ print("%s,开始删除%s的数据..."%(datetime_now(),yesterday))
|
|
|
+ delete_sql = DELETE_SQL% (project_id,yesterday,today)
|
|
|
+ MysqlUtil.update(delete_sql)
|
|
|
+ # sqls = []
|
|
|
+ energy_cooling = "0"
|
|
|
+ energy_heating = "0"
|
|
|
+ energy_ac_terminal = "0"
|
|
|
+ energy_light = "0"
|
|
|
+ sum_data_value = "0"
|
|
|
+ for i in datas:
|
|
|
+
|
|
|
+ if i["energyModelNodeSign"] == "EI1001":
|
|
|
+ sum_data_value = i["data_value"]
|
|
|
+
|
|
|
+ #冷热源
|
|
|
+ if i["energyModelNodeSign"] == "EI10102010101001":
|
|
|
+ energy_cooling = i["data_value"]
|
|
|
+ # #热
|
|
|
+ # if i["energyModelNodeSign"] == "EI1010201010101001":
|
|
|
+ # energy_heating = i["data_value"]
|
|
|
+ #空调末端
|
|
|
+ if i["energyModelNodeSign"] == "EI10102010102001":
|
|
|
+ energy_ac_terminal = i["data_value"]
|
|
|
+ #照明
|
|
|
+ if i["energyModelNodeSign"] == "EI1010202001":
|
|
|
+ energy_light = i["data_value"]
|
|
|
+ energy_other = float(sum_data_value) - float(energy_cooling) - float(energy_heating) -float(energy_light) - float(energy_ac_terminal)
|
|
|
+ sql = "('%s','%s','%s','%s','%s','%s','%s','%s','%s')" % (
|
|
|
+ project_id, yesterday,energy_cooling,energy_heating,energy_ac_terminal,energy_light,energy_other, datetime_now(), datetime_now())
|
|
|
+ inser_sql = INSERT_SQL % (my_database, "energy_week_day") + sql
|
|
|
+ print("%s,开始插入数据..."%datetime_now())
|
|
|
+ MysqlUtil.update(inser_sql)
|
|
|
+
|
|
|
+
|
|
|
+ else:
|
|
|
+ print("%s,没有查询到数据...")
|
|
|
+ time.sleep(2)
|
|
|
+ print("%s,等待下次程序执行时间%s" % (datetime_now(), set_time))
|
|
|
|
|
|
|
|
|
|
|
|
-
|
|
|
-
|
|
|
-now_month = datetime.date.today().strftime("%Y%m")
|
|
|
-last_month = (datetime.date.today() - relativedelta(months=1)).strftime("%Y%m")
|
|
|
-
|
|
|
-
|
|
|
-# #连接hbase
|
|
|
-zillionUtil = ZillionUtil(url)
|
|
|
-# #连接hbase
|
|
|
-MysqlUtil = MysqlUtils(**mysql)
|
|
|
-datas = get_data_time(hbase_database, "data_energydata_1d", building, last_month, now_month)
|
|
|
-if datas:
|
|
|
- project_id = "Pj" + building
|
|
|
- # 删除上月数据
|
|
|
- print("%s,开始删除%s的数据..."%(datetime_now(),last_month))
|
|
|
- delete_sql = DELETE_SQL% (project_id,last_month,now_month)
|
|
|
- MysqlUtil.update(delete_sql)
|
|
|
- sqls = []
|
|
|
- for i in datas:
|
|
|
- dt = i["data_time"][0:8]
|
|
|
- data_value = i["data_value"]
|
|
|
- sql = "('%s','%s','0','0','0','0','%s','%s','%s')" % (project_id, dt, data_value, datetime_now(),datetime_now())
|
|
|
- sqls.append(sql)
|
|
|
- insert_mysql(sqls,building,"energy_week_day")
|
|
|
-else:
|
|
|
- print("%s,没有查询到数据...")
|
|
|
-
|
|
|
-
|