Mit Blockly mehrere DP in ein influxDB Measurement schreiben

[prorun]

Hi, ich möchte mit Blockly zwei Datenpunkte in ein influxDB Measurement schreiben.

mit folgender Vorlage arbeite ich.

@legro said in Daten nur einmal täglich in Influxdb schreiben:

Hier ein Beispiel, wie ich das mit meinem Gasverbrauch geregelt habe ..

sendTo('influxdb.0', 'storeState', {
    id: ziel,
    state: {ts: zeit, val: wert, ack: false, from: ziel}
}, result => console.log('added'));

Ergebnis:

Nun möchte ich aber zwei Datenpunkte in das gleiche Measurement schreiben sodass es wie folgt aussieht:

Folgendes hatte ich ausprobiert. Leider ohne Erfolg:

sendTo('influxdb.0', 'storeState', {
    id: ziel,
    state: {ts: zeit, latitude: latitude, longitude: longitude, ack: false, from: ziel}
}, result => console.log('added'));

Weiß jemand wie der Quelltext des Java-Blockly-Blocks abgeändert werde muss sodass ich zwei Datenpunkte in das gleiche Measurement schreiben kann?

[TT-Tom]

@prorun

vielleicht hilft das hier weiter

[prorun]

@tt-tom
Vielen Dank für den Hinweis jedoch wird in all diesen Beispielen immer nur ein Wert pro Measurement übergeben (val: 123). Ich benötige jedoch zwei Fields in diesem einen Measurement.

Folgendes

sendTo('influxdb.0', 'storeState', {
    id: ziel,
    state: {ts: zeit, fields: {lati: latitude, longi: longitude}, ack: false, from: ziel}
}, result => console.log('added'));

funktioniert leider auch nicht.

[TT-Tom]

@prorun

dann kann ich dir das noch empfehlen, ist zwar nicht Blockly aber sehr gut erklärt.
Link Text

[prorun]

Ja, per JavaScript klappt es grundsätzlich jedoch bekomme ich in Java die nötigen Bedingen die ich zusätzlich brauche nicht hin.

Aber vlt kann davon jemand ableiten wie es in Blockly aussehen müsste?

// 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();

Wenn mir bzgl. des JavaScripts jemand aufzeigen könnte wie ich dafür sorgen kann dass die Abfrage der Datenpunkte (latitude und longitude) verzögert zum trigger erfolgt wäre mein Problem auch gelöst.

Hintergrund: 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. Ich brauche quasi eine Verzögerungszeit zwischen triggern des Skripts und auslesen der Werte aus den beiden Datenpunkten.

In Blockly eine Warteschleife mit der entsprechenden Verzögerung zu bauen die unterbrochen wird wenn 2ms später der zweite Wert das skript triggert und dann erst das Auslesen der Datenpunkte erfolgt bekomme ich hin. Das in Java umzusetzen übersteigt jedoch meine java-Kenntnisse.

[Marc Berg]

@prorun sagte in Mit Blockly mehrere DP in ein influxDB Measurement schreiben:

Weiß jemand wie der Quelltext des Java-Blockly-Blocks abgeändert werde muss sodass ich zwei Datenpunkte in das gleiche Measurement schreiben kann?

Du kannst mit dem InfluxDB Adapter nicht mehrere Fields und/oder Tags schreiben, und Blockly setzt meines Wissens nur auf dem Adapter auf.

@Kamikaze hatte aber ein ähnliches Problem per Skript gelöst:

https://forum.iobroker.net/post/1092536

[prorun]

@marc-berg Schade dass es mit Blockly nicht funktioniert. Dann wäre ich damit besser klargekommen.

Habe es jetzt aber per JavaScript 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 neuste Kombination der Werte in die influxDB geschrieben.

Funktioniert jetzt einwandfrei. Hiermit ist es möglich Koordinaten sauber in der influxDB abzulegen um mit Grafana beispielsweise 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"])

[Marc Berg]

@prorun
Über welchen Mechanismus bekommst du die GPS-Daten rein?

[prorun]

@marc-berg

Unterschiedlich.

Fahrzeuge kommen über BMW-Adapter

iPhone kommt über zwei iot-services (lati, longi) die über zahlreiche Kurzbefehle (für jede halbe stunde einen), die den aktuellen Standort abfragen, befüttert werden. Leider nicht mehr über den apple-find-me-Adapter
Wenn fürs iPhone jemand eine Alternative kennt, immer her damit.

Und Android-Geräte habe ich endlich allesamt abgeschafft. Also fast. Meine Frau darf sich Heiligabend freuen - ob sie will oder nicht