82 lines
3.7 KiB
Python
82 lines
3.7 KiB
Python
import influxdb_api
|
||
import globals
|
||
from datetime import datetime, timedelta, timezone
|
||
import schedule
|
||
import time
|
||
from influxdb_client import InfluxDBClient, BucketsApi, WriteApi, OrganizationsApi, Point, QueryApi
|
||
|
||
|
||
# 2025/02/06
|
||
def get_next_period_time() -> str:
|
||
"""
|
||
获取下一个6小时时间点，返回格式为字符串'YYYY-MM-DDTHH:00:00+08:00'
|
||
:return: 返回字符串格式的时间，表示下一个6小时执行时间点
|
||
"""
|
||
# 获取当前时间，并设定为北京时间
|
||
now = datetime.now() # now 类型为 datetime，表示当前本地时间
|
||
# 获取当前的小时数并计算下一个6小时时间点
|
||
next_period_hour = (now.hour // 6 + 1) * 6 - 6 # next_period_hour 类型为 int，表示下一个6小时时间点的小时部分
|
||
# 如果计算的小时大于23，表示进入第二天，调整为00:00
|
||
if next_period_hour >= 24:
|
||
next_period_hour = 0
|
||
now = now + timedelta(days=1) # 如果超过24小时，日期增加1天
|
||
# 将秒和微秒部分清除，构建出下一个6小时点的datetime对象
|
||
next_period_time = now.replace(hour=next_period_hour, minute=0, second=0, microsecond=0)
|
||
return next_period_time.strftime('%Y-%m-%dT%H:%M:%S+08:00') # 格式化为指定的字符串格式并返回
|
||
|
||
|
||
# 2025/02/06
|
||
def store_non_realtime_SCADA_data_job() -> None:
|
||
"""
|
||
定义的任务2，每6小时执行一次，在0点、6点、12点、18点执行，执行时，更新get_history_data_end_time并调用store_non_realtime_SCADA_data_to_influxdb函数
|
||
:return: None
|
||
"""
|
||
# 获取当前时间
|
||
current_time = datetime.now()
|
||
# 只在0点、6点、12点、18点执行任务
|
||
if current_time.hour % 6 == 0 and current_time.minute == 0:
|
||
# 获取下一个6小时的时间点，并更新get_history_data_end_time
|
||
get_history_data_end_time: str = get_next_period_time() # get_history_data_end_time 类型为 str，格式为'2025-02-06T12:00:00+08:00'
|
||
print(get_history_data_end_time)
|
||
# 调用函数执行任务
|
||
influxdb_api.store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time)
|
||
print('{} -- Successfully store non realtime SCADA data.'.format(
|
||
datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
|
||
|
||
|
||
# 2025/02/06
|
||
def store_non_realtime_SCADA_data_task() -> None:
|
||
"""
|
||
定时执行6小时的任务，使用schedule库每分钟执行一次store_non_realtime_SCADA_data_job函数。
|
||
该任务会一直运行，定期调用store_non_realtime_SCADA_data_job获取SCADA数据。
|
||
:return:
|
||
"""
|
||
# 等待到整分对齐
|
||
now = datetime.now()
|
||
wait_seconds = 60 - now.second
|
||
time.sleep(wait_seconds)
|
||
try:
|
||
# 每分钟检查一次，执行store_non_realtime_SCADA_data_job
|
||
schedule.every(1).minute.at(":00").do(store_non_realtime_SCADA_data_job)
|
||
|
||
# 持续执行任务，检查是否有待执行的任务
|
||
while True:
|
||
schedule.run_pending() # 执行所有待处理的定时任务
|
||
time.sleep(1) # 暂停1秒，避免过于频繁的任务检查
|
||
pass
|
||
except Exception as e:
|
||
print(f"Error occurred in store_non_realtime_SCADA_data_task: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
url = "http://localhost:8086" # 替换为你的InfluxDB实例地址
|
||
token = "Z4UZj9HuLwLlwoApywvT2nGVP3bwLy18y-sJQ7enzZlJd8YMzMWbBA6F-q4gBiZ-7-IqdxR5aR9LvicKiSNmnA==" # 替换为你的InfluxDB Token
|
||
org_name = "beibei" # 替换为你的Organization名称
|
||
|
||
client = InfluxDBClient(url=url, token=token)
|
||
|
||
# step2: 先查询pg数据库中scada_info的信息，然后存储SCADA数据到SCADA_data这个bucket里
|
||
influxdb_api.query_pg_scada_info_non_realtime('bb')
|
||
# 自动执行
|
||
store_non_realtime_SCADA_data_task()
|