merge-trees-to-database

bin/merge.js

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

package.json

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

db/db-config.js → src/db/db-config.js

@@ -1,7 +1,7 @@
 /**
  * https://github.com/vitaly-t/pg-promise/wiki/Connection-Syntax
  */
- const dbConfig = {
+export const dbConfig = {
   host: process.env.POSTGRES_HOST,
   port: process.env.POSTGRES_PORT,
   database: process.env.POSTGRES_DB,
@@ -11,4 +11,4 @@
 
 // console.log('dbConfig',dbConfig, process.env);
 
-module.exports = dbConfig;
+//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);
+

db/pg-promise-config.js → src/db/pg-promise-config.js

@@ -1,6 +1,6 @@
 /* eslint-disable guard-for-in */
 /* eslint-disable no-restricted-syntax */
-const pgp = require('pg-promise');
+import pgp from 'pg-promise';
 
 /**
  * source:
@@ -20,9 +20,8 @@ function camelizeColumns(data) {
   }
 }
 
-const pgPromiseConfig = {
+export const pgPromiseConfig = {
   capSQL: true,
   receive: (data) => camelizeColumns(data),
 };
 
-module.exports = pgPromiseConfig;

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 concatenate from "./stages/concatenate.js";
 import * as tile from "./stages/tile.js";
 import * as upload from "./stages/upload.js";
@@ -22,6 +23,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);
+    }
+  });
+};