transform_to_csv_with_fips.js

const fs = require('fs')
const stream = require('stream')
const es = require('event-stream')
const parse = require('csv-parse')
const stringify = require('csv-stringify')

const fipsLookup = JSON.parse(fs.readFileSync('data/fips_lookup_by_state.json'))

const convertStream = new stream.Transform({
  objectMode: true,
  transform: async (row, encoding, callback) => {
    try {
      const location = row[0]
      let fips
      row[1] = row[1].replace('%', '')
      const pieces = location.split('(')
      if (pieces.length > 1) {
        const county = pieces[0].trim()
        const state = pieces[1].replace(')', '')
        fips = fipsLookup[state][county]
        row.push(fips)
      } else {
        console.log('\nUH OH', row)
      }
      if (fips) {
        callback(null, row)
      } else {
        callback()
      }
    } catch (err) {
      callback(err)
    }
  }
})

function go() {
  const rawPath = 'PP-Data-Store-HPI-Impact-of-Tax-Plan.csv'
  const rawStream = fs.createReadStream(rawPath, { flags: 'r', encoding: 'utf8' })
  const fipsPath = `${rawPath.replace('.csv', '')}-with-FIPS.csv`
  const fipsStream = fs.createWriteStream(fipsPath, { encoding: 'utf8' })

  rawStream
    .pipe(es.split(/(\r?\n)/))
    .pipe(parse())
    .pipe(convertStream)
    .pipe(stringify())
    .pipe(fipsStream)
    .on('end', () => console.log('finished'))
    .on('error', err => console.log('error adding FIPS codes', error))
}

go()