@dp20eic stimmt, ChatGPT ist einfach schneller als doku lesen (rtfm ist oldschool im zeitalter von KI), heute geht man auch nicht mehr in die Bibliothek, sondern fragt das Internet, ebenso kann man ChatGPt alles auf einmal machen lassen:
doku für mich lesen
das relevante davon extrahieren
und gleich für meinen use-case anwenden und das script schreiben.
... nach ein paar weiteren Iterationen wars dann auch schon perfekt:
# manuell ausführen mit
# /bin/python3 /volume1/Christof/Daten/Organisation/030_Infrastruktur/020_Devices/Smart-Home/Logging/csv-export/export_influxdb_to_csv_tmp-ist-wohn_6.py
from influxdb_client import InfluxDBClient, Point, WriteOptions
import pandas as pd
import csv
import warnings
from influxdb_client.client.warnings import MissingPivotFunction
from pathlib import Path
# InfluxDB credentials
url = "http://10.ab.xy.rs:8086/"
token = "Rea..."
myorg = "smarthome"
mybucket = "logging1"
# Output file path
output_file_path = "/volume1/Christof/Daten/Organisation/030_Infrastruktur/020_Devices/Smart-Home/Logging/influx_output_tmp-ist-wohn_2.csv"
# Create InfluxDB client
client = InfluxDBClient(url=url, token=token)
# Suppress the warning about the missing pivot function
warnings.simplefilter("ignore", MissingPivotFunction)
# Check connection
try:
health = client.health()
if health.message == 'ready for queries and writes':
print("Connection successful.")
# Specify your InfluxDB query here
query = 'from(bucket: "logging1") |> range(start: -1d)'
# Run the query
tables = client.query_api().query(org=myorg, query=query)
# Specify the values you want to filter for
filter_measurement = "knx.0.Heizung__Klima__Lüftung.EG.Heiz-Wohn-Ess_Temp-IST"
filter_field = "value"
# Extract and export data to CSV
with open(output_file_path, 'w', newline='') as csvfile:
csv_writer = csv.writer(csvfile)
# Write header
csv_writer.writerow(["_time", "_value", "_field", "_measurement"])
# Print all measurement names for debugging
measurement_names = set()
for table in tables:
for record in table.records:
measurement_names.add(record.get_measurement())
print(f"All measurement names: {measurement_names}")
# Write data with filters
for table in tables:
for record in table.records:
# Check if both conditions are met
if record.get_measurement() == filter_measurement and record.get_field() == filter_field:
csv_writer.writerow([record.get_time(), record.get_value(), record.get_field(), record.get_measurement()])
print(f"Data exported to {output_file_path}")
else:
print(f"Connection failed. InfluxDB status: {health.message}")
except Exception as e:
print(f"An error occurred: {e}")
KI ist echt ein Gamechanger.... 🙂
