main.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. import datetime
  2. import json
  3. import os
  4. import time
  5. from MyUtils.ZillionUtil import ZillionUtil
  6. from MyUtils.MysqlUtils import MysqlUtils
  7. import pytz
  8. INSERT_SQL = "replace into %s.%s(project_id,date,energy_cooling,energy_heating,energy_ac_terminal,energy_light,energy_others,create_time,update_time) values "
  9. DELETE_SQL = "DELETE FROM `energy_week_day` WHERE `project_id` = '%s' AND `date` >= '%s' AND `date` < '%s' "
  10. def datetime_now():
  11. # datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
  12. #容器时间
  13. # tz = pytz.timezone('Asia/Shanghai') # 东八区
  14. datetime_now = datetime.datetime.fromtimestamp(int(time.time()),
  15. pytz.timezone('Asia/Shanghai')).strftime('%Y%m%d%H%M%S')
  16. return datetime_now
  17. # 获取能耗数据
  18. def get_data_time(hbase_database, hbase_table, building, from_time, to_time):
  19. Criteria = {
  20. "building": building,
  21. "energyModelSign": building + "EM01",
  22. "energyModelNodeSign": {
  23. "in":[
  24. "EI10102010101001",
  25. "EI10102010102001",
  26. "EI1010202001",
  27. "EI1001"
  28. ]
  29. },
  30. "data_time": {
  31. "$gte": from_time,
  32. "$lt": to_time
  33. }
  34. }
  35. datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
  36. return datas
  37. # # mysql插入数据
  38. # def insert_mysql(sqls,building,my_table):
  39. # print("%s,开始往mysql插入%s的%s数据..."%(datetime_now(),building,my_table))
  40. # for i in range(0, len(sqls), 1000):
  41. # sqlranges = sqls[i:i + 1000]
  42. # sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
  43. # MysqlUtil.update(sqlranges)
  44. # print("%s,mysql数据%s,项目%s插入成功,合计%s条..." % (datetime_now(),my_table,building,len(sqls)))
  45. # with open("config.json", "r") as f:
  46. # data = json.load(f)
  47. # hbase_database = data["metadata"]["database"]
  48. # url = data["metadata"]["url"]
  49. # building = data["building"]
  50. # mysql = data["mysql"]
  51. # my_database = mysql["database"]
  52. hbase_database = os.getenv("hbase_database")
  53. url = os.getenv("url")
  54. building = os.getenv("building")
  55. mysql = {
  56. "host": os.getenv("host"),
  57. "port": os.getenv("port"),
  58. "user": os.getenv("user"),
  59. "passwd": os.getenv("passwd"),
  60. "database": os.getenv("database")
  61. }
  62. my_database = os.getenv("database")
  63. print("同步%s项目数据"%(building))
  64. while True:
  65. # 容器里获取时间
  66. # tz = pytz.timezone('Asia/Shanghai') # 东八区
  67. time_now = datetime.datetime.fromtimestamp(int(time.time()),
  68. pytz.timezone('Asia/Shanghai')).strftime('%H:%M:%S')
  69. # time_now = time.strftime("%H:%M:%S", time.localtime()) # 刷新
  70. am_set_time = "16:34:00"
  71. pm_set_time = "22:00:00"
  72. if time_now == am_set_time or time_now ==pm_set_time:#此处设置每天定时的时间
  73. today = datetime.date.today().strftime("%Y%m%d")
  74. yesterday = (datetime.date.today()-datetime.timedelta(days=1)).strftime("%Y%m%d")
  75. # #连接hbase
  76. zillionUtil = ZillionUtil(url)
  77. # #连接hbase
  78. MysqlUtil = MysqlUtils(**mysql)
  79. datas = get_data_time(hbase_database, "data_energydata_1d", building, yesterday, today)
  80. if datas:
  81. project_id = "Pj" + building
  82. # 删除上月数据
  83. print("%s,开始删除%s的数据..."%(datetime_now(),yesterday))
  84. delete_sql = DELETE_SQL% (project_id,yesterday,today)
  85. MysqlUtil.update(delete_sql)
  86. # sqls = []
  87. energy_cooling = "0"
  88. energy_heating = "0"
  89. energy_ac_terminal = "0"
  90. energy_light = "0"
  91. sum_data_value = "0"
  92. for i in datas:
  93. if i["energyModelNodeSign"] == "EI1001":
  94. sum_data_value = i["data_value"]
  95. #冷热源
  96. if i["energyModelNodeSign"] == "EI10102010101001":
  97. energy_cooling = i["data_value"]
  98. # #热
  99. # if i["energyModelNodeSign"] == "EI1010201010101001":
  100. # energy_heating = i["data_value"]
  101. #空调末端
  102. if i["energyModelNodeSign"] == "EI10102010102001":
  103. energy_ac_terminal = i["data_value"]
  104. #照明
  105. if i["energyModelNodeSign"] == "EI1010202001":
  106. energy_light = i["data_value"]
  107. energy_other = float(sum_data_value) - float(energy_cooling) - float(energy_heating) -float(energy_light) - float(energy_ac_terminal)
  108. sql = "('%s','%s','%s','%s','%s','%s','%s','%s','%s')" % (
  109. project_id, yesterday,energy_cooling,energy_heating,energy_ac_terminal,energy_light,energy_other, datetime_now(), datetime_now())
  110. inser_sql = INSERT_SQL % (my_database, "energy_week_day") + sql
  111. print("%s,开始插入数据..."%datetime_now())
  112. MysqlUtil.update(inser_sql)
  113. else:
  114. print("%s,没有查询到数据...")
  115. time.sleep(2)
  116. print("%s,等待下次程序执行" % (datetime_now()))