laermsensor-stack/sensorapi/utilities/victoria2json.js

// 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
}