from influxdb_client import InfluxDBClient, BucketsApi, WriteApi, OrganizationsApi, Point, QueryApi from typing import List, Dict from datetime import datetime, timedelta, timezone from dateutil import parser # influxdb数据库连接信息 url = "http://localhost:8086" # 替换为你的InfluxDB实例地址 token = "xGDM5RZqRJAuzAGS-otXUdC2NFdY75qJAjRLqAB4p5WcIIAlIUpOpT8_yA16AOHmJWerwQ_08gwb84sy42jnZQ==" # 替换为你的InfluxDB Token org_name = "TJWATEORG" # 替换为你的Organization名称 client = InfluxDBClient(url=url, token=token, org=org_name) def create_and_initialize_buckets(client: InfluxDBClient, org_name: str) -> None: bucket_api = BucketsApi(client) write_api = client.write_api() org_api = OrganizationsApi(client) # 获取 org_id org = next((o for o in org_api.find_organizations() if o.name == org_name), None) if not org: raise ValueError(f"Organization '{org_name}' not found.") org_id = org.id print(f"Using Organization ID: {org_id}") # 定义 Buckets 信息 buckets = [ {"name": "SCADA_data", "retention_rules": []}, {"name": "realtime_data", "retention_rules": []}, {"name": "scheme_simulation", "retention_rules": []} ] # 创建 Buckets 并初始化数据 for bucket in buckets: # 创建 Bucket created_bucket = bucket_api.create_bucket( bucket_name=bucket["name"], retention_rules=bucket["retention_rules"], org_id=org_id ) print(f"Bucket '{bucket['name']}' created with ID: {created_bucket.id}") # 根据 Bucket 初始化数据 if bucket["name"] == "SCADA_data": point = Point("SCADA") \ .tag("date", None) \ .tag("type", None) \ .tag("device_ID", None) \ .field("monitored_value", 0.0) \ .field("datacleaning_value", 0.0) \ .field("simulation_value", 0.0) \ .time("2024-11-21T00:00:00Z") write_api.write(bucket="SCADA_data", org=org_name, record=point) print("Initialized SCADA_data with default structure.") elif bucket["name"] == "realtime_data": link_point = Point("link") \ .tag("date", None) \ .tag("ID", None) \ .field("flow", 0.0) \ .field("leakage", 0.0) \ .field("velocity", 0.0) \ .field("headloss", 0.0) \ .field("status", None) \ .field("setting", 0.0) \ .field("quality", 0.0) \ .time("2024-11-21T00:00:00Z") node_point = Point("node") \ .tag("date", None) \ .tag("ID", None) \ .field("head", 0.0) \ .field("pressure", 0.0) \ .field("actualdemand", 0.0) \ .field("demanddeficit", 0.0) \ .field("totalExternalOutflow", 0.0) \ .field("quality", 0.0) \ .time("2024-11-21T00:00:00Z") write_api.write(bucket="realtime_data", org=org_name, record=link_point) write_api.write(bucket="realtime_data", org=org_name, record=node_point) print("Initialized realtime_data with default structure.") elif bucket["name"] == "scheme_simulation": link_point = Point("link") \ .tag("date", None) \ .tag("ID", None) \ .tag("scheme_Type", None) \ .tag("scheme_Name", None) \ .field("flow", 0.0) \ .field("leakage", 0.0) \ .field("velocity", 0.0) \ .field("headloss", 0.0) \ .field("status", None) \ .field("setting", 0.0) \ .field("quality", 0.0) \ .time("2024-11-21T00:00:00Z") node_point = Point("node") \ .tag("date", None) \ .tag("ID", None) \ .tag("scheme_Type", None) \ .tag("scheme_Name", None) \ .field("head", 0.0) \ .field("pressure", 0.0) \ .field("actualdemand", 0.0) \ .field("demanddeficit", 0.0) \ .field("totalExternalOutflow", 0.0) \ .field("quality", 0.0) \ .time("2024-11-21T00:00:00Z") write_api.write(bucket="scheme_simulation", org=org_name, record=link_point) write_api.write(bucket="scheme_simulation", org=org_name, record=node_point) print("Initialized scheme_simulation with default structure.") print("All buckets created and initialized successfully.") def store_realtime_data_to_influxdb(node_result_list: List[Dict[str, any]], link_result_list: List[Dict[str, any]], result_start_time: str, bucket: str = "realtime_data", client: InfluxDBClient = client): """ 将实时数据存储到 InfluxDB 的realtime_data这个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_data"。 :param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例。 :return: """ if client.ping(): print("Successfully connected to InfluxDB.") else: print("Failed to connect to InfluxDB.") # 开始写入数据 try: write_api = client.write_api() date_str = result_start_time.split('T')[0] time_beijing = datetime.strptime(result_start_time, '%Y-%m-%dT%H:%M:%S%z').isoformat() for result in node_result_list: # 提取节点信息和结果数据 node_id = result.get('node') data_list = result.get('result', []) for data in data_list: # 构建 Point 数据，多个 field 存在于一个数据点中 node_point = Point("node") \ .tag("date", date_str) \ .tag("ID", node_id) \ .field("head", data.get('head', 0.0)) \ .field("pressure", data.get('pressure', 0.0)) \ .field("actualdemand", data.get('demand', 0.0)) \ .field("demanddeficit", None) \ .field("totalExternalOutflow", None) \ .field("quality", data.get('quality', 0.0)) \ .time(time_beijing) # 写入数据到 InfluxDB，多个 field 在同一个 point 中 write_api.write(bucket=bucket, org=org_name, record=node_point) print(f"成功将 {len(node_result_list)} 条node数据写入 InfluxDB。") for result in link_result_list: link_id = result.get('link') data_list = result.get('result', []) for data in data_list: link_point = Point("link") \ .tag("date", date_str) \ .tag("ID", link_id) \ .field("flow", data.get('flow', 0.0)) \ .field("velocity", data.get('velocity', 0.0)) \ .field("headloss", data.get('headloss', 0.0)) \ .field("quality", data.get('quality', 0.0)) \ .field("status", data.get('status', "UNKNOWN")) \ .field("setting", data.get('setting', 0.0)) \ .field("reaction", data.get('reaction', 0.0)) \ .field("friction", data.get('friction', 0.0)) \ .time(time_beijing) write_api.write(bucket=bucket, org=org_name, record=link_point) print(f"成功将 {len(link_result_list)} 条link数据写入 InfluxDB。") except Exception as e: raise RuntimeError(f"数据写入 InfluxDB 时发生错误: {e}") def query_latest_record_by_ID(ID: str, type: str, bucket: str="realtime_data", client: InfluxDBClient=client) -> dict: """ 查询指定ID的最新的一条记录 :param ID: (str): 要查询的 ID。 :param type: (str): "node"或“link” :param bucket: (str): 数据存储的 bucket 名称。 :param client: (InfluxDBClient): 已初始化的 InfluxDB 客户端实例。 :return: dict: 最新记录的数据，如果没有找到则返回 None。 """ if client.ping(): print("Successfully connected to InfluxDB.") else: print("Failed to connect to InfluxDB.") query_api = client.query_api() if type == "node": flux_query = f''' from(bucket: "{bucket}") |> range(start: -30d) // 查找最近一月的记录 |> filter(fn: (r) => r["_measurement"] == "node") |> filter(fn: (r) => r["ID"] == "{ID}") |> pivot( rowKey:["_time"], columnKey:["_field"], valueColumn:"_value" ) |> sort(columns: ["_time"], desc: true) |> limit(n: 1) ''' tables = query_api.query(flux_query) # 解析查询结果 for table in tables: for record in table.records: return { "time": record["_time"], "nodeID": ID, "head": record["head"], "pressure": record["pressure"], "actualdemand": record["actualdemand"], # "demanddeficit": record["demanddeficit"], # "totalExternalOutflow": record["totalExternalOutflow"], "quality": record["quality"] } elif type == "link": flux_query = f''' from(bucket: "{bucket}") |> range(start: -30d) // 查找最近一月的记录 |> filter(fn: (r) => r["_measurement"] == "link") |> filter(fn: (r) => r["ID"] == "{ID}") |> pivot( rowKey:["_time"], columnKey:["_field"], valueColumn:"_value" ) |> sort(columns: ["_time"], desc: true) |> limit(n: 1) ''' tables = query_api.query(flux_query) # 解析查询结果 for table in tables: for record in table.records: return { "time": record["_time"], "linkID": ID, "flow": record["flow"], "velocity": record["velocity"], "headloss": record["headloss"], "quality": record["quality"], "status": record["status"], "setting": record["setting"], "reaction": record["reaction"], "friction": record["friction"] } return None # 如果没有找到记录 def query_all_record_by_time(query_time: str, bucket: str="realtime_data", client: InfluxDBClient=client) -> 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) """ if client.ping(): print("Successfully connected to InfluxDB.") else: print("Failed to connect to InfluxDB.") query_api = client.query_api() # 将北京时间转换为 UTC 时间 beijing_time = datetime.fromisoformat(query_time) utc_time = beijing_time.astimezone(timezone.utc) utc_start_time = utc_time - timedelta(seconds=1) utc_stop_time = utc_time + timedelta(seconds=1) # 构建 Flux 查询语句 flux_query = f''' from(bucket: "{bucket}") |> range(start: {utc_start_time.isoformat()}, stop: {utc_stop_time.isoformat()}) |> filter(fn: (r) => r["_measurement"] == "node" or r["_measurement"] == "link") |> pivot( rowKey:["_time"], columnKey:["_field"], valueColumn:"_value" ) ''' # 执行查询 tables = query_api.query(flux_query) node_records = [] link_records = [] # 解析查询结果 for table in tables: for record in table.records: # print(record.values) # 打印完整记录内容 measurement = record["_measurement"] # 处理 node 数据 if measurement == "node": node_records.append({ "time": record["_time"], "ID": record["ID"], "head": record["head"], "pressure": record["pressure"], "actualdemand": record["actualdemand"], "quality": record["quality"] }) # 处理 link 数据 elif measurement == "link": link_records.append({ "time": record["_time"], "linkID": record["ID"], "flow": record["flow"], "velocity": record["velocity"], "headloss": record["headloss"], "quality": record["quality"], "status": record["status"], "setting": record["setting"], "reaction": record["reaction"], "friction": record["friction"] }) return node_records, link_records # DingZQ return list of dict { 'time': 'timevalue', 'value', 'valuevalue' } def query_curve_by_ID_property_daterange(ID: str, type: str, property: str, start_date: str, end_date: str, bucket: str="realtime_data", client: InfluxDBClient=client) -> list: """ 根据 type 查询对应的 measurement，根据 ID 和 date 查询对应的 tag，根据 property 查询对应的 field。 :param ID: (str): 要查询的 ID（tag） :param type: (str): 查询的类型（决定 measurement） :param property: (str): 查询的字段名称（field） :param start_date: (str): 查询的开始日期，格式为 'YYYY-MM-DD' :param end_date: (str): 查询的结束日期，格式为 'YYYY-MM-DD' :param bucket: (str): 数据存储的 bucket 名称，默认值为 "realtime_data" :param client: (InfluxDBClient): 已初始化的 InfluxDBClient 实例 :return: 查询结果的列表 """ if client.ping(): print("Successfully connected to InfluxDB.") else: print("Failed to connect to InfluxDB.") query_api = client.query_api() # 确定 measurement if type == "node": measurement = "node" elif type == "link": measurement = "link" else: raise ValueError(f"不支持的类型: {type}") # 解析日期范围（当天的 UTC 开始和结束时间） # previous_day = datetime.strptime(start_date, "%Y-%m-%d") - timedelta(days=1) # start_time = previous_day.isoformat() + "T16:00:00Z" # stop_time = datetime.strptime(end_date, "%Y-%m-%d").isoformat() + "T15:59:59Z" # 将 start_date 的北京时间转换为 UTC 时间范围 start_time = (datetime.strptime(start_date, "%Y-%m-%d") - timedelta(days=1)).replace(hour=16, minute=0, second=0, tzinfo=timezone.utc).isoformat() stop_time = datetime.strptime(end_date, "%Y-%m-%d").replace(hour=15, minute=59, second=59, tzinfo=timezone.utc).isoformat() # 构建 Flux 查询语句 flux_query = f''' from(bucket: "{bucket}") |> range(start: {start_time}, stop: {stop_time}) |> filter(fn: (r) => r["_measurement"] == "{measurement}") |> filter(fn: (r) => r["ID"] == "{ID}") |> filter(fn: (r) => r["_field"] == "{property}") ''' # 执行查询 tables = query_api.query(flux_query) # 解析查询结果 results = [] for table in tables: for record in table.records: results.append({ "time": record["_time"], "value": record["_value"] }) return results # 示例调用 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) # # 检查连接状态 # try: # create_and_initialize_buckets(client, org_name) # except Exception as e: # print(f"连接失败: {e}") # finally: # client.close() with InfluxDBClient(url=url, token=token, org=org_name) as client: bucket_name = "realtime_data" # 数据存储的 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("未找到符合条件的记录。") # node_records, link_records = query_all_record_by_time(query_time="2024-11-25T06:00:00+08:00") # print("Node 数据：", node_records) # print("Link 数据：", link_records) 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)