Merge pull request #11 from openaq/provider/clarity

Add Clarity provider

Author: russbiggs

README.md

@@ -4,9 +4,9 @@
 
 ## Deploy
 
- * `yarn cdk deploy`      deploy this stack to your default AWS account/region
- * `yarn cdk diff`        compare deployed stack with current state
- * `yarn cdk synth`       emits the synthesized CloudFormation template
+- `yarn cdk deploy` deploy this stack to your default AWS account/region
+- `yarn cdk diff` compare deployed stack with current state
+- `yarn cdk synth` emits the synthesized CloudFormation template
 
 ## Development
 
@@ -26,12 +26,12 @@ yarn test
 
 Configuration for the ingestion is provided via environment variables.
 
-* `BUCKET`: The bucket to which the ingested data should be written. **Required**
-* `SOURCE`: The [data source](#data-sources) to ingest. **Required**
-* `LCS_API`: The API used when fetching supported measurands. _Default: `'https://api.openaq.org'`_
-* `STACK`: The stack to which the ingested data should be associated. This is mainly used to apply a prefix to data uploaded to S3 in order to separate it from production data. _Default: `'local'`_
-* `SECRET_STACK`: The stack to which the used [Secrets](#provider-secrets) are associated. At times, a developer may want to use credentials relating to a different stack (e.g. a devloper is testing the script, they want output data uploaded to the `local` stack but want to use the production stack's secrets). _Default: the value from the `STACK` env variable_
-* `VERBOSE`: Enable verbose logging. _Default: disabled_
+- `BUCKET`: The bucket to which the ingested data should be written. **Required**
+- `SOURCE`: The [data source](#data-sources) to ingest. **Required**
+- `LCS_API`: The API used when fetching supported measurands. _Default: `'https://api.openaq.org'`_
+- `STACK`: The stack to which the ingested data should be associated. This is mainly used to apply a prefix to data uploaded to S3 in order to separate it from production data. _Default: `'local'`_
+- `SECRET_STACK`: The stack to which the used [Secrets](#provider-secrets) are associated. At times, a developer may want to use credentials relating to a different stack (e.g. a devloper is testing the script, they want output data uploaded to the `local` stack but want to use the production stack's secrets). _Default: the value from the `STACK` env variable_
+- `VERBOSE`: Enable verbose logging. _Default: disabled_
 
 ### Running locally
 
@@ -56,15 +56,12 @@ outline what is necessary to create and a new source.
 
 The first step for a new source is to add JSON config file to the the `fetcher/sources` directory.
 
-
 ```json
 {
-    "schema": "v1",
-    "provider": "example",
-    "frequency": "hour",
-    "meta": {
-
-    }
+  "schema": "v1",
+  "provider": "example",
+  "frequency": "hour",
+  "meta": {}
 }
 ```
 
@@ -73,7 +70,6 @@ The first step for a new source is to add JSON config file to the the `fetcher/s
 | `provider`  | Unique provider name       |
 | `frequency` | `day`, `hour`, or `minute` |
 
-
 The config file can contain any properties that should be configurable via the
 provider script. The above table however outlines the attributes that are required.
 
@@ -86,32 +82,33 @@ The script here should expose a function named `processor`. This function should
 
 The script below is a basic example of a new source:
 
-
 ```js
-const Providers = require('../providers');
-const { Sensor, SensorNode, SensorSystem } = require('../station');
-const { Measures, FixedMeasure, MobileMeasure } = require('../measure');
+const Providers = require("../providers");
+const { Sensor, SensorNode, SensorSystem } = require("../station");
+const { Measures, FixedMeasure, MobileMeasure } = require("../measure");
 
 async function processor(source_name, source) {
-    // Get Locations/Sensor Systems via http/s3 etc.
-    const locs = await get_locations()
+  // Get Locations/Sensor Systems via http/s3 etc.
+  const locs = await get_locations();
 
-    // Map locations into SensorNodes
-    const station = new SensorNode();
+  // Map locations into SensorNodes
+  const station = new SensorNode();
 
-    await Providers.put_stations(source_name, [ station ]);
+  await Providers.put_stations(source_name, [station]);
 
-    const fixed_measures = new Measures(FixedMeasure);
-    // or
-    const mobile_measures = new Measures(MobileMeasure);
+  const fixed_measures = new Measures(FixedMeasure);
+  // or
+  const mobile_measures = new Measures(MobileMeasure);
 
-    fixed_measures.push(new FixedMeasure({
-        sensor_id: 'PurpleAir-123',
-        measure: 123,
-        timestamp: Math.floor(new Date() / 1000) //UNIX Timestamp
-    }));
+  fixed_measures.push(
+    new FixedMeasure({
+      sensor_id: "PurpleAir-123",
+      measure: 123,
+      timestamp: Math.floor(new Date() / 1000), //UNIX Timestamp
+    })
+  );
 
-    await Providers.put_measures(source_name, fixed_measures);
+  await Providers.put_measures(source_name, fixed_measures);
 }
 
 module.exports = { processor };
@@ -120,3 +117,28 @@ module.exports = { processor };
 ### Provider Secrets
 
 For data providers that require credentials, credentials should be store on AWS Secrets Manager with an ID composed of the stack name and provider name, such as `:stackName/:providerName`.
+
+#### Google Keys
+
+Some providers (e.g. CMU, Clarity) require us to read data from Google services (e.g. Drive, Sheets). To do this, the organization hosting the data should do the following:
+
+1. [create a project & enable access to the required APIs](https://developers.google.com/workspace/guides/create-project)
+1. [create a service account](https://cloud.google.com/iam/docs/creating-managing-service-accounts)
+1. [generate service account keys](https://cloud.google.com/iam/docs/creating-managing-service-account-keys)
+
+The should look something like the following and be stored in its entirety within the AWS Secrets Manager.
+
+```json
+{
+  "type": "service_account",
+  "project_id": "project-id",
+  "private_key_id": "key-id",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nprivate-key\n-----END PRIVATE KEY-----\n",
+  "client_email": "service-account-email",
+  "client_id": "client-id",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://accounts.google.com/o/oauth2/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/service-account-email"
+}
+```

fetcher/lib/measurand.js

@@ -17,11 +17,13 @@ class Measurand {
      * @returns { Object } normalizer
      */
     get _normalizer() {
-        return {
-            ppb: ['ppm', (val) => (val / 1000)],
-            'ng/m³': ['µg/m³', (val) => (val / 1000)],
-            'pp100ml': ['particles/cm³', (val) => (val / 100)]
-        }[this.unit] || [this.unit, (val) => val];
+        return (
+            {
+                ppb: ['ppm', (val) => val / 1000],
+                'ng/m³': ['µg/m³', (val) => val / 1000],
+                pp100ml: ['particles/cm³', (val) => val / 100]
+            }[this.unit] || [this.unit, (val) => val]
+        );
     }
 
     get normalized_unit() {
@@ -35,7 +37,7 @@ class Measurand {
     /**
      * Given a map of lookups from an input parameter (i.e. how a data provider
      * identifies a measurand) to a tuple of a measurand parameter (i.e. how we
-     * idenify a measurand internally) and a measurand unit, generate an array
+     * identify a measurand internally) and a measurand unit, generate an array
      * Measurand objects that are supported by the OpenAQ API.
      *
      * @param {*} lookups, e.g. {'CO': ['co', 'ppb'] }
@@ -47,9 +49,14 @@ class Measurand {
         let morePages;
         let page = 1;
         do {
-            const url = new URL('/v2/parameters', process.env.LCS_API || 'https://api.openaq.org');
+            const url = new URL(
+                '/v2/parameters',
+                process.env.LCS_API || 'https://api.openaq.org'
+            );
             url.searchParams.append('page', page++);
-            const { body: { meta, results } } = await request({
+            const {
+                body: { meta, results }
+            } = await request({
                 json: true,
                 method: 'GET',
                 url
@@ -59,24 +66,52 @@ class Measurand {
             }
             morePages = meta.found > meta.page * meta.limit;
         } while (morePages);
-        if (VERBOSE) console.debug(`Fetched ${supportedMeasurandParameters.length} supported measurement parameters.`);
+        if (VERBOSE)
+            console.debug(
+                `Fetched ${supportedMeasurandParameters.length} supported measurement parameters.`
+            );
 
         // Filter provided lookups
         const supportedLookups = Object.entries(lookups).filter(
             // eslint-disable-next-line no-unused-vars
-            ([input_param, [measurand_parameter, measurand_unit]]) => supportedMeasurandParameters.includes(measurand_parameter)
+            ([input_param, [measurand_parameter, measurand_unit]]) =>
+                supportedMeasurandParameters.includes(measurand_parameter)
         );
         if (!supportedLookups.length) throw new Error('No measurands supported.');
         if (VERBOSE) {
             Object.values(lookups)
                 .map(([measurand_parameter]) => measurand_parameter)
-                .filter((measurand_parameter) => !supportedMeasurandParameters.includes(measurand_parameter))
-                .map((measurand_parameter) => console.debug(`warning - ignoring unsupported parameters: ${measurand_parameter}`));
+                .filter(
+                    (measurand_parameter) =>
+                        !supportedMeasurandParameters.includes(measurand_parameter)
+                )
+                .map((measurand_parameter) =>
+                    console.debug(
+                        `warning - ignoring unsupported parameters: ${measurand_parameter}`
+                    )
+                );
         }
         return supportedLookups.map(
-            ([input_param, [parameter, unit]]) => (
+            ([input_param, [parameter, unit]]) =>
                 new Measurand({ input_param, parameter, unit })
-            )
+        );
+    }
+
+    /**
+     * Given a map of lookups from an input parameter (i.e. how a data provider
+     * identifies a measurand) to a tuple of a measurand parameter (i.e. how we
+     * identify a measurand internally) and a measurand unit, generate an object
+     * of Measurand objects that are supported by the OpenAQ API, indexed by their
+     * input parameter.
+     *
+     * @param {*} lookups  e.g. {'CO': ['co', 'ppb'] }
+     * @returns {object}
+     */
+    static async getIndexedSupportedMeasurands(lookups) {
+        const measurands = await Measurand.getSupportedMeasurands(lookups);
+        return Object.assign(
+            {},
+            ...measurands.map((measurand) => ({ [measurand.input_param]: measurand }))
         );
     }
 }

fetcher/lib/providers.js

@@ -1,15 +1,11 @@
 const fs = require('fs');
-const zlib = require('zlib');
-const { promisify } = require('util');
 const path = require('path');
 const AWS = require('aws-sdk');
-const { VERBOSE } = require('./utils');
+const { VERBOSE, gzip, unzip } = require('./utils');
 
 const s3 = new AWS.S3({
     maxRetries: 10
 });
-const gzip = promisify(zlib.gzip);
-const unzip = promisify(zlib.unzip);
 
 /**
  * Runtime handler for each of the custom provider scripts, as well

fetcher/lib/utils.js

@@ -1,3 +1,4 @@
+const zlib = require('zlib');
 const { promisify } = require('util');
 const request = promisify(require('request'));
 const AWS = require('aws-sdk');
@@ -17,19 +18,47 @@ async function fetchSecret(source_name) {
 
     if (!process.env.STACK) throw new Error('STACK Env Var Required');
 
-    const SecretId = `${process.env.SECRET_STACK || process.env.STACK}/${source_name}`;
+    const SecretId = `${
+        process.env.SECRET_STACK || process.env.STACK
+    }/${source_name}`;
 
     if (VERBOSE) console.debug(`Fetching ${SecretId}...`);
 
-    const { SecretString } = await secretsManager.getSecretValue({
-        SecretId
-    }).promise();
+    const { SecretString } = await secretsManager
+        .getSecretValue({
+            SecretId
+        })
+        .promise();
 
     return JSON.parse(SecretString);
 }
 
+/**
+ * Transform phrase to camel case.
+ * e.g. toCamelCase("API Key") === "apiKey"
+ *
+ * @param {string} phrase
+ * @returns {string}
+ */
+function toCamelCase(phrase) {
+    return phrase
+        .split(' ')
+        .map((word) => word.toLowerCase())
+        .map((word, i) => {
+            if (i === 0) return word;
+            return word.replace(/^./, word[0].toUpperCase());
+        })
+        .join('');
+}
+
+const gzip = promisify(zlib.gzip);
+const unzip = promisify(zlib.unzip);
+
 module.exports = {
     fetchSecret,
     request,
+    toCamelCase,
+    gzip,
+    unzip,
     VERBOSE
 };