diff --git a/influxdb_api.py b/influxdb_api.py index 747b460..c436bf9 100644 --- a/influxdb_api.py +++ b/influxdb_api.py @@ -1,5 +1,4 @@ from influxdb_client import InfluxDBClient, BucketsApi, WriteApi, OrganizationsApi, Point, QueryApi, WriteOptions -from influxdb_client.client.exceptions import InfluxDBError from typing import List, Dict from datetime import datetime, timedelta, timezone from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS @@ -103,7 +102,8 @@ def query_pg_scada_info_non_realtime(name: str) -> None: open_project(name) dic_time = get_time(name) globals.hydraulic_timestep = dic_time['HYDRAULIC TIMESTEP'] - close_project(name) + # DingZQ, 2025-03-21 + #close_project(name) # 连接数据库 conn_string = f"dbname={name} host=127.0.0.1" try: @@ -231,7 +231,8 @@ def create_and_initialize_buckets(org_name: str) -> None: if not client.ping(): print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) # 先删除原有的，然后再进行初始化 - delete_buckets(org_name) + # delete_buckets(org_name) + bucket_api = BucketsApi(client) # 本地变量，用于记录成功写入的数据点数量 points_written = 0 @@ -284,8 +285,7 @@ def create_and_initialize_buckets(org_name: str) -> None: .tag("device_ID", None) \ .field("monitored_value", 0.0) \ .field("datacleaning_value", 0.0) \ - .field("datafilling_value", 0.0) \ - .field("cleaned_value", 0.0) \ + .field("simulation_value", 0.0) \ .time("2024-11-21T00:00:00Z", write_precision='s') points_to_write.append(point) # write_api.write(bucket="SCADA_data", org=org_name, record=point) @@ -1500,6 +1500,7 @@ def query_all_SCADA_records_by_date(query_date: str, bucket: str="SCADA_data") - return SCADA_results + def query_SCADA_data_by_device_ID_and_time(query_ids_list: List[str], query_time: str, bucket: str="SCADA_data") -> Dict[str, float]: """ 根据SCADA设备的ID和时间查询值 @@ -1545,6 +1546,7 @@ def query_SCADA_data_by_device_ID_and_time(query_ids_list: List[str], query_time print(f"Error querying InfluxDB for device ID {device_id}: {e}") SCADA_result_dict[device_id] = None client.close() + return SCADA_result_dict @@ -1593,7 +1595,9 @@ def query_scheme_SCADA_data_by_device_ID_and_time(query_ids_list: List[str], que except Exception as e: print(f"Error querying InfluxDB for device ID {device_id}: {e}") SCADA_result_dict[device_id] = None + client.close() + return SCADA_result_dict # 2025/03/14 @@ -1672,10 +1676,11 @@ def query_SCADA_data_by_device_ID_and_date(query_ids_list: List[str], query_date return query_SCADA_data_by_device_ID_and_time_range(query_ids_list, str(start_time), str(end_time), bucket) + # 2025/04/17 -def query_cleaning_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start_time: str, end_time: str, bucket: str="SCADA_data"): +def query_cleaned_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start_time: str, end_time: str, bucket: str="SCADA_data"): """ - 查询指定时间范围内，多个SCADA设备的修复的单个的数据 + 查询指定时间范围内，多个SCADA设备的清洗后的数据 :param query_ids_list: SCADA设备ID的列表 :param start_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 :param end_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 @@ -1722,106 +1727,6 @@ def query_cleaning_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[st return SCADA_dict -# 2025/04/22 -def query_filling_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start_time: str, end_time: str, bucket: str="SCADA_data"): - """ - 查询指定时间范围内，多个SCADA设备的填补的单个的数据 - :param query_ids_list: SCADA设备ID的列表 - :param start_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 - :param end_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 - :param bucket: InfluxDB 的 bucket 名称，默认值为 "SCADA_data"。 - :return: - """ - client = get_new_client() - if not client.ping(): - print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) - - query_api = client.query_api() - print('start_time', start_time) - print('end_time', end_time) - # 将北京时间转换为 UTC 时间 - beijing_start_time = datetime.fromisoformat(start_time) - print('beijing_start_time', beijing_start_time) - utc_start_time = time_api.to_utc_time(beijing_start_time) - print('utc_start_time', utc_start_time) - beijing_end_time = datetime.fromisoformat(end_time) - print('beijing_end_time', beijing_end_time) - utc_stop_time = time_api.to_utc_time(beijing_end_time) - print('utc_stop_time', utc_stop_time) - SCADA_dict = {} - for device_id in query_ids_list: - flux_query = f''' - from(bucket: "{bucket}") - |> range(start: {utc_start_time.isoformat()}, stop: {utc_stop_time.isoformat()}) - |> filter(fn: (r) => r["device_ID"] == "{device_id}" and r["_field"] == "datafilling_value") - |> sort(columns: ["_time"]) - ''' - # 执行查询，返回一个 FluxTable 列表 - tables = query_api.query(flux_query) - print(tables) - records_list = [] - for table in tables: - for record in table.records: - # 获取记录的时间和监测值 - records_list.append({ - "time": record["_time"], - "value": record["_value"] - }) - SCADA_dict[device_id] = records_list - client.close() - return SCADA_dict - - -# 2025/04/22 -def query_cleaned_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start_time: str, end_time: str, bucket: str="SCADA_data"): - """ - 查询指定时间范围内，多个SCADA设备的清洗完毕后的完整数据 - :param query_ids_list: SCADA设备ID的列表 - :param start_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 - :param end_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。 - :param bucket: InfluxDB 的 bucket 名称，默认值为 "SCADA_data"。 - :return: - """ - client = get_new_client() - if not client.ping(): - print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) - - query_api = client.query_api() - print('start_time', start_time) - print('end_time', end_time) - # 将北京时间转换为 UTC 时间 - beijing_start_time = datetime.fromisoformat(start_time) - print('beijing_start_time', beijing_start_time) - utc_start_time = time_api.to_utc_time(beijing_start_time) - print('utc_start_time', utc_start_time) - beijing_end_time = datetime.fromisoformat(end_time) - print('beijing_end_time', beijing_end_time) - utc_stop_time = time_api.to_utc_time(beijing_end_time) - print('utc_stop_time', utc_stop_time) - SCADA_dict = {} - for device_id in query_ids_list: - flux_query = f''' - from(bucket: "{bucket}") - |> range(start: {utc_start_time.isoformat()}, stop: {utc_stop_time.isoformat()}) - |> filter(fn: (r) => r["device_ID"] == "{device_id}" and r["_field"] == "cleaned_value") - |> sort(columns: ["_time"]) - ''' - # 执行查询，返回一个 FluxTable 列表 - tables = query_api.query(flux_query) - print(tables) - records_list = [] - for table in tables: - for record in table.records: - # 获取记录的时间和监测值 - records_list.append({ - "time": record["_time"], - "value": record["_value"] - }) - SCADA_dict[device_id] = records_list - client.close() - return SCADA_dict - - # 2025/02/01 def store_realtime_simulation_result_to_influxdb(node_result_list: List[Dict[str, any]], link_result_list: List[Dict[str, any]], result_start_time: str, bucket: str = "realtime_simulation_result"): @@ -2123,8 +2028,7 @@ def query_all_records_by_date(query_date: str, bucket: str="realtime_simulation_ client = get_new_client() # 记录开始时间 time_cost_start = time.perf_counter() - print('{} -- Hydraulic simulation started.'.format( - datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S'))) + print('{} -- query_all_records_by_date started.'.format(datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S'))) if not client.ping(): print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) @@ -3401,8 +3305,6 @@ def upload_cleaned_SCADA_data_to_influxdb(file_path: str, bucket: str="SCADA_dat datetime_value = datetime.strptime(row['time'], '%Y-%m-%d %H:%M:%S%z') # 处理datacleaning_value为空的情况 datacleaning_value = float(row['datacleaning_value']) if row['datacleaning_value'] else None - datafilling_value = float(row['datafilling_value']) if row['datafilling_value'] else None - cleaned_value = float(row['cleaned_value']) if row['cleaned_value'] else None # 处理monitored_value字段类型错误 try: monitored_value = float(row['monitored_value']) if row['monitored_value'] else None @@ -3416,8 +3318,6 @@ def upload_cleaned_SCADA_data_to_influxdb(file_path: str, bucket: str="SCADA_dat 'description': row['description'], 'monitored_value': monitored_value, 'datacleaning_value': datacleaning_value, - 'datafilling_value': datafilling_value, - 'cleaned_value': cleaned_value, 'datetime': datetime_value }) @@ -3429,6 +3329,7 @@ def upload_cleaned_SCADA_data_to_influxdb(file_path: str, bucket: str="SCADA_dat write_api = client.write_api(write_options=SYNCHRONOUS) # 写入数据 for data in data_list: + print(data) # 创建Point对象 point = ( Point(data['measurement']) # measurement为mpointName @@ -3437,8 +3338,6 @@ def upload_cleaned_SCADA_data_to_influxdb(file_path: str, bucket: str="SCADA_dat .tag('description', data['description']) .field("monitored_value", data['monitored_value']) # field key为dataValue .field('datacleaning_value', data['datacleaning_value']) - .field('datafilling_value', data['datafilling_value']) - .field('cleaned_value', data['cleaned_value']) .time(data['datetime']) # 时间以datetime为准 ) @@ -3462,11 +3361,11 @@ if __name__ == "__main__": client = InfluxDBClient(url=url, token=token) # step1: 检查连接状态，初始化influxdb的buckets - # try: - # # delete_buckets(org_name) - # create_and_initialize_buckets(org_name) - # except Exception as e: - # print(f"连接失败: {e}") + try: + # delete_buckets(org_name) + create_and_initialize_buckets(org_name) + except Exception as e: + print(f"连接失败: {e}") # step2: 先查询pg数据库中scada_info的信息，然后存储SCADA数据到SCADA_data这个bucket里 @@ -3485,7 +3384,7 @@ if __name__ == "__main__": # store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time='2025-03-08T12:00:00+08:00') # 示例3：download_history_data_manually - # download_history_data_manually(begin_time='2025-04-17T00:00:00+08:00', end_time='2025-04-17T23:59:00+08:00') + # download_history_data_manually(begin_time='2025-04-16T00:00:00+08:00', end_time='2025-04-16T23:59:00+08:00') # step3: 查询测试示例