Merge pull request #38 from developmentseed/new-pipeline

New changeset parsing pipeline

Author: sunu

.github/workflows/pipeline.yaml

@@ -0,0 +1,63 @@
+name: Hourly Pipeline
+
+on:
+  workflow_dispatch:
+  schedule:
+  - cron: "50 * * * *"
+
+
+jobs:
+  generate-hourly-fgb:
+    runs-on: ubuntu-latest
+
+    steps:
+        - name: Checkout code
+          uses: actions/checkout@v2
+
+        - name: Build docker image
+          run: docker build -t gradient-pipeline ./pipeline
+
+        - name: set date
+          id: set_date
+          run: |
+            DATE_TWO_HOURS_AGO=$(date -d "2 hours ago" +%Y-%m-%d)
+            echo "DATE_TWO_HOURS_AGO=$DATE_TWO_HOURS_AGO" >> $GITHUB_OUTPUT
+        
+        - name: set hour
+          id: set_hour
+          run: |
+            HOUR_TWO_HOURS_AGO=$(date -d "2 hours ago" +%H)
+            echo "HOUR_TWO_HOURS_AGO=$HOUR_TWO_HOURS_AGO" >> $GITHUB_OUTPUT
+        
+        - name: Check if the date and hour are set
+          run: cat $GITHUB_OUTPUT
+        
+        - name: process changesets for the hour ${{ steps.set_hour.outputs.HOUR_TWO_HOURS_AGO }}:00 on ${{ steps.set_date.outputs.DATE_TWO_HOURS_AGO }}
+          run: |
+            docker run -v ./data:/tmp gradient-pipeline sh -c "node cli.js process-hour $DATE_TWO_HOURS_AGO $HOUR_TWO_HOURS_AGO"
+          env:
+            DATE_TWO_HOURS_AGO: ${{ steps.set_date.outputs.DATE_TWO_HOURS_AGO }}
+            HOUR_TWO_HOURS_AGO: ${{ steps.set_hour.outputs.HOUR_TWO_HOURS_AGO }}
+        
+        - name: convert to fgb for the hour ${{ steps.set_hour.outputs.HOUR_TWO_HOURS_AGO }}:00 on ${{ steps.set_date.outputs.DATE_TWO_HOURS_AGO }}
+          run: |
+            docker run -v ./data:/tmp gradient-pipeline sh -c "ogr2ogr -f FlatGeobuf /tmp/${DATE_TWO_HOURS_AGO}T${HOUR_TWO_HOURS_AGO}:00.fgb /tmp/${DATE_TWO_HOURS_AGO}T${HOUR_TWO_HOURS_AGO}:00.geojson -skipfailures"
+          env:
+            DATE_TWO_HOURS_AGO: ${{ steps.set_date.outputs.DATE_TWO_HOURS_AGO }}
+            HOUR_TWO_HOURS_AGO: ${{ steps.set_hour.outputs.HOUR_TWO_HOURS_AGO }}
+        
+        - name: Check if the pipeline ran successfully
+          run: ls -lh ./data/*.fgb
+
+        - name: Authenticate with GCP
+          uses: google-github-actions/auth@v2
+          with:
+            credentials_json: ${{ secrets.GCS_SERVICE_ACCOUNT_KEY }}
+        
+        - name: Upload to GCS
+          uses: google-github-actions/upload-cloud-storage@v2
+          with:
+            path: data
+            destination: osm-tardis
+            parent: false
+            glob: '*.fgb'

.gitignore

@@ -2,4 +2,8 @@ node_modules/
 .yarn
 out
 package-lock.json
-.pnp*
+.pnp*
+
+pipeline/node_modules
+pipeline/.yarn
+pipeline/data

pipeline/Dockerfile

@@ -1,33 +1,15 @@
-# Start with an official Node image
-FROM node:18
+FROM ghcr.io/osgeo/gdal:alpine-small-latest
 
-# 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/*
+RUN apk add --no-cache nodejs yarn git
 
-# Create an app directory to hold the application code inside the image
-WORKDIR /usr/src/app
+WORKDIR /app
 
-# Copy your package.json and package-lock.json (if you have one) into the container
-COPY package*.json ./
+COPY package.json ./
+COPY yarn.lock ./
 
-# Install your Node dependencies
-RUN npm install
+RUN yarn 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" ]
+COPY src/ ./src/
+COPY cli.js ./
+RUN chmod +x cli.js

pipeline/cli.js

@@ -0,0 +1,36 @@
+#!/usr/bin/env node
+
+const { program } = require('commander');
+
+const { getHourlyReplicationFileURL, getChangesetIDs } = require('./src/utils');
+const { processChangesets, processChangeset } = require('./src/process');
+
+/**
+ * Runs the process to retrieve and process changesets for a given date and hour.
+ * @param {string} date - The date in the format 'YYYY-MM-DD'.
+ * @param {number} hour - The hour of the day (0-23).
+ * @returns {void}
+ */
+async function run(date, hour) {
+    let url = getHourlyReplicationFileURL(date, hour);
+    let changesets = await getChangesetIDs(url);
+    // changesets = changesets.slice(0, 2);
+    // console.log(changesets);
+    processChangesets(changesets, date, hour);
+}
+
+async function processSingleChangeset(changeset) {
+    await processChangeset(changeset);
+}
+
+program
+    .command('process-hour <date> <hour>')
+    .description('Process an hour of changesets starting from a given date and hour (in UTC) and combine the changed features into a single GeoJSON file.')
+    .action(run);
+
+program
+    .command('process-changeset <changeset>')
+    .description('Process a single changeset and save the features to a JSON file - for debugging purposes.')
+    .action(processSingleChangeset);
+
+program.parseAsync(process.argv);

pipeline/fetchOsc.js

@@ -1,18 +1,17 @@
 {
-  "name": "osmgradient",
+  "name": "osm-gradient-pipeline",
   "version": "1.0.0",
   "description": "Minutely metrics for OSM using FlatGeoBuff",
-  "main": "index.js",
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1"
   },
   "author": "",
   "license": "ISC",
   "dependencies": {
     "axios": "^1.5.0",
-    "osm-adiff-parser": "^1.1.0",
+    "commander": "12.0.0",
     "real-changesets-parser": "https://github.com/developmentseed/real-changesets-parser.git",
-    "serve": "^14.2.1"
+    "sax": "^1.3.0"
   },
   "devDependencies": {
     "geojson-validation": "^1.0.2"

pipeline/package.json

@@ -0,0 +1,5 @@
+const config = {
+    DATA_PATH: process.env.DATA_PATH || '/tmp',
+};
+
+module.exports = config;