tim/bmspy
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Update from influxdb_client to influxdb_client_3

Browse Source
This commit is contained in:
tim
2026-05-02 17:40:31 +02:00
parent a782e19563
commit 827f59cf49
3 changed files with 318 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bmspy/influxdb.py
+52 -69
View File
@@ -1,32 +1,26 @@
import atexit, datetime, os, sys, time
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from influxdb_client_3 import InfluxDBClient3, Point
from bmspy import client
DAEMON_UPDATE_PERIOD = 30
''' Close remote connections '''
def influx_shutdown(influxclient):
if influxclient is not None:
influxclient.close()
def writeapi_shutdown(writeapi):
if writeapi is not None:
writeapi.close()
def influxdb_export(bucket, url=None, org=None, token=None, socket_path=None, daemonize=True, debug=0):
data = dict()
if not url:
url = os.environ["INFLUXDB_V2_URL"]
org = os.environ.get("INFLUXDB_V2_ORG")
token = os.environ["INFLUXDB_V2_TOKEN"]
if url:
influxclient = InfluxDBClient(url=url, token=token, org=org)
else:
influxclient = InfluxDBClient.from_env_properties()
influxclient = InfluxDBClient3(host=url, org=org, database=bucket, token=token)
atexit.register(influx_shutdown, influxclient)
if daemonize:
while True:
# collect new data, delay, and start again
data = client.read_data(socket_path, 'influxdb')
influxdb_write_snapshot(influxclient, bucket, data, debug)
time.sleep(DAEMON_UPDATE_PERIOD)
@@ -37,85 +31,71 @@ def influxdb_export(bucket, url=None, org=None, token=None, socket_path=None, da
influxclient.close()
atexit.unregister(influx_shutdown)
return
def influxdb_write_snapshot(influxclient, bucket, data, debug=0):
writeapi = influxclient.write_api(write_options=SYNCHRONOUS)
atexit.register(writeapi_shutdown, writeapi)
# Populate the data structure this period
if debug > 1:
print("influxdb: creating snapshot")
points = influxdb_create_snapshot(data, debug)
if debug > 1:
print("influxdb: writing snapshot")
try:
writeapi.write(bucket=bucket, record=points)
influxclient.write(record=points, database=bucket)
except Exception as e:
print(e)
writeapi.close()
atexit.unregister(writeapi_shutdown)
return
def influxdb_create_snapshot(data, debug=0):
points = []
now = datetime.datetime.now(datetime.timezone.utc).isoformat()
now = datetime.datetime.now(datetime.timezone.utc)
for kind, contains in data.items():
helpmsg = ''
if contains.get('help'):
helpmsg = contains.get('help')
units = None
if contains.get('units'):
units = contains.get('units')
# Simple values
value = None
helpmsg = contains.get('help') or ''
units = contains.get('units')
if contains.get('raw_value') is not None:
value = contains.get('raw_value')
if debug > 2:
print("value: {} : {}".format(kind, value));
point = Point(kind) \
.tag("units", units) \
.tag("help", helpmsg) \
.field("value", value) \
.time(now)
points.append(point)
# Doesn't have a value, but multiple values, each with a label:
label = None
print("value: {} : {}".format(kind, value))
points.append(
Point(kind)
.tag("units", units)
.tag("help", helpmsg)
.field("value", value)
.time(now)
)
if contains.get('raw_values') is not None and isinstance(contains.get('raw_values'), dict):
label = contains.get('label')
for idx, label_value in contains.get('raw_values').items():
if debug > 2:
print("labels: {} [{}] : {}".format(kind, idx, label_value));
point = Point(kind) \
.tag(label, idx) \
.tag("units", units) \
.tag("help", helpmsg) \
.field("value", label_value) \
.time(now)
points.append(point)
# Information (like a manufacturing date or a serial number)
print("labels: {} [{}] : {}".format(kind, idx, label_value))
points.append(
Point(kind)
.tag(label, idx)
.tag("units", units)
.tag("help", helpmsg)
.field("value", label_value)
.time(now)
)
if contains.get('info') is not None:
value = contains.get('info')
if debug > 2:
print("info: {} : {}".format(kind, value));
point = Point(kind) \
.tag("units", units) \
.tag("help", helpmsg) \
.field("value", value) \
.time(now)
points.append(point)
else:
pass
print("info: {} : {}".format(kind, value))
points.append(
Point(kind)
.tag("units", units)
.tag("help", helpmsg)
.field("value", value)
.time(now)
)
return points
def main():
import argparse
parser = argparse.ArgumentParser(
description='Query JBD BMS and report status',
add_help=True,
@@ -134,13 +114,13 @@ def main():
default=0, help='Print more verbose information (can be specified multiple times)')
args = parser.parse_args()
debug=args.verbose
debug = args.verbose
try:
if not os.getenv('INFLUXDB_V2_URL') and not args.influx_url:
raise argparse.ArgumentTypeError('Missing value for --url')
if not os.getenv('INFLUXDB_V2_ORG') and not args.influx_org:
raise argparse.ArgumentTypeError('Missing value for --url')
raise argparse.ArgumentTypeError('Missing value for --org')
if not os.getenv('INFLUXDB_V2_TOKEN') and not args.influx_token:
raise argparse.ArgumentTypeError('Missing value for --token')
except Exception as e:
@@ -152,13 +132,16 @@ def main():
client.handle_registration(args.socket, 'influxdb', debug)
atexit.register(client.handle_registration, args.socket, 'influxdb', debug)
influxdb_export(bucket=args.influx_bucket, \
url=args.influx_url, \
org=args.influx_org, \
token=args.influx_token, \
socket_path=args.socket, \
daemonize=True, \
debug=debug)
influxdb_export(
bucket=args.influx_bucket,
url=args.influx_url or None,
org=args.influx_org or None,
token=args.influx_token or None,
socket_path=args.socket,
daemonize=True,
debug=debug,
)
if __name__ == "__main__":
main()