OrgTJWater/TJWaterServer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update influxdb api

Browse Source
This commit is contained in:
DingZQ
2025-03-24 21:59:17 +08:00
parent 4a824707de
commit 49be1c7089
2 changed files with 168 additions and 3009 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2865
influxdb_api(2).py
View File

File diff suppressed because it is too large Load Diff

312
influxdb_api.py
View File

@@ -168,14 +168,20 @@ def query_pg_scada_info_non_realtime(name: str) -> None:
print(f"查询时发生错误：{e}")
# 2025/03/23
def get_new_client() -> InfluxDBClient:
"""每次调用返回一个新的 InfluxDBClient 实例。"""
return InfluxDBClient(url=url, token=token, org=org_name)
# 2025/02/01
def delete_buckets(client: InfluxDBClient, org_name: str) -> None:
def delete_buckets(org_name: str) -> None:
"""
删除InfluxDB中指定organization下的所有buckets。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
:param org_name: InfluxDB中organization的名称。
:return: None
"""
client = get_new_client()
# 定义需要删除的 bucket 名称列表
buckets_to_delete = ['SCADA_data', 'realtime_simulation_result', 'scheme_simulation_result']
buckets_api = client.buckets_api()
@@ -193,18 +199,19 @@ def delete_buckets(client: InfluxDBClient, org_name: str) -> None:
print(f"Skipping bucket {bucket.name}. Not in the deletion list.")
else:
print("未找到 buckets 属性，无法迭代 buckets。")
client.close()
# 2025/02/01
def create_and_initialize_buckets(client: InfluxDBClient, org_name: str) -> None:
def create_and_initialize_buckets(org_name: str) -> None:
"""
初始化influxdb的三个数据存储库，分别为SCADA_data、realtime_simulation_result、scheme_simulation_result
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
:param org_name: InfluxDB中organization的名称
:return:
"""
client = get_new_client()
# 先删除原有的，然后再进行初始化
delete_buckets(client, org_name)
delete_buckets(org_name)
bucket_api = BucketsApi(client)
write_api = client.write_api()
org_api = OrganizationsApi(client)
@@ -321,16 +328,17 @@ def create_and_initialize_buckets(client: InfluxDBClient, org_name: str) -> None
write_api.write(bucket=bucket, org=org_name, record=points_to_write)
write_api.flush() # 刷新缓存一次
print("All buckets created and initialized successfully.")
client.close()
def store_realtime_SCADA_data_to_influxdb(get_real_value_time: str, bucket: str = "SCADA_data", client: InfluxDBClient = client) -> None:
def store_realtime_SCADA_data_to_influxdb(get_real_value_time: str, bucket: str = "SCADA_data") -> None:
"""
将SCADA数据通过数据接口导入数据库
:param get_real_value_time: 获取数据的时间，格式如'2024-11-25T09:00:00+08:00'
:param bucket: (str): InfluxDB 的 bucket 名称，默认值为 "SCADA_data"。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
: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')))
@@ -638,6 +646,7 @@ def store_realtime_SCADA_data_to_influxdb(get_real_value_time: str, bucket: str
if points_to_write:
write_api.write(bucket=bucket, org=org_name, record=points_to_write)
write_api.flush() # 刷新缓存一次
client.close()
def convert_time_format(original_time: str) -> str:
@@ -652,14 +661,14 @@ def convert_time_format(original_time: str) -> str:
# 2025/01/10
def store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time: str, bucket: str = "SCADA_data", client: InfluxDBClient = client) -> None:
def store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time: str, bucket: str = "SCADA_data") -> None:
"""
获取某段时间内传回的scada数据
:param get_history_data_end_time: 获取历史数据的终止时间时间，格式如'2024-11-25T09:00:00+08:00'
:param bucket: (str): InfluxDB 的 bucket 名称，默认值为 "SCADA_data"。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
: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')))
@@ -891,19 +900,19 @@ def store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time: str, bu
if points_to_write:
write_api.write(bucket=bucket, org=org_name, record=points_to_write)
write_api.flush() # 刷新缓存一次
client.close()
# 2025/03/01
def download_history_data_manually(begin_time: str, end_time: str, bucket: str = "SCADA_data",
client: InfluxDBClient = client) -> None:
def download_history_data_manually(begin_time: str, end_time: str, bucket: str = "SCADA_data") -> None:
"""
获取某个时间段内所有SCADA设备的历史数据，非实时执行，手动补充数据版
:param begin_time: 获取历史数据的开始时间，格式如'2024-11-25T09:00:00+08:00'
:param end_time: 获取历史数据的结束时间，格式如'2024-11-25T09:00:00+08:00'
:param bucket: InfluxDB 的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -1384,15 +1393,15 @@ def query_all_SCADA_records_by_date(query_date: str, bucket: str="SCADA_data", c
return SCADA_results
def query_SCADA_data_by_device_ID_and_time(query_ids_list: List[str], query_time: str, bucket: str="SCADA_data", client: InfluxDBClient=client) -> Dict[str, float]:
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和时间查询值
:param query_ids_list: SCADA设备ID的列表
:param query_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。
:param bucket: InfluxDB 的 bucket 名称，默认值为 "SCADA_data"。
:param client: 已初始化的 InfluxDBClient 实例。
: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')))
@@ -1428,20 +1437,20 @@ def query_SCADA_data_by_device_ID_and_time(query_ids_list: List[str], query_time
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
def query_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start_time: str, end_time: str,
bucket: str="SCADA_data", client: InfluxDBClient=client):
def query_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"。
:param client: 已初始化的 InfluxDBClient 实例。
: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')))
@@ -1453,14 +1462,14 @@ def query_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start
beijing_end_time = datetime.fromisoformat(end_time)
utc_end_time = beijing_end_time.astimezone(timezone.utc) + timedelta(seconds=1)
print(utc_end_time)
SCADA_dict = {}
for device_id in query_ids_list:
flux_query = f'''
from(bucket: "{bucket}")
|> range(start: {utc_start_time.isoformat()}, stop: {utc_end_time.isoformat()})
|> filter(fn: (r) => r["device_ID"] == "{device_id}" and r["_field"] == "monitored_value")
|> filter(fn: (r) => r["_measurement"] == "SCADA_data" and r["device_ID"] = {device_id} and r["_field"] == "monitored_value")
|> pivot(rowKey: ["_time"], columnKey: ["device_ID"], valueColumn: "_value")
|> sort(columns: ["_time"])
'''
# 执行查询，返回一个 FluxTable 列表
tables = query_api.query(flux_query)
@@ -1470,10 +1479,10 @@ def query_SCADA_data_by_device_ID_and_timerange(query_ids_list: List[str], start
# 获取记录的时间和监测值
records_list.append({
"time": record["_time"],
"value": record.get_value()
"value": record["_value"]
})
SCADA_dict[device_id] = records_list
client.close()
return SCADA_dict
@@ -1494,17 +1503,16 @@ def query_SCADA_data_by_device_ID_and_date(query_ids_list: List[str], query_date
# 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", client: InfluxDBClient = client):
result_start_time: str, bucket: str = "realtime_simulation_result"):
"""
将实时模拟计算结果数据存储到 InfluxDB 的realtime_simulation_result这个bucket中。
:param node_result_list: (List[Dict[str, any]]): 包含节点和结果数据的字典列表。
:param link_result_list: (List[Dict[str, any]]): 包含连接和结果数据的字典列表。
:param result_start_time: (str): 计算结果的模拟开始时间。
:param bucket: (str): InfluxDB 的 bucket 名称，默认值为 "realtime_simulation_result"。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
: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')))
@@ -1539,7 +1547,7 @@ def store_realtime_simulation_result_to_influxdb(node_result_list: List[Dict[str
# 写入数据到 InfluxDB，多个 field 在同一个 point 中
# write_api.write(bucket=bucket, org=org_name, record=node_point)
# write_api.flush()
print(f"成功将 {len(node_result_list)} 条node数据写入 InfluxDB。")
# print(f"成功将 {len(node_result_list)} 条node数据写入 InfluxDB。")
for result in link_result_list:
link_id = result.get('link')
data_list = result.get('result', [])
@@ -1566,18 +1574,19 @@ def store_realtime_simulation_result_to_influxdb(node_result_list: List[Dict[str
write_api.flush() # 刷新缓存一次
except Exception as e:
raise RuntimeError(f"数据写入 InfluxDB 时发生错误: {e}")
client.close()
# 2025/02/01
def query_latest_record_by_ID(ID: str, type: str, bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> dict:
def query_latest_record_by_ID(ID: str, type: str, bucket: str="realtime_simulation_result") -> dict:
"""
查询指定ID的最新的一条记录
:param ID: (str): 要查询的 ID。
:param type: (str): "node"或“link”
:param bucket: (str): 数据存储的 bucket 名称。
:param client: (InfluxDBClient): 已初始化的 InfluxDB 客户端实例。
:return: dict: 最新记录的数据，如果没有找到则返回 None。
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -1668,14 +1677,14 @@ def query_latest_record_by_ID(ID: str, type: str, bucket: str="realtime_simulati
# 2025/02/01
def query_all_record_by_time(query_time: str, bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> tuple:
def query_all_record_by_time(query_time: str, bucket: str="realtime_simulation_result") -> tuple:
"""
查询指定北京时间的所有记录，包括 'node' 和 'link' measurement，分别以指定格式返回。
:param query_time: (str): 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。
:param bucket: (str): 数据存储的 bucket 名称。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -1729,21 +1738,21 @@ def query_all_record_by_time(query_time: str, bucket: str="realtime_simulation_r
"reaction": record["reaction"],
"friction": record["friction"]
})
client.close()
return node_records, link_records
# 2025/03/03
def query_all_record_by_time_property(query_time: str, type: str, property: str, bucket: str="realtime_simulation_result",
client: InfluxDBClient=client) -> list:
def query_all_record_by_time_property(query_time: str, type: str, property: str, bucket: str="realtime_simulation_result") -> list:
"""
查询指定北京时间的所有记录，查询 'node' 或 'link' 的某一属性值，以指定格式返回。
:param query_time: (str): 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。
:param type: (str): 查询的类型（决定 measurement）
:param property: (str): 查询的字段名称（field）
:param bucket: (str): 数据存储的 bucket 名称。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
:return: list(dict): result_records
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -1777,18 +1786,19 @@ def query_all_record_by_time_property(query_time: str, type: str, property: str,
"ID": record["ID"],
"value": record["_value"]
})
client.close()
return result_records
# 2025/02/21
def query_all_record_by_date(query_date: str, bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> tuple:
def query_all_record_by_date(query_date: str, bucket: str="realtime_simulation_result") -> tuple:
"""
查询指定日期的所有记录，包括‘node’和‘link’，分别以指定的格式返回
:param query_date: 输入的日期，格式为‘2025-02-14’
:param bucket: 数据存储的bucket名称
:param client: 已初始化的InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
# 记录开始时间
time_cost_start = time.perf_counter()
print('{} -- Hydraulic simulation started.'.format(
@@ -1849,6 +1859,7 @@ def query_all_record_by_date(query_date: str, bucket: str="realtime_simulation_r
print('{} -- Hydraulic simulation finished, cost time: {:.2f} s.'.format(
datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S'),
time_cost_end - time_cost_start))
client.close()
return node_records, link_records
# 2025/03/15 DingZQ
@@ -1936,19 +1947,17 @@ def query_all_records_by_date_with_type(query_date: str, query_type: str, bucket
time_cost_end - time_cost_start))
return result_records
# 2025/02/21
def query_all_record_by_date_property(query_date: str, type: str, property: str,
bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> list:
def query_all_record_by_date_property(query_date: str, type: str, property: str, bucket: str="realtime_simulation_result") -> list:
"""
查询指定日期的‘node’或‘link’的某一属性值的所有记录，以指定的格式返回
:param query_date: 输入的日期，格式为‘2025-02-14’
:param type: (str): 查询的类型（决定 measurement）
:param property: (str): 查询的字段名称（field）
:param bucket: 数据存储的bucket名称
:param client: 已初始化的InfluxDBClient 实例。
:return: list(dict): result_records
"""
client = get_new_client()
# 记录开始时间
time_cost_start = time.perf_counter()
print('{} -- Hydraulic simulation started.'.format(
@@ -1989,11 +1998,12 @@ def query_all_record_by_date_property(query_date: str, type: str, property: str,
print('{} -- Hydraulic simulation finished, cost time: {:.2f} s.'.format(
datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S'),
time_cost_end - time_cost_start))
client.close()
return result_records
# 2025/02/01
def query_curve_by_ID_property_daterange(ID: str, type: str, property: str, start_date: str, end_date: str, bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> list:
def query_curve_by_ID_property_daterange(ID: str, type: str, property: str, start_date: str, end_date: str, bucket: str="realtime_simulation_result") -> list:
"""
根据 type 查询对应的 measurement，根据 ID 和 date 查询对应的 tag，根据 property 查询对应的 field。
:param ID: (str): 要查询的 ID（tag）
@@ -2002,9 +2012,9 @@ def query_curve_by_ID_property_daterange(ID: str, type: str, property: str, star
:param start_date: (str): 查询的开始日期，格式为 'YYYY-MM-DD'
:param end_date: (str): 查询的结束日期，格式为 'YYYY-MM-DD'
:param bucket: (str): 数据存储的 bucket 名称，默认值为 "realtime_simulation_result"
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例
: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')))
@@ -2039,13 +2049,14 @@ def query_curve_by_ID_property_daterange(ID: str, type: str, property: str, star
"time": record["_time"],
"value": record["_value"]
})
client.close()
return results
# 2025/02/13
def store_scheme_simulation_result_to_influxdb(node_result_list: List[Dict[str, any]], link_result_list: List[Dict[str, any]],
scheme_start_time: str, num_periods: int = 1, scheme_Type: str = None, scheme_Name: str = None,
bucket: str = "scheme_simulation_result", client: InfluxDBClient = client):
bucket: str = "scheme_simulation_result"):
"""
将方案模拟计算结果存入 InfluxuDb 的scheme_simulation_result这个bucket中。
:param node_result_list: (List[Dict[str, any]]): 包含节点和结果数据的字典列表。
@@ -2055,9 +2066,9 @@ def store_scheme_simulation_result_to_influxdb(node_result_list: List[Dict[str,
:param scheme_Type: (str): 方案类型
:param scheme_Name: (str): 方案名称
:param bucket: (str): InfluxDB 的 bucket 名称，默认值为 "scheme_simulation_result"。
:param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。
: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')))
@@ -2127,6 +2138,7 @@ def store_scheme_simulation_result_to_influxdb(node_result_list: List[Dict[str,
write_api.flush() # 刷新缓存一次
except Exception as e:
raise RuntimeError(f"数据写入 InfluxDB 时发生错误: {e}")
client.close()
# 2025/03/12
@@ -2171,17 +2183,25 @@ def query_corresponding_query_id_and_element_id(name: str) -> None:
print(f"数据库连接或查询出错: {e}")
# 2025/03/22
# def auto_get_burst_flow():
# 2025/03/22
# def manually_get_burst_flow():
# 2025/03/11
def fill_scheme_simulation_result_to_SCADA(scheme_Type: str = None, scheme_Name: str = None, query_date: str = None,
bucket: str = "scheme_simulation_result", client: InfluxDBClient = client):
bucket: str = "scheme_simulation_result"):
"""
:param scheme_Type: 方案类型
:param scheme_Name: 方案名称
:param query_date: 查询日期，格式为 'YYYY-MM-DD'
:param bucket: InfluxDB 的 bucket 名称，默认值为 "scheme_simulation_result"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2277,19 +2297,20 @@ def fill_scheme_simulation_result_to_SCADA(scheme_Type: str = None, scheme_Name:
if points_to_write:
write_api.write(bucket=bucket, org=org_name, record=points_to_write)
write_api.flush() # 刷新缓存一次
client.close()
# 2025/02/15
def query_SCADA_data_curve(api_query_id: str, start_date: str, end_date: str, bucket: str="SCADA_data", client: InfluxDBClient=client) -> list:
def query_SCADA_data_curve(api_query_id: str, start_date: str, end_date: str, bucket: str="SCADA_data") -> list:
"""
根据SCADA设备的api_query_id和时间范围，查询得到曲线，查到的数据为0时区时间
:param api_query_id: SCADA设备的api_query_id
:param start_date: 查询开始的时间，格式为 'YYYY-MM-DD'
:param end_date: 查询结束的时间，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2313,21 +2334,21 @@ def query_SCADA_data_curve(api_query_id: str, start_date: str, end_date: str, bu
"time": record["_time"],
"value": record["_value"]
})
client.close()
return results
# 2025/02/18
def query_scheme_all_record_by_time(scheme_Type: str, scheme_Name: str, query_time: str,
bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> tuple:
def query_scheme_all_record_by_time(scheme_Type: str, scheme_Name: str, query_time: str, bucket: str="scheme_simulation_result") -> tuple:
"""
查询指定方案某一时刻的所有记录，包括‘node'和‘link’，分别以指定格式返回。
:param scheme_Type: 方案类型
:param scheme_Name: 方案名称
:param query_time: 输入的北京时间，格式为 '2024-11-24T17:30:00+08:00'。
:param bucket: 数据存储的 bucket 名称。
:param client: 已初始化的 InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -2381,12 +2402,13 @@ def query_scheme_all_record_by_time(scheme_Type: str, scheme_Name: str, query_ti
"reaction": record["reaction"],
"friction": record["friction"]
})
client.close()
return node_records, link_records
# 2025/03/04
def query_scheme_all_record_by_time_property(scheme_Type: str, scheme_Name: str, query_time: str, type: str, property: str,
bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> list:
bucket: str="scheme_simulation_result") -> list:
"""
查询指定方案某一时刻‘node'或‘link’某一属性值，以指定格式返回。
:param scheme_Type: 方案类型
@@ -2395,9 +2417,9 @@ def query_scheme_all_record_by_time_property(scheme_Type: str, scheme_Name: str,
:param type: 查询的类型（决定 measurement）
:param property: 查询的字段名称（field）
:param bucket: 数据存储的 bucket 名称。
:param client: 已初始化的 InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -2430,12 +2452,13 @@ def query_scheme_all_record_by_time_property(scheme_Type: str, scheme_Name: str,
"ID": record["ID"],
"value": record["_value"]
})
client.close()
return result_records
# 2025/02/19
def query_scheme_curve_by_ID_property(scheme_Type: str, scheme_Name: str, query_date: str, ID: str, type: str, property: str,
bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> list:
bucket: str="scheme_simulation_result") -> list:
"""
根据scheme_Type和scheme_Name,查询该模拟方案中，某一node或link的某一属性值的所有时间的结果
:param scheme_Type: 方案类型
@@ -2445,9 +2468,9 @@ def query_scheme_curve_by_ID_property(scheme_Type: str, scheme_Name: str, query_
:param type: 元素的类型，node或link
:param property: 元素的属性值
:param bucket: 数据存储的 bucket 名称，默认值为 "scheme_simulation_result"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2478,21 +2501,21 @@ def query_scheme_curve_by_ID_property(scheme_Type: str, scheme_Name: str, query_
"time": record["_time"],
"value": record["_value"]
})
client.close()
return results
# 2025/02/21
def query_scheme_all_record(scheme_Type: str, scheme_Name: str, query_date: str, bucket: str="scheme_simulation_result",
client: InfluxDBClient=client) -> tuple:
def query_scheme_all_record(scheme_Type: str, scheme_Name: str, query_date: str, bucket: str="scheme_simulation_result") -> tuple:
"""
查询指定方案的所有记录，包括‘node'和‘link’，分别以指定格式返回。
:param scheme_Type: 方案类型
:param scheme_Name: 方案名称
:param query_date: 查询日期，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称。
:param client: 已初始化的 InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -2543,14 +2566,13 @@ def query_scheme_all_record(scheme_Type: str, scheme_Name: str, query_date: str,
"reaction": record["reaction"],
"friction": record["friction"]
})
client.close()
return node_records, link_records
# 2025/03/04
# burst_Analysis
def query_scheme_all_record_property(scheme_Type: str, scheme_Name: str, query_date: str, type: str, property: str,
bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> list:
bucket: str="scheme_simulation_result") -> list:
"""
查询指定方案的‘node'或‘link’的某一属性值，以指定格式返回。
:param scheme_Type: 方案类型
@@ -2559,9 +2581,9 @@ def query_scheme_all_record_property(scheme_Type: str, scheme_Name: str, query_d
:param type: 查询的类型（决定 measurement）
:param property: 查询的字段名称（field）
:param bucket: 数据存储的 bucket 名称。
:param client: 已初始化的 InfluxDBClient 实例。
:return: dict: tuple: (node_records, link_records)
"""
client = get_new_client()
if not client.ping():
print("{} -- Failed to connect to InfluxDB.".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
@@ -2592,19 +2614,20 @@ def query_scheme_all_record_property(scheme_Type: str, scheme_Name: str, query_d
"ID": record["ID"],
"value": record["_value"]
})
client.close()
return result_records
# 2025/02/16
def export_SCADA_data_to_csv(start_date: str, end_date: str, bucket: str="SCADA_data", client: InfluxDBClient=client) -> None:
def export_SCADA_data_to_csv(start_date: str, end_date: str, bucket: str="SCADA_data") -> None:
"""
导出influxdb中SCADA_data这个bucket的数据到csv中
:param start_date: 查询开始的时间，格式为 'YYYY-MM-DD'
:param end_date: 查询结束的时间，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2642,18 +2665,19 @@ def export_SCADA_data_to_csv(start_date: str, end_date: str, bucket: str="SCADA_
writer.writeheader()
writer.writerows(rows)
print(f"Data exported to {csv_filename} successfully.")
client.close()
# 2025/02/17
def export_realtime_simulation_result_to_csv(start_date: str, end_date: str, bucket: str="realtime_simulation_result", client: InfluxDBClient=client) -> None:
def export_realtime_simulation_result_to_csv(start_date: str, end_date: str, bucket: str="realtime_simulation_result") -> None:
"""
导出influxdb中realtime_simulation_result这个bucket的数据到csv中
:param start_date: 查询开始的时间，格式为 'YYYY-MM-DD'
:param end_date: 查询结束的时间，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2724,18 +2748,19 @@ def export_realtime_simulation_result_to_csv(start_date: str, end_date: str, buc
writer.writeheader()
writer.writerows(node_rows)
print(f"Data exported to {csv_filename_link} and {csv_filename_node} successfully.")
client.close()
# 2025/02/18
def export_scheme_simulation_result_to_csv_time(start_date: str, end_date: str, bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> None:
def export_scheme_simulation_result_to_csv_time(start_date: str, end_date: str, bucket: str="scheme_simulation_result") -> None:
"""
导出influxdb中scheme_simulation_result这个bucket的数据到csv中
:param start_date: 查询开始的时间，格式为 'YYYY-MM-DD'
:param end_date: 查询结束的时间，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2809,20 +2834,20 @@ def export_scheme_simulation_result_to_csv_time(start_date: str, end_date: str,
writer.writeheader()
writer.writerows(node_rows)
print(f"Data exported to {csv_filename_link} and {csv_filename_node} successfully.")
client.close()
# 2025/02/18
def export_scheme_simulation_result_to_csv_scheme(scheme_Type: str, scheme_Name: str, query_date: str,
bucket: str="scheme_simulation_result", client: InfluxDBClient=client) -> None:
def export_scheme_simulation_result_to_csv_scheme(scheme_Type: str, scheme_Name: str, query_date: str, bucket: str="scheme_simulation_result") -> None:
"""
导出influxdb中scheme_simulation_result这个bucket的数据到csv中
:param scheme_Type: 查询的方案类型
:param scheme_Name: 查询的方案名
:param query_date: 查询日期，格式为 'YYYY-MM-DD'
:param bucket: 数据存储的 bucket 名称，默认值为 "SCADA_data"
:param client: 已初始化的 InfluxDBClient 实例
: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')))
@@ -2895,6 +2920,7 @@ def export_scheme_simulation_result_to_csv_scheme(scheme_Type: str, scheme_Name:
writer.writeheader()
writer.writerows(node_rows)
print(f"Data exported to {csv_filename_link} and {csv_filename_node} successfully.")
client.close()
# 示例调用
@@ -2904,15 +2930,13 @@ if __name__ == "__main__":
org_name = influxdb_info.org
client = InfluxDBClient(url=url, token=token)
# step1: 检查连接状态，初始化influxdb的buckets
# try:
# # delete_buckets(client, org_name)
# create_and_initialize_buckets(client, org_name)
# # delete_buckets(org_name)
# create_and_initialize_buckets(org_name)
# except Exception as e:
# print(f"连接失败: {e}")
# finally:
# client.close()
# step2: 先查询pg数据库中scada_info的信息，然后存储SCADA数据到SCADA_data这个bucket里
query_pg_scada_info_realtime('bb')
@@ -2930,98 +2954,98 @@ 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-03-04T00:00:00+08:00', end_time='2025-03-10T00:00:00+08:00')
# download_history_data_manually(begin_time='2025-03-21T00:00:00+08:00', end_time='2025-03-22T00:00:00+08:00')
# step3: 查询测试示例
with InfluxDBClient(url=url, token=token, org=org_name) as client:
# # 示例1：query_latest_record_by_ID
# bucket_name = "realtime_simulation_result" # 数据存储的 bucket 名称
# node_id = "ZBBDTZDP000022" # 查询的节点 ID
# link_id = "ZBBGXSZW000002"
# 示例1：query_latest_record_by_ID
# bucket_name = "realtime_simulation_result" # 数据存储的 bucket 名称
# node_id = "ZBBDTZDP000022" # 查询的节点 ID
# link_id = "ZBBGXSZW000002"
#
# latest_record = query_latest_record_by_ID(ID=node_id, type="node", bucket=bucket_name, client=client)
# # # latest_record = query_latest_record_by_ID(ID=link_id, type="link", bucket=bucket_name, client=client)
# #
# if latest_record:
# print("最新记录：", latest_record)
# else:
# print("未找到符合条件的记录。")
# latest_record = query_latest_record_by_ID(ID=node_id, type="node", bucket=bucket_name)
# # # latest_record = query_latest_record_by_ID(ID=link_id, type="link", bucket=bucket_name)
# #
# if latest_record:
# print("最新记录：", latest_record)
# else:
# print("未找到符合条件的记录。")
# 示例2：query_all_record_by_time
# node_records, link_records = query_all_record_by_time(query_time="2025-02-14T10:30:00+08:00")
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例2：query_all_record_by_time
# node_records, link_records = query_all_record_by_time(query_time="2025-02-14T10:30:00+08:00")
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例3：query_curve_by_ID_property_daterange
# curve_result = query_curve_by_ID_property_daterange(ID=node_id, type="node", property="head",
# start_date="2024-11-25", end_date="2024-11-25")
# print(curve_result)
# 示例3：query_curve_by_ID_property_daterange
# curve_result = query_curve_by_ID_property_daterange(ID=node_id, type="node", property="head",
# start_date="2024-11-25", end_date="2024-11-25")
# print(curve_result)
# 示例4：query_SCADA_data_by_device_ID_and_time
# SCADA_result_dict = query_SCADA_data_by_device_ID_and_time(globals.fixed_pump_realtime_ids, query_time='2025-03-09T23:45:00+08:00')
# print(SCADA_result_dict)
# 示例4：query_SCADA_data_by_device_ID_and_time
# SCADA_result_dict = query_SCADA_data_by_device_ID_and_time(globals.fixed_pump_realtime_ids, query_time='2025-03-09T23:45:00+08:00')
# print(SCADA_result_dict)
# 示例5：query_SCADA_data_curve
# SCADA_result = query_SCADA_data_curve(api_query_id='9519', start_date='2025-03-08', end_date='2025-03-08')
# print(SCADA_result)
# 示例5：query_SCADA_data_curve
# SCADA_result = query_SCADA_data_curve(api_query_id='9519', start_date='2025-03-08', end_date='2025-03-08')
# print(SCADA_result)
# 示例6：export_SCADA_data_to_csv
# export_SCADA_data_to_csv(start_date='2025-02-13', end_date='2025-02-15')
# 示例6：export_SCADA_data_to_csv
# export_SCADA_data_to_csv(start_date='2025-02-13', end_date='2025-02-15')
# 示例7：export_realtime_simulation_result_to_csv
# export_realtime_simulation_result_to_csv(start_date='2025-02-13', end_date='2025-02-15')
# 示例7：export_realtime_simulation_result_to_csv
# export_realtime_simulation_result_to_csv(start_date='2025-02-13', end_date='2025-02-15')
# 示例8：export_scheme_simulation_result_to_csv_time
# export_scheme_simulation_result_to_csv_time(start_date='2025-02-13', end_date='2025-02-15')
# 示例8：export_scheme_simulation_result_to_csv_time
# export_scheme_simulation_result_to_csv_time(start_date='2025-02-13', end_date='2025-02-15')
# 示例9：export_scheme_simulation_result_to_csv_scheme
# export_scheme_simulation_result_to_csv_scheme(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10')
# 示例9：export_scheme_simulation_result_to_csv_scheme
# export_scheme_simulation_result_to_csv_scheme(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10')
# 示例10：query_scheme_all_record_by_time
# node_records, link_records = query_scheme_all_record_by_time(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_time="2025-02-14T10:30:00+08:00")
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例10：query_scheme_all_record_by_time
# node_records, link_records = query_scheme_all_record_by_time(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_time="2025-02-14T10:30:00+08:00")
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例11：query_scheme_curve_by_ID_property
# curve_result = query_scheme_curve_by_ID_property(scheme_Type='burst_Analysis', scheme_Name='scheme1', ID='ZBBDTZDP000022',
# type='node', property='head')
# print(curve_result)
# 示例11：query_scheme_curve_by_ID_property
# curve_result = query_scheme_curve_by_ID_property(scheme_Type='burst_Analysis', scheme_Name='scheme1', ID='ZBBDTZDP000022',
# type='node', property='head')
# print(curve_result)
# 示例12：query_all_record_by_date
node_records, link_records = query_all_record_by_date(query_date='2025-02-27')
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例12：query_all_record_by_date
# node_records, link_records = query_all_record_by_date(query_date='2025-02-27')
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例13：query_scheme_all_record
# node_records, link_records = query_scheme_all_record(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10')
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例13：query_scheme_all_record
# node_records, link_records = query_scheme_all_record(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10')
# print("Node 数据：", node_records)
# print("Link 数据：", link_records)
# 示例14：query_all_record_by_time_property
# result_records = query_all_record_by_time_property(query_time='2025-02-25T23:45:00+08:00', type='node', property='head')
# print(result_records)
# 示例14：query_all_record_by_time_property
# result_records = query_all_record_by_time_property(query_time='2025-02-25T23:45:00+08:00', type='node', property='head')
# print(result_records)
# 示例15：query_all_record_by_date_property
# result_records = query_all_record_by_date_property(query_date='2025-02-14', type='node', property='head')
# print(result_records)
# 示例15：query_all_record_by_date_property
# result_records = query_all_record_by_date_property(query_date='2025-02-14', type='node', property='head')
# print(result_records)
# 示例16：query_scheme_all_record_by_time_property
# result_records = query_scheme_all_record_by_time_property(scheme_Type='burst_Analysis', scheme_Name='scheme1',
# query_time='2025-02-14T10:30:00+08:00', type='node', property='head')
# print(result_records)
# 示例16：query_scheme_all_record_by_time_property
# result_records = query_scheme_all_record_by_time_property(scheme_Type='burst_Analysis', scheme_Name='scheme1',
# query_time='2025-02-14T10:30:00+08:00', type='node', property='head')
# print(result_records)
# 示例17：query_scheme_all_record_property
# result_records = query_scheme_all_record_property(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10', type='node', property='head')
# print(result_records)
# 示例17：query_scheme_all_record_property
# result_records = query_scheme_all_record_property(scheme_Type='burst_Analysis', scheme_Name='scheme1', query_date='2025-03-10', type='node', property='head')
# print(result_records)
# 示例18：fill_scheme_simulation_result_to_SCADA
# fill_scheme_simulation_result_to_SCADA(scheme_Type='burst_Analysis', scheme_Name='burst_scheme', query_date='2025-03-10')
# 示例18：fill_scheme_simulation_result_to_SCADA
# fill_scheme_simulation_result_to_SCADA(scheme_Type='burst_Analysis', scheme_Name='burst_scheme', query_date='2025-03-10')
# 示例19：query_SCADA_data_by_device_ID_and_timerange
# result = query_SCADA_data_by_device_ID_and_timerange(query_ids_list=globals.fixed_pump_realtime_ids, start_time='2025-03-09T12:00:00+08:00',
# end_time='2025-03-09T12:10:00+08:00')
# print(result)
# 示例19：query_SCADA_data_by_device_ID_and_timerange
# result = query_SCADA_data_by_device_ID_and_timerange(query_ids_list=globals.fixed_pump_realtime_ids, start_time='2025-03-09T12:00:00+08:00',
# end_time='2025-03-09T12:10:00+08:00')
# print(result)