Add data loader endpoint

jason-fox · jason-fox · commit 9a8f2b4c5800 · 2025-09-09T19:11:32.000+02:00
diff --git a/app/controllers/csv.js b/app/controllers/csv.js
@@ -0,0 +1,235 @@
+const fs = require('fs');
+const csv = require('fast-csv');
+const BatchUpdate = require('../lib/batchUpdate');
+const debug = require('debug')('tutorial:csv');
+const _ = require('lodash');
+const Status = require('http-status-codes');
+const path = require('path');
+const moment = require('moment-timezone');
+
+/*
+ * Delete the temporary file
+ */
+function removeCsvFile(path) {
+    fs.unlink(path, (err) => {
+        if (err) {
+            throw err;
+        }
+    });
+}
+
+/*
+ * Read the CSV data from the temporary file.
+ * This returns an in memory representation of the raw CSV file
+ */
+function readCsvFile(path) {
+    return new Promise((resolve, reject) => {
+        const rows = [];
+
+        fs.createReadStream(path)
+            .pipe(csv.parse({ headers: true }))
+            .on('error', (error) => {
+                reject(error.message);
+            })
+            .on('data', (row) => {
+                rows.push(row);
+            })
+            .on('end', () => {
+                resolve(rows);
+            });
+    });
+}
+
+/*
+ *  Strip the id and an key from the header row.
+ */
+function parseId(input) {
+    const regexId = /^[^\s]+/;
+    const regexKey = /[\w]+$/;
+    const id = regexId.exec(input)[0];
+    const key = regexKey.exec(input)[0];
+
+    return { id, key };
+}
+
+function createEntitiesFromRows(rows) {
+    const allEntities = [];
+    const timestamp = new Date().toISOString();
+
+    rows.forEach((row) => {
+        const timestamp = moment.tz(row.annee, 'Etc/UTC').toISOString();
+        const entity = {
+            id: row.id,
+            type: row.type
+        };
+
+        Object.keys(row).forEach((key, index) => {
+            const value = row[key];
+            if (value !== '') {
+                switch (key) {
+                    case 'birthdate':
+                    case 'comment':
+                    case 'fedWith':
+                    case 'giveName':
+                    case 'legalId':
+                    case 'name':
+                    case 'species':
+                        entity[key] = { value: value, type: 'Property' };
+                        break;
+
+                    case 'temperature':
+                        entity[key] = {
+                            value: Number(value),
+                            type: 'Property',
+                            unitCode: 'CEL',
+                            observedAt: timestamp
+                        };
+                        break;
+                    case 'heartRate':
+                        entity[key] = { value: Number(value), type: 'Property', unitCode: '5K', observedAt: timestamp };
+                        break;
+                    case 'weight':
+                        entity[key] = {
+                            value: Number(value),
+                            type: 'Property',
+                            unitCode: 'KGM',
+                            observedAt: timestamp
+                        };
+                        break;
+
+                    case 'jobTitle_name':
+                        entity.jobTitle = {
+                            type: 'Property',
+                            value: {
+                                name: row.jobTitle_name,
+                                inDefinedTermSet: row.jobTitle_inDefinedTermSet,
+                                termCode: row.jobTitle_termCode,
+                                url: row.jobTitle_url
+                            }
+                        };
+                        break;
+                    case 'streetAddress':
+                        entity.address = {
+                            type: 'Property',
+                            value: {
+                                addressLocality: row.addressLocality,
+                                addressRegion: row.addressRegion,
+                                postalCode: row.postalCode,
+                                streetAddress: row.streetAddress
+                            }
+                        };
+                        break;
+                    case 'location_type':
+                        entity.location = {
+                            type: 'GeoProperty',
+                            observedAt: timestamp,
+                            value: {
+                                type: row.location_type,
+                                coordinates: [Number(row.lng), Number(row.lat)]
+                            }
+                        };
+                        break;
+
+                    case 'category':
+                    case 'cropStatus':
+                    case 'gender':
+                    case 'healthCondition':
+                    case 'phenologicalCondition':
+                    case 'reproductiveCondition':
+                    case 'sex':
+                        entity[key] = { vocab: value, type: 'VocabProperty' };
+                        break;
+                    case 'calvedBy':
+                    case 'cropType':
+                    case 'owns':
+                    case 'owner':
+                    case 'ownedBy':
+                    case 'siredBy':
+                        entity[key] = { object: value, type: 'Relationship' };
+                        break;
+                    case 'locatedAt':
+                    case 'observation':
+                    case 'prediction':
+                        entity[key] = { object: value, type: 'Relationship', observedAt: timestamp };
+                        break;
+                    case 'id':
+                    case 'type':
+                        break;
+                    default:
+                        debug('unknown : ' + key);
+                        break;
+                }
+            }
+
+        });
+
+        allEntities.push(entity);
+    });
+    return allEntities;
+}
+
+/*
+ * Create an array of promises to send data to the context broker.
+ * Each insert represents a series of readings at a given timestamp
+ */
+function createContextRequests(entities) {
+    const promises = [];
+    entities.forEach((entitiesAtTimeStamp) => {
+        promises.push(BatchUpdate.sendAsHTTP(entitiesAtTimeStamp));
+    });
+    return promises;
+}
+
+/**
+ * Actions when uploading a CSV file. The CSV file holds an array of
+ * measurements each at a given timestamp.
+ */
+const upload = (req, res) => {
+    if (req.file === undefined) {
+        return res.status(Status.UNSUPPORTED_MEDIA_TYPE).send('Please upload a CSV file!');
+    }
+
+    const file = path.join(__dirname, '../resources/', req.file.filename);
+
+    return readCsvFile(file)
+        .then((rows) => {
+            removeCsvFile(file);
+            //console.log(rows[0])
+            return createEntitiesFromRows(rows);
+        })
+        .then((entities) => {
+            //console.log(JSON.stringify(entities[0], null, 2))
+
+            batchEntities = [];
+            const chunkSize = 10;
+
+            for (let i = 0; i < entities.length; i += chunkSize) {
+                const chunk = entities.slice(i, i + chunkSize);
+                batchEntities.push(chunk);
+            }
+
+            return createContextRequests(batchEntities);
+        })
+        .then(async (promises) => {
+            const results = [];
+            for (const promise of promises) {
+                const result = await promise;
+                results.push(result);
+            }
+            return results;
+        })
+        .then((results) => {
+            const errors = _.filter(results, function (o) {
+                return o.status === 'rejected';
+            });
+            return errors.length ? res.status(Status.BAD_REQUEST).json(errors) : res.status(Status.NO_CONTENT).send();
+        })
+        .catch((err) => {
+            debug(err.message);
+            return res.status(Status.INTERNAL_SERVER_ERROR).send(err.message);
+        });
+};
+
+module.exports = {
+    upload
+};
diff --git a/app/controllers/iot/command-listener.js b/app/controllers/iot/command-listener.js
@@ -18,6 +18,7 @@ const port = process.env.WEB_APP_PORT || '3000';
 const devicesPort = process.env.DUMMY_DEVICES_PORT || 3001;
 const devices = process.env.DUMMY_DEVICES || `http://localhost:${devicesPort}`;
 const autoMoveTractors = process.env.MOVE_TRACTOR || 10000;
+const devicesOff = process.env.DUMMY_OFF || false;
 
 const dataModelContext =
     process.env.IOTA_JSON_LD_CONTEXT || 'http://localhost:' + port + '/data-models/ngsi-context.jsonld';
@@ -203,20 +204,25 @@ function updateTractorStatus() {
     });
 }
 
-setInterval(() => {
-    fireDevices('tractor');
-}, 3361);
-setInterval(fireAnimalCollars, 5099);
-setInterval(() => {
-    fireDevices('temperature');
-}, 7001);
-setInterval(fireOverallFarmStatus, 10000);
-setInterval(() => {
-    fireDevices('humidity');
-}, 8009);
-
-if (autoMoveTractors > 0) {
-    setInterval(updateTractorStatus, autoMoveTractors);
+if (!devicesOff) {
+    debug(`Enabling dummy device updates on ${devices}`);
+    setInterval(() => {
+        fireDevices('tractor');
+    }, 3361);
+    setInterval(fireAnimalCollars, 5099);
+    setInterval(() => {
+        fireDevices('temperature');
+    }, 7001);
+    setInterval(fireOverallFarmStatus, 10000);
+    setInterval(() => {
+        fireDevices('humidity');
+    }, 8009);
+
+    if (autoMoveTractors > 0) {
+        setInterval(updateTractorStatus, autoMoveTractors);
+    }
+} else {
+    debug('Dummy device updates are disabled.');
 }
 
 // The temperature Gauge does not accept commands,
diff --git a/app/lib/batchUpdate.js b/app/lib/batchUpdate.js
@@ -0,0 +1,48 @@
+const CONTEXT_BROKER_URL = process.env.CONTEXT_BROKER || 'http://localhost:1026/ngsi-ld/v1';
+const LINKED_DATA = process.env.IOTA_JSON_LD_CONTEXT || config.contextBroker.jsonLdContext;
+
+async function parse(response) {
+    let text = '';
+    try {
+        text = await response.text();
+        const data = JSON.parse(text);
+        return data;
+    } catch (err) {
+        return text;
+    }
+}
+
+function is2xxSuccessful(status) {
+    return Math.floor(status / 100) === 2;
+}
+
+// measures sent over HTTP are POST requests with params
+async function sendAsHTTP(state) {
+    const url = CONTEXT_BROKER_URL + '/entityOperations/upsert';
+    const headers = {
+        'Content-Type': 'application/json',
+        Link: '<' + LINKED_DATA + '>; rel="http://www.w3.org/ns/json-ld#context"; type="application/ld+json"'
+    };
+    const body = Array.isArray(state) ? state : [state];
+    try {
+        return fetch(`${url}?${new URLSearchParams({ options: 'update' })}`, {
+            method: 'POST',
+            headers,
+            body: JSON.stringify(body)
+        })
+            .then((r) => parse(r).then((data) => ({ status: r.status, body: data })))
+            .then((data) => {
+                if (!is2xxSuccessful(data.status)) {
+                    throw new Error(data.body);
+                }
+                return data.body;
+            });
+    } catch (e) {
+        console.log(e);
+        return null;
+    }
+}
+
+module.exports = {
+    sendAsHTTP
+};
diff --git a/app/lib/upload.js b/app/lib/upload.js
@@ -0,0 +1,22 @@
+const multer = require('multer');
+const path = require('path');
+
+const csvFilter = (req, file, callback) => {
+    if (file.mimetype.includes('csv') || file.mimetype.includes('excel') || file.mimetype.includes('spreadsheetml')) {
+        callback(null, true);
+    } else {
+        callback('Please upload only csv and excel file.', false);
+    }
+};
+
+const storage = multer.diskStorage({
+    destination: (req, file, callback) => {
+        callback(null, path.join(__dirname, '../resources/'));
+    },
+    filename: (req, file, callback) => {
+        callback(null, `${Date.now()}-${file.originalname}`);
+    }
+});
+
+const uploadFile = multer({ storage, fileFilter: csvFilter });
+module.exports = uploadFile;
diff --git a/app/package.json b/app/package.json
@@ -35,14 +35,17 @@
     "express": "~4.17.3",
     "express-healthcheck": "^0.1.0",
     "express-session": "^1.17.2",
+    "fast-csv": "^4.3.2",
     "http-errors": "^2.0.0",
+    "http-status-codes": "^2.1.4",
     "jsonld": "^8.3.3",
     "lodash": "^4.17.21",
-    "moment": "^2.29.1",
+    "moment": "~2.29.1",
+    "moment-timezone": "~0.5.25",
     "mongoose": "^6.2.2",
     "morgan": "^1.10.0",
     "mqtt": "^4.3.6",
-    "node-cache": "^5.1.2",
+    "multer": "^2.0.2",
     "parse-links": "^0.1.0",
     "pug": "3.0.2",
     "request": "^2.88.2",
diff --git a/app/resources/.gitkeep b/app/resources/.gitkeep
@@ -0,0 +1 @@
+.gitkeep
diff --git a/app/routes/index.js b/app/routes/index.js
