first

2023-03-08 16:17:02 +01:00
commit 0b6959d4f6
35 changed files with 5804 additions and 0 deletions
@@ -0,0 +1,71 @@
+// Access to influxDB vie HTTP
+
+import axios from 'axios'
+import { DateTime } from 'luxon'
+// import csvParse from 'csv-parser'
+import { logit, logerror } from '../utilities/logit.js'
+import {returnOnError} from "../utilities/reporterror.js";
+
+let INFLUXHOST = process.env.INFLUXHOST || "localhost"
+let INFLUXPORT = process.env.INFLUXPORT || 8086
+let INFLUXTOKEN = process.env.INFLUXTOKEN ||
+    //"rklEClT22KfdXZhA47eyJhbqcvekb8bcKCqlUG7n72uDSmR2xGvif0CmGJe0WQtXB96y29mmt-9BdsgWA5npfg=="
+    "BNR6cGdb006O1T6hQkGcfB8tgH-UPO6QkOPToeAvrP7LATJbCuWi1wYf3HBpVdZQEBxHxNSrNenZsOSMogX-lg=="
+let INFLUXDATABUCKET = process.env.INFLUXDATABUCKET || "sensor_data"
+let INFLUXORG = process.env.INFLUXORG || "citysensor"
+
+const INFLUXURL_READ = `http://${INFLUXHOST}:${INFLUXPORT}/api/v2/query?org=${INFLUXORG}`
+const INFLUXURL_WRITE = `http://${INFLUXHOST}:${INFLUXPORT}/api/v2/write?org=${INFLUXORG}&bucket=${INFLUXDATABUCKET}`
+
+export const influxRead = async (query) => {
+    let start = DateTime.now()
+    let erg = { values: [], err: null}
+    try {
+        let ret = await axios({
+            method: 'post',
+            url: INFLUXURL_READ,
+            data: query,
+            headers: {
+                Authorization: `Token ${INFLUXTOKEN}`,
+                Accept: 'application/csv',
+                'Content-type': 'application/vnd.flux'
+            },
+            timeout: 10000,
+        })
+        if (ret.status !== 200) {
+            let error = ERR.RESPSTATUS.replace('xx',ret.status)
+            return returnOnError(erg,  error, influxRead.name)
+        }
+        erg.values = ret.data
+    } catch (e) {
+        return returnOnError(erg,  e, influxRead.name)
+    }
+    logit(`Influx read time: ${start.diffNow('seconds').toObject().seconds * -1} sec`)
+    return erg
+}
+
+
+export const influxWrite = async (data) => {
+    let start = DateTime.now()
+    let ret
+    try {
+        ret = await axios({
+            method: 'post',
+            url: INFLUXURL_WRITE,
+            data: data,
+            headers: {
+                Authorization: `Token ${INFLUXTOKEN}`,
+                Accept: 'application/json',
+                'Content-Type': 'text/plain; charset=utf-8'
+            },
+            timeout: 10000,
+        })
+        if (ret.status !== 204) {
+            logerror(`doWrite2API Status: ${ret.status}`)
+        }
+    } catch (e) {
+        logerror(`doWrite2API  ${e}`)
+    }
+    logit(`Influx-Write-Time: ${start.diffNow('seconds').toObject().seconds * -1} sec`)
+    return ret
+}
@@ -0,0 +1,252 @@
+/* Interface for MongoDB
+*/
+import  { MongoClient } from  'mongodb'
+import { logit, logerror } from '../utilities/logit.js'
+import { DateTime } from 'luxon'
+import {returnOnError} from "../utilities/reporterror.js";
+
+// const nodemailer = require('nodemailer');
+
+let MONGOHOST = process.env.MONGOHOST;
+let MONGOPORT = process.env.MONGOPORT;
+let MONGOAUTH = process.env.MONGOAUTH;
+let MONGOUSRP = process.env.MONGOUSRP;
+let MONGOBASE =  process.env.MONGOBASE;
+
+if (MONGOHOST === undefined) { MONGOHOST = 'localhost';}
+if (MONGOPORT === undefined) { MONGOPORT =  27097; }
+if (MONGOAUTH === undefined) { MONGOAUTH =  'false'; }
+if (MONGOBASE === undefined) { MONGOBASE =  'allsensors'; }
+
+let MONGO_URL = 'mongodb://'+MONGOHOST+':'+MONGOPORT;  	// URL to mongo database
+if (MONGOAUTH === 'true') {
+    MONGO_URL = 'mongodb://'+MONGOUSRP+'@' + MONGOHOST + ':' + MONGOPORT + '/?authSource=' + MONGOBASE;          // URL to mongo database
+}
+
+let statistics = {}
+let client
+
+export const connectMongo = async () => {
+    try {
+        logit(`Try to connect to ${MONGO_URL}`)
+//        let client = await MongoClient.connect(MONGO_URL, { useNewUrlParser: true , useUnifiedTopology: true })
+        let client = await MongoClient.connect(MONGO_URL)
+        logit(`Mongodbase connected to ${MONGO_URL}`)
+        return client
+    }
+    catch(error){
+        throw(error)
+    }
+}
+
+export const closeMongo = async (client) => {
+    client.close()
+}
+
+const listDatabases = async (client) => {
+    let databasesList = await client.db().admin().listDatabases();
+
+    console.log("Databases:");
+    databasesList.databases.forEach(db => console.log(` - ${db.name}`));
+}
+
+/* ***************************************************
+// READ routines
+******************************************************/
+
+// Read properties from the database
+export const readProperties = async (query, limit = 0) => {
+    let ret = {err: null, values: null}
+        client = await connectMongo()
+    try {
+        if ("sid" in query) {                               // if sid is given, read property for sid
+            ret.values = await client.db(MONGOBASE).collection("properties").findOne({_id: query.sid})
+        } else {                                            // otherwise read props corresponding to query
+            ret.values = await client.db(MONGOBASE).collection("properties").find(query).limit(limit).toArray()
+        }
+    } catch (e) {
+        ret.err = e
+    }
+    finally {
+            await closeMongo(client)
+            client = null
+    }
+    return ret
+}
+
+// read mapdata from database
+export const readMapdata = async (query, limit) => {
+    let ret = {err: null, values: []}
+    client = await connectMongo()
+    try {
+            ret.values = await client.db(MONGOBASE).collection("mapdata").find(query).limit(limit).toArray()
+    } catch (e) {
+        ret.err = e
+    }
+    finally {
+        await closeMongo(client)
+        client = null
+    }
+    return ret
+}
+
+
+export const getallProperties = async (client) => {
+    return await client.db(MONGOBASE).collection("properties")
+        .find().toArray()
+
+}
+
+export const getOneproperty = async (sid) => {
+    let ret = {error: false}
+    client = await connectMongo()
+    try {
+        ret.properties = await client.db(MONGOBASE).collection("properties")
+            .findOne({_id: sid})
+    } catch (e) {
+        ret =  {error: true, errortext: e}
+    }
+    finally {
+        await closeMongo(client)
+        client = null
+    }
+    return ret
+}
+
+
+export const doAggregate = async (coll, query, conn, clos) => {
+    let ret = {error: false}
+    if (conn) {
+        client = await connectMongo()
+    }
+    try {
+        ret.values = await client.db(MONGOBASE).collection(coll).aggregate(query).toArray()
+    } catch (e) {
+        ret =  {error: true, errortext: e}
+    }
+    finally {
+        if (clos) {
+            await closeMongo(client)
+            client = null
+        }
+    }
+    return ret
+}
+
+
+export const writeOneproperty = async (client, prop) => {
+    try {
+        let result = await client.db(MONGOBASE).collection("properties")
+            .insertOne(prop)
+    } catch(e) {
+        if (e.code === 11000) {
+            return false
+        } else {
+            throw(e)
+        }
+    }
+    return true
+}
+
+export const writeProperties = async (client, props) => {
+    let start = DateTime.now();
+    if (props.new.length !== 0) {
+        try {
+            let result = await client.db(MONGOBASE).collection("properties")
+                .insertMany(props.new)
+        } catch (e) {
+            console.error(e)
+        }
+    }
+    if (props.loc.length !== 0) {
+        try {
+            for (let item of props.loc) {
+                let result = await client.db(MONGOBASE).collection("properties")
+                    .updateOne({_id: item._id}, {
+                        $set: {location_id: item.location_id},
+                        $push: {location: { $each: [item.location[0]], $position: 0}}
+                    })
+            }
+        } catch (e) {
+            console.error(e)
+        }
+    }
+    if (props.sname.lenth !== 0) {
+        try {
+            for (let item of props.sname) {
+                let result = await client.db(MONGOBASE).collection("properties")
+                    .updateOne({_id: item._id}, {
+                        $set: {name: item.name, since: item.since}
+                    })
+            }
+        } catch (e) {
+            console.error(e)
+        }
+    }
+    statistics.writePropsTime = DateTime.now().diff(start, ['seconds']).toObject().seconds
+    logit(`Write properties to mongoDB: ${statistics.writePropsTime} sec.`)
+}
+
+export const writeDataArray = async (data, collection, conn, clos) => {
+    let start = DateTime.now();
+    let result
+    if (conn) {
+        client = await connectMongo()
+    }
+     try {
+        result = await client.db(MONGOBASE).collection(collection)
+            .insertMany(data)
+    } catch (e) {
+        console.error(e)
+    } finally {
+        if(clos) {
+            await closeMongo(client)
+            client = null
+        }
+    }
+    statistics.writeDataTime = DateTime.now().diff(start, ['seconds']).toObject().seconds
+    logit(`Write Data to mongoDB: ${statistics.writeDataTime} sec.`)
+    return result
+}
+
+export const writeStatistic = async (client, stat) => {
+    let start = DateTime.now();
+    let entry = { timestamp: new Date(), ...stat }
+    try {
+        let result = await client.db(MONGOBASE).collection("statistics")
+            .insertOne(entry)
+    } catch (e) {
+        console.error(e)
+    }
+    statistics.writeStatisticTime = DateTime.now().diff(start, ['seconds']).toObject().seconds
+    logit(`Write statistics to mongoDB: ${statistics.writeStatisticTime} sec.`)
+}
+
+export const readAKWs = async (options) => {
+    let ret = {values: { akws: [], th1_akws: []}, err: null}
+    let erg = []
+    client = await connectMongo()
+    try {
+       let docs = await client.db(MONGOBASE).collection("akws")
+            .find().toArray()
+        if(docs == null) {
+            return returnOnError(ret, 'akws - docs == null', readAKWs.name)
+        }
+        logit(`getawkdata: data fetched from akws, length= ${docs.length}`);
+        ret.values.akws = docs
+        let docs1 = await client.db(MONGOBASE).collection("th1_akws")
+            .find().toArray()
+        if(docs1 == null) {
+            return returnOnError(ret, 'th1_akws - docs == null', readAKWs.name)
+        }
+        logit(`getawkdata: data fetched from th1_akws, length= ${docs1.length}`)
+        ret.values.th1_akws = docs1
+    } catch (e) {
+        return returnOnError(ret, e, readAKWs.name)
+    }
+    finally {
+        await closeMongo(client)
+        client = null
+    }
+    return ret
+}