WIP: use spatial service for PIP

cmd/import.sh

@@ -3,7 +3,7 @@ set -e;
 
 # per-source imports
 function import_wof(){ compose_run 'whosonfirst' './bin/start'; }
-function import_oa(){ compose_run 'openaddresses' "./bin/parallel ${OPENADDRESSES_PARALLELISM:-1}"; }
+function import_oa(){ compose_run -T 'openaddresses' "./bin/parallel ${OPENADDRESSES_PARALLELISM:-1}"; }
 function import_osm(){ compose_run 'openstreetmap' './bin/start'; }
 function import_polylines(){ compose_run 'polylines' './bin/start'; }
 function import_geonames(){ compose_run 'geonames' './bin/start'; }

cmd/prepare.sh

@@ -9,9 +9,34 @@ function prepare_placeholder(){
   compose_run -T 'placeholder' ./cmd/build.sh;
 }
 
+function prepare_spatial(){
+  compose_run -T --entrypoint='bash' 'spatial' << 'EOF'
+    #!/bin/bash
+    set -euo pipefail
+
+    TARGET_DB='/data/spatial/docker.spatial.db'
+    WOF_SQLITE_DIR='/data/whosonfirst/sqlite'
+    SQL_EXTRACT_QUERY="SELECT json_extract(body, '$') FROM geojson"
+
+    mkdir -p $(dirname "${TARGET_DB}")
+    rm -rf "${TARGET_DB}"
+
+    export_all_wof_databases_as_geojson() {
+      for db in $(find "${WOF_SQLITE_DIR}" -name '*.db' -type f -maxdepth 1); do
+        1>&2 echo "[reading] $(basename ${db})"
+        /opt/spatial/bin/sqlite3 "${db}" "${SQL_EXTRACT_QUERY}"
+      done
+    }
+
+    export_all_wof_databases_as_geojson \
+      | node bin/spatial.js import whosonfirst --db="${TARGET_DB}"
+EOF
+}
+
 register 'prepare' 'polylines' 'export road network from openstreetmap into polylines format' prepare_polylines
 register 'prepare' 'interpolation' 'build interpolation sqlite databases' prepare_interpolation
 register 'prepare' 'placeholder' 'build placeholder sqlite databases' prepare_placeholder
+register 'prepare' 'spatial' 'build spatial database' prepare_spatial
 
 # prepare all the data to be used by imports
 function prepare_all(){

projects/portland-metro/Dockerfile.oa.plus.spatial

@@ -0,0 +1,8 @@
+FROM pelias/spatial:pip_beta AS spatial
+FROM pelias/openaddresses:master
+
+# copy spatial runtime
+COPY --from=spatial --chown=pelias /opt/spatial /opt/spatial
+
+# copy spatial code
+COPY --from=spatial --chown=pelias /code /code/spatial

projects/portland-metro/admin-lookup/SpatialServiceConfig.js

@@ -0,0 +1,23 @@
+const pmw = require('pelias-microservice-wrapper')
+
+class SpatialServiceConfig extends pmw.ServiceConfiguration {
+  constructor() {
+    super('spatial', {
+      url: 'http://spatial:4770/'
+    })
+  }
+  getUrl() {
+    return `${this.baseUrl}query/pip/beta`;
+  }
+  getParameters(params) {
+    return params;
+  }
+  isEnabled() {
+    return true
+  }
+  createClient() {
+    return pmw.service(this)
+  }
+};
+
+module.exports = SpatialServiceConfig

projects/portland-metro/admin-lookup/codec.js

@@ -0,0 +1,23 @@
+const _ = require('lodash')
+const Document = require('pelias-model').Document
+
+module.exports.marshal = (doc) => {
+  return doc
+}
+
+module.exports.unmarshal = (data) => {
+
+  // assign document prototype
+  // let doc = new Document(data.source, data.layer, data.source_id)
+  // doc = _.merge(doc, data)
+
+  let doc = Object.setPrototypeOf(data, Document.prototype)
+
+  // create a non-enumerable property for metadata
+  Object.defineProperty(doc, '_meta', { writable: true, value: {} });
+
+  // create a non-enumerable property for post-processing scripts
+  Object.defineProperty(doc, '_post', { writable: true, value: [] });
+
+  return doc
+}

projects/portland-metro/admin-lookup/localClient.js

@@ -0,0 +1,45 @@
+const _ = require('lodash')
+const codec = require('./codec')
+const cpuCount = require('os').cpus().length
+const Pool = require('worker-thread-pool');
+
+// // const threads = require('worker_threads')
+// const worker = new threads.Worker('./node_modules/pelias-wof-admin-lookup/queryServiceWorkerThread.js', {
+//   workerData: {
+//     readonly: true,
+//     filename: '/data/spatial/docker.spatial.db'
+//   }
+// });
+
+
+const pool = new Pool({
+  size: cpuCount,
+  path: './node_modules/pelias-wof-admin-lookup/queryServiceWorkerThread.js'
+})
+
+function lookup(doc, cb){
+  // console.error('lookup', query)
+  // message.doc = codec.marshal(message.doc)
+
+  // const crypto = require('crypto')
+  // const token = crypto.randomBytes(64).toString('hex').substr(0, 8);
+
+  pool.run({ centroid: doc.getCentroid() })
+    .then((message) => {
+      // doc = codec.unmarshal(doc)
+      // console.error('worker message', message)
+
+      message.addParent.forEach(ap => {
+        // console.error('addParent', ap)
+        doc.addParent(...ap)
+      })
+
+      cb()
+    })
+    .catch(({ err }) => {
+      // console.error(message)
+      cb({ err })
+    })
+}
+
+module.exports = lookup

projects/portland-metro/admin-lookup/main.js

@@ -0,0 +1,40 @@
+const _ = require('lodash')
+const cpuCount = require('os').cpus().length
+// const through = require('through2');
+const transform = require('parallel-transform');
+const parallelism = cpuCount
+const logger = require('pelias-logger').get('admin-lookup');
+const streamOptions = { highwaterMark: 2000, ordered: false  }
+
+// network client
+// const SpatialServiceConfig = require('./SpatialServiceConfig');
+// const client = new SpatialServiceConfig().createClient()
+
+// local client
+const client = require('./localClient')
+
+function create(){
+  // return through.obj(streamOptions, (doc, enc, next) => {
+  return transform(parallelism, streamOptions, (doc, next) => {
+    client(doc, ({ err }) => {
+      if (err) {
+        console.error('PIP error', err)
+
+        // if there's an error, just log it and move on
+        logger.error(`PIP server failed: ${(err.message || JSON.stringify(err))}`, {
+          id: doc.getGid(),
+          lat: centroid.lat,
+          lon: centroid.lon
+        });
+
+        // don't pass the unmodified doc along
+        return next()
+      }
+
+      // console.error(query)
+      next(null, doc);
+    });
+  })
+}
+
+module.exports = { create }