VictoriaMetrics als zusätzliche Datenbank-Option für Messwerte ergänzt (STORE/DBASE=victoria)

Dritte, zu mongo/influx exklusive Auswahl für die laufenden Messwerte. Schreibpfad
nutzt das bestehende Influx-Line-Protocol unverändert (common/victoria_post.js);
Lesepfad (sensorapi/databases/victoria.js + victoria2json.js) holt Rohdaten per
VictoriaMetrics' /api/v1/export und bucketet/aggregiert stundenweise clientseitig,
nach Mongo-Konvention (Stunden-Start als Label, kein Zeit-Shift nötig wie bei Influx).
Scope bewusst auf die schon heute per DBASE umschaltbaren Funktionen begrenzt
(getActData/getNoiseAVGData) - getAvgData/getLongAvg/getGeigerData bleiben wie bisher.

Docker-Compose um victoriametrics-Service ergänzt (Retention explizit auf 100y
gesetzt, da VictoriaMetrics sonst nach 1 Monat Daten löscht).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.gitignore

@@ -9,3 +9,9 @@ log/
 
 # Produktionsstand vom Server, nur Referenz
 noisesensors/vonRemote/
+
+# Laufzeitdaten der lokal getesteten Container (DB-Dateien, aktuelle Messwerte)
+noisesensors/data/mongo/data/
+noisesensors/data/influx/
+noisesensors/data/victoria/
+noisesensors/data/aktdata.json

README.md

@@ -20,6 +20,8 @@ Die Datei **noise.tgz** enthält die folgenden Dateien und Verzeichnisse:
        |  |  |  +- <f>create.js
        |  + <d>influx
        |  |  + <d>data   
+       |  + <d>victoria
+       |  |  + <d>data   
 
 <f>: file,  <d>: directory       
 ~~~
@@ -42,6 +44,10 @@ hier liegt die Datei **create.js**, mit deren Hilfe bei ersten Start der Datenba
 Verzeichnis für die Influx-Datenbank
 * **data/influx/data**  
 Hier dann die eingelesenen Daten
+* **data/victoria**  
+Verzeichnis für die VictoriaMetrics-Datenbank (optionale Alternative zu Influx, siehe unten)
+* **data/victoria/data**  
+Hier dann die eingelesenen Daten
 
 
 ### Aufrufe
@@ -60,6 +66,8 @@ STORE=influx
 ~~~
 \<username>, \<passwort>, \<username1>, \<passwort1> und \<token> sind anzupassen.
 
+**STORE** legt fest, in welche Datenbank(en) *readin* die laufenden Messwerte schreibt: `influx`, `mongo`, `victoria` oder `both` (= mongo **und** influx gleichzeitig; Default). `victoria` ist dabei ein zu `influx` alternativer, exklusiver Wert (nicht über `both` kombinierbar). *sensorapi* liest die Messwerte unabhängig davon über **DBASE** (dieselben Werte `influx`/`mongo`/`victoria`, siehe Container-Beschreibung unten).
+
 Danach einloggen in das Docker-Registry auf *citysensor.de*:
 
 ~~~
@@ -91,7 +99,7 @@ Im **Portainer** als eigenes Stack anlegen (Inhalt von `readarchive-stack.yml`)
  `-s` | `START` | gestern
  `-e` | `END` | Startdatum + 1 Tag (**exklusiv**)
  `-t` | `TYP` | `noise`
- `-d` | `DBASE` | `both`, in der Stack-Datei auf `influx` vorbelegt
+ `-d` | `DBASE` | `mongo`, `influx`, `victoria` oder `both` (= mongo+influx); in der Stack-Datei auf `influx` vorbelegt
 
 *Deploy the stack* startet den Lauf, der Container endet danach und bleibt als „Exited" stehen. Für den nächsten Zeitraum nur die Variablen ändern und *Update the stack*. Der Haupt-Stack wird dabei nicht angefasst.
 
@@ -143,13 +151,19 @@ Die Influx-Datenbank.  Hier werden in dem Bucket *sensor_data* die reinen Messwe
 http://\<server>:8086  
 Logindaten:  Entsprechen der im Portainer hinterlegten.  
 
+* **victoriametrics**  
+Alternative zur Influx-Datenbank für die reinen Messwerte, wählbar über `STORE`/`DBASE=victoria`. Läuft als Single-Node-Instanz auf **Port 8428**, ohne Login/Auth.  
+**Wichtig:** VictoriaMetrics löscht standardmäßig Daten, die älter als 1 Monat sind (`-retentionPeriod=1`). Im `docker-compose.yml` ist das Flag deshalb explizit auf `--retentionPeriod=100y` gesetzt — das **nicht** entfernen, sonst verschwinden alte Messwerte nach 30 Tagen kommentarlos.  
+**Zugriff:**  
+http://\<server>:8428 (z.B. `/api/v1/export?match[]={sid="..."}` zum Rohdaten-Export)
+
 
 ## Sourcen für die Container
 
 Das gesamte Projekt ist im GitHub unter **laermsensor-stack** abgelegt. Für jeden der 4 Container existiert darunter ein Verzeichnis:
 **readin**, **readarchive**, **sensorapi** und **noise**. In diesen Verzeichnissen sind alle benötigten Sourcen enthalten. 
 
-Daneben gibt es **common** mit den Modulen, die sich *readin* und *readarchive* teilen: `mongo.js`, `influx_post.js`, `logit.js` und `statistics.js`. Sie lagen früher in beiden Komponenten doppelt und sind auseinandergelaufen. Wer dort etwas ändert, ändert es für beide — nach einer Änderung also **beide** Container neu bauen.
+Daneben gibt es **common** mit den Modulen, die sich *readin* und *readarchive* teilen: `mongo.js`, `influx_post.js`, `victoria_post.js`, `logit.js` und `statistics.js`. Sie lagen früher in beiden Komponenten doppelt und sind auseinandergelaufen. Wer dort etwas ändert, ändert es für beide — nach einer Änderung also **beide** Container neu bauen.
 
 Sollte was geändert werden, so muss mit **deploy.sh** der Container neu erzeugt und auf die Registry (siehe oben) gepushed werden. Der Build-Kontext von *readin* und *readarchive* ist wegen `common/` das Wurzelverzeichnis des Repositories; `deploy.sh` wechselt selbst dorthin und kann aus dem Komponentenverzeichnis aufgerufen werden.
 

common/victoria_post.js

@@ -0,0 +1,55 @@
+/* Zugriff auf VictoriaMetrics per HTTP (InfluxDB-Line-Protocol-kompatibler Write-Endpunkt)
+
+Gemeinsam genutzt von readin (Live-API) und readarchive (CSV-Archiv).
+*/
+
+import axios from 'axios'
+import { logit, logerror } from './logit.js'
+import { statistics } from './statistics.js'
+import { DateTime } from 'luxon'
+
+let VICTORIAHOST = process.env.VICTORIAHOST || "localhost"
+let VICTORIAPORT = process.env.VICTORIAPORT || 8428
+
+const VICTORIAURL_WRITE = `http://${VICTORIAHOST}:${VICTORIAPORT}/write?precision=ms`
+
+// `${e}` liefert bei einem AggregateError nur "AggregateError" - die einzelnen
+// Verbindungsfehler (je einer pro aufgeloester IP) stecken in e.errors bzw. e.cause.errors
+function describeError(e) {
+    const sub = e.errors || e.cause?.errors
+    if (sub?.length) {
+        return `${e.message || e.name}: ` + sub.map(s => `${s.code} ${s.address}:${s.port}`).join(', ')
+    }
+    if (e.response) {
+        return `HTTP ${e.response.status} ${JSON.stringify(e.response.data)}`
+    }
+    return `${e.code ? e.code + ' ' : ''}${e.message || e}`
+}
+
+// liefert true, wenn VictoriaMetrics die Daten uebernommen hat, sonst false
+export const victoriaWrite = async (data) => {
+    let start = DateTime.now()
+    let ok = false
+    try {
+        const ret = await axios({
+            method: 'post',
+            url: VICTORIAURL_WRITE,
+            data: data,
+            headers: {
+                Accept: 'application/json',
+                'Content-Type': 'text/plain; charset=utf-8'
+            },
+            timeout: 10000,
+        })
+        if (ret.status != 204) {
+            logerror(`doWrite2API Status: ${ret.status}`)
+        } else {
+            ok = true
+        }
+    } catch (e) {
+        logerror(`doWrite2API  ${VICTORIAURL_WRITE}  ${describeError(e)}`)
+    }
+    statistics['writeVictoriaData[sensor_data]Time'] = DateTime.now().diff(start, ['seconds']).toObject().seconds
+    logit(`Victoria-Write-Time: ${start.diffNow('seconds').toObject().seconds * -1} sec`)
+    return ok
+}

noisesensors/.env.example

@@ -9,9 +9,12 @@ MONGO_ROOT_PASSWORD=<Mongo_Passwort>
 
 LOCALDIR=<Aktuelle Direcrory ab />
 
-# Datenbank, in welche die laufenden Messwerte gespeichert werden (kannn 'influx' oder 'mongo' sein)
+# Datenbank, in welche die laufenden Messwerte gespeichert werden (kann 'influx', 'mongo' oder 'victoria' sein)
 STORE=influx
 
+# Datenbank, aus welcher sensorapi die Messwerte liest (kann 'influx', 'mongo' oder 'victoria' sein)
+DBASE=influx
+
 # Secret zum Signieren der Session-Cookies von esp2sensor (langer Zufallsstring)
 ESP2SENSOR_SESSION_SECRET=<zufaelliger_string>
 

noisesensors/docker-compose.yml

@@ -26,20 +26,33 @@ services:
       DOCKER_INFLUXDB_INIT_ORG: citysensor
       DOCKER_INFLUXDB_INIT_BUCKET: sensor_data
       DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: ${DOCKER_INFLUXTOKEN}  
-    restart: 
-      unless-stopped 
-    container_name: influxdb  
+    restart:
+      unless-stopped
+    container_name: influxdb
+
+  victoriametrics:
+    image: victoriametrics/victoria-metrics:v1.102.0
+    ports:
+      - "8428:8428"
+    volumes:
+      - ${LOCALDIR}/noisesensors/data/victoria/data:/victoria-metrics-data
+    command:
+      - "--storageDataPath=/victoria-metrics-data"
+      - "--retentionPeriod=100y"
+    restart: unless-stopped
+    container_name: victoriametrics
 
   readin:
     image: docker.citysensor.de/readin
     environment:
       MONGOHOST: mongodb
       INFLUXHOST: influxdb
+      VICTORIAHOST: victoriametrics
       TYP: "[\"noise\"]"
       STORE: ${STORE:-mongo}
       MONGOAUTH: true
       MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD}
-      INFLUXTOKEN: ${DOCKER_INFLUXTOKEN}  
+      INFLUXTOKEN: ${DOCKER_INFLUXTOKEN}
     volumes:
       - ${LOCALDIR}/noisesensors/log:/var/log
       # Pfad im Image ist jetzt /opt/app/readin, weil readin und readarchive
@@ -67,7 +80,8 @@ services:
     environment:
       INFLUXHOST: influxdb
       INFLUXTOKEN: ${DOCKER_INFLUXTOKEN}
-      DBASE: influx
+      VICTORIAHOST: victoriametrics
+      DBASE: ${DBASE:-influx}
       MONGOHOST: mongodb
       MONGOAUTH: true
       MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD}

noisesensors/readarchive-stack.yml

@@ -16,7 +16,8 @@
 #
 # readFromcsv.js liest jeden Parameter auch aus einer Env-Variablen:
 #   -s = START (Default: gestern), -e = END (Default: Startdatum + 1 Tag,
-#   exklusiv), -t = TYP (Default: noise), -d = DBASE (Default: both),
+#   exklusiv), -t = TYP (Default: noise), -d = DBASE ('mongo', 'influx',
+#   'victoria' oder 'both' = mongo+influx; Default: both),
 #   -p = PARALLEL (Default: 4) - so viele Sensoren werden gleichzeitig geholt.
 #
 # Voraussetzungen:
@@ -39,6 +40,7 @@ services:
     environment:
       MONGOHOST: mongodb
       INFLUXHOST: influxdb
+      VICTORIAHOST: victoriametrics
       MONGOAUTH: "true"
       MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD}
       INFLUXTOKEN: ${DOCKER_INFLUXTOKEN}

readarchive/readFromcsv.js

@@ -33,7 +33,7 @@ const DBASE = process.env.DBASE || 'both'
 // Mehr als 4 bringt nichts: ab da liefert das Archiv nicht mehr schneller.
 const PARALLEL = parseInt(process.env.PARALLEL) || 4
 
-const DATABASE = ['mongo', 'influx', 'both']
+const DATABASE = ['mongo', 'influx', 'victoria', 'both']
 
 const DEVELOP = process.env.DEVELOP || false;
 const LIVE = (process.env.LIVE == "true") || true
@@ -47,6 +47,7 @@ import axios from 'axios'
 import https from 'https'
 import * as mongo  from '../common/mongo.js'
 import * as influx from '../common/influx_post.js'
+import * as victoria from '../common/victoria_post.js'
 import fs from 'fs'
 import csv from 'csvtojson'
 import pkg from  './package.json' with { type: "json" }
@@ -67,6 +68,8 @@ const newProps = []
 
 // Sensor/Tag-Paare, deren Influx-Write fehlgeschlagen ist - Auswertung am Ende von main()
 const failedInfluxWrites = []
+// Sensor/Tag-Paare, deren VictoriaMetrics-Write fehlgeschlagen ist - Auswertung am Ende von main()
+const failedVictoriaWrites = []
 
 const saveList = (list) => {
     fs.writeFileSync('data/list.json', JSON.stringify(list))
@@ -397,6 +400,12 @@ async function enterOneSensorinDB(client, sid, erg, dataline, day, styp, args) {
             failedInfluxWrites.push({sensorid: sid, day: day})
         }
     }
+    if ((args.database === 'victoria') && (dataline !== '')) {
+        const ok = await victoria.victoriaWrite(dataline)
+        if (!ok) {
+            failedVictoriaWrites.push({sensorid: sid, day: day})
+        }
+    }
     if((args.database === 'both') || (args.database === 'mongo')) {
         if(erg.length !== 0) {
             await mongo.writeDataArray(client, mongo.dataCollName(styp), erg)
@@ -507,7 +516,7 @@ function parse_cmdline(argv) {
                 console.log("       will be used and endDate = startDate + 1 day")
                 console.log("    -t sensorType: if given, only those sensors will be used; default: 'noise'");
                 console.log("       allowed types: 'noise', 'radiactivity', 'pm' and 'thp'")
-                console.log("    -d database: 'mongo', 'influx' or 'both'; default: 'both'")
+                console.log("    -d database: 'mongo', 'influx', 'victoria' or 'both' ('both' = mongo + influx); default: 'both'")
                 console.log(`    -p parallel: number of sensors fetched at the same time; default: ${PARALLEL}`)
                 console.log("    -v version: show version");
                 console.log("    -h this help text");
@@ -555,6 +564,13 @@ async function main() {
         }
         process.exitCode = 1                            // damit cron/docker den Fehlschlag sieht
     }
+    if (failedVictoriaWrites.length > 0) {
+        logerror(`${failedVictoriaWrites.length} Victoria-Writes fehlgeschlagen - diese Daten fehlen in VictoriaMetrics:`)
+        for (const f of failedVictoriaWrites) {
+            logerror(`   Sensor ${f.sensorid}, Tag ${f.day}`)
+        }
+        process.exitCode = 1                            // damit cron/docker den Fehlschlag sieht
+    }
     let dauer  = starttime.diffNow('seconds').toObject().seconds * -1
     const duration1 = Duration.fromObject({ seconds: dauer });
     const output1 = duration1.toFormat('hh:mm:ss');

readin/fetchnewdata.js

@@ -14,13 +14,14 @@
 // TYP: Ist ein Array von Strings, wenn nur einzelne Typen gespeichert werden sollen, also z.B.:
 //  ['pm', 'noise']
 const TYP = process.env.TYP || ''
-const STORE = (process.env.STORE || 'both').toLowerCase()  // 'mongo' | 'influx' | 'both'
+const STORE = (process.env.STORE || 'both').toLowerCase()  // 'mongo' | 'influx' | 'victoria' | 'both' ('both' = mongo + influx)
 
 import { doReadfromAPI as  readin } from './readdata.js'
 import { constructDBaseEntries as parse} from './parse.js'
 
 import * as mongo from '../common/mongo.js'
 import * as influx from '../common/influx_post.js'
+import * as victoria from '../common/victoria_post.js'
 import { logit, logerror } from '../common/logit.js'
 import { statistics } from '../common/statistics.js'
 import { DateTime } from 'luxon'
@@ -62,6 +63,11 @@ const fetchNewData = async (args) => {
                 await influx.influxWrite(idata)
             }
 
+            // write sensor data to VictoriaMetrics
+            if(args.victoria) {
+                await victoria.victoriaWrite(idata)
+            }
+
             // write properties to mongoDB
             await mongo.bulkWrite(client, mongo.property_coll, props)
         }
@@ -78,7 +84,7 @@ const fetchNewData = async (args) => {
 function parse_cmdline(argv) {
     let parser = new mod_getopt.BasicParser('i(influx)m(mongo)t:(typ)h(help)v(version)',argv);
     let option;
-    let ret = {influx: STORE === 'both' || STORE === 'influx', mongo: STORE === 'both' || STORE === 'mongo', typ: TYP}
+    let ret = {influx: STORE === 'both' || STORE === 'influx', mongo: STORE === 'both' || STORE === 'mongo', victoria: STORE === 'victoria', typ: TYP}
     while((option = parser.getopt()) !== undefined) {
         switch(option.option) {
             case 'i':
@@ -111,6 +117,7 @@ function parse_cmdline(argv) {
                 console.log("Params:");
                 console.log("    -i use only InfluxDB to store the data (default use both, InfluxDB and MongoDB))");
                 console.log("    -m use only MongoDB to store the data (default use both, InfluxDB and MongoDB)");
+                console.log("    (VictoriaMetrics storage is selected exclusively via STORE=victoria, no CLI flag)");
                 console.log("    -t [ sensorType, ..]: if given, only those sensors will be used (ex: laerm) default: all");
                 console.log("       MUST BE AN ARRAY!; allowed types: 'pm', 'noise', 'radiactivity', 'thp', 'gps'.")
                 console.log("    -v version: show version");

sensorapi/actions/getsensorData.js

@@ -4,6 +4,7 @@ const DBASE = process.env.DBASE || 'mongo'
 import {DateTime} from "luxon"
 import * as influx from "../databases/influx.js"
 import * as mongo from "../databases/mongo.js"
+import * as victoria from "../databases/victoria.js"
 import {returnOnError} from "../utilities/reporterror.js"
 import {csv2Json} from "../utilities/csv2json.js"
 import checkParams from "../utilities/checkparams.js"
@@ -155,8 +156,10 @@ export async function getActData(opts) {
         return await mongo.fetchActData(opts)
     } else if (DBASE === 'influx') {
        return await influx.fetchActData(opts)
+    } else if (DBASE === 'victoria') {
+       return await victoria.fetchActData(opts)
     }
-    return {err: 'DBASEUNKNOWN', values: []}    
+    return {err: 'DBASEUNKNOWN', values: []}
 }
 
 

sensorapi/databases/victoria.js

@@ -0,0 +1,67 @@
+// Access to VictoriaMetrics via HTTP (Prometheus-compatible /api/v1/export,
+// no Flux support - see common/victoria_post.js for the write side, which
+// reuses the InfluxDB line-protocol format on VictoriaMetrics' /write endpoint).
+
+import axios from 'axios'
+import { logit, logerror } from '../utilities/logit.js'
+import { returnOnError } from "../utilities/reporterror.js"
+import { exportToRows, bucketNoiseAVG } from "../utilities/victoria2json.js"
+
+let VICTORIAHOST = process.env.VICTORIAHOST || "localhost"
+let VICTORIAPORT = process.env.VICTORIAPORT || 8428
+
+const VICTORIAURL_EXPORT = `http://${VICTORIAHOST}:${VICTORIAPORT}/api/v1/export`
+
+// opts.start/opts.stop arrive as Flux range() fragments ("start: <iso>" /
+// "stop: <iso>") built by calcRange() in getsensorData.js - strip the Flux
+// keyword the same way sensorapi/databases/mongo.js already does.
+const isoStart = (opts) => opts.start.slice(7)
+const isoStop = (opts) => opts.stop.slice(6)
+
+const victoriaExport = async (matchSelector, start, end) => {
+    let erg = { values: '', err: null }
+    try {
+        let ret = await axios({
+            method: 'get',
+            url: VICTORIAURL_EXPORT,
+            params: { 'match[]': matchSelector, start, end },
+            timeout: 10000,
+            transformResponse: [(data) => data],   // response body is ndjson, not a single JSON document - keep it raw
+        })
+        if (ret.status !== 200) {
+            return returnOnError(erg, 'RESPSTATUS', victoriaExport.name, ret.status)
+        }
+        erg.values = ret.data
+    } catch (e) {
+        return returnOnError(erg, e, victoriaExport.name)
+    }
+    return erg
+}
+
+export const fetchActData = async (opts) => {
+    let ret = { err: null, values: [] }
+    const match = `{__name__=~"noise_(LAeq|LA_min|LA_max|E10tel_eq)", sid="${opts.sensorid}"}`
+    let { values, err } = await victoriaExport(match, isoStart(opts), isoStop(opts))
+    if (err) {
+        return returnOnError(ret, err, fetchActData.name)
+    }
+    ret.values = exportToRows(values, opts.sort)
+    if (ret.values.length === 0) {
+        return returnOnError(ret, 'NODATA', fetchActData.name)
+    }
+    return ret
+}
+
+export const fetchNoiseAVGData = async (opts) => {
+    let ret = { err: null, values: [] }
+    const match = `{__name__=~"noise_(E10tel_eq|LA_max)", sid="${opts.sensorid}"}`
+    let { values, err } = await victoriaExport(match, isoStart(opts), isoStop(opts))
+    if (err) {
+        return returnOnError(ret, err, fetchNoiseAVGData.name)
+    }
+    ret.values = bucketNoiseAVG(values, opts.peak, opts.long)
+    if (ret.values.length === 0) {
+        return returnOnError(ret, 'NODATA', fetchNoiseAVGData.name)
+    }
+    return ret
+}

sensorapi/sensorspecials/noise.js

@@ -9,6 +9,7 @@ import {DateTime} from 'luxon'
 import { translate as trans } from '../routes/api.js'
 import * as influx from "../databases/influx.js"
 import * as mongo from "../databases/mongo.js"
+import * as victoria from "../databases/victoria.js"
 import { setoptionfromtable } from "../utilities/chartoptions.js"
 
 export const getNoiseData = async (params, possibles, props) => {
@@ -451,6 +452,9 @@ const getNoiseAVGData = async (opts) => {
         for (let x=0; x < ret.values.length; x++) {
             ret.values[x].datetime = DateTime.fromISO(ret.values[x].datetime).toUTC().minus({hours:1}).toFormat("yyyy-LL-dd'T'HH:mm:ss'Z'")
         }
+    } else if (DBASE === 'victoria') {
+        // victoria.js buckets by hour-start itself (like mongo), so no shift is needed here
+        ret = await victoria.fetchNoiseAVGData(opts)
     } else {
         ret.err = 'DBASEUNKNOWN'
     }

sensorapi/utilities/victoria2json.js

@@ -0,0 +1,97 @@
+// Parse VictoriaMetrics' /api/v1/export ndjson responses into the same row
+// shapes influx.js builds via csv2Json() / pivot(), for the noise-specific
+// fields (measurement "noise", separator "_", i.e. metric names noise_LAeq,
+// noise_LA_min, noise_LA_max, noise_E10tel_eq).
+
+import { DateTime } from 'luxon'
+
+const parseExportLines = (ndjsonBody) => {
+    return String(ndjsonBody)
+        .split('\n')
+        .filter((line) => line.trim() !== '')
+        .map((line) => {
+            try {
+                return JSON.parse(line)
+            } catch (e) {
+                return null
+            }
+        })
+        .filter((series) => series !== null)
+}
+
+const fieldName = (metricName) => metricName.replace(/^noise_/, '')
+
+// Merges the per-field series (one ndjson line per field) back into rows of
+// {datetime, LAeq, LA_min, LA_max, E10tel_eq}, matched by raw timestamp -
+// mirrors influx's pivot(rowKey:["_time"], columnKey:["_field"]).
+export const exportToRows = (ndjsonBody, sort) => {
+    const rowsByTime = new Map()
+    for (const series of parseExportLines(ndjsonBody)) {
+        const field = fieldName(series.metric.__name__)
+        const { values, timestamps } = series
+        for (let i = 0; i < timestamps.length; i++) {
+            const ts = timestamps[i]
+            let row = rowsByTime.get(ts)
+            if (!row) {
+                row = { datetime: DateTime.fromMillis(ts).toUTC().toISO() }
+                rowsByTime.set(ts, row)
+            }
+            row[field] = values[i]
+        }
+    }
+    const rows = Array.from(rowsByTime.values())
+    rows.sort((a, b) => (a.datetime < b.datetime ? -1 : a.datetime > b.datetime ? 1 : 0))
+    if (sort === -1) {
+        rows.reverse()
+    }
+    return rows
+}
+
+// Buckets raw E10tel_eq/LA_max samples by hour (truncated to the start of the
+// hour of each sample, same convention as sensorapi/databases/mongo.js's
+// $dateToString hour-grouping) and computes n_AVG/n_sum/count/peakcount -
+// equivalent to influx.js's aggregateWindow()/reduce() Flux pipeline.
+// Bucketing by hour-start (not hour-end, as Influx does) means no extra
+// 1-hour shift is needed downstream, unlike the DBASE === 'influx' branch.
+export const bucketNoiseAVG = (ndjsonBody, peak, long) => {
+    const buckets = new Map()
+    for (const series of parseExportLines(ndjsonBody)) {
+        const field = fieldName(series.metric.__name__)
+        if (field !== 'E10tel_eq' && field !== 'LA_max') {
+            continue
+        }
+        const { values, timestamps } = series
+        for (let i = 0; i < timestamps.length; i++) {
+            const key = DateTime.fromMillis(timestamps[i]).toUTC().startOf('hour').toISO()
+            let b = buckets.get(key)
+            if (!b) {
+                b = { sum: 0, count: 0, peakcount: 0 }
+                buckets.set(key, b)
+            }
+            if (field === 'E10tel_eq') {
+                b.sum += values[i]
+                b.count += 1
+            } else {
+                if (values[i] >= peak) {
+                    b.peakcount += 1
+                }
+            }
+        }
+    }
+    const rows = Array.from(buckets.entries())
+        .filter(([, b]) => b.count > 0)          // matches Influx's inner join: hours without E10tel_eq samples are dropped
+        .map(([datetime, b]) => {
+            let row = {
+                datetime,
+                n_AVG: 10 * Math.log10(b.sum / b.count),
+                peakcount: b.peakcount,
+            }
+            if (long) {
+                row.count = b.count
+                row.n_sum = b.sum
+            }
+            return row
+        })
+    rows.sort((a, b) => (a.datetime < b.datetime ? -1 : a.datetime > b.datetime ? 1 : 0))
+    return rows
+}