merge-trees-to-database

bin/merge.js

@@ -0,0 +1,2 @@
+import { runMerge } from "../src/index.js";
+await runMerge();

package.json

@@ -12,6 +12,7 @@
     "convert": "node bin/convert.js",
     "normalize": "node bin/normalize.js",
     "concatenate": "node bin/concatenate.js",
+    "merge": "node bin/merge.js",
     "save": "node bin/save.js",
     "tile": "node bin/tile.js",
     "upload": "node bin/upload.js",

src/db/db-config.js

@@ -0,0 +1,14 @@
+/**
+ * https://github.com/vitaly-t/pg-promise/wiki/Connection-Syntax
+ */
+export const dbConfig = {
+  host: process.env.POSTGRES_HOST,
+  port: process.env.POSTGRES_PORT,
+  database: process.env.POSTGRES_DB,
+  user: process.env.POSTGRES_USER,
+  password: process.env.POSTGRES_PASSWORD,
+};
+
+// console.log('dbConfig',dbConfig, process.env);
+
+//module.exports = dbConfig;

src/db/index.js

@@ -0,0 +1,7 @@
+import pgp from 'pg-promise';
+import { dbConfig } from './db-config.js';
+import { pgPromiseConfig } from './pg-promise-config.js';
+
+export const pgPromise = pgp(pgPromiseConfig);
+export const db = pgPromise(dbConfig);
+

src/db/pg-promise-config.js

@@ -0,0 +1,27 @@
+/* eslint-disable guard-for-in */
+/* eslint-disable no-restricted-syntax */
+import pgp from 'pg-promise';
+
+/**
+ * source:
+ * https://github.com/vitaly-t/pg-promise/issues/78#issuecomment-171951303
+ */
+function camelizeColumns(data) {
+  const tmp = data[0];
+  for (const prop in tmp) {
+    const camel = pgp.utils.camelize(prop);
+    if (!(camel in tmp)) {
+      for (let i = 0; i < data.length; i++) {
+        const d = data[i];
+        d[camel] = d[prop];
+        delete d[prop];
+      }
+    }
+  }
+}
+
+export const pgPromiseConfig = {
+  capSQL: true,
+  receive: (data) => camelizeColumns(data),
+};
+

src/index.js

@@ -2,6 +2,7 @@ import path from "path";
 import * as download from "./stages/download.js";
 import * as convert from "./stages/convert.js";
 import * as normalize from "./stages/normalize.js";
+import * as merge from "./stages/merge.js";
 import * as save from "./stages/save.js";
 import * as concatenate from "./stages/concatenate.js";
 import * as tile from "./stages/tile.js";
@@ -23,6 +24,10 @@ export const runNormalize = async () => {
   await normalize.normalizeSources(sources);
 };
 
+export const runMerge = async () => {
+  await merge.mergeSources(sources);
+};
+
 export const runConcatenate = async () => {
   const filenames = await utils.asyncReadDir(config.NORMALIZED_DIRECTORY);
   const filepaths = filenames.map((f) =>

src/stages/merge.js

@@ -0,0 +1,76 @@
+import { db } from '../db/index.js';
+import { exec } from 'child_process';
+import { dbConfig } from '../db/db-config.js';
+import * as utils from "../core/utils.js";
+
+
+export const mergeSource = async (source) => {
+  console.log('source.id', source.id);
+  if (
+    !source.destinations ||
+    !source.destinations.geojson ||
+    !source.destinations.normalized
+  ) {
+    throw new Error(`No destinations for source: "${source}"`);
+  }
+
+  const normalizedExists = await utils.asyncFileExists(
+    source.destinations.normalized.path
+  );
+  if (!normalizedExists) {
+    console.log(
+      `The normalized file '${source.destinations.normalized.path}' does not exists. Skipping...`
+    );
+    return `NO FILE for ${source.id}`; // Early Return
+  }
+
+    console.log(`Running for ${source.destinations.normalized.path}`);
+    // FIXME use the async version of exec, but that means a new dependecy
+    const command = `ogr2ogr -f "PostgreSQL" PG:"host=${dbConfig.host} user=${dbConfig.user} password=${dbConfig.password} dbname=${dbConfig.database}" ${source.destinations.normalized.path} -nln tree_staging -geomfield geom -append`
+    exec(command, async (error, stdout, stderr) => {
+        if (error) {
+            console.log(`error: ${error.message}`);
+            return;
+        }
+        if (stderr) {
+            console.log(`stderr: ${stderr}`);
+            return;
+        }
+
+        console.log(`Running the stored proc for ${source.id}`);
+        const result = await db.proc('public.merge_treedata', [source.id]);
+        console.log(result);
+        console.log(`stdout: ${stdout}`);
+    });
+    /*
+    */
+
+  const context = {
+    source,
+    nullGeometry: 0,
+    invalidGeometry: 0,
+  };
+
+  return context;
+};
+
+
+export const mergeSources = async (list) => {
+  if (process.argv.length > 2) {
+    list = list.filter(m => m.id == process.argv[2]);
+  }
+
+  const limit = pLimit(5);
+  const promises = list.map((source) =>
+    limit(() => mergeSource(source))
+  );
+  const results = await Promise.allSettled(promises);
+  console.log("Finished uploading to the database...");
+  results.forEach((l) => {
+    if (l && l.forEach) {
+      l.forEach(console.log);
+    } else {
+      console.log(l);
+    }
+  });
+};