Switched from influx to victoria metrics (more light weight for pi)
This commit is contained in:
@@ -1,152 +1,107 @@
|
||||
import {InfluxDB} from '@influxdata/influxdb-client';
|
||||
import {cfg} from './config.mjs';
|
||||
|
||||
const MEASUREMENTS = ['accumulation', 'environment', 'light', 'lightning', 'rain', 'seismic', 'wind'];
|
||||
|
||||
let client, currentUrl, currentToken;
|
||||
|
||||
function getClient() {
|
||||
const c = cfg();
|
||||
if(c.INFLUX_URL !== currentUrl || c.INFLUX_TOKEN !== currentToken) {
|
||||
client = new InfluxDB({url: c.INFLUX_URL, token: c.INFLUX_TOKEN});
|
||||
currentUrl = c.INFLUX_URL;
|
||||
currentToken = c.INFLUX_TOKEN;
|
||||
}
|
||||
return {client, c};
|
||||
async function query(c, q) {
|
||||
const url = new URL('/api/v1/query_range', c.INFLUX_URL);
|
||||
url.searchParams.set('query', q);
|
||||
url.searchParams.set('step', '60s');
|
||||
const res = await fetch(url);
|
||||
const json = await res.json();
|
||||
return json.data?.result || [];
|
||||
}
|
||||
|
||||
async function query(flux) {
|
||||
const {client, c} = getClient();
|
||||
const api = client.getQueryApi(c.INFLUX_ORG);
|
||||
const rows = [];
|
||||
return new Promise((resolve, reject) => {
|
||||
api.queryRows(flux, {
|
||||
next(row, meta) { rows.push(meta.toObject(row)); },
|
||||
error(err) { reject(err); },
|
||||
complete() { resolve(rows); },
|
||||
});
|
||||
});
|
||||
async function queryInstant(c, q) {
|
||||
const url = new URL('/api/v1/query', c.INFLUX_URL);
|
||||
url.searchParams.set('query', q);
|
||||
const res = await fetch(url);
|
||||
const json = await res.json();
|
||||
return json.data?.result || [];
|
||||
}
|
||||
|
||||
function truncateToHour(iso) {
|
||||
const d = new Date(iso);
|
||||
d.setMinutes(0, 0, 0);
|
||||
const pad = n => String(n).padStart(2, '0');
|
||||
return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}T${pad(d.getHours())}:00`;
|
||||
}
|
||||
|
||||
function truncateToDay(iso) {
|
||||
return new Date(iso).toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
function extractRow(row) {
|
||||
function metricToFields(results) {
|
||||
const out = {};
|
||||
for(const [k, v] of Object.entries(row)) {
|
||||
if(!k.startsWith('_') && k !== 'result' && k !== 'table') out[k] = v;
|
||||
for (const r of results) {
|
||||
const field = r.metric.__name__ || r.metric.field;
|
||||
const val = r.value?.[1] ?? r.values?.at(-1)?.[1];
|
||||
if (val !== undefined) out[field] = parseFloat(val);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
export async function queryCurrent() {
|
||||
const {c} = getClient();
|
||||
const flux = `
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: -1h)
|
||||
|> filter(fn: (r) => ${MEASUREMENTS.map(m => `r._measurement == "${m}"`).join(' or ')})
|
||||
|> last()
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`;
|
||||
const rows = await query(flux);
|
||||
const result = {time: new Date()};
|
||||
for(const row of rows) {
|
||||
Object.assign(result, extractRow(row));
|
||||
const c = cfg();
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
const out = {time: new Date()};
|
||||
|
||||
for (const m of MEASUREMENTS) {
|
||||
const results = await queryInstant(c, `{measurement="${m}"}`);
|
||||
Object.assign(out, metricToFields(results));
|
||||
}
|
||||
return result;
|
||||
return out;
|
||||
}
|
||||
|
||||
export async function queryHourly(start, end) {
|
||||
const {c} = getClient();
|
||||
const flux = `
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: ${start.toISOString()}, stop: ${end.toISOString()})
|
||||
|> filter(fn: (r) => ${MEASUREMENTS.map(m => `r._measurement == "${m}"`).join(' or ')})
|
||||
|> group(columns: ["_field"])
|
||||
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`;
|
||||
const rows = await query(flux);
|
||||
return rows.map(row => ({
|
||||
time: truncateToHour(row._time),
|
||||
...extractRow(row),
|
||||
}));
|
||||
const c = cfg();
|
||||
const s = Math.floor(start.getTime() / 1000);
|
||||
const e = Math.floor(end.getTime() / 1000);
|
||||
const byTime = {};
|
||||
|
||||
for (const m of MEASUREMENTS) {
|
||||
const results = await query(c, `avg_over_time({measurement="${m}"}[1h])`);
|
||||
results.forEach(r => {
|
||||
const field = r.metric.__name__;
|
||||
r.values.forEach(([ts, val]) => {
|
||||
const time = new Date(ts * 1000).toISOString().slice(0, 13) + ':00';
|
||||
if (!byTime[time]) byTime[time] = {time};
|
||||
byTime[time][field] = parseFloat(val);
|
||||
});
|
||||
});
|
||||
}
|
||||
return Object.values(byTime).sort((a, b) => a.time.localeCompare(b.time));
|
||||
}
|
||||
|
||||
export async function queryDaily(start, end) {
|
||||
const {c} = getClient();
|
||||
const measurements = ['accumulation', 'environment', 'light', 'lightning', 'seismic'];
|
||||
const results = {};
|
||||
const c = cfg();
|
||||
const byTime = {};
|
||||
|
||||
for(const m of measurements) {
|
||||
for (const m of ['accumulation', 'environment', 'light', 'lightning', 'seismic']) {
|
||||
const [means, mins, maxs] = await Promise.all([
|
||||
query(`
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: ${start.toISOString()}, stop: ${end.toISOString()})
|
||||
|> filter(fn: (r) => r._measurement == "${m}")
|
||||
|> group(columns: ["_field"])
|
||||
|> aggregateWindow(every: 1d, fn: mean, createEmpty: false)
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`),
|
||||
query(`
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: ${start.toISOString()}, stop: ${end.toISOString()})
|
||||
|> filter(fn: (r) => r._measurement == "${m}")
|
||||
|> group(columns: ["_field"])
|
||||
|> aggregateWindow(every: 1d, fn: min, createEmpty: false)
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`),
|
||||
query(`
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: ${start.toISOString()}, stop: ${end.toISOString()})
|
||||
|> filter(fn: (r) => r._measurement == "${m}")
|
||||
|> group(columns: ["_field"])
|
||||
|> aggregateWindow(every: 1d, fn: max, createEmpty: false)
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`),
|
||||
query(c, `avg_over_time({measurement="${m}"}[1d])`),
|
||||
query(c, `min_over_time({measurement="${m}"}[1d])`),
|
||||
query(c, `max_over_time({measurement="${m}"}[1d])`),
|
||||
]);
|
||||
|
||||
for(const row of means) {
|
||||
const time = truncateToDay(row._time);
|
||||
if(!results[time]) results[time] = {time};
|
||||
Object.assign(results[time], extractRow(row));
|
||||
}
|
||||
means.forEach(r => {
|
||||
r.values.forEach(([ts, val]) => {
|
||||
const time = new Date(ts * 1000).toISOString().slice(0, 10);
|
||||
if (!byTime[time]) byTime[time] = {time};
|
||||
byTime[time][r.metric.__name__] = parseFloat(val);
|
||||
});
|
||||
});
|
||||
|
||||
for(const row of mins) {
|
||||
const time = truncateToDay(row._time);
|
||||
if(!results[time]) results[time] = {time};
|
||||
if(row.temperature != null) results[time].env_temp_min_c = row.temperature;
|
||||
}
|
||||
for(const row of maxs) {
|
||||
const time = truncateToDay(row._time);
|
||||
if(!results[time]) results[time] = {time};
|
||||
if(row.temperature != null) results[time].env_temp_max_c = row.temperature;
|
||||
}
|
||||
const applyMinMax = (results, suffix) => results.forEach(r => {
|
||||
if (r.metric.__name__ !== 'temperature') return;
|
||||
r.values.forEach(([ts, val]) => {
|
||||
const time = new Date(ts * 1000).toISOString().slice(0, 10);
|
||||
if (!byTime[time]) byTime[time] = {time};
|
||||
byTime[time][suffix] = parseFloat(val);
|
||||
});
|
||||
});
|
||||
|
||||
applyMinMax(mins, 'env_temp_min_c');
|
||||
applyMinMax(maxs, 'env_temp_max_c');
|
||||
}
|
||||
|
||||
return Object.values(results).sort((a, b) => a.time.localeCompare(b.time));
|
||||
return Object.values(byTime).sort((a, b) => a.time.localeCompare(b.time));
|
||||
}
|
||||
|
||||
export async function getCoords() {
|
||||
const {c} = getClient();
|
||||
const flux = `
|
||||
from(bucket: "${c.INFLUX_BUCKET}")
|
||||
|> range(start: -24h)
|
||||
|> filter(fn: (r) => r._measurement == "gps")
|
||||
|> last()
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
`;
|
||||
const rows = await query(flux);
|
||||
if(rows.length && rows[0].latitude && rows[0].longitude) {
|
||||
return {latitude: rows[0].latitude, longitude: rows[0].longitude, altitude: rows[0].altitude || c.ALTITUDE};
|
||||
const c = cfg();
|
||||
const results = await queryInstant(c, `{measurement="gps"}`);
|
||||
const fields = metricToFields(results);
|
||||
|
||||
if (fields.latitude && fields.longitude) {
|
||||
return {latitude: fields.latitude, longitude: fields.longitude, altitude: fields.altitude || c.ALTITUDE};
|
||||
}
|
||||
return {latitude: c.LATITUDE, longitude: c.LONGITUDE, altitude: c.ALTITUDE};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user