Initial monorepo: readin, sensorapi, noise (Node22, npm ci, axios1.x, i18next-fs-backend)
This commit is contained in:
rxf
@@ -0,0 +1,10 @@
|
||||
node-modules
|
||||
.gitignore
|
||||
.dockerignore
|
||||
build_and_copy.sh
|
||||
docker-compose.yml
|
||||
Dockerfile*
|
||||
INFLUXSERVER
|
||||
TRANSFERE
|
||||
data
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
data/
|
||||
node_modules
|
||||
.idea
|
||||
.env
|
||||
log
|
||||
docs/Readme.md_x
|
||||
docs/Readme_tmp.html
|
||||
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
FROM node:22-alpine
|
||||
|
||||
ADD package.json /tmp/package.json
|
||||
ADD package-lock.json /tmp/package-lock.json
|
||||
RUN cd /tmp && npm ci
|
||||
RUN mkdir -p /opt/app && cp -a /tmp/node_modules /tmp/package.json /opt/app/
|
||||
|
||||
WORKDIR /opt/app
|
||||
ADD . /opt/app
|
||||
ADD crontab.tmp /opt/app
|
||||
|
||||
RUN mkdir -p data
|
||||
|
||||
#RUN apk add busybox-initscripts
|
||||
#RUN apk add --no-cache tzdata
|
||||
#ENV TZ Europe/Berlin
|
||||
#RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
|
||||
|
||||
RUN crontab crontab.tmp
|
||||
RUN deluser --remove-home node
|
||||
|
||||
RUN touch cmds.sh \
|
||||
&& echo 'crond -f' >>cmds.sh
|
||||
|
||||
CMD sh ./cmds.sh
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
FROM node:alpine
|
||||
|
||||
ADD package.json /tmp/package.json
|
||||
RUN cd /tmp && npm install
|
||||
RUN mkdir -p /opt/app && cp -a /tmp/node_modules /opt/app/
|
||||
|
||||
WORKDIR /opt/app
|
||||
ADD . /opt/app
|
||||
ADD crontab.tmp /opt/app
|
||||
|
||||
RUN mkdir -p data
|
||||
|
||||
#RUN apk add busybox-initscripts
|
||||
#RUN apk add --no-cache tzdata
|
||||
#ENV TZ Europe/Berlin
|
||||
#RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
|
||||
|
||||
RUN crontab crontab.tmp
|
||||
RUN deluser --remove-home node
|
||||
|
||||
RUN touch cmds.sh \
|
||||
&& echo 'crond -f' >>cmds.sh
|
||||
|
||||
CMD sh ./cmds.sh
|
||||
|
||||
|
||||
|
||||
Executable
+63
@@ -0,0 +1,63 @@
|
||||
#!/bin/bash
|
||||
# Build Docker-Container
|
||||
#
|
||||
# Call: buildit.sh name [target]
|
||||
#
|
||||
# The Dockerfile must be named like Dockerfile_name
|
||||
#
|
||||
# 2018-09-20 rxf
|
||||
# - before sending docker image to remote, tag actual remote image
|
||||
#
|
||||
# 2018-09-14 rxf
|
||||
# - first Version
|
||||
#
|
||||
|
||||
set -x
|
||||
port=""
|
||||
orgName=readin
|
||||
name=readin
|
||||
|
||||
usage()
|
||||
{
|
||||
echo "Usage build_and_copy.sh [-p port] [-n name] target"
|
||||
echo " Build docker container $name and copy to target"
|
||||
echo "Params:"
|
||||
echo " target: Where to copy the container to "
|
||||
echo " -p port: ssh port (default 22)"
|
||||
echo " -n name: new name for container (default: $orgName)"
|
||||
}
|
||||
|
||||
while getopts n:p:h? o
|
||||
do
|
||||
case "$o" in
|
||||
n) name="$OPTARG";;
|
||||
p) port="-p $OPTARG";;
|
||||
h) usage; exit 0;;
|
||||
*) usage; exit 1;;
|
||||
esac
|
||||
done
|
||||
shift $((OPTIND-1))
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
if [[ -z "$target" ]]; then
|
||||
target=$1
|
||||
shift
|
||||
else
|
||||
echo "bad option $1"
|
||||
# exit 1
|
||||
shift
|
||||
fi
|
||||
done
|
||||
|
||||
docker build -f Dockerfile_$orgName -t $name .
|
||||
|
||||
dat=`date +%Y%m%d%H%M`
|
||||
|
||||
if [ "$target" == "localhost" ]
|
||||
then
|
||||
docker tag $name $name:V_$dat
|
||||
exit
|
||||
fi
|
||||
|
||||
ssh $port $target "docker tag $name $name:V_$dat"
|
||||
docker save $name | bzip2 | pv | ssh $port $target 'bunzip2 | docker load'
|
||||
@@ -0,0 +1,2 @@
|
||||
3-59/5 * * * * cd /opt/app && npm start
|
||||
|
||||
Executable
+58
@@ -0,0 +1,58 @@
|
||||
#!/bin/bash
|
||||
|
||||
set -e
|
||||
|
||||
# Angepasst aus /Projekte/Logbuch/deploy.sh
|
||||
# Unterschiede: nur linux/amd64, nativer Build auf esprimo (amd64) -> kein buildx/qemu,
|
||||
# eigener Dockerfile-Name (Dockerfile_readin).
|
||||
|
||||
REGISTRY="docker.citysensor.de"
|
||||
IMAGE_NAME="readin"
|
||||
TAG="${1:-latest}"
|
||||
PLATFORM="linux/amd64"
|
||||
DOCKERFILE="Dockerfile_readin"
|
||||
FULL_IMAGE="${REGISTRY}/${IMAGE_NAME}:${TAG}"
|
||||
BUILD_DATE=$(date +%d.%m.%Y)
|
||||
|
||||
echo "=========================================="
|
||||
echo "readin Deploy Script (amd64)"
|
||||
echo "=========================================="
|
||||
echo "Registry: ${REGISTRY}"
|
||||
echo "Image: ${IMAGE_NAME}"
|
||||
echo "Tag: ${TAG}"
|
||||
echo "Platform: ${PLATFORM}"
|
||||
echo "Dockerfile: ${DOCKERFILE}"
|
||||
echo "Build-Datum: ${BUILD_DATE}"
|
||||
echo "=========================================="
|
||||
echo ""
|
||||
|
||||
# Hinweis: Login wird uebersprungen, weil die Credentials fuer ${REGISTRY}
|
||||
# bereits in ~/.docker/config.json liegen (Container laufen bereits aus dieser Registry).
|
||||
# Bei Bedarf manuell: docker login ${REGISTRY}
|
||||
|
||||
echo ">>> Baue Image (nur ${PLATFORM}, nativ) ..."
|
||||
docker build \
|
||||
--platform "${PLATFORM}" \
|
||||
-f "${DOCKERFILE}" \
|
||||
-t "${FULL_IMAGE}" \
|
||||
.
|
||||
echo ""
|
||||
|
||||
echo ">>> Pushe ${FULL_IMAGE} ..."
|
||||
docker push "${FULL_IMAGE}"
|
||||
echo ""
|
||||
|
||||
# Falls ein anderer Tag als 'latest' gebaut wurde, zusaetzlich latest setzen+pushen
|
||||
if [ "${TAG}" != "latest" ]; then
|
||||
echo ">>> Tagge zusaetzlich als :latest ..."
|
||||
docker tag "${FULL_IMAGE}" "${REGISTRY}/${IMAGE_NAME}:latest"
|
||||
docker push "${REGISTRY}/${IMAGE_NAME}:latest"
|
||||
echo ""
|
||||
fi
|
||||
|
||||
echo "=========================================="
|
||||
echo "Build + Push fertig: ${FULL_IMAGE}"
|
||||
echo "=========================================="
|
||||
echo ""
|
||||
echo "Danach Stack aktualisieren:"
|
||||
echo " cd /opt/stacks/esprimo/noisesensors && docker compose up -d readin"
|
||||
@@ -0,0 +1,55 @@
|
||||
version: '3.9'
|
||||
|
||||
volumes:
|
||||
mongo_vol:
|
||||
influx_vol:
|
||||
|
||||
services:
|
||||
timeseries:
|
||||
image: timeseries
|
||||
environment:
|
||||
DEVELOP: "true"
|
||||
MONGOHOST: mongodb
|
||||
MONGOUSRP: "rexfue:noise4mongo"
|
||||
MONGOAUTH: "true"
|
||||
INFLUXHOST: influxdb
|
||||
INFLUXTOKEN: "q35XUBaElzcy8dDd9HF2_mpeHvYCampZg_9mJNP5jeBQRopq3EWIzNTZ555QLSIXhZC05RXCoFgjiaT7VzyNkQ=="
|
||||
TYP: "[\"noise\", \"thp\"]"
|
||||
STORE: "mongo"
|
||||
volumes:
|
||||
- ${PWD}/log:/var/log
|
||||
container_name: timeseries
|
||||
restart: unless-stopped
|
||||
|
||||
mongodb:
|
||||
image: mongo:6
|
||||
volumes:
|
||||
- ${PWD}/entries:/docker-entrypoint-initdb.d
|
||||
- mongo_vol:/data/db
|
||||
ports:
|
||||
- "27017:27017"
|
||||
container_name: mongodb
|
||||
environment:
|
||||
- MONGO_INITDB_DATABASE=sensor_data
|
||||
- MONGO_INITDB_ROOT_USERNAME=${MONGO_ROOT_USERNAME}
|
||||
- MONGO_INITDB_ROOT_PASSWORD=${MONGO_ROOT_PASSWORD}
|
||||
restart: unless-stopped
|
||||
|
||||
influxdb:
|
||||
image: influxdb:2.0
|
||||
ports:
|
||||
- '8086:8086'
|
||||
volumes:
|
||||
- influx_vol:/var/lib/influxdb2
|
||||
environment:
|
||||
DOCKER_INFLUXDB_INIT_MODE: setup
|
||||
DOCKER_INFLUXDB_INIT_USERNAME: ${DOCKER_INFLUXDB_INIT_USERNAME}
|
||||
DOCKER_INFLUXDB_INIT_PASSWORD: ${DOCKER_INFLUXDB_INIT_PASSWORD}
|
||||
DOCKER_INFLUXDB_INIT_ORG: citysensor
|
||||
DOCKER_INFLUXDB_INIT_BUCKET: sensor_data
|
||||
DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: ${DOCKER_INFLUXTOKEN}
|
||||
restart:
|
||||
unless-stopped
|
||||
container_name: influxdb
|
||||
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
// Einlesen aller Sensordaten von sensor.comunity und abspeicher in einer Inlux- bzw. einer Mongo-DB
|
||||
|
||||
// Ausgehend von Version 1.x.x 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 2.0.0 2023-10-13 rxf
|
||||
// - Erste Version mit der Trennung nach Typen
|
||||
//
|
||||
// V 2.0.1 2023-10-15 rxf
|
||||
// - Auswahl der Typen via Commandline (-t) oder Environment TYP
|
||||
|
||||
// TYP: Ist ein Array von Strings, wenn nur einzelne Typen gespeichert werden sollen, also z.B.:
|
||||
// ['pm', 'noise']
|
||||
const TYP = process.env.TYP || ''
|
||||
const STORE = (process.env.STORE || 'both').toLowerCase() // 'mongo' | 'influx' | 'both'
|
||||
|
||||
import { doReadfromAPI as readin, statistics } from './readdata.js'
|
||||
import { constructDBaseEntries as parse} from './parse.js'
|
||||
|
||||
import * as mongo from './mongo.js'
|
||||
import * as influx from './influx_post.js'
|
||||
import { logit, logerror } from './logit.js'
|
||||
import { DateTime } from 'luxon'
|
||||
import fs from 'fs'
|
||||
import mod_getopt from 'posix-getopt'
|
||||
import pkg from './package.json' with { type: "json" }
|
||||
|
||||
|
||||
// import nodeSchedule from 'node-schedule'
|
||||
|
||||
const fetchNewData = async (args) => {
|
||||
let client
|
||||
let start = DateTime.now();
|
||||
try {
|
||||
client = await mongo.connectMongo()
|
||||
} catch(e) {
|
||||
logerror(`Connect to Mongo ${e}`);
|
||||
logit(`Programmend - Error in connecting mongo\n`)
|
||||
process.exit(-1);
|
||||
}
|
||||
try {
|
||||
// read data from API or from disk
|
||||
let dat = await readin()
|
||||
if (dat.length !== 0) {
|
||||
// parse the data
|
||||
let [props, data, idata] = await parse(client, dat, args)
|
||||
|
||||
// write sensor data to mongoDB
|
||||
if(args.mongo) {
|
||||
for (const [k,v] of Object.entries(data)) {
|
||||
if(v.length !== 0) {
|
||||
await mongo.writeDataArray(client, k+'_sensors', v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// write sensor data to influxDB
|
||||
if(args.influx) {
|
||||
await influx.influxWrite(idata)
|
||||
}
|
||||
|
||||
// write properties to mongoDB
|
||||
await mongo.bulkWrite(client, mongo.property_coll, props)
|
||||
}
|
||||
} catch (e) {
|
||||
logerror(`Catch in main ${e}`);
|
||||
} finally {
|
||||
statistics.totalTime = DateTime.now().diff(start,['seconds']).toObject().seconds
|
||||
await mongo.writeStatistic(client, statistics)
|
||||
await mongo.closeMongo(client)
|
||||
}
|
||||
}
|
||||
|
||||
// Parse command line options
|
||||
function parse_cmdline(argv) {
|
||||
let parser = new mod_getopt.BasicParser('i(influx)m(mongo)t:(typ)h(help)v(version)',argv);
|
||||
let option;
|
||||
let ret = {influx: STORE === 'both' || STORE === 'influx', mongo: STORE === 'both' || STORE === 'mongo', typ: TYP}
|
||||
while((option = parser.getopt()) !== undefined) {
|
||||
switch(option.option) {
|
||||
case 'i':
|
||||
ret.mongo = false
|
||||
break;
|
||||
|
||||
case 'm':
|
||||
ret.influx = false
|
||||
break;
|
||||
|
||||
case 't':
|
||||
let x = option.optarg.trim()
|
||||
let y = []
|
||||
if(x[0] === '[') {
|
||||
y = JSON.parse(x)
|
||||
} else {
|
||||
y.push(x)
|
||||
}
|
||||
ret.typ = y
|
||||
break;
|
||||
|
||||
case 'v':
|
||||
console.log(`Version: ${pkg.version} from ${pkg.date}`);
|
||||
console.log();
|
||||
process.exit();
|
||||
break;
|
||||
|
||||
case 'h':
|
||||
console.log("Usage: node fetchnewdata.js [-i] [-m] [-t [typ, typ, ..]] [-v] [-h]");
|
||||
console.log("Params:");
|
||||
console.log(" -i use only InfluxDB to store the data (default use both, InfluxDB and MongoDB))");
|
||||
console.log(" -m use only MongoDB to store the data (default use both, InfluxDB and MongoDB)");
|
||||
console.log(" -t [ sensorType, ..]: if given, only those sensors will be used (ex: laerm) default: all");
|
||||
console.log(" MUST BE AN ARRAY!; allowed types: 'pm', 'noise', 'radiactivity', 'thp', 'gps'.")
|
||||
console.log(" -v version: show version");
|
||||
console.log(" -h this help text");
|
||||
console.log("All parameters are optional.");
|
||||
console.log();
|
||||
process.exit();
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
logit(JSON.stringify(ret));
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
|
||||
const main = async () => {
|
||||
const json = JSON.parse(fs.readFileSync('package.json', 'utf8'))
|
||||
logit(`Programmstart V ${json.version} vom ${json.date}.`);
|
||||
let args = parse_cmdline(process.argv);
|
||||
await fetchNewData(args)
|
||||
logit(`Program end - running time: ${statistics.totalTime} sec\n\n`)
|
||||
}
|
||||
|
||||
main().catch(console.error)
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
// Access to influxDB vie HTTP
|
||||
|
||||
import axios from 'axios'
|
||||
import { logit, logerror } from './logit.js'
|
||||
import { DateTime } from 'luxon'
|
||||
import { statistics } from'./readdata.js'
|
||||
|
||||
let DEVELOP = process.env.DEVELOP || 'false'
|
||||
|
||||
let INFLUXHOST = process.env.INFLUXHOST || "localhost"
|
||||
let INFLUXPORT = process.env.INFLUXPORT || 8086
|
||||
let INFLUXTOKEN = process.env.INFLUXTOKEN || 'empty'
|
||||
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}&precision=ms`
|
||||
|
||||
export const influxRead = async (query) => {
|
||||
let start = DateTime.now()
|
||||
let data = []
|
||||
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) {
|
||||
logerror(`doReadfromAPI Status: ${ret.status}`)
|
||||
}
|
||||
data = ret.data
|
||||
} catch (e) {
|
||||
logerror(`doReadfromAPI ${e}`)
|
||||
}
|
||||
logit(`ReadIn-Time: ${start.diffNow('seconds').toObject().seconds * -1} sec`)
|
||||
return data
|
||||
}
|
||||
|
||||
|
||||
export const influxWrite = async (data) => {
|
||||
let start = DateTime.now()
|
||||
let ret
|
||||
logit(INFLUXURL_WRITE)
|
||||
if (DEVELOP === 'true') {
|
||||
logit(`Token: ${INFLUXTOKEN}`)
|
||||
}
|
||||
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}`)
|
||||
}
|
||||
let statname = `writeInfluxData[sensor_data]Time`
|
||||
statistics[statname] = DateTime.now().diff(start, ['seconds']).toObject().seconds
|
||||
logit(`Influx-Write-Time: ${start.diffNow('seconds').toObject().seconds * -1} sec`)
|
||||
return ret
|
||||
}
|
||||
|
||||
/*
|
||||
async function main() {
|
||||
let data = `
|
||||
pm,sid=140 P1=12,P2=13
|
||||
pm,sid=142 P1=42,P2=13
|
||||
pm,sid=143 P1=43,P2=13
|
||||
pm,sid=144 P1=44,P2=13
|
||||
thp,sid=141 temperature=23.5,humidity=48,pressure=998
|
||||
`
|
||||
let ret = await influxWrite(data)
|
||||
process.exit()
|
||||
|
||||
let query = `from(bucket:"sensor_data")
|
||||
|> range(start: -1mo)
|
||||
|> filter(fn: (r) => r._measurement == "pm")
|
||||
|> filter(fn: (r) => r.sid == "140")
|
||||
`
|
||||
let erg = await influxRead(query)
|
||||
console.log(erg)
|
||||
}
|
||||
|
||||
|
||||
main().catch(console.error)
|
||||
*/
|
||||
@@ -0,0 +1,12 @@
|
||||
import { DateTime} from 'luxon'
|
||||
|
||||
export function logit(str) {
|
||||
let s = `${DateTime.now().toISO()} => ${str}`;
|
||||
console.log(s);
|
||||
}
|
||||
|
||||
export function logerror(str) {
|
||||
let s = `${DateTime.utc().toISO()} => *** ERROR *** ${str}`;
|
||||
console.log(s);
|
||||
}
|
||||
|
||||
+202
@@ -0,0 +1,202 @@
|
||||
/* Interface for MongoDB
|
||||
*/
|
||||
import { MongoClient } from 'mongodb'
|
||||
import { logit, logerror } from './logit.js'
|
||||
import { statistics } from './readdata.js'
|
||||
import { DateTime } from 'luxon'
|
||||
|
||||
let DEVELOP = process.env.DEVELOP || 'false'
|
||||
|
||||
let MONGOHOST = process.env.MONGOHOST || 'localhost'
|
||||
let MONGOPORT = process.env.MONGOPORT || 27017
|
||||
let MONGOAUTH = process.env.MONGOAUTH || 'false'
|
||||
let MONGOUSRP = process.env.MONGOUSRP || ''
|
||||
let MONGOBASE = process.env.MONGOBASE || 'sensor_data'
|
||||
|
||||
|
||||
let MONGO_URL = 'mongodb://' + MONGOHOST + ':' + MONGOPORT; // URL to mongo database
|
||||
if (MONGOAUTH == 'true') {
|
||||
MONGO_URL = 'mongodb://' + MONGOUSRP + '@' + MONGOHOST + ':' + MONGOPORT + '/?authSource=admin'; // URL to mongo database
|
||||
}
|
||||
|
||||
export const property_coll = 'properties'
|
||||
|
||||
const addandshowstatistics = (client, text, field, start) => {
|
||||
statistics[field] = DateTime.now().diff(start, ['seconds']).toObject().seconds
|
||||
logit(`Write ${text} to mongoDB: Time: ${statistics[field]} sec.`)
|
||||
}
|
||||
|
||||
export const connectMongo = async () => {
|
||||
try {
|
||||
if(DEVELOP === 'true') {
|
||||
logit(`Try to connect to ${MONGO_URL}`)
|
||||
} else {
|
||||
logit(`Try to connect to ${'mongodb://' + MONGOHOST + ':' + MONGOPORT}`)
|
||||
}
|
||||
let client = await MongoClient.connect(MONGO_URL, { useNewUrlParser: true, useUnifiedTopology: true })
|
||||
if ( DEVELOP === 'true') {
|
||||
logit(`Mongodbase connected to ${MONGO_URL}`)
|
||||
} else {
|
||||
logit('Mongodbase connected')
|
||||
}
|
||||
return client
|
||||
}
|
||||
catch (error) {
|
||||
throw (error)
|
||||
}
|
||||
}
|
||||
|
||||
export const closeMongo = async (client) => {
|
||||
client.close()
|
||||
}
|
||||
|
||||
|
||||
export const getallProperties = async (client) => {
|
||||
return await client.db(MONGOBASE).collection(property_coll)
|
||||
.find().sort({ _id: 1 }).toArray()
|
||||
|
||||
}
|
||||
|
||||
export const checkOneproperty = async (client, sid) => {
|
||||
return await client.db(MONGOBASE).collection("properties")
|
||||
.findOne({ _id: sid })
|
||||
}
|
||||
|
||||
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 result
|
||||
let startAll = DateTime.now();
|
||||
let start
|
||||
let coll = client.db(MONGOBASE).collection("properties")
|
||||
if (props.new.length !== 0) {
|
||||
start = DateTime.now();
|
||||
try {
|
||||
result = await coll.insertMany(props.new)
|
||||
} catch (e) {
|
||||
logerror(`Write properties new ${e}`)
|
||||
}
|
||||
logit(`Write ${props.new.length} properties NEW to mongoDB: Result: ${result.acknowledged}, Time: ${start.diffNow('seconds').toObject().seconds * -1} sec.`)
|
||||
}
|
||||
if (props.loc.length !== 0) {
|
||||
start = DateTime.now()
|
||||
try {
|
||||
for (let item of props.loc) {
|
||||
result = await coll.updateOne({ _id: item._id }, {
|
||||
$set: { location_id: item.location_id },
|
||||
$push: { location: { $each: [item.location[0]], $position: 0 } }
|
||||
})
|
||||
}
|
||||
} catch (e) {
|
||||
logerror(`Write properties location ${e}`)
|
||||
}
|
||||
logit(`Write ${props.loc.length} properties LOC to mongoDB: Result: ${result.acknowledged}, Time: ${start.diffNow('seconds').toObject().seconds * -1} sec.`)
|
||||
}
|
||||
if (props.sname.length !== 0) {
|
||||
start = DateTime.now()
|
||||
try {
|
||||
for (let item of props.sname) {
|
||||
result = await coll.updateOne({ _id: item._id }, {
|
||||
$push: { name: { $each: [item.name[0]], $position: 0 } }
|
||||
})
|
||||
}
|
||||
} catch (e) {
|
||||
logerror(`Write properties samename ${e}`)
|
||||
}
|
||||
logit(`Write ${props.sname.length} properties NAME to mongoDB: Result: ${result.acknowledged}, Time: ${start.diffNow('seconds').toObject().seconds * -1} sec.`)
|
||||
}
|
||||
addandshowstatistics(client, 'properties', 'writePropsTime', startAll)
|
||||
}
|
||||
|
||||
export const writeDataArray = async (client, coll, data) => {
|
||||
let result
|
||||
let start = DateTime.now();
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection(coll)
|
||||
.insertMany(data, { ordered: false })
|
||||
} catch (e) {
|
||||
if (e.code !== 11000) {
|
||||
console.error(e)
|
||||
}
|
||||
}
|
||||
let statname = `writeMongoData[${coll}]Time`
|
||||
addandshowstatistics(client, `${data.length} entries for ${coll}`, `writeMongoData[${coll}]Time`, start)
|
||||
|
||||
// statistics[statname] = DateTime.now().diff(start, ['seconds']).toObject().seconds
|
||||
// logit(`Write Data for ${coll} to mongoDB: Time: ${statistics[statname]} sec.`)
|
||||
}
|
||||
|
||||
export const writeStatistic = async (client, stat) => {
|
||||
let result
|
||||
let start = DateTime.now();
|
||||
let entry = { timestamp: new Date(), ...stat }
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection("statistics")
|
||||
.insertOne(entry)
|
||||
} catch (e) {
|
||||
console.error(e)
|
||||
}
|
||||
addandshowstatistics(client, `statistics`, `writeStatisticTime`, start)
|
||||
}
|
||||
|
||||
export const dropColl = async (client, coll) => {
|
||||
let start = DateTime.now()
|
||||
let result
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection(coll).drop()
|
||||
} catch (e) {
|
||||
console.error(e)
|
||||
}
|
||||
logit(`Drop collection ${coll}: Result: ${result}, Time: ${start.diffNow('second').toObject().seconds * -1} sec.`)
|
||||
}
|
||||
|
||||
|
||||
export const createIndex = async (client, coll) => {
|
||||
let result
|
||||
let start = DateTime.now()
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection(coll).createIndex({ "location.loc": "2dsphere" })
|
||||
} catch (e) {
|
||||
console.error(e)
|
||||
}
|
||||
logit(`Create-Index: Result: ${result}, Time: ${start.diffNow('second').toObject().seconds * -1} sec.`)
|
||||
}
|
||||
|
||||
export const bulkUpdateMapdata = async (client, data) => {
|
||||
let start = DateTime.now()
|
||||
let result
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection("mapdata")
|
||||
.bulkWrite(data, { ordered: false })
|
||||
} catch (e) {
|
||||
console.error(e)
|
||||
}
|
||||
logit(`Write MapData: Result: ${result}, Time: ${start.diffNow('second').toObject().seconds * -1} sec.`)
|
||||
return result
|
||||
}
|
||||
|
||||
|
||||
export const bulkWrite = async (client, coll, data) => {
|
||||
let start = DateTime.now()
|
||||
let result
|
||||
try {
|
||||
result = await client.db(MONGOBASE).collection(coll)
|
||||
.bulkWrite(data, { ordered: false })
|
||||
} catch (e) {
|
||||
console.error(e)
|
||||
}
|
||||
addandshowstatistics(client, `Data for ${coll}`, `writeMongoProperties[${coll}]Time`, start)
|
||||
return result
|
||||
}
|
||||
Generated
+1020
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,21 @@
|
||||
{
|
||||
"name": "timeseries_mongo",
|
||||
"version": "2.0.2",
|
||||
"date": "2023-12-17",
|
||||
"description": "",
|
||||
"main": "main.js",
|
||||
"scripts": {
|
||||
"test": "echo \"Error: no test specified\" && exit 1",
|
||||
"start": "node fetchnewdata.js >>/var/log/readin.log 2>&1"
|
||||
},
|
||||
"type": "module",
|
||||
"keywords": [],
|
||||
"author": "",
|
||||
"license": "ISC",
|
||||
"dependencies": {
|
||||
"axios": "^1.12.0",
|
||||
"luxon": "^2.3.1",
|
||||
"mongodb": "^4.4.1",
|
||||
"posix-getopt": "^1.2.1"
|
||||
}
|
||||
}
|
||||
+243
@@ -0,0 +1,243 @@
|
||||
// import logit from './logit.js'
|
||||
import { DateTime } from 'luxon'
|
||||
import * as mongo from './mongo.js'
|
||||
import { statistics } from'./readdata.js'
|
||||
import { logit, logerror } from './logit.js'
|
||||
|
||||
let actualProps = []
|
||||
let newProps = []
|
||||
|
||||
// Check lat/lon and convert to float
|
||||
function checkLatLon(w) {
|
||||
let loc = 0.0;
|
||||
if (!((w === undefined) || (w == null) || (w == ''))) {
|
||||
try {
|
||||
loc = parseFloat(w)
|
||||
} catch (e) {
|
||||
logerror(`Math error with lat/lon, ${e}`)
|
||||
}
|
||||
}
|
||||
return loc
|
||||
}
|
||||
|
||||
// Check, if altitude is there. If so, use it, else use 0
|
||||
function checkAltitude(alt) {
|
||||
let altitude = 0;
|
||||
if(!((alt === undefined) || (alt === ''))) {
|
||||
try {
|
||||
altitude = Math.floor(parseFloat(alt))
|
||||
} catch (e) {
|
||||
logerror(`Math error with altitude, ${e}`)
|
||||
}
|
||||
}
|
||||
return altitude
|
||||
}
|
||||
|
||||
// binary search fir sensor id in property array
|
||||
const binarySearch = (arr, element, x) => {
|
||||
let start = 0, end = arr.length - 1
|
||||
// Iterate while start not meets end
|
||||
while (start <= end) {
|
||||
// Find the mid index
|
||||
let mid = Math.floor((start + end) / 2)
|
||||
// If element is present at mid, return True
|
||||
if (arr[mid][element] === x) return mid
|
||||
// Else look in left or right half accordingly
|
||||
else if (arr[mid][element] < x)
|
||||
start = mid + 1
|
||||
else
|
||||
end = mid - 1
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
|
||||
const checkProperties = (item, mapvalues, typ, dt) => {
|
||||
let now = DateTime.utc().toJSDate()
|
||||
let entry
|
||||
// read entry from actualprops
|
||||
let idx = binarySearch(actualProps, '_id', item.sensor.id)
|
||||
if (idx === -1) { // not in properties => new sensor
|
||||
entry = buildNewEntry(item, typ, dt, now) // so build a new entry
|
||||
} else {
|
||||
entry = actualProps[idx] // get actual properties
|
||||
// check for change of location
|
||||
if (entry.location[0].id !== item.location.id) { // have got a new location
|
||||
if (newProps.findIndex((obj) => {
|
||||
return (obj.replaceOne.replacement.location[0].id === item.location.id)
|
||||
}) === -1) {
|
||||
const newloc = {
|
||||
loc: {
|
||||
type: "Point",
|
||||
coordinates: [
|
||||
checkLatLon(item.location.longitude),
|
||||
checkLatLon(item.location.latitude)
|
||||
]
|
||||
},
|
||||
id: item.location.id,
|
||||
altitude: checkAltitude(item.location.altitude),
|
||||
since: now,
|
||||
exact_loc: item.location.exact_location,
|
||||
indoor: item.location.indoor,
|
||||
country: item.location.country
|
||||
}
|
||||
entry.location.splice(0, 0, newloc) // insert new location at pos 0 in location array
|
||||
}
|
||||
} else { // same location check for change in indoor or exact_location
|
||||
if (entry.location[0].indoor !== item.location.indoor) {
|
||||
entry.location[0].indoor = item.location.indoor // update indoor
|
||||
}
|
||||
if (entry.location[0].exact_loc !== item.location.exact_location) {
|
||||
entry.location[0].exact_loc = item.location.exact_location // update exact_location
|
||||
}
|
||||
if (entry.location[0].country === '') {
|
||||
entry.location[0].country = item.location.country // update country
|
||||
}
|
||||
}
|
||||
// Check für new name
|
||||
if (entry.name[0].name !== item.sensor.sensor_type.name) { // have got a new name
|
||||
if (newProps.findIndex((obj) => {
|
||||
return (obj.replaceOne.replacement.name[0].name === item.sensor.sensor_type.name)
|
||||
}) === -1) {
|
||||
let newname = {
|
||||
name: item.sensor.sensor_type.name,
|
||||
since: now
|
||||
}
|
||||
entry.name.splice(0, 0, newname)
|
||||
}
|
||||
}
|
||||
}
|
||||
// set new mapvalues
|
||||
entry.values = mapvalues
|
||||
delete entry.location_id
|
||||
delete entry.last_seen
|
||||
delete entry.since
|
||||
// push this entry to the new proerties array
|
||||
newProps.push({replaceOne: { filter: {_id: entry._id}, replacement: entry, upsert: true}})
|
||||
}
|
||||
|
||||
|
||||
const buildNewEntry = (item, typ, dt, now) => {
|
||||
return {
|
||||
_id: item.sensor.id,
|
||||
type: typ,
|
||||
name: [{
|
||||
name: item.sensor.sensor_type.name,
|
||||
since: now,
|
||||
}],
|
||||
location: [{
|
||||
loc: {
|
||||
type: "Point",
|
||||
coordinates: [
|
||||
checkLatLon(item.location.longitude),
|
||||
checkLatLon(item.location.latitude)
|
||||
]
|
||||
},
|
||||
id: item.location.id,
|
||||
altitude: checkAltitude(item.location.altitude),
|
||||
since: now,
|
||||
exact_loc: item.location.exact_location,
|
||||
indoor: item.location.indoor,
|
||||
country: item.location.country
|
||||
}]
|
||||
}
|
||||
}
|
||||
|
||||
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'
|
||||
};
|
||||
|
||||
|
||||
function getType(typ) {
|
||||
if(typ in types) {
|
||||
return types[typ];
|
||||
} else {
|
||||
return 'unknown'
|
||||
}
|
||||
}
|
||||
|
||||
export const constructDBaseEntries = async (client, body, args) => {
|
||||
logit(`Number of entries: ${body.length}`)
|
||||
logit('Parsing ...')
|
||||
let start = DateTime.now()
|
||||
let allLines = {
|
||||
pm: [], thp: [], noise: [], radioactivity: [], gps: [], unknown: []
|
||||
}
|
||||
let datalines = ''
|
||||
actualProps = await mongo.getallProperties(client)
|
||||
try {
|
||||
for (let item of body) { // check all entries
|
||||
const dt = item.timestamp.split(' ')
|
||||
const datetime = new Date(dt[0] + 'T' + dt[1] + 'Z') // extract date of entry as utc
|
||||
let values = {}
|
||||
let ival = '' // fetch values
|
||||
let typ = 'unknown'
|
||||
let mapvalue = {}
|
||||
for (let v of item.sensordatavalues) { // for all values
|
||||
let vtyp = v.value_type; // extract value type
|
||||
if (typ === 'unknown') { // extract measurement type
|
||||
typ = getType(vtyp)
|
||||
}
|
||||
if(typ === 'noise') {
|
||||
vtyp = vtyp.slice(6)
|
||||
}
|
||||
let val = v.value; // and value
|
||||
let x
|
||||
try {
|
||||
x = parseFloat(val); // 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
|
||||
}
|
||||
values[vtyp] = x
|
||||
ival += `${vtyp}=${x},`
|
||||
// if noise sensor precalculate pow10
|
||||
if (vtyp == 'LAeq') {
|
||||
let e10 = Math.pow(10, x / 10)
|
||||
values.E10tel_eq = e10
|
||||
ival += `E10tel_eq=${e10},`
|
||||
mapvalue.E10tel_eq = e10
|
||||
}
|
||||
mapvalue[vtyp] = x
|
||||
}
|
||||
let store = true
|
||||
if(args.typ) {
|
||||
if(!args.typ.includes(typ)) {
|
||||
store = false
|
||||
}
|
||||
}
|
||||
if (store) {
|
||||
allLines[typ].push({sensorid: item.sensor.id, datetime: datetime, values: values})
|
||||
datalines += `${typ},sid=${item.sensor.id} ${ival.slice(0,-1)} ${datetime.getTime()}\n`
|
||||
mapvalue.timestamp = datetime
|
||||
checkProperties(item, mapvalue, typ, datetime); // check if new or new location or new sensor
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
logerror(`constructDBaseEntries ${e}`);
|
||||
}
|
||||
// sort allLines on sensorID
|
||||
|
||||
for (const [k,v] of Object.entries(allLines)) {
|
||||
allLines[k].sort((a, b) => {
|
||||
if (a.sensorid < b.sensorid) {
|
||||
return -1
|
||||
} else if (a.sensorid > b.sensorid) {
|
||||
return 1
|
||||
} else {
|
||||
return 0
|
||||
}
|
||||
})
|
||||
}
|
||||
statistics.parseTime = DateTime.now().diff(start, ['seconds']).toObject().seconds
|
||||
logit(`Parse time: ${statistics.parseTime} sec`)
|
||||
return [newProps, allLines, datalines]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import axios from 'axios'
|
||||
import * as fs from 'fs'
|
||||
import { logit, logerror} from'./logit.js'
|
||||
import { DateTime } from 'luxon'
|
||||
const API_URL = 'https://api.sensor.community/static/v1/data.json'; // URL to API on 'luftdaten.info'
|
||||
const SAVE_NAME = './data/aktdata.json'; // filename for actual data
|
||||
|
||||
let LIVE = process.env.LIVE || 'true'
|
||||
export let statistics = {};
|
||||
|
||||
export const doReadfromAPI = async () => {
|
||||
logit(`LIVE = ${LIVE}`)
|
||||
let start = DateTime.now()
|
||||
let data = []
|
||||
if (LIVE === 'true') {
|
||||
logit(`Start Reading from API`)
|
||||
let body
|
||||
for(let count = 1; count <= 3; count++) {
|
||||
try {
|
||||
logit(`Try - ${count}`)
|
||||
let ret = await axios(API_URL, {timeout: 10000})
|
||||
if (ret.status != 200) {
|
||||
continue
|
||||
}
|
||||
data = ret.data
|
||||
saveDatatoFile(SAVE_NAME, JSON.stringify(data))
|
||||
break
|
||||
} catch (e) {
|
||||
logerror(`doReadfromAPI ${e}`)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
logit('Start using data on disk')
|
||||
data = readDatafromFile(SAVE_NAME)
|
||||
}
|
||||
statistics.readInTime = start.diffNow('seconds').toObject().seconds * -1
|
||||
statistics.entries = data.length
|
||||
logit(`ReadIn-Time: ${statistics.readInTime} sec`)
|
||||
return data
|
||||
}
|
||||
|
||||
|
||||
// die Daten in eimnr Datei zwischenspeichern
|
||||
function saveDatatoFile(fn, data) {
|
||||
fs.writeFileSync(fn, data)
|
||||
}
|
||||
|
||||
// Daten wieder vom File lesen
|
||||
function readDatafromFile(fn) {
|
||||
return JSON.parse(fs.readFileSync(fn))
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
/var/log/timeseries.log {
|
||||
daily
|
||||
rotate 7
|
||||
delaycompress
|
||||
compress
|
||||
notifempty
|
||||
missingok
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user