copy CITY field to STREET field for streetless addresses

lib/streams/recordStream.js

@@ -10,6 +10,7 @@ const config = require('pelias-config').generate();
 
 const CleanupStream = require('./cleanupStream');
 const ContentHashStream = require('./contentHashStream');
+const StreetlessAddressStream = require('./streetlessAddressStream');
 const ValidRecordFilterStream = require('./validRecordFilterStream');
 const DocumentStream = require('./documentStream');
 
@@ -56,6 +57,7 @@ function createRecordStream( filePath, dirPath ){
   });
 
   const contentHashStream = ContentHashStream.create();
+  const streetlessAddressStream = StreetlessAddressStream.create();
   const validRecordFilterStream = ValidRecordFilterStream.create();
   const cleanupStream = CleanupStream.create();
   const idPrefix = getIdPrefix(filePath, dirPath);
@@ -68,6 +70,7 @@ function createRecordStream( filePath, dirPath ){
   return fs.createReadStream( filePath )
     .pipe( csvParser )
     .pipe( contentHashStream )
+    .pipe( streetlessAddressStream )
     .pipe( validRecordFilterStream )
     .pipe( cleanupStream )
     .pipe( documentStream );

lib/streams/streetlessAddressStream.js

@@ -0,0 +1,21 @@
+const _ = require('lodash');
+const through2 = require('through2');
+
+/*
+ * a stream that uses the city name in place of a missing street name.
+ *
+ * see: https://github.com/pelias/pelias/issues/642
+ */
+
+function createStreetlessAddressStream() {
+  return through2.obj((record, enc, next) => {
+    if (_.has(record, 'NUMBER') && _.has(record, 'CITY') && !_.has(record, 'STREET')){
+      _.set(record, 'STREET', _.get(record, 'CITY'));
+    }
+    next(null, record);
+  });
+}
+
+module.exports = {
+  create: createStreetlessAddressStream
+};

test/streams/streetlessAddressStream.js

@@ -0,0 +1,30 @@
+const tape = require('tape');
+const stream_mock = require('stream-mock');
+const StreetlessAddressStream = require('../../lib/streams/streetlessAddressStream');
+
+function test_stream(input, testedStream, callback) {
+  const reader = new stream_mock.ObjectReadableMock(input);
+  const writer = new stream_mock.ObjectWritableMock();
+  writer.on('error', (e) => callback(e));
+  writer.on('finish', () => callback(null, writer.data));
+  reader.pipe(testedStream).pipe(writer);
+}
+
+tape('streetlessAddressStream copies CITY to STREET', function (test) {
+  var input = {
+    NUMBER: '5',
+    LAT: 5,
+    LON: 6,
+    CITY: 'example city'
+  };
+
+  var streetlessAddressStream = StreetlessAddressStream.create();
+
+  test_stream([input], streetlessAddressStream, function (err, records) {
+    test.equal(records.length, 1, '1 record emitted');
+
+    var record = records[0];
+    test.equal(record.STREET, 'example city', 'STREET field generated');
+    test.end();
+  });
+});

test/test.js

@@ -9,6 +9,7 @@ require( './importPipeline');
 require( './parameters' );
 require( './streams/cleanupStream' );
 require( './streams/contentHashStream' );
+require( './streams/streetlessAddressStream' );
 require( './streams/documentStream' );
 require( './streams/germanicAbbreviationStream');
 require( './streams/isUSorCAHouseNumberZero' );