sensorapi/databases/mongo.js

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

export const properties_collection = 'pptest'

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

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, properties: null}
        let client = await connectMongo()
    try {
        if ("sid" in query) {                               // if sid is given, read property for sid
            ret.properties = await client.db(MONGOBASE).collection(properties_collection).findOne({_id: query.sid})
        } else {                                            // otherwise read props corresponding to query
            ret.properties = await client.db(MONGOBASE).collection(properties_collection).find(query).limit(limit).toArray()
        }
    } catch (e) {
        ret.err = e
    }
    finally {
        client.close()
    }
    return ret
}

// read mapdata from database
export const readMapdata = async (query, limit) => {
    let ret = {err: null, mapdata: []}
    let client = await connectMongo()
    try {
        ret.mapdata = await client.db(MONGOBASE).collection("mapdata").find(query).limit(limit).toArray()
    } catch (e) {
        ret.err = e
    }
    finally {
        client.close()
    }
    return ret
}


export const getallProperties = async (coll, query) => {
    let ret = {err: null, properties: []}
    let client = await connectMongo()
    try {
        ret.properties = await client.db(MONGOBASE).collection(coll)
            .find(query).toArray()
    } catch (e) {
        ret.err = e
    }
    finally {
        client.close()
    }
    return ret
}


export const getOneproperty = async (sid) => {
    let ret = {error: false}
    let client = await connectMongo()
    try {
        ret.properties = await client.db(MONGOBASE).collection("properties_collection")
            .findOne({_id: sid})
    } catch (e) {
        ret =  {error: true, errortext: e}
    }
    finally {
        client.close()
    }
    return ret
}


export const readAKWs = async (options) => {
    let ret = {values: { akws: [], th1_akws: []}, err: null}
    let erg = []
    let 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 {
        client.close()
    }
    return ret
}

export const fetchActData = async (opts) => {
    let ret = {err: null, values: []}
    let start = opts.start.slice(7)
    let end = opts.stop.slice(6)
    start = DateTime.fromISO(start).toJSDate()
    end = DateTime.fromISO(end).toJSDate()
    let query = {sensorid: opts.sensorid, datetime: {$gte: start, $lt: end}}
    let options = { projection: {_id: 0, values: 1, datetime: 1}, sort: {datetime: 1}}
    let client = await connectMongo()
    try {
        ret.values = await client.db(MONGOBASE).collection('sensors')
            .find(query, options).toArray()
    }
    catch(e) {
        ret.err = e
    }
    finally {
        client.close()
    }
    return ret
}
/*
    let docs = await collection.find(
        { datetime:
            { $gte: start.toDate(),  $lt: end.toDate() }
        },
        { projection:
            {_id:0, E_eq:0, E_mx:0, E_mi:0, E10tel_mx:0, E10tel_mi:0}, sort: {datetime: sort}
        },
    ).toArray();
*/


export const fetchActDataxx = async (opts) => {
    let ret = {err: null, values: []}
    let sorting = ''
    if(opts.sort) {
        if (opts.sort === 1) {
            sorting = '|> sort(columns: ["_time"], desc: false)'
        } else if (opts.sort === -1) {
            sorting = '|> sort(columns: ["_time"], desc: true)'
        }
    }
    // build the flux query
    let query = `
    from(bucket: "sensor_data")     
|> range(${opts.start}, ${opts.stop})  
|> filter(fn: (r) => r.sid == "${opts.sensorid}")  
${sorting}
|> keep(columns: ["_time","_field","_value"]) 
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
`
    return await fetchFromInflux(ret, query)
}