Quick hacky way to get Gadgetbridge exported data to InfluxDB

upload.py

#! /usr/bin/env python3

import argparse
import datetime
import influxdb
import sqlite3


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-d", "--database",
        help="Gadgetbridge exported database file",
        required=True,
    )
    parser.add_argument(
        "-t", "--host",
        help="Your influxdb host",
        type=str,
        required=True,
    )
    parser.add_argument(
        "-p", "--port",
        help="Your influxdb port",
        type=int,
        default=8086
    )
    parser.add_argument(
        "-i", "--influxdb",
        help="Your influxdb database",
        type=str,
        required=True
    )
    # TODO: Fetch this automatically
    parser.add_argument(
        "--last",
        help="Last uploaded timestamp",
        type=int,
        default=0
    )
    parser.add_argument(
        "--username",
        default="root"
    )
    parser.add_argument(
        "--password",
        default="root"
    )
    args = parser.parse_args()

    client = influxdb.InfluxDBClient(
        host=args.host,
        port=args.port,
        # Passing these as command line arguments is dangerous, don't do that
        # unless you are sure you know what you are doing
        username=args.username,
        password=args.password,
    )
    client.switch_database(args.influxdb)

    sql = sqlite3.connect(args.database)

    # TODO: This is really limited
    json_body = []
    for timestamp, steps, heart_rate in sql.execute("SELECT TIMESTAMP, STEPS, HEART_RATE FROM MI_BAND_ACTIVITY_SAMPLE WHERE timestamp > ? ORDER BY timestamp ASC", [args.last]):
        isodate = datetime.datetime.fromtimestamp(timestamp).isoformat()
        json_body.append({
            "measurement": "steps",
            "time": isodate,
            "fields": {
                "value": steps
            }
        })
        if heart_rate not in {-1, 0, 255}:  # Undefined data?
            json_body.append({
                "measurement": "heart_rate",
                "time": isodate,
                "fields": {
                    "value": heart_rate
                }
            })
    if json_body:
        client.write_points(json_body)
        print("Inserted {} points".format(len(json_body)))
        print("Last valid timestamp {}".format(timestamp))
    else:
        print("No new data found")

Unfortunately there doesn't seem to be an easy way to do this automatically at the moment. Hopefully this will become possible by https://github.com/Freeyourgadget/nextcloud-client at some point.