123456789101112131415161718192021222324252627282930 |
- from influxdb_client import InfluxDBClient, Point
- from influxdb_client.client.write_api import SYNCHRONOUS
- import dataclasses
- class InfluxDB:
- def __init__(self):
- self.client = InfluxDBClient(url="http://localhost:8086", token="XPBViJ3s4JL9_wPffwd5M2EgXj5hcUgT0n4jNhv7m6-NC-6SSxQ3run4kXtWBvOk-FYr1VG5Tj5WcoHgjge9jw==", org="laempe")
- self.bucket = "energy-monitor"
- self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
- self.query_api = self.client.query_api()
- def write(self, values):
- points = []
- for meas in values:
- p = Point(meas.series).time(meas.timestamp).tag("source", meas.source)
- for field in dataclasses.fields(meas):
- if not field.name in ["timestamp", "series", "source"]:
- value = getattr(meas, field.name)
- if not type(value) is tuple:
- p.field(field.name, value)
- else:
- for i, v in enumerate(value):
- p.field(F"{field.name}_{i}", v)
- points.append(p)
- try:
- self.write_api.write(bucket=self.bucket, record=points)
- except Exception as ex:
- print("Influx DB write failed", ex)
|