Basic method for transforming files

.eslintrc.json

@@ -12,7 +12,7 @@
         "no-var": "error",
         "prefer-const": "error",
         "array-bracket-spacing": [ "error", "never" ],
-        "comma-dangle": [ "error", "never" ],
+        "comma-dangle": [ "error", "only-multiline" ],
         "computed-property-spacing": [ "error", "never" ],
         "eol-last": "error",
         "eqeqeq": [ "error", "smart" ],

fetcher/index.js

@@ -4,7 +4,6 @@ require('dotenv').config({ path });
 const providers = new (require('./lib/providers'))();
 const sources = require('./sources');
 
-
 if (require.main === module) {
     handler();
 }

fetcher/lib/measurand.js

@@ -47,7 +47,7 @@ class Measurand {
         // Fetch from API
         const supportedMeasurandParameters = [
             'pm10',
-            'pm25','o3','co','no2','so2','no2','co','so2','o3','bc','co2','no2','bc','pm1','co2','wind_direction','nox','no','rh','nox','ch4','pn','o3','ufp','wind_speed','no','pm','ambient_temp','pressure','pm25-old','relativehumidity','temperature','so2','co','um003','um010','temperature','um050','um025','pm100','pressure','um005','humidity','um100','voc','ozone','nox','bc','no','pm4','so4','ec','oc','cl','no3','pm25'];
+            'pm25','o3','co','no2','so2','no2','co','so2','o3','bc','co2','no2','bc','pm1','co2','wind_direction','nox','no','rh','nox','ch4','pn','o3','ufp','wind_speed','no','pm','ambient_temp','pressure','pm25-old','relativehumidity','temperature','so2','co','um003','um010','temperature','um050','um025','pm100','pressure','um005','humidity','um100','voc','ozone','nox','bc','no','pm4','so4','ec','oc','cl','no3','pm25','bc_375','bc_470','bc_528','bc_625','bc_880', 'as', 'cd', 'fe', 'k', 'ni', 'pb', 'v'];
 
         // Filter provided lookups
         const supportedLookups = Object.entries(lookups).filter(

fetcher/lib/measure.js

@@ -32,6 +32,14 @@ class Measures {
         return this.measures.length;
     }
 
+    json() {
+        return this.measures.map((m) => ({
+            sensor_id: m.sensor_id,
+            timestamp: m.timestamp.utc().format(),
+            measure: m.measure,
+        }));
+    }
+
     csv() {
         const csvStringifier = createCsvStringifier({
             header: this.headers.map((head) => ({

fetcher/lib/providers.js

@@ -35,14 +35,15 @@ class Providers {
      * @param {Object} source
      */
     async processor(source) {
-        if (VERBOSE) console.debug('Processing', source.provider);
-        if (!this[source.provider]) throw new Error(`${source.provider} is not a supported provider`);
+        if (VERBOSE) console.debug(`Processing ${source.provider} with ${source.processor} processor`);
+        const p = this[source.processor];
+        if (!p) throw new Error(`${source.processor} is not a supported processor`);
         // fetch any secrets we may be storing for the provider
         if (VERBOSE) console.debug('Fetching secret: ', source.secretKey);
         const config = await fetchSecret(source);
         // and combine them with the source config for more generic access
         if (VERBOSE) console.log('Starting processor', { ...source, ...config });
-        const log = await this[source.provider].processor({ ...source, ...config });
+        const log = await p.processor({ ...source, ...config });
         // source_name is more consistent with our db schema
         if (typeof(log) == 'object' && !Array.isArray(log) && !log.source_name) {
             log.source_name = source.provider;

fetcher/lib/utils.js

@@ -4,13 +4,15 @@ const request = promisify(require('request'));
 const fs = require('node:fs');
 const path = require('path');
 const homedir = require('os').homedir();
+const csv = require('csv-parser');
 
 const { SecretsManagerClient, GetSecretValueCommand } = require('@aws-sdk/client-secrets-manager');
 const { S3Client, GetObjectCommand, PutObjectCommand } = require('@aws-sdk/client-s3');
 
 const VERBOSE = !!process.env.VERBOSE;
 const DRYRUN = !!process.env.DRYRUN;
 
+
 const s3 = new S3Client({
     maxRetries: 10
 });
@@ -185,11 +187,11 @@ function checkResponseData(data, start_timestamp, end_timestamp) {
     const fdata = data.filter((d) => {
         return (
             (d.time / 1000 >= start_timestamp || !start_timestamp) &&
-      d.time / 1000 <= end_timestamp
+                d.time / 1000 <= end_timestamp
         );
     });
     if (fdata.length < n) {
-    // submit warning so we can track this
+        // submit warning so we can track this
         const requested_start = new Date(
             start_timestamp * 1000
         ).toISOString();
@@ -205,9 +207,39 @@ function checkResponseData(data, start_timestamp, end_timestamp) {
     return fdata;
 }
 
+const fetchFileLocal = async (file) => {
+    const filepath = file.path.replace('$CWD', process.cwd());
+    const data = [];
+    console.debug('fetching file', filepath);
+    return new Promise((resolve, reject) => {
+        fs.createReadStream(filepath)
+            .on('error', (e) => {
+                reject(e);
+            })
+            .pipe(csv())
+            .on('data', (row) => {
+                data.push(row);
+            })
+            .on('end', () => {
+                resolve(data);
+            });
+    });
+};
+
+const fetchFile = async (file) => {
+    const source = file.source;
+    let data;
+    if (['local','undefined',undefined,null].includes(source)) {
+        data = await fetchFileLocal(file);
+    } else {
+        throw new Error(`No support for ${source}`);
+    }
+    return data;
+};
 
 module.exports = {
     fetchSecret,
+    fetchFile,
     request,
     toCamelCase,
     gzip,