Ordentlich Logging GPS-Daten influxDB Grafana Geomap Routen

[prorun]

Moin Leute, ich möchte GPS-Daten per Java in die influxDB schreiben sodass latitude und longitude mit exakt gleichen Zeitstempel in ein Measurement geschrieben werden was mit folgendem Skript auch sehr gut klappt (vielen Dank an der Stelle an Matthias Kleine von haus-automatisierung).

Jedoch habe ich folgendes Problem: die Werte latitude und longitude kommen etwas Zeitversetzt rein, manchmal nur mit einem Versatz von 2ms, aber das ist ausreichend dass zuerst eine falsche Kombination in die influx geschrieben wird - neue latitude mit altem longitude und 2ms später die richtige Kombination.

Wie kann ich in dem folgendem Skript eine Verzögerung einbauen sodass nach Datenpunktaktualisierung 500ms abgewartet wird und danach erneut abgefragte Werte in die influx übertragen werden. Sodass zwar die erste Änderung an einem der beiden Datenpunkte (latitude, longitude) weiterhin das Skript trigger aber dann eine feste Zeit abgewartet wird bis die Werte geschrieben werden.

Leider sind meine Java Kenntnisse nur begrenzt. Bitte helft mir. Dieses Thema interessiert bestimmt auch den ein oder anderen. Ein ordentliches Loggin von GPS-Daten ist schließlich ne feine Sache. Leider ist die Umsetzung gar nicht mal so trivial.

// v0.1
const axios = require('axios').default;

const influxDbInstance = 'influxdb.0';
const token = 'TOKEN';
const measurement = 'Measurement';

const loggingTemplate = {
    '0_userdata.0.test-number2': 'latitude',
    '0_userdata.0.test-number3': 'longitude',
};

const loggingObj = {};

async function start() {
    const influxDbInstanceConfig = await getObjectAsync(`system.adapter.${influxDbInstance}`);

    const protocol = influxDbInstanceConfig.native.protocol;
    const host = influxDbInstanceConfig.native.host;
    const port = influxDbInstanceConfig.native.port;
    const org = influxDbInstanceConfig.native.organization;
    const bucket = influxDbInstanceConfig.native.dbname;

    console.log(`Starting "${measurement}" logging to ${protocol}://${host}:${port} into bucket "${bucket}" by org ${org}`);

    // Init loggingObj with current values
    for (let [objId, key] of Object.entries(loggingTemplate)) {
        const state = await getStateAsync(objId);
        if (state && !isNaN(state.val)) {
            loggingObj[key] = state.val;
        } else {
            loggingObj[key] = 0;
        }
    }

    on({ id: Object.keys(loggingTemplate), change: 'ne' }, async (obj) => {
        // Update value in loggingObj
        const key = loggingTemplate[obj.id];
        loggingObj[key] = obj.state.val;

        // Save Data
        const data = `${measurement} ${Object.keys(loggingObj)
            .filter(key => !isNaN(loggingObj[key]))
            .map((key) => `${key}=${loggingObj[key]}`)
            .join(',')}`;

        if (data) {
            // console.log(`Saving "${data}" to InfluxDB @ ${protocol}://${host}:${port}/`);

            axios.post(`${protocol}://${host}:${port}/api/v2/write?bucket=${bucket}&org=${org}`, data, {
                headers: {
                    'Content-Type': 'text/plain',
                    'Authorization': `Token ${token}`
                }
            }).catch(err => {
                console.error(err);
            });
        }
    });
}
start();

[prorun]

Habe es jetzt gelöst bekommen.

Mit folgendem Skript erhält man ein einziges Measurement was latitude und longitude mit exakt gleichen Zeitstempel enthält. Zusätzlich können latitude und longitude zeitversetzt eintrudeln - welcher Wert zuerst kommt ist dabei egal. In diesem Beispiel hier bis zu 10 Sekunden Versatz. Es wird also obwohl latitude und longitude in unterschiedlichen Datenpunkten liegen und bei Änderung beider Werte das Skript 2x triggert (bei Änderung von latitude und noch mal bei Änderung von longitude) immer nur einmal und nur die Kombination aus beiden neuen Werten in die influxDB geschrieben.

Funktioniert jetzt einwandfrei. Hiermit ist es möglich Koordinaten sauber in der influxDB abzulegen um beispielsweise mit Grafana Routen daraus zu basteln.

// v0.2
const axios = require('axios').default;
 
const influxDbInstance = 'influxdb.0';
const token = 'TOKEN';
const measurement = 'GPS-Test';
 
const loggingTemplate = {
    '0_userdata.0.test-number2': 'lati',
    '0_userdata.0.test-number3': 'longi',
};

const loggingObj = {};

async function start() {
    const influxDbInstanceConfig = await getObjectAsync(`system.adapter.${influxDbInstance}`);

    const protocol = influxDbInstanceConfig.native.protocol;
    const host = influxDbInstanceConfig.native.host;
    const port = influxDbInstanceConfig.native.port;
    const org = influxDbInstanceConfig.native.organization;
    const bucket = influxDbInstanceConfig.native.dbname;

    console.log(`Starting "${measurement}" logging to ${protocol}://${host}:${port} into bucket "${bucket}" by org ${org}`);

    // Init loggingObj with current values
    for (let [objId, key] of Object.entries(loggingTemplate)) {
        const state = await getStateAsync(objId);
        if (state && !isNaN(state.val)) {
            loggingObj[key] = state.val;
        } else {
            loggingObj[key] = 0;
        }
    }
// Variable für Timeout sodass bei Wiederholung der vorherige Timeout abgebrochen werden kann
let timeoutHandler;

// Trigger-Bedingung (bei Änderung eines der DP (loggingTemplate))
on({ id: Object.keys(loggingTemplate), change: 'ne' }, async (obj) => {
    // Update value in loggingObj
    const key = loggingTemplate[obj.id];
    loggingObj[key] = obj.state.val;

    // Lösche vorherigen Timeout, falls vorhanden
    if (timeoutHandler) {
        clearTimeout(timeoutHandler);
    }

    // Setze neuen Timeout
    timeoutHandler = setTimeout(async () => {
        // Save Data
        const data = `${measurement} ${Object.keys(loggingObj)
            .filter(key => !isNaN(loggingObj[key]))
            .map((key) => `${key}=${loggingObj[key]}`)
            .join(',')}`;

        if (data) {
            // console.log(`Saving "${data}" to InfluxDB @ ${protocol}://${host}:${port}/`);

            axios.post(`${protocol}://${host}:${port}/api/v2/write?bucket=${bucket}&org=${org}`, data, {
                headers: {
                    'Content-Type': 'text/plain',
                    'Authorization': `Token ${token}`
                }
            }).catch(err => {
                console.error(err);
            });
        }
    }, 10000); // 10.000 Millisekunden = 10 Sekunden
});
}
start();

Uns so hier sieht das ganze dann in Grafana mit dem Widget Geomap aus:

from(bucket: "iobroker")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r)   => 
  r._measurement == "GPS-Test" and 
  r._field == "lati" or
  r._field == "longi" )
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> drop(columns: ["_start", "_stop", "_measurement"])