admin
bd5cf9978d
mongo.js, influx_post.js und logit.js lagen in beiden Komponenten doppelt
und waren auseinandergelaufen - so ist der Zeitzonen-Fehler entstanden, den
readin seit jeher richtig loest und readFromcsv.js zwei Jahre lang nicht.
Sie liegen jetzt einmal unter common/.
Wo die Fassungen sich widersprachen:
- MONGOAUTH wird als String verglichen. Die readarchive-Fassung pruefte nur
auf truthy, dadurch schaltete auch MONGOAUTH=false die Auth ein.
- writeDataArray(client, coll, data) nimmt den Collection-Namen direkt;
readarchive baut ihn mit dem neuen dataCollName(styp).
- getallProperties liefert das Array selbst. Die readarchive-Fassung mit
{error, properties} war dort ungenutzt.
- properties_collection heisst einheitlich property_coll.
Das Sammelobjekt statistics lag in readin/readdata.js und wurde von
mongo.js und influx_post.js importiert. Es liegt jetzt in
common/statistics.js, damit die gemeinsamen Module readin nicht kennen
muessen.
Der Build-Kontext beider Images ist dadurch das Repository-Wurzelverzeichnis;
das Layout im Image spiegelt das Repository, damit ../common/... unveraendert
aufgeht. Die Volume-Zeile von readin im Compose zieht deshalb auf
/opt/app/readin/data um.
readarchive folgt jetzt den Konventionen des Stacks: deploy.sh statt
build_and_copy.sh, Dockerfile_readarchive statt Dockerfile_rfcsv, und die
eigene docker-compose.yml entfaellt - der Dienst haengt als Profil "tools"
im Compose des Stacks und wird mit Parametern gestartet:
docker compose run --rm readarchive -t noise -s 2026-07-25 -e 2026-07-29
Getestet: beide Images gebaut, beide gegen Testdatenbanken laufen lassen.
Nicht deployt - die Images in der Registry sind unveraendert.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
540 lines
19 KiB
JavaScript
540 lines
19 KiB
JavaScript
// readFromcsv - Alte Daten vom Luftsdaten per CSV einlesen
|
|
// rxf 2017-12-12
|
|
//
|
|
// Ausgehend von Version 2.0.0 werden hier nun die Daten nach den Sensortypen getrennt gespeichert
|
|
// und zwar einstellbar in einer Mongo-DB oder in einer Influx-DB (oder auch in beide).
|
|
|
|
// Version:
|
|
//
|
|
// V 3.1.1 2023-11-14 rxf
|
|
// - Enddatum eingeführt
|
|
//
|
|
// V 3.1.0 2023-11-01 rxf
|
|
// - Anpassung, so dass es als Docker-Container mit dem timeseries-Container läuft
|
|
//
|
|
// V 3.0.1 2023-10-18 rxf
|
|
// - mehrer Typen gleichzeitig auswählbar
|
|
|
|
// V 3.0.0 2023-10-07 rxf
|
|
// - Erste Version mit der Trennung nach Typen
|
|
//
|
|
// TYP: Ist der Sensortyp, der eingelesen wird. Default ist 'pm', d.h. alle Feinstaubsensoren. Mögliche Werte sind
|
|
// ['radioactivity', 'noise']. Die jeweils dazugehörenden THP-Sensoren werden mit eingelesen.
|
|
const TYP = process.env.TYP || 'noise'
|
|
const START = process.env.START
|
|
const END = process.env.END
|
|
const DBASE = process.env.DBASE || 'both'
|
|
|
|
const DATABASE = ['mongo', 'influx', 'both']
|
|
|
|
const DEVELOP = process.env.DEVELOP || false;
|
|
const LIVE = (process.env.LIVE == "true") || true
|
|
|
|
const MAXTRIES = 10
|
|
|
|
// MUSS der erste Import bleiben: mongo.js und influx_post.js lesen ihre
|
|
// Konfiguration beim Laden aus process.env - .env muss vorher drin sein
|
|
import 'dotenv/config'
|
|
import axios from 'axios'
|
|
import https from 'https'
|
|
import * as mongo from '../common/mongo.js'
|
|
import * as influx from '../common/influx_post.js'
|
|
import fs from 'fs'
|
|
import csv from 'csvtojson'
|
|
import pkg from './package.json' with { type: "json" }
|
|
import mod_getopt from 'posix-getopt'
|
|
import { logit, logerror } from '../common/logit.js'
|
|
import { DateTime, Duration } from 'luxon'
|
|
import { checkProperties} from "./utilities/checkprops.js";
|
|
import { version } from "os"
|
|
|
|
const DEFAULTPORT = 3008;
|
|
const PORT = process.env.SERVERPORT || DEFAULTPORT; // Port for server
|
|
|
|
const API_URL = 'https://archive.sensor.community/'; // URL to API on 'luftdaten.info'
|
|
|
|
const valueTypes = ["P1","P2","P0","P4","N0","N1","N4","N25","N10","TS", "co2_ppm", "temperature","humidity","pressure","pressure_at_sealevel","noise_LA_min","noise_LA_max","noise_LAeq","counts_per_minute","hv_pulses","counts","sample_time_ms"]
|
|
|
|
const newProps = []
|
|
|
|
// Sensor/Tag-Paare, deren Influx-Write fehlgeschlagen ist - Auswertung am Ende von main()
|
|
const failedInfluxWrites = []
|
|
|
|
const saveList = (list) => {
|
|
fs.writeFileSync('data/list.json', JSON.stringify(list))
|
|
}
|
|
|
|
const getListFromDisk = () => {
|
|
let buffer = fs.readFileSync('data/list.json')
|
|
return JSON.parse(buffer)
|
|
}
|
|
|
|
const types = {
|
|
P1: 'pm', P2: 'pm', P0: 'pm',
|
|
temperature: 'thp', humidity: 'thp', pressure: 'thp',
|
|
noise_LAeq: 'noise', noise_LA_max: 'noise', noise_LA_min: 'noise',
|
|
counts_per_minute: 'radioactivity',
|
|
lat: 'gps'
|
|
};
|
|
const sensorTypes = {
|
|
"bme280": "thp",
|
|
"bmp180": "thp",
|
|
"bmp280": "thp",
|
|
"dht22": "thp",
|
|
"ds18b20": "thp",
|
|
"hpm": "pm",
|
|
"htu21d": "thp",
|
|
"laerm": "noise",
|
|
"nextpm": "pm",
|
|
"pms1003": "pm",
|
|
"pms3003": "pm",
|
|
"pms5003": "pm",
|
|
"pms6003": "pm",
|
|
"pms7003": "pm",
|
|
"ppd42ns": "pm",
|
|
"radiation": "radioactivity",
|
|
"scd30": "co2",
|
|
"sds011": "pm",
|
|
"sht11": "thp",
|
|
"sht15": "thp",
|
|
"sht30": "thp",
|
|
"sht31": "thp",
|
|
"sht35": "thp",
|
|
"sht85": "thp",
|
|
"sps30": "pm",
|
|
}
|
|
|
|
const getSensorType = (item) => {
|
|
let p = item.split('_')
|
|
let t = sensorTypes[p[1]]
|
|
if (t === undefined) {
|
|
console.log("Type not known: " + p[1])
|
|
t = 'unknown'
|
|
}
|
|
return t
|
|
}
|
|
|
|
const checkType = (styp, args) => {
|
|
if ((args.sensorType === styp) || (styp === 'thp')) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
function getType(typ) {
|
|
if(typ in types) {
|
|
return types[typ];
|
|
} else {
|
|
return 'unknown'
|
|
}
|
|
}
|
|
|
|
// get the sensorids of all THP sensors belonging to the sensors of the give type
|
|
const getTHPList = async (client, stype) => {
|
|
let locids = await mongo.getLocationIDs(client, stype) // { sensorid: 12345, locid: 34567}
|
|
let thplist = []
|
|
for (let lid of locids.locations) {
|
|
let x = await mongo.getTHPSensors(client, lid)
|
|
if (x.sensors) {
|
|
thplist.push(x.sensors._id)
|
|
}
|
|
}
|
|
return thplist
|
|
}
|
|
|
|
// get list of all saved sensors for this day
|
|
const getdirlistOfOneDay = async (day) => {
|
|
let tries = 5
|
|
let ret = { err: false, list : []}
|
|
|
|
while(true) {
|
|
try {
|
|
const erg = await axios.get(API_URL + day);
|
|
// logit(`getdirListOfOneDay: Status = ${erg.status}`);
|
|
let a = erg.data.split('"'); // parse the list
|
|
for (let i = 0; i < a.length; i++) {
|
|
if (a[i].startsWith(day.substr(0, 4))) { // extract the sensor names
|
|
ret.list.push(a[i]);
|
|
}
|
|
}
|
|
logit(`Read ${ret.list.length} entries for ${day}`)
|
|
break
|
|
} catch (e) {
|
|
if(tries-- === 0) {
|
|
ret.err = e
|
|
return ret
|
|
}
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
const parseDayList = (list, thplist, args) => {
|
|
let sl = []
|
|
for (let elem of list) {
|
|
let oneEntry = {url: elem, indoor: false}
|
|
let name = elem
|
|
if (elem.indexOf('indoor') != -1) { // there is 'indoor'
|
|
name = elem.replace('_indoor', '');
|
|
oneEntry.indoor = true;
|
|
}
|
|
let n = name.lastIndexOf('_');
|
|
oneEntry.sensorType = getSensorType(elem)
|
|
oneEntry.sensorid = parseInt(name.substr(n + 1).replace('.csv', ''))
|
|
if(!((oneEntry.sensorType === args.sensorType) || (oneEntry.sensorType === 'thp'))){
|
|
continue
|
|
}
|
|
if ((oneEntry.sensorType === 'thp') && (thplist.indexOf(oneEntry.sensorid) === -1)) {
|
|
continue
|
|
}
|
|
sl.push(oneEntry)
|
|
}
|
|
return sl
|
|
}
|
|
|
|
|
|
// do the whole work
|
|
async function readSensorsperDay(client, args) {
|
|
let list
|
|
let insertedCount = 0
|
|
let sensorList = []
|
|
let st = DateTime.fromFormat(args.startDate, 'yyyy-LL-dd') // startdate
|
|
let end = DateTime.fromFormat(args.endDate, 'yyyy-LL-dd')
|
|
let thplist = await getTHPList(client, args.sensorType) // find all THP sensors belonging to the sensors of the given type
|
|
for (let d = st; d < end; d = d.plus({days: 1})) { // loop thru days
|
|
let dstr = d.toFormat('yyyy-LL-dd')
|
|
logit(`*************** ${dstr}`) // log every day
|
|
let mist = false;
|
|
// fetch sensors list of current day
|
|
if(LIVE) {
|
|
let erg = await getdirlistOfOneDay(dstr)
|
|
if (erg.err) {
|
|
logerror(erg.err)
|
|
mist = true
|
|
} else {
|
|
list = erg.list
|
|
};
|
|
} else {
|
|
list = getListFromDisk()
|
|
logit(`Read ${list.length} entries from Disk for ${dstr}`)
|
|
}
|
|
if (mist) continue; // if day doesn't exist, continue
|
|
if (LIVE) {
|
|
saveList(list)
|
|
}
|
|
sensorList = parseDayList(list, thplist, args)
|
|
insertedCount += await enterSensors(client, sensorList, dstr, args); // fetch and enter sensor data
|
|
}
|
|
return insertedCount
|
|
}
|
|
|
|
const scanOneSensor = async (client, sensorList, dt, args) => {
|
|
let inserted = 0
|
|
let missed = []
|
|
let listcnt = sensorList.length
|
|
for (let item of sensorList) { // iterate the list
|
|
let sid = item.sensorid
|
|
if(args.sensornbr !== undefined) {
|
|
if(args.sensornbr !== sid) {
|
|
continue
|
|
}
|
|
}
|
|
logit(`${item.url}`)
|
|
let icount = await putOneSensorInDb(client, item.url, sid, dt, missed, item.indoor, item.sensorType, args) // put one sensor data into DB
|
|
logit(`${listcnt} -- Sensor ${sid}: ${icount} Einträge`)
|
|
inserted += icount
|
|
listcnt--
|
|
if (args.nbrOfEntries !== undefined) {
|
|
if (--args.nbrOfEntries === 0) {
|
|
break
|
|
}
|
|
|
|
}
|
|
}
|
|
return {inserted: inserted, missed: missed}
|
|
}
|
|
|
|
// Iterate thru the list and enter every sensor that matches data into db
|
|
async function enterSensors(client, sensorList, dt, args) {
|
|
let inserted = 0
|
|
// check which types to scan
|
|
while (sensorList.length > 0) {
|
|
let missed = []
|
|
let erg = await scanOneSensor(client, sensorList, dt, args)
|
|
inserted += erg.inserted
|
|
if(erg.missed.length > 0) {
|
|
logit(`Error reading ${erg.missed.length} missed files `)
|
|
}
|
|
logit(`enterSensors: ${inserted} entries handled`)
|
|
sensorList = erg.missed
|
|
}
|
|
return inserted
|
|
}
|
|
|
|
// read CSV file and enter data
|
|
async function putOneSensorInDb(client, name, sid, dt, missed, indoor, styp, args) {
|
|
let z = 0
|
|
try {
|
|
let [all, dataline] = await readOneSensorOneDay(client, name, sid, dt, missed, indoor, args);
|
|
const erg = await enterOneSensorinDB(client, sid, all, dataline, dt, styp, args)
|
|
z = erg === undefined ? 0 : erg
|
|
}
|
|
catch(err) {
|
|
console.log("Error in putOneSensorInDB()");
|
|
}
|
|
return z
|
|
}
|
|
|
|
// read the CSV-File and parse it into right format for DB
|
|
async function readOneSensorOneDay(client, name, sid, dt, missed, indoor, args) {
|
|
let all = [];
|
|
let url = API_URL + dt + '/' + name; // construct URL
|
|
const sname = name.split('_')[1]
|
|
let firstItem
|
|
let typ = 'unknown'
|
|
let dataline = ''
|
|
|
|
try {
|
|
const erg = await axios(url);
|
|
if ((erg.status != 200) || (erg.data == "")) {
|
|
return {error: {status: erg.status, data: erg.data}};
|
|
}
|
|
|
|
const data = await csv({delimiter: ';'}).fromString(erg.data);
|
|
firstItem = data[0]
|
|
for (let item of data) {
|
|
let entry = {values: {}, sensorid: sid}
|
|
// Die Zeitstempel im CSV-Archiv sind UTC, haben aber keine Zonenangabe.
|
|
// Ohne zone:'utc' wuerde luxon sie in der Zeitzone der Maschine lesen.
|
|
let date = DateTime.fromISO(item.timestamp, { zone: 'utc' }) // extract date of entry
|
|
entry.datetime = date.toJSDate() // make date for Mongo (== ISODate)
|
|
for (let val of valueTypes) {
|
|
if (item[val] !== undefined) {
|
|
let x
|
|
let v = item[val]
|
|
try {
|
|
x = parseFloat(v) // convert value to float
|
|
if (Number.isNaN(x)) {
|
|
x = -9999.9 // default if value is invalid or unknown
|
|
}
|
|
} catch (err) {
|
|
console.log('Math parse float error on value');
|
|
x = -9999.9 // default if value is invalid or unknown
|
|
}
|
|
if (typ === 'unknown') { // extract measurement type
|
|
typ = getType(val)
|
|
}
|
|
if(typ === 'noise') {
|
|
val = val.slice(6)
|
|
}
|
|
entry.values[val] = x
|
|
if (val == "LAeq") {
|
|
entry.values.E10tel_eq = Math.pow(10, entry.values.LAeq / 10);
|
|
}
|
|
}
|
|
}
|
|
all.push(entry)
|
|
if ((args.database === 'influx') || (args.database === 'both')) {
|
|
let influxvalue = ''
|
|
for (const [key, value] of Object.entries(entry.values)) {
|
|
influxvalue = influxvalue + `${key}=${value},`
|
|
}
|
|
influxvalue = influxvalue.slice(0,-1)
|
|
dataline += `${typ},sid=${entry.sensorid} ${influxvalue} ${entry.datetime.valueOf()}\n`
|
|
}
|
|
}
|
|
checkProperties(client, firstItem, indoor, sname, typ, dt, newProps)
|
|
} catch (e) {
|
|
console.log(e)
|
|
missed.push(name)
|
|
}
|
|
return [all, dataline] // return all the data
|
|
}
|
|
|
|
// Neue Einträge aus dem CSV mit denen in der DB vergleichn, identische aus dem CSV raus streichen
|
|
const condenseEntry = async (csvData, dbData ) => {
|
|
for( let i = csvData.length - 1; i >= 0; i--) {
|
|
const cvt = csvData[i].datetime.valueOf()
|
|
if( dbData.findIndex( (item) => item.datetime.valueOf() == cvt) !== -1) {
|
|
csvData.splice(i,1)
|
|
}
|
|
}
|
|
return csvData
|
|
}
|
|
|
|
// enter all data for one sensor into DB
|
|
async function enterOneSensorinDB(client, sid, erg, dataline, day, styp, args) {
|
|
const dataOneDay = await mongo.getOneSensorOneday(client, sid, day, styp)
|
|
if (dataOneDay.data.length > 0) {
|
|
await condenseEntry(erg, dataOneDay.data)
|
|
}
|
|
if (((args.database === 'both') || (args.database === 'influx')) && (dataline !== '')) {
|
|
const ok = await influx.influxWrite(dataline)
|
|
if (!ok) {
|
|
failedInfluxWrites.push({sensorid: sid, day: day})
|
|
}
|
|
}
|
|
if((args.database === 'both') || (args.database === 'mongo')) {
|
|
if(erg.length !== 0) {
|
|
await mongo.writeDataArray(client, mongo.dataCollName(styp), erg)
|
|
}
|
|
}
|
|
return erg.length
|
|
}
|
|
|
|
/*
|
|
let std = moment.utc(dt).startOf('day');
|
|
let endd = moment.utc(dt).startOf('day').add(1,'day');
|
|
let docs = await coll.find({datetime: {$gte: new Date(std), $lt: new Date(endd)}}, {sort: {datetime: 1}}).toArray();
|
|
for (let i = docs.length - 1; i >= 0; i--) {
|
|
let dt = docs[i].datetime.valueOf();
|
|
for (let a = all.length - 1; a >= 0; a--) {
|
|
let at = all[a].datetime.valueOf();
|
|
if (dt == at) {
|
|
all.splice(a, 1);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
*/
|
|
/*
|
|
for(let item of erg) {
|
|
// Einlesen der Einträge für diesen Sensor und diesen Tag
|
|
// {sensorid: "11999", datetime: {$gte: ISODate("2023-04-07"),$lt: ISODate("2023-04-08")}}
|
|
// check, ob es den Eintrag schon gibt
|
|
const ret = await mongo.checkOneSensor(client, sid, item.datetime)
|
|
if(ret.err) {
|
|
logit(`Èrror checking sensor ${sid} for ${item.datetime}\n${ret.errortext}`)
|
|
return
|
|
}
|
|
if (!ret.exists) {
|
|
const r = await mongo.writeOneSensor(client,item)
|
|
count += r.inserted
|
|
if(r.err) {
|
|
logit(`Error writing sensor ${sid} for ${item.datetime}\n${r.errortext}`)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
*/
|
|
|
|
// Parse command line options
|
|
function parse_cmdline(argv) {
|
|
let parser = new mod_getopt.BasicParser('s:(start)e:(end)t:(type)h(help)v(version)a:(entries)n:(sensorid)d:(dbase)',argv);
|
|
let option;
|
|
let std = START ? START : DateTime.now().startOf('day').minus({days: 1}).toFormat('yyyy-LL-dd'); // yesterday
|
|
let ret = {startDate: std, endDate: '', sensorType: TYP, database: DBASE};
|
|
|
|
while((option = parser.getopt()) !== undefined) {
|
|
switch(option.option) {
|
|
case 's':
|
|
ret.startDate = option.optarg.trim()
|
|
break;
|
|
|
|
case 'e':
|
|
ret.endDate = option.optarg.trim()
|
|
break;
|
|
|
|
case 'a':
|
|
ret.nbrOfEntries = parseInt(option.optarg.trim())
|
|
break;
|
|
|
|
case 'n':
|
|
ret.sensornbr = parseInt(option.optarg.trim())
|
|
break;
|
|
|
|
case 't':
|
|
let x = option.optarg.trim()
|
|
if (( x === 'noise') || ( x === 'radioactivity') ||
|
|
(x === 'pm') || ( x === 'thp')) {
|
|
ret.sensorType = x
|
|
}
|
|
break;
|
|
|
|
case 'd':
|
|
let db = option.optarg.trim()
|
|
if (DATABASE.indexOf(db) !== -1) {
|
|
ret.database = db
|
|
}
|
|
break;
|
|
|
|
case 'v':
|
|
console.log(`Version: ${pkg.version} from ${pkg.date}`);
|
|
console.log();
|
|
process.exit();
|
|
break;
|
|
|
|
case 'h':
|
|
console.log("Usage: node readFromcvs.js [-h] [-s startDate] [-e endDate] [-t sensorType] [-v version] [-h help]")
|
|
if (DEVELOP) {
|
|
console.log(" [-a entries] [-n sensorid]");
|
|
}
|
|
console.log("Params:");
|
|
console.log(" -s startDate: date to begin calculation (ex: 2023-10-23); default: yesterday");
|
|
console.log(" -e endDate: date to stop calculatien (exclusive !). If not given, only the startDate")
|
|
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(" -v version: show version");
|
|
console.log(" -h this help text");
|
|
if(DEVELOP) {
|
|
console.log(" -a entries: only work on 'entries' entries (only for debug)")
|
|
console.log(" -n check only this sensor number (only for debug)")
|
|
}
|
|
console.log("All parameters are optional.");
|
|
console.log();
|
|
process.exit();
|
|
break;
|
|
|
|
default:
|
|
break;
|
|
}
|
|
}
|
|
if (ret.endDate === '') {
|
|
if (END) {
|
|
ret.endDate = END
|
|
} else {
|
|
let ed = DateTime.fromISO(ret.startDate)
|
|
ret.endDate = ed.plus({days: 1}).toFormat('yyyy-LL-dd')
|
|
}
|
|
}
|
|
logit(JSON.stringify(ret));
|
|
return ret;
|
|
}
|
|
|
|
async function main() {
|
|
logit(`Programm V: ${pkg.version} from ${pkg.date} - Start`)
|
|
let starttime = DateTime.now()
|
|
let args = parse_cmdline(process.argv)
|
|
const client = await mongo.connectMongo()
|
|
let inserted = await readSensorsperDay(client, args)
|
|
logit(`Inserted: ${inserted}`)
|
|
// enter changed properties into database
|
|
if(newProps.length > 0) {
|
|
await mongo.bulkWrite(client, mongo.property_coll, newProps)
|
|
}
|
|
await mongo.closeMongo(client)
|
|
if (failedInfluxWrites.length > 0) {
|
|
logerror(`${failedInfluxWrites.length} Influx-Writes fehlgeschlagen - diese Daten fehlen in Influx:`)
|
|
for (const f of failedInfluxWrites) {
|
|
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');
|
|
logit(`Dauer: ${output1} Sekunden`)
|
|
logit("Programm - Ende")
|
|
}
|
|
|
|
main().catch((e) => {
|
|
console.error(e)
|
|
process.exitCode = 1
|
|
})
|
|
|
|
|