Allow for CSV FHIR ingestions - observation example included

examples/csv_format/csv_observation_input.json

@@ -0,0 +1,57 @@
+{
+  "fieldMaps": [
+    {
+      "columnName": "id",
+      "jpath": "id"
+    },
+    {
+      "value": "Observation",
+      "jpath": "resourceType"
+    },
+    {
+      "columnName": "subject",
+      "jpath": "subject.reference"
+    },
+    {
+      "columnName": "date",
+      "jpath": "effectiveDateTime"
+    },
+    {
+      "value": "http://lifeomic.com/fhir/source",
+      "jpath": "meta.tag[0].system"
+    },
+    {
+      "value": "CLI CSV",
+      "jpath": "meta.tag[0].code"
+    },
+    {
+      "value": "final",
+      "jpath": "status"
+    },
+    {
+      "columnName": "code",
+      "jpath": "valueCodeableConcept.coding[0].code"
+    },
+    {
+      "columnName": "codeSystem",
+      "jpath": "valueCodeableConcept.coding[0].system"
+    },
+    {
+      "columnName": "codeDisplay",
+      "jpath": "valueCodeableConcept.coding[0].display"
+    },
+    {
+      "columnName": "value",
+      "isNumber": true,
+      "jpath": "valueQuantity.value"
+    },
+    {
+      "columnName": "valueUnitSystem",
+      "jpath": "valueQuantity.system"
+    },
+    {
+      "columnName": "valueUnit",
+      "jpath": "valueQuantity.unit"
+    }
+  ]
+}

examples/csv_input/observations.csv

@@ -0,0 +1,2 @@
+id,subject,date,codeSystem,code,codeDisplay,valueUnitSystem,valueUnit,value
+"3abfdf2f-7af6-47ca-aebc-d28a8587efe2","34baff70-ff0d-433b-9f16-8196efd9e1cc","2019-07-12T16:10:28-04:00","http://loinc.org","1974-5","Bilirubin","http://unitsofmeasure.org/","mg/dL",0.8629134740075964

lib/cmds/fhir_cmds/ingest.js

@@ -4,8 +4,17 @@ const { post, getAccount } = require('../../fhir');
 const print = require('../../print');
 const stdin = require('../../stdin');
 const _chunk = require('lodash/chunk');
+const _set = require('lodash/set');
 const {chain} = require('stream-chain');
 const StreamValues = require('stream-json/streamers/StreamValues');
+const {parser: csvParser} = require('stream-csv-as-json');
+const {asObjects} = require('stream-csv-as-json/AsObjects');
+const fs = require('fs');
+
+function numberOrString (value) {
+  const numericValue = parseFloat(value);
+  return (!isNaN(numericValue) && isFinite(numericValue)) ? numericValue : value;
+}
 
 function setDataset (data, options) {
   if (options.project) {
@@ -51,6 +60,29 @@ async function batchUpload (options, resources) {
   }
 }
 
+function getCSVChain () {
+  return chain([
+    csvParser(),
+    asObjects(),
+    StreamValues.streamValues()
+  ]);
+}
+
+function parseCSVValue (csvConfig, value) {
+  const resource = {};
+  csvConfig.fieldMaps.forEach(fieldMap => {
+    if (fieldMap.value) {
+      _set(resource, fieldMap.jpath, fieldMap.value);
+    } else if (fieldMap.columnName && value[fieldMap.columnName]) {
+      const fieldValue = fieldMap.isNumber
+        ? numberOrString(value[fieldMap.columnName])
+        : value[fieldMap.columnName];
+      _set(resource, fieldMap.jpath, fieldValue);
+    }
+  });
+  return resource;
+}
+
 exports.command = 'ingest';
 exports.desc = 'Create or update one or more FHIR resources. The resources are read from stdin.';
 exports.builder = yargs => {
@@ -61,16 +93,24 @@ exports.builder = yargs => {
   }).option('project', {
     describe: 'Tag the resource with the given project ID',
     type: 'string'
+  }).option('csv', {
+    describe: 'CSV Format Configuration file in json',
+    type: 'string'
   });
 };
 
 exports.handler = async argv => {
   let resources = [];
 
+  const csvConfig = argv.csv ? JSON.parse(fs.readFileSync(argv.csv, { encoding: 'utf8' })) : null;
+
   const pipeline = chain([
     stdin(),
-    StreamValues.withParser(),
+    csvConfig ? getCSVChain() : StreamValues.withParser(),
     async ({value}) => {
+      if (csvConfig) {
+        value = parseCSVValue(csvConfig, value);
+      }
       if (Array.isArray(value)) {
         resources = resources.concat(value);
       } else {

package.json

@@ -41,6 +41,7 @@
     "recursive-readdir": "^2.2.2",
     "stoppable": "^1.1.0",
     "stream-chain": "^2.1.0",
+    "stream-csv-as-json": "^1.0.1",
     "stream-json": "^1.2.1",
     "yargs": "^12.0.5"
   },

test/unit/commands/fhir-test.js

@@ -173,6 +173,42 @@ test.serial.cb('The "fhir-ingest" with project command should update a fhir reso
     .parse('ingest --project abc');
 });
 
+test.serial.cb('The "fhir-ingest" command can accept CSV stdin given a CSV configuration', t => {
+  const res = {data: {entry: [{response: {location: '/account/dstu3/Patient/1234', status: '200'}}]}};
+  postStub.returns(res);
+
+  // eslint-disable-next-line security/detect-non-literal-fs-filename
+  const stdin = fs.createReadStream(`${__dirname}/fhir/csvInput.csv`);
+  stdinStub.returns(stdin);
+
+  callback = () => {
+    t.is(postStub.callCount, 1);
+    t.is(postStub.getCall(0).args[1], 'account/dstu3');
+    t.deepEqual(postStub.getCall(0).args[2], {
+      resourceType: 'Bundle',
+      type: 'collection',
+      entry: [
+        {
+          resource:
+          {
+            resourceType: 'Patient',
+            id: '1234',
+            name: [{
+              family: 'LastName'
+            }]
+          }
+        }
+      ]
+    });
+    t.is(printSpy.callCount, 1);
+    t.deepEqual(printSpy.getCall(0).args[0], [{response: {location: '/account/dstu3/Patient/1234', status: '200'}}]);
+    t.end();
+  };
+
+  yargs.command(ingest)
+    .parse('ingest --csv ./test/unit/commands/fhir/csvFormat.json');
+});
+
 test.serial.cb('The "fhir-delete" command should delete a fhir resource', t => {
   callback = () => {
     t.is(delStub.callCount, 1);

test/unit/commands/fhir/csvFormat.json

@@ -0,0 +1,16 @@
+{
+  "fieldMaps": [
+    {
+      "columnName": "id",
+      "jpath": "id"
+    },
+    {
+      "value": "Patient",
+      "jpath": "resourceType"
+    },
+    {
+      "columnName": "lastName",
+      "jpath": "name[0].family"
+    }
+  ]
+}

test/unit/commands/fhir/csvInput.csv

@@ -0,0 +1,2 @@
+id,lastName
+"1234","LastName"

yarn.lock

@@ -4886,6 +4886,20 @@ stream-chain@^2.1.0:
   resolved "https://registry.yarnpkg.com/stream-chain/-/stream-chain-2.1.0.tgz#a058a7104d7261a71bf7eb4c5f496f13dc665320"
   integrity sha512-PAUXdRGm0G8P0+/+JEd3O9kfmB9kwmr2nKIc5zhcsHn0KdBByD5PJ2po21iDzc+TZsOSEbU8j4JbAevJsZkLyQ==
 
+stream-csv-as-json@^1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/stream-csv-as-json/-/stream-csv-as-json-1.0.1.tgz#2531183b73676d2cf64fa3b8d3dbfe8f492040b2"
+  integrity sha512-nnfIA8Xbbo5H4jMUQKEiwf4jfzOkyYXzQiD5O3LwKt52iTrrDYqAP9/thQAUeUj6GoLkwZwbOq+3jd/TB+/SIw==
+  dependencies:
+    stream-json "^1.0.1"
+
+stream-json@^1.0.1:
+  version "1.3.0"
+  resolved "https://registry.yarnpkg.com/stream-json/-/stream-json-1.3.0.tgz#c7c0f40a808ce91181c15018fc3dbcd09bcd6cf8"
+  integrity sha512-3qLDv/xnwmleb5kssgmKbGPqcLou2tbFIgj3CM3fy5PxKaAmeCv103Zp4LKIALdE30zMHTJn09yVwxq773btaw==
+  dependencies:
+    stream-chain "^2.1.0"
+
 stream-json@^1.2.1:
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/stream-json/-/stream-json-1.2.1.tgz#7b58112358dca3496b3bb1fc9c2a3099a53b34f4"