Merge pull request #7 from developmentseed/process-pipeline

Add process file and Docker run for batch

Author: kamicut

.yarnrc.yml

@@ -0,0 +1 @@
+nodeLinker: node-modules

README.md

@@ -3,4 +3,4 @@
 Visualize statistics and browse OSM diffs over a map.
 
 - [Front-end app](web-vite/README.md)
-- Pipeline
+- [Pipeline](pipeline/README.md)

parser.js

@@ -0,0 +1,33 @@
+# Start with an official Node image
+FROM node:18
+
+# Update the system and install necessary dependencies
+RUN apt-get update && \
+  apt-get install -y parallel jq && \
+  apt-get clean && \
+  rm -rf /var/lib/apt/lists/*
+
+# Create an app directory to hold the application code inside the image
+WORKDIR /usr/src/app
+
+# Copy your package.json and package-lock.json (if you have one) into the container
+COPY package*.json ./
+
+# Install your Node dependencies
+RUN npm install
+
+# Copy your Node scripts into the container
+COPY fetchOsc.js ./
+COPY parser.js ./
+
+# The main script to run the tasks
+COPY process.sh ./
+
+# Install geojson-merge
+RUN npm install -g geojson-merge
+
+# Give execute permissions to the script
+RUN chmod +x process.sh
+
+# The command to run when the container starts
+ENTRYPOINT [ "./process.sh" ]

pipeline/Dockerfile

@@ -0,0 +1,9 @@
+## Usage
+
+Generate the past 10 minutes of OSM data starting at a sequence number (5652386 in this example)
+
+```
+docker build osm-tardis .
+docker run osm-tardis 5652386 10
+docker run -it -v ./data:/tmp ghcr.io/osgeo/gdal:alpine-small-latest sh ogr2ogr -f "FlatGeobuf" /tmp/test.fgb /tmp/2023-09-01T01_00.geojsonld -skipfailures
+```

pipeline/README.md

@@ -0,0 +1,29 @@
+const maxRetries = 10;
+const retryInterval = 5000;
+const fs = require("fs");
+const axios = require("axios");
+
+const OSC_SOURCE_URL =
+  "http://s3-eu-west-1.amazonaws.com/overpass-db-eu-west-1/augmented-diffs";
+
+const sequenceNumber = process.argv[2];
+const outputFileName = process.argv[3];
+const url = `${OSC_SOURCE_URL}/${sequenceNumber}.osc`;
+
+async function downloadFile(url, outputFileName) {
+  const oscFile = fs.createWriteStream(outputFileName);
+  try {
+    const res = await axios({
+      method: "get",
+      url,
+      responseType: "stream",
+    });
+    res.data.pipe(oscFile);
+  } catch (err) {
+    console.error(err);
+  }
+}
+
+downloadFile(url, outputFileName).then(() => {
+  console.log(`Downloaded ${url} to ${outputFileName}`);
+});

pipeline/fetchOsc.js

@@ -9,7 +9,12 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
+    "axios": "^1.5.0",
     "osm-adiff-parser": "^1.1.0",
-    "real-changesets-parser": "https://github.com/developmentseed/real-changesets-parser.git"
+    "real-changesets-parser": "https://github.com/developmentseed/real-changesets-parser.git",
+    "serve": "^14.2.1"
+  },
+  "devDependencies": {
+    "geojson-validation": "^1.0.2"
   }
 }

package.json renamed to pipeline/package.json

@@ -0,0 +1,31 @@
+const adiffParser = require("osm-adiff-parser");
+const changesetParser = require("real-changesets-parser");
+const fs = require("fs");
+
+// Read the file name from the command-line argument
+const fileName = process.argv[2];
+
+// Read the file
+fs.readFile(fileName, "utf8", (err, data) => {
+  if (err) {
+    console.error("Error reading the file:", err);
+    return;
+  }
+  let ranOnce = false;
+  adiffParser(data, null, (err, result) => {
+    if (ranOnce) {
+      return;
+    }
+    ranOnce = true;
+    Object.keys(result).forEach((changesetId) => {
+      result[changesetId].forEach((element) => {
+        const change = changesetParser.elementParser(element);
+
+        change.forEach((feature) => {
+          // do some validation here
+          console.log(JSON.stringify(feature));
+        });
+      });
+    });
+  });
+});

pipeline/parser.js

@@ -0,0 +1,66 @@
+#!/bin/bash
+
+# Check if the necessary arguments were provided
+if [[ -z "$1" ]] || [[ -z "$2" ]]; then
+    echo "Usage: $0 <date> <hour> [num_files]"
+    exit 1
+fi
+
+# Starting number provided as the first argument
+start_date=$1
+
+start_hour=$2
+
+# Add a leading zero if the hour is less than 10
+[[ $start_hour -lt 10 && $start_hour -ge 0 ]] && start_hour="0$start_hour"
+
+# Number of files to fetch
+num_files=${3:-60}
+if (( num_files < 1 )); then
+    num_files=1
+fi
+if (( num_files > 60 )); then
+    num_files=60
+fi
+
+echo "Number of files to fetch: ${num_files}"
+
+start_num=$(( $(date -d "${start_date}T${start_hour}:00:00Z" +%s)/60 - 22457216))
+
+end_num=$((start_num + num_files - 1))
+
+if [[ $start_num != $end_num ]]; then
+    echo "Fetching files from number ${start_num} to ${end_num}"
+else
+    echo "Fetching file number ${start_num}"
+fi
+
+# Fetch .osc files using fetchOsc.js
+fetchOscFunction() {
+    sequence_num=$1
+    file_path="/tmp/${sequence_num}.osc"
+    node fetchOsc.js $sequence_num $file_path
+}
+
+export -f fetchOscFunction
+
+seq $start_num $end_num | parallel fetchOscFunction
+
+# Parse .osc files and collect into a single geojson using parser.js and jq
+echo 'Parsing files'
+parallel -j +0 "
+    echo {};
+    file_path=\"/tmp/{}.osc\";
+    node parser.js \$file_path > \"/tmp/${start_date}-${start_hour}-{}.geojsonld\"
+" ::: $(seq $start_num $((end_num-1)))
+
+echo 'Done parsing files'
+
+echo "Combining files"
+output_file="/tmp/${start_date}T${start_hour}:00.geojsonld"
+cat /tmp/${start_date}-${start_hour}-*.geojsonld > $output_file
+echo "Done combining files"
+
+echo 'Replace nulls with 0'
+sed -i 's/null/0/g' $output_file
+echo 'Done replacing nulls with 0'