diff --git a/pyproject.toml b/pyproject.toml index 5ace62e..f91e629 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,6 +21,7 @@ dependencies = [ "anthropic>=0.84.0", "click>=8.3.1", "geopandas>=1.1.2", + "httpx>=0.27", "laser-generic>=1.0.1", "matplotlib>=3.10.8", "openai>=2.24.0", @@ -48,6 +49,7 @@ dev = [ [project.scripts] laser-init = "laser.init.cli:cli" +laser-generate = "laser.init.generate:main" [tool.ruff] line-length = 100 diff --git a/services/AKS/DEPLOYMENT.md b/services/AKS/DEPLOYMENT.md new file mode 100644 index 0000000..006dc24 --- /dev/null +++ b/services/AKS/DEPLOYMENT.md @@ -0,0 +1,135 @@ +# Geodata Services — AKS Deployment Guide + +All five geodata microservices deploy into the existing `laser-ai` namespace. +Each gets its own LoadBalancer IP (matching the jenner-mcp pattern). +Each service mounts a PVC at `/cache` for persistent raster/shapefile storage +that survives pod restarts and redeployments. + +## Services + +| Name | Image | Port | Cache PVC | Notes | +|---|---|---|---|---| +| `laser-unwpp-svc` | `laser-unwpp-service` | 80 | 2 Gi | WPP CSVs; downloads on startup | +| `laser-gadm-svc` | `laser-gadm-service` | 80 | 20 Gi | GADM GDB per country; on-demand | +| `laser-geoboundaries-svc` | `laser-geoboundaries-service` | 80 | 10 Gi | geoBoundaries ZIPs; on-demand | +| `laser-unocha-svc` | `laser-unocha-service` | 80 | 5 Gi | UNOCHA global GDB (~2 GB); on first request | +| `laser-worldpop-svc` | `laser-worldpop-service` | 80 | 100 Gi | WorldPop rasters; on-demand or pre-warmed | + +Registry: `idm-docker-staging.packages.idmod.org/` + +--- + +## Quick Start + +### 1. Push images + +```bash +cd services/ +make push-all-geodata TAG=v0.5.1a +``` + +### 2. Deploy PVCs, Deployments, and Services + +```bash +make k8s-deploy +# or directly: +kubectl apply -f AKS/geodata-services.yaml --kubeconfig AKS/kube.conf +``` + +### 3. Wait for all pods to be Ready + +```bash +make k8s-status +# or: +kubectl get deployment,svc,pvc -n laser-ai -l app.kubernetes.io/part-of=laser-geodata \ + --kubeconfig AKS/kube.conf +``` + +### 4. Pre-warm the WorldPop raster cache (optional but recommended) + +Wait for `laser-worldpop-svc` to be Ready, then: + +```bash +make k8s-prewarm +# or directly: +kubectl apply -f AKS/worldpop-prewarm-job.yaml --kubeconfig AKS/kube.conf + +# Monitor progress (~50 countries, 4 parallel workers): +kubectl logs -f job/laser-worldpop-prewarm -n laser-ai --kubeconfig AKS/kube.conf +``` + +The Job downloads ~50 commonly-used country rasters (~2–500 MB each) in parallel. +Total time ~15–30 min depending on network. Completes once, then stays in a +terminal state (safe to leave or delete). + +--- + +## Get service IPs + +After deployment, retrieve LoadBalancer IPs: + +```bash +kubectl get svc -n laser-ai -l app.kubernetes.io/part-of=laser-geodata \ + --kubeconfig AKS/kube.conf \ + -o custom-columns='NAME:.metadata.name,IP:.status.loadBalancer.ingress[0].ip' +``` + +Use those IPs in `generate.py`: + +```bash +python services/generate.py NGA 1 2015 2020 \ + --shape-source unocha \ + --shapes-url http:// \ + --worldpop-url http:// \ + --unwpp-url http:// \ + --emit-scripts +``` + +--- + +## Update a service + +```bash +make push-worldpop TAG=v0.5.2a +kubectl set image deployment/laser-worldpop-svc \ + laser-worldpop-svc=idm-docker-staging.packages.idmod.org/laser-worldpop-service:v0.5.2a \ + -n laser-ai --kubeconfig AKS/kube.conf +kubectl rollout status deployment/laser-worldpop-svc -n laser-ai --kubeconfig AKS/kube.conf +``` + +## Rollback + +```bash +kubectl rollout undo deployment/laser-worldpop-svc -n laser-ai --kubeconfig AKS/kube.conf +``` + +## Tail logs + +```bash +kubectl logs -f deployment/laser-worldpop-svc -n laser-ai --kubeconfig AKS/kube.conf +``` + +## Tear down (preserves PVCs and cached data) + +```bash +kubectl delete deployments,services \ + laser-unwpp-svc laser-gadm-svc laser-geoboundaries-svc laser-unocha-svc laser-worldpop-svc \ + -n laser-ai --kubeconfig AKS/kube.conf +``` + +To also delete the cached data (irreversible): + +```bash +kubectl delete pvc \ + laser-unwpp-cache laser-gadm-cache laser-geoboundaries-cache \ + laser-unocha-cache laser-worldpop-cache \ + -n laser-ai --kubeconfig AKS/kube.conf +``` + +--- + +## Image Tagging Convention + +Matches the jenner-mcp pattern: `v0..` + +Examples: `v0.5.1a`, `v0.5.7b` diff --git a/services/AKS/geodata-services.yaml b/services/AKS/geodata-services.yaml new file mode 100644 index 0000000..a7a0ea2 --- /dev/null +++ b/services/AKS/geodata-services.yaml @@ -0,0 +1,444 @@ +# geodata-services.yaml — PVCs + Deployments + LoadBalancer Services +# +# Deploys all 5 geodata microservices into the laser-ai namespace. +# +# Deploy: +# kubectl apply -f AKS/geodata-services.yaml --kubeconfig AKS/kube.conf +# +# Push images first: +# make push-all-geodata TAG=v0.5.1a (from services/ directory) + +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: laser-unwpp-cache + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + accessModes: [ReadWriteOnce] + storageClassName: managed-csi + resources: + requests: + storage: 2Gi +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: laser-gadm-cache + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + accessModes: [ReadWriteOnce] + storageClassName: managed-csi + resources: + requests: + storage: 20Gi +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: laser-geoboundaries-cache + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + accessModes: [ReadWriteOnce] + storageClassName: managed-csi + resources: + requests: + storage: 10Gi +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: laser-unocha-cache + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + accessModes: [ReadWriteOnce] + storageClassName: managed-csi + resources: + requests: + storage: 5Gi +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: laser-worldpop-cache + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + accessModes: [ReadWriteOnce] + storageClassName: managed-csi + resources: + requests: + storage: 100Gi + +# ── unwpp-service ───────────────────────────────────────────────────────────── + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: laser-unwpp-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + replicas: 1 + revisionHistoryLimit: 3 + strategy: + type: Recreate + selector: + matchLabels: + app: laser-unwpp-svc + template: + metadata: + labels: + app: laser-unwpp-svc + app.kubernetes.io/part-of: laser-geodata + spec: + imagePullSecrets: + - name: artifactory-login + containers: + - name: laser-unwpp-svc + image: idm-docker-staging.packages.idmod.org/laser-unwpp-service:v0.5.1a + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: CACHE_DIR + value: /cache + resources: + requests: + cpu: "500m" + memory: "3Gi" + limits: + cpu: "2" + memory: "6Gi" + volumeMounts: + - name: cache + mountPath: /cache + readinessProbe: + httpGet: + path: /health + port: 8000 + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 5 + volumes: + - name: cache + persistentVolumeClaim: + claimName: laser-unwpp-cache +--- +apiVersion: v1 +kind: Service +metadata: + name: laser-unwpp-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + type: LoadBalancer + ports: + - name: http + port: 80 + targetPort: 8000 + selector: + app: laser-unwpp-svc + +# ── gadm-service ────────────────────────────────────────────────────────────── + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: laser-gadm-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + replicas: 1 + revisionHistoryLimit: 3 + strategy: + type: Recreate + selector: + matchLabels: + app: laser-gadm-svc + template: + metadata: + labels: + app: laser-gadm-svc + app.kubernetes.io/part-of: laser-geodata + spec: + imagePullSecrets: + - name: artifactory-login + containers: + - name: laser-gadm-svc + image: idm-docker-staging.packages.idmod.org/laser-gadm-service:v0.3.2a + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: CACHE_DIR + value: /cache + resources: + requests: + cpu: "250m" + memory: "1Gi" + limits: + cpu: "2" + memory: "2Gi" + volumeMounts: + - name: cache + mountPath: /cache + readinessProbe: + httpGet: + path: /health + port: 8000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + volumes: + - name: cache + persistentVolumeClaim: + claimName: laser-gadm-cache +--- +apiVersion: v1 +kind: Service +metadata: + name: laser-gadm-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + type: LoadBalancer + ports: + - name: http + port: 80 + targetPort: 8000 + selector: + app: laser-gadm-svc + +# ── geoboundaries-service ───────────────────────────────────────────────────── + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: laser-geoboundaries-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + replicas: 1 + revisionHistoryLimit: 3 + strategy: + type: Recreate + selector: + matchLabels: + app: laser-geoboundaries-svc + template: + metadata: + labels: + app: laser-geoboundaries-svc + app.kubernetes.io/part-of: laser-geodata + spec: + imagePullSecrets: + - name: artifactory-login + containers: + - name: laser-geoboundaries-svc + image: idm-docker-staging.packages.idmod.org/laser-geoboundaries-service:v0.5.1a + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: CACHE_DIR + value: /cache + resources: + requests: + cpu: "250m" + memory: "512Mi" + limits: + cpu: "2" + memory: "2Gi" + volumeMounts: + - name: cache + mountPath: /cache + readinessProbe: + httpGet: + path: /health + port: 8000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + volumes: + - name: cache + persistentVolumeClaim: + claimName: laser-geoboundaries-cache +--- +apiVersion: v1 +kind: Service +metadata: + name: laser-geoboundaries-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + type: LoadBalancer + ports: + - name: http + port: 80 + targetPort: 8000 + selector: + app: laser-geoboundaries-svc + +# ── unocha-service ──────────────────────────────────────────────────────────── + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: laser-unocha-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + replicas: 1 + revisionHistoryLimit: 3 + strategy: + type: Recreate + selector: + matchLabels: + app: laser-unocha-svc + template: + metadata: + labels: + app: laser-unocha-svc + app.kubernetes.io/part-of: laser-geodata + spec: + imagePullSecrets: + - name: artifactory-login + containers: + - name: laser-unocha-svc + image: idm-docker-staging.packages.idmod.org/laser-unocha-service:v0.5.1a + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: CACHE_DIR + value: /cache + resources: + requests: + cpu: "250m" + memory: "1Gi" + limits: + cpu: "2" + memory: "3Gi" + volumeMounts: + - name: cache + mountPath: /cache + readinessProbe: + httpGet: + path: /health + port: 8000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + volumes: + - name: cache + persistentVolumeClaim: + claimName: laser-unocha-cache +--- +apiVersion: v1 +kind: Service +metadata: + name: laser-unocha-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + type: LoadBalancer + ports: + - name: http + port: 80 + targetPort: 8000 + selector: + app: laser-unocha-svc + +# ── worldpop-service ────────────────────────────────────────────────────────── + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: laser-worldpop-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + replicas: 1 + revisionHistoryLimit: 3 + strategy: + type: Recreate + selector: + matchLabels: + app: laser-worldpop-svc + template: + metadata: + labels: + app: laser-worldpop-svc + app.kubernetes.io/part-of: laser-geodata + spec: + imagePullSecrets: + - name: artifactory-login + containers: + - name: laser-worldpop-svc + image: idm-docker-staging.packages.idmod.org/laser-worldpop-service:v0.5.12a + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: CACHE_DIR + value: /cache + resources: + requests: + cpu: "500m" + memory: "2Gi" + limits: + cpu: "4" + memory: "4Gi" + volumeMounts: + - name: cache + mountPath: /cache + readinessProbe: + httpGet: + path: /health + port: 8000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + volumes: + - name: cache + persistentVolumeClaim: + claimName: laser-worldpop-cache +--- +apiVersion: v1 +kind: Service +metadata: + name: laser-worldpop-svc + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + type: LoadBalancer + ports: + - name: http + port: 80 + targetPort: 8000 + selector: + app: laser-worldpop-svc diff --git a/services/AKS/worldpop-prewarm-job.yaml b/services/AKS/worldpop-prewarm-job.yaml new file mode 100644 index 0000000..8f23d57 --- /dev/null +++ b/services/AKS/worldpop-prewarm-job.yaml @@ -0,0 +1,52 @@ +# worldpop-prewarm-job.yaml — one-shot Job that pre-populates the worldpop PVC. +# +# Run AFTER laser-worldpop-svc is Ready. The Job calls POST /prewarm/{iso} for +# each country in the default humanitarian list (~50 ISOs, 4 workers). +# Re-applying is safe: completed jobs are not re-run; delete and re-apply to retry. +# +# Apply: +# kubectl apply -f AKS/worldpop-prewarm-job.yaml --kubeconfig AKS/kube.conf +# +# Monitor: +# kubectl logs -f job/laser-worldpop-prewarm -n laser-ai --kubeconfig AKS/kube.conf +# +# Delete once complete (optional): +# kubectl delete job laser-worldpop-prewarm -n laser-ai --kubeconfig AKS/kube.conf + +apiVersion: batch/v1 +kind: Job +metadata: + name: laser-worldpop-prewarm + namespace: laser-ai + labels: + app.kubernetes.io/part-of: laser-geodata +spec: + # Do not restart on success; retry up to 2× on failure. + backoffLimit: 2 + template: + spec: + imagePullSecrets: + - name: artifactory-login + restartPolicy: OnFailure + containers: + - name: prewarm + image: idm-docker-staging.packages.idmod.org/laser-worldpop-service:v0.5.12a + imagePullPolicy: Always + # Override the default uvicorn CMD with the prewarm script. + command: ["python", "-u", "prewarm_countries.py"] + args: + - "--url" + - "http://laser-worldpop-svc" + - "--year" + - "2020" + - "--workers" + - "4" + resources: + requests: + cpu: "250m" + memory: "256Mi" + limits: + cpu: "1" + memory: "512Mi" + # activeDeadlineSeconds: 7200 on the Job spec handles overall timeout. + activeDeadlineSeconds: 7200 diff --git a/services/Makefile b/services/Makefile new file mode 100644 index 0000000..bdc8676 --- /dev/null +++ b/services/Makefile @@ -0,0 +1,215 @@ +CACHE_ROOT ?= $(HOME)/.laser/cache +PYTHON ?= python3.11 + +# ── unwpp-service (port 8100) ────────────────────────────────────────────────── + +build-unwpp: + docker build -t laser-unwpp-service unwpp/ + +run-unwpp: build-unwpp + @mkdir -p $(CACHE_ROOT)/unwpp + docker run -d \ + --name laser-unwpp-service \ + -p 8100:8000 \ + -v $(CACHE_ROOT)/unwpp:/cache \ + laser-unwpp-service + @echo "unwpp-service running on http://localhost:8100" + +stop-unwpp: + docker stop laser-unwpp-service 2>/dev/null || true + docker rm laser-unwpp-service 2>/dev/null || true + +restart-unwpp: stop-unwpp run-unwpp + +logs-unwpp: + docker logs -f laser-unwpp-service + +test-unwpp: + $(PYTHON) unwpp/test_client.py --url http://localhost:8100 + +# ── gadm-service (port 8101) ─────────────────────────────────────────────────── + +build-gadm: + docker build -t laser-gadm-service gadm/ + +run-gadm: build-gadm + @mkdir -p $(CACHE_ROOT)/gadm + docker run -d \ + --name laser-gadm-service \ + -p 8101:8000 \ + -v $(CACHE_ROOT)/gadm:/cache \ + laser-gadm-service + @echo "gadm-service running on http://localhost:8101" + +stop-gadm: + docker stop laser-gadm-service 2>/dev/null || true + docker rm laser-gadm-service 2>/dev/null || true + +restart-gadm: stop-gadm run-gadm + +logs-gadm: + docker logs -f laser-gadm-service + +test-gadm: + $(PYTHON) gadm/test_client.py --url http://localhost:8101 + +# ── geoboundaries-service (port 8102) ───────────────────────────────────────── + +build-geoboundaries: + docker build -t laser-geoboundaries-service geoboundaries/ + +run-geoboundaries: build-geoboundaries + @mkdir -p $(CACHE_ROOT)/geoboundaries + docker run -d \ + --name laser-geoboundaries-service \ + -p 8102:8000 \ + -v $(CACHE_ROOT)/geoboundaries:/cache \ + laser-geoboundaries-service + @echo "geoboundaries-service running on http://localhost:8102" + +stop-geoboundaries: + docker stop laser-geoboundaries-service 2>/dev/null || true + docker rm laser-geoboundaries-service 2>/dev/null || true + +restart-geoboundaries: stop-geoboundaries run-geoboundaries + +logs-geoboundaries: + docker logs -f laser-geoboundaries-service + +test-geoboundaries: + $(PYTHON) geoboundaries/test_client.py --url http://localhost:8102 + +# ── unocha-service (port 8103) ──────────────────────────────────────────────── + +build-unocha: + docker build -t laser-unocha-service unocha/ + +run-unocha: build-unocha + @mkdir -p $(CACHE_ROOT)/unocha + docker run -d \ + --name laser-unocha-service \ + -p 8103:8000 \ + -v $(CACHE_ROOT)/unocha:/cache \ + laser-unocha-service + @echo "unocha-service running on http://localhost:8103" + @echo "Note: first run downloads ~1-2 GB global GDB — use 'make logs-unocha' to monitor." + +stop-unocha: + docker stop laser-unocha-service 2>/dev/null || true + docker rm laser-unocha-service 2>/dev/null || true + +restart-unocha: stop-unocha run-unocha + +logs-unocha: + docker logs -f laser-unocha-service + +test-unocha: + $(PYTHON) unocha/test_client.py --url http://localhost:8103 --wait + +# ── worldpop-service (port 8104) ────────────────────────────────────────────── + +build-worldpop: + docker build -t laser-worldpop-service worldpop/ + +run-worldpop: build-worldpop + @mkdir -p $(CACHE_ROOT)/worldpop + docker run -d \ + --name laser-worldpop-service \ + -p 8104:8000 \ + -v $(CACHE_ROOT)/worldpop:/cache \ + laser-worldpop-service + @echo "worldpop-service running on http://localhost:8104" + @echo "Note: rasters are downloaded on demand via POST /prewarm/{iso} or POST /aggregate/{iso}." + +stop-worldpop: + docker stop laser-worldpop-service 2>/dev/null || true + docker rm laser-worldpop-service 2>/dev/null || true + +restart-worldpop: stop-worldpop run-worldpop + +logs-worldpop: + docker logs -f laser-worldpop-service + +test-worldpop: + $(PYTHON) worldpop/test_client.py --url http://localhost:8104 + +# ── all (local Docker) ──────────────────────────────────────────────────────── + +run-all: run-unwpp run-gadm run-geoboundaries run-unocha run-worldpop +stop-all: stop-unwpp stop-gadm stop-geoboundaries stop-unocha stop-worldpop + +# ── AKS: build + push to registry ──────────────────────────────────────────── +# Usage: make push-all-geodata TAG=v0.5.1a +# make push-worldpop TAG=v0.5.2a + +REGISTRY ?= idm-docker-staging.packages.idmod.org +TAG ?= latest +KUBECONFIG ?= AKS/kube.conf +KUBECTL = kubectl --kubeconfig $(KUBECONFIG) +NAMESPACE = laser-ai + +push-unwpp: + docker build -t $(REGISTRY)/laser-unwpp-service:$(TAG) unwpp/ + docker push $(REGISTRY)/laser-unwpp-service:$(TAG) + +push-gadm: + docker build -t $(REGISTRY)/laser-gadm-service:$(TAG) gadm/ + docker push $(REGISTRY)/laser-gadm-service:$(TAG) + +push-geoboundaries: + docker build -t $(REGISTRY)/laser-geoboundaries-service:$(TAG) geoboundaries/ + docker push $(REGISTRY)/laser-geoboundaries-service:$(TAG) + +push-unocha: + docker build -t $(REGISTRY)/laser-unocha-service:$(TAG) unocha/ + docker push $(REGISTRY)/laser-unocha-service:$(TAG) + +push-worldpop: + docker build -t $(REGISTRY)/laser-worldpop-service:$(TAG) worldpop/ + docker push $(REGISTRY)/laser-worldpop-service:$(TAG) + +push-all-geodata: push-unwpp push-gadm push-geoboundaries push-unocha push-worldpop + +# ── AKS: deploy / status / manage ──────────────────────────────────────────── + +k8s-deploy: + $(KUBECTL) apply -f AKS/geodata-services.yaml -n $(NAMESPACE) + +k8s-status: + $(KUBECTL) get deployment,svc,pvc -n $(NAMESPACE) \ + -l app.kubernetes.io/part-of=laser-geodata + +k8s-ips: + $(KUBECTL) get svc -n $(NAMESPACE) \ + -l app.kubernetes.io/part-of=laser-geodata \ + -o custom-columns='NAME:.metadata.name,IP:.status.loadBalancer.ingress[0].ip' + +k8s-prewarm: + $(KUBECTL) apply -f AKS/worldpop-prewarm-job.yaml -n $(NAMESPACE) + +k8s-logs-%: + $(KUBECTL) logs -f deployment/laser-$*-svc -n $(NAMESPACE) + +# Tear down Deployments + Services but preserve PVCs (cached data survives). +k8s-delete: + $(KUBECTL) delete deployment,service \ + laser-unwpp-svc laser-gadm-svc laser-geoboundaries-svc \ + laser-unocha-svc laser-worldpop-svc \ + -n $(NAMESPACE) --ignore-not-found + +# Also wipe PVCs (destroys all cached rasters/shapefiles — use with care). +k8s-delete-all: k8s-delete + $(KUBECTL) delete pvc \ + laser-unwpp-cache laser-gadm-cache laser-geoboundaries-cache \ + laser-unocha-cache laser-worldpop-cache \ + -n $(NAMESPACE) --ignore-not-found + +.PHONY: build-unwpp run-unwpp stop-unwpp restart-unwpp logs-unwpp test-unwpp \ + build-gadm run-gadm stop-gadm restart-gadm logs-gadm test-gadm \ + build-geoboundaries run-geoboundaries stop-geoboundaries \ + restart-geoboundaries logs-geoboundaries test-geoboundaries \ + build-unocha run-unocha stop-unocha restart-unocha logs-unocha test-unocha \ + build-worldpop run-worldpop stop-worldpop restart-worldpop logs-worldpop test-worldpop \ + run-all stop-all \ + push-unwpp push-gadm push-geoboundaries push-unocha push-worldpop push-all-geodata \ + k8s-deploy k8s-status k8s-ips k8s-prewarm k8s-delete k8s-delete-all diff --git a/services/gadm/Dockerfile b/services/gadm/Dockerfile new file mode 100644 index 0000000..49f8db1 --- /dev/null +++ b/services/gadm/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.11-slim + +WORKDIR /app + +# pyogrio bundles its own GDAL — no system GDAL install needed +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY main.py . + +ENV CACHE_DIR=/cache + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/gadm/README.md b/services/gadm/README.md new file mode 100644 index 0000000..f08f63d --- /dev/null +++ b/services/gadm/README.md @@ -0,0 +1,112 @@ +# gadm-service + +A lightweight HTTP microservice that serves [GADM 4.1](https://gadm.org) administrative boundary GeoJSON for use by `laser-generate` and the browser choropleth client. + +## Why it exists + +`laser-generate` needs administrative boundary polygons (shapes + names) as a prerequisite for population aggregation and model-grid construction. GADM provides high-quality, globally consistent boundaries at levels 0–4, but the raw shapefiles are large (hundreds of MB per country) and require GDAL/GeoPandas to parse. Running that machinery inside a shared microservice means: + +- The shapefile is downloaded and cached **once**, then served instantly to all clients. +- No GDAL installation is required on the user's machine. +- The browser choropleth client can fetch GeoJSON directly with a simple `fetch()` call. + +## API + +### `GET /health` + +Liveness check. Returns `{"status": "ok"}`. + +### `GET /boundaries/{ISO}/{level}` + +Returns a GeoJSON `FeatureCollection` of administrative boundaries. + +| Parameter | Description | +|-----------|-------------| +| `ISO` | ISO 3166-1 alpha-3 country code (case-insensitive, e.g. `NGA`, `nga`) | +| `level` | Admin level 0–5 (0 = country, 1 = province, 2 = district, …) | + +**Feature properties:** + +| Property | Description | +|----------|-------------| +| `nodeid` | Zero-based integer index (stable within a request, matches WorldPop service) | +| `name` | Human-readable region name | +| `gid` | GADM GID string (e.g. `NGA.1_1`) | + +**Geometry:** simplified to ~100 m (0.001°) for levels ≥ 1, which reduces payload size significantly for large countries while preserving topology at WorldPop raster resolution (~1 km). + +**Error responses:** + +| Status | Condition | +|--------|-----------| +| 400 | `level` outside 0–5 | +| 404 | GADM has no data for the requested ISO or admin level | + +**Example:** + +``` +GET /boundaries/NGA/1 +``` + +```json +{ + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": { "nodeid": 0, "name": "Abia", "gid": "NGA.1_1" }, + "geometry": { "type": "MultiPolygon", "coordinates": [...] } + }, + ... + ] +} +``` + +## Caching + +On first request for a country, the service downloads `gadm41_{ISO}_shp.zip` from `geodata.ucdavis.edu` (~1–500 MB depending on country) and caches it under `$CACHE_DIR/{ISO}/`. All admin levels for that country are served from the single cached zip — no re-download on subsequent requests or different level queries. + +The cache directory defaults to `/cache` in the container (backed by a Kubernetes PVC in AKS) and `~/.laser/cache/gadm` locally. + +## Running locally + +```bash +# Install dependencies +pip install -r requirements.txt + +# Start the service on port 8101 +CACHE_DIR=~/.laser/cache/gadm uvicorn main:app --host 0.0.0.0 --port 8101 + +# Run the test suite against it +python test_client.py --url http://localhost:8101 +``` + +First request for a country triggers the shapefile download. Luxembourg (`LUX`, ~1 MB) is used by the test suite for speed. + +## Docker + +```bash +# Build +docker build -t laser-gadm-service . + +# Run +docker run -p 8101:8000 -v gadm-cache:/cache laser-gadm-service +``` + +## AKS deployment + +Deployed as part of `services/AKS/geodata-services.yaml` in the `laser-ai` namespace. The deployment uses a 20 Gi PVC (`laser-gadm-cache`) for the shapefile cache and exposes a `LoadBalancer` service on port 80. + +```bash +# Deploy / update +kubectl apply -f services/AKS/geodata-services.yaml --kubeconfig services/AKS/kube.conf + +# Check status +kubectl get pods -n laser-ai -l app=laser-gadm-svc --kubeconfig services/AKS/kube.conf +``` + +## Known limitations + +- **BRA admin-2**: 5 572 municipalities. Geometry simplification keeps the response under 50 MB, but the request takes ~20 s on first read from zip. Subsequent requests are equally slow (no in-memory cache between requests). +- **Level 4–5**: Available only for countries where GADM provides sub-district data; most countries stop at level 2 or 3. +- **Single replica**: The deployment uses `strategy: Recreate` because the PVC is `ReadWriteOnce`. Scaling requires switching to a shared file store. diff --git a/services/gadm/client.html b/services/gadm/client.html new file mode 100644 index 0000000..1382196 --- /dev/null +++ b/services/gadm/client.html @@ -0,0 +1,150 @@ + + + + + gadm-service explorer + + + + + + + + + + +
+

gadm-service

+ + + + + + + + ready +
+ +
+ +
+

+
+
+ + + + diff --git a/services/gadm/main.py b/services/gadm/main.py new file mode 100644 index 0000000..2c6ab73 --- /dev/null +++ b/services/gadm/main.py @@ -0,0 +1,113 @@ +""" +gadm-service — serves GADM 4.1 admin boundary GeoJSON. + +GET /health +GET /boundaries/{ISO}/{admin_level} → GeoJSON FeatureCollection + +Downloads gadm41_{ISO}_shp.zip on first request for a country, caches it. +All admin levels for that country are then served from the single cached zip. +""" + +import logging +import os +import shutil +import tempfile +from pathlib import Path + +import geopandas as gpd +import httpx +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path.home() / ".laser" / "cache" / "gadm")) + +_GADM_BASE = "https://geodata.ucdavis.edu/gadm/gadm4.1/shp" + + +def _zip_path(iso: str) -> Path: + return CACHE_DIR / iso / f"gadm41_{iso}_shp.zip" + + +def _download(iso: str) -> Path: + dest = _zip_path(iso) + if dest.exists(): + logger.info("Cache hit: %s", dest) + return dest + + dest.parent.mkdir(parents=True, exist_ok=True) + url = f"{_GADM_BASE}/gadm41_{iso}_shp.zip" + logger.info("Downloading %s ...", url) + + # Write to a temp file first — avoids a partial file being treated as cached + tmp_fd, tmp_name = tempfile.mkstemp(dir=dest.parent, suffix=".tmp") + os.close(tmp_fd) + tmp = Path(tmp_name) + try: + with httpx.stream("GET", url, follow_redirects=True, timeout=600) as r: + if r.status_code == 404: + raise HTTPException(404, f"GADM has no data for ISO {iso!r}") + r.raise_for_status() + with tmp.open("wb") as f: + for chunk in r.iter_bytes(chunk_size=65_536): + f.write(chunk) + shutil.move(str(tmp), str(dest)) + except Exception: + tmp.unlink(missing_ok=True) + raise + + logger.info("Saved %s (%.1f MB)", dest, dest.stat().st_size / 1e6) + return dest + + +def _read(iso: str, level: int) -> gpd.GeoDataFrame: + zip_path = _download(iso) + layer = f"gadm41_{iso}_{level}" + logger.info("Reading %s from %s ...", layer, zip_path.name) + + try: + gdf = gpd.read_file(f"zip://{zip_path}!{layer}.shp", engine="pyogrio") + except Exception as exc: + raise HTTPException(404, f"Admin level {level} not available for {iso}: {exc}") from exc + + gdf["nodeid"] = range(len(gdf)) + + if level == 0: + gdf["name"] = gdf["GID_0"] + elif level <= 3: + gdf["name"] = gdf[f"NAME_{level}"] + else: + gdf["name"] = gdf["NAME_3"].astype(str) + ":" + gdf["NAME_4"].astype(str) + + gid_col = f"GID_{level}" + gdf = gdf[["nodeid", "name", gid_col, "geometry"]].rename(columns={gid_col: "gid"}) + + # Simplify to ~100 m (0.001°). Matches WorldPop raster resolution, removes + # the coordinate bloat that OOMkills the pod on large level-2+ countries. + if level >= 1: + gdf.geometry = gdf.geometry.simplify(0.001, preserve_topology=True) + + return gdf + + +app = FastAPI(title="gadm-service") +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"]) + + +@app.get("/health") +def health(): + return {"status": "ok"} + + +@app.get("/boundaries/{iso}/{level}") +def boundaries(iso: str, level: int): + if not 0 <= level <= 5: + raise HTTPException(400, f"admin_level must be 0–5, got {level}") + gdf = _read(iso.upper(), level) + # to_crs ensures WGS-84; set_crs first if the shapefile has no CRS metadata. + if gdf.crs is None: + gdf = gdf.set_crs("EPSG:4326") + return JSONResponse(content=gdf.to_crs("EPSG:4326").__geo_interface__) diff --git a/services/gadm/requirements.txt b/services/gadm/requirements.txt new file mode 100644 index 0000000..1e48c37 --- /dev/null +++ b/services/gadm/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.110 +uvicorn[standard]>=0.27 +httpx>=0.27 +geopandas>=0.14 +pyogrio>=0.7 diff --git a/services/gadm/test_client.py b/services/gadm/test_client.py new file mode 100644 index 0000000..f7fef9b --- /dev/null +++ b/services/gadm/test_client.py @@ -0,0 +1,125 @@ +""" +Test client for gadm-service. + +Usage: + python test_client.py [--url http://localhost:8101] + +Uses Luxembourg (LUX) — tiny shapefile (~1 MB) — for tests that trigger a +real download. Checks GeoJSON structure, property presence, and error cases. +""" + +import argparse +import sys + +import httpx + +BASE = "http://localhost:8101" +TEST_ISO = "LUX" # Luxembourg — small file, fast download + + +def get(path: str, timeout: float = 120) -> httpx.Response: + return httpx.get(f"{BASE}{path}", timeout=timeout) + + +def check(label: str, cond: bool, detail: str = "") -> None: + status = "PASS" if cond else "FAIL" + print(f" [{status}] {label}" + (f" — {detail}" if detail else "")) + if not cond: + sys.exit(1) + + +def test_health(): + print("── /health ─────────────────────────────────────────────") + r = get("/health") + check("HTTP 200", r.status_code == 200) + check("status == ok", r.json().get("status") == "ok") + + +def test_boundaries_level0(): + print(f"── GET /boundaries/{TEST_ISO}/0 (country outline) ─────") + r = get(f"/boundaries/{TEST_ISO}/0") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + fc = r.json() + check("type == FeatureCollection", fc.get("type") == "FeatureCollection") + features = fc.get("features", []) + check("exactly 1 feature at level 0", len(features) == 1, f"got {len(features)}") + props = features[0].get("properties", {}) + check("nodeid present", "nodeid" in props) + check("name present", "name" in props) + check("gid present", "gid" in props) + geom = features[0].get("geometry", {}) + check("geometry present", bool(geom)) + check("geometry type is Polygon/MultiPolygon", geom.get("type") in ("Polygon", "MultiPolygon")) + + +def test_boundaries_level1(): + print(f"── GET /boundaries/{TEST_ISO}/1 ────────────────────────") + r = get(f"/boundaries/{TEST_ISO}/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + features = r.json().get("features", []) + check("at least 1 feature", len(features) >= 1, f"got {len(features)}") + for f in features: + props = f.get("properties", {}) + check("each feature has nodeid", "nodeid" in props) + check("each feature has name", "name" in props) + break # spot-check first feature only + + +def test_boundaries_level2(): + print(f"── GET /boundaries/{TEST_ISO}/2 ────────────────────────") + r = get(f"/boundaries/{TEST_ISO}/2") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + features = r.json().get("features", []) + check("more features at level 2 than level 1", len(features) > 1, f"got {len(features)}") + + +def test_lowercase_iso(): + print(f"── GET /boundaries/lux/1 (lowercase ISO) ───────────────") + r = get("/boundaries/lux/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + + +def test_second_request_is_cached(): + print(f"── Second request uses cache (fast) ────────────────────") + import time + t0 = time.monotonic() + r = get(f"/boundaries/{TEST_ISO}/1") + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200) + check("responded in < 10s (cache hit)", elapsed < 10, f"{elapsed:.2f}s") + + +def test_invalid_level(): + print("── GET /boundaries/LUX/9 (invalid level) ───────────────") + r = get("/boundaries/LUX/9") + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def test_unknown_iso(): + print("── GET /boundaries/ZZZ/1 (unknown ISO) ─────────────────") + r = get("/boundaries/ZZZ/1") + check("HTTP 404", r.status_code == 404, f"got {r.status_code}") + + +def main(): + global BASE + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=BASE) + args = parser.parse_args() + BASE = args.url.rstrip("/") + + print(f"\nRunning gadm-service tests against {BASE}\n") + print(f"Note: first run downloads {TEST_ISO} shapefile — may take a moment.\n") + test_health() + test_boundaries_level0() + test_boundaries_level1() + test_boundaries_level2() + test_lowercase_iso() + test_second_request_is_cached() + test_invalid_level() + test_unknown_iso() + print("\nAll tests passed.") + + +if __name__ == "__main__": + main() diff --git a/services/generate.py b/services/generate.py new file mode 100644 index 0000000..9a36b5a --- /dev/null +++ b/services/generate.py @@ -0,0 +1,13 @@ +"""Thin shim — delegates to the installed laser-init package. + +Run directly: + python3 services/generate.py NGA 1 2020 2020 --emit-scripts + +Or use the installed entry point: + laser-generate NGA 1 2020 2020 --emit-scripts +""" + +from laser.init.generate import main + +if __name__ == "__main__": + main() diff --git a/services/geoboundaries/Dockerfile b/services/geoboundaries/Dockerfile new file mode 100644 index 0000000..377fddd --- /dev/null +++ b/services/geoboundaries/Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY main.py . + +ENV CACHE_DIR=/cache + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/geoboundaries/README.md b/services/geoboundaries/README.md new file mode 100644 index 0000000..4216225 --- /dev/null +++ b/services/geoboundaries/README.md @@ -0,0 +1,92 @@ +# geoboundaries-service + +A microservice that serves [geoBoundaries v6.0.0](https://www.geoboundaries.org) administrative boundary GeoJSON. It implements the same API contract as `gadm-service` and is interchangeable with it in `laser-generate` via `--shape-source geoboundaries`. + +## Why it exists + +geoBoundaries is a fully open, CC-BY licensed boundary dataset maintained by the College of William & Mary. It is an alternative to GADM for contexts where license terms matter or where geoBoundaries' boundary definitions better match operational humanitarian usage. This service provides: + +- On-demand download and caching of per-(ISO, level) boundary zips from GitHub. +- The same GeoJSON `FeatureCollection` response format as `gadm-service`, so the caller doesn't need to know which source was used. + +## API + +### `GET /health` + +Returns `{"status": "ok"}`. + +### `GET /boundaries/{ISO}/{level}` + +Returns a GeoJSON `FeatureCollection` of administrative boundaries. + +| Parameter | Description | +|-----------|-------------| +| `ISO` | ISO 3166-1 alpha-3 country code (e.g. `ETH`) | +| `level` | Admin level 0–5 | + +**Feature properties:** + +| Property | Description | +|----------|-------------| +| `nodeid` | Zero-based integer index | +| `name` | Region name (`shapeName` from geoBoundaries) | +| `gid` | geoBoundaries shape ID (`shapeID`) | + +**Error responses:** + +| Status | Condition | +|--------|-----------| +| 400 | `level` outside 0–5 | +| 404 | geoBoundaries has no data for the requested ISO / level | + +## How it differs from gadm-service + +| | gadm-service | geoboundaries-service | +|---|---|---| +| Source | GADM 4.1 | geoBoundaries v6.0.0 | +| License | Non-commercial | CC-BY (open) | +| Coverage | ~250 countries, up to level 5 | ~200+ countries, typically 0–3 | +| Cache unit | One zip per country (all levels) | One zip per (ISO, level) | +| Geometry simplification | Yes — 0.001° for level ≥ 1 | No | +| Boundary definitions | Academic/statistical | Humanitarian (COD-AB aligned) | + +geoBoundaries uses P-codes (`shapeID`) consistent with OCHA's Common Operational Datasets, making it the preferred source when integrating with humanitarian data pipelines. + +## Caching + +Each `(ISO, level)` combination is a separate zip file on GitHub. The service downloads and caches each on first request at `$CACHE_DIR/{ISO}/ADM{level}/geoBoundaries-{ISO}-ADM{level}-all.zip`. Subsequent requests for the same (ISO, level) are served from the cached zip. + +The cache directory defaults to `/cache` in the container (backed by a 10 Gi Kubernetes PVC in AKS) and `~/.laser/cache/geoboundaries` locally. + +## Running locally + +```bash +pip install -r requirements.txt + +CACHE_DIR=~/.laser/cache/geoboundaries uvicorn main:app --host 0.0.0.0 --port 8102 + +# Test +curl http://localhost:8102/boundaries/ETH/1 +``` + +## Docker + +```bash +docker build -t laser-geoboundaries-service . +docker run -p 8102:8000 -v geoboundaries-cache:/cache laser-geoboundaries-service +``` + +## AKS deployment + +Deployed as part of `services/AKS/geodata-services.yaml`. Uses a 10 Gi PVC (`laser-geoboundaries-cache`) and a `LoadBalancer` service on port 80. + +```bash +kubectl apply -f services/AKS/geodata-services.yaml --kubeconfig services/AKS/kube.conf +kubectl get pods -n laser-ai -l app=laser-geoboundaries-svc --kubeconfig services/AKS/kube.conf +``` + +## Known limitations + +- **No geometry simplification**: Unlike `gadm-service`, geometries are served at full resolution. For countries with very detailed coastlines or borders, responses can be large (tens of MB). +- **Level availability varies**: geoBoundaries only includes levels for which data has been validated. Many countries have data only at levels 0–2. +- **GitHub rate limits**: Downloads come from `github.com/wmgeolab/geoBoundaries`. Repeated cold-start requests from many clients could hit GitHub's unauthenticated rate limit. diff --git a/services/geoboundaries/main.py b/services/geoboundaries/main.py new file mode 100644 index 0000000..465a2e6 --- /dev/null +++ b/services/geoboundaries/main.py @@ -0,0 +1,94 @@ +""" +geoboundaries-service — serves geoBoundaries v6.0.0 admin boundary GeoJSON. + +GET /health +GET /boundaries/{ISO}/{admin_level} → GeoJSON FeatureCollection + +Each (ISO, level) combination is a separate zip on GitHub. Downloads on first +request and caches. Same endpoint contract as gadm-service. +""" + +import logging +import os +import shutil +import tempfile +from pathlib import Path + +import geopandas as gpd +import httpx +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path.home() / ".laser" / "cache" / "geoboundaries")) + +_GB_BASE = "https://github.com/wmgeolab/geoBoundaries/raw/refs/tags/v6.0.0/releaseData/gbOpen" + + +def _zip_path(iso: str, level: int) -> Path: + return CACHE_DIR / iso / f"ADM{level}" / f"geoBoundaries-{iso}-ADM{level}-all.zip" + + +def _download(iso: str, level: int) -> Path: + dest = _zip_path(iso, level) + if dest.exists(): + logger.info("Cache hit: %s", dest) + return dest + + dest.parent.mkdir(parents=True, exist_ok=True) + url = f"{_GB_BASE}/{iso}/ADM{level}/geoBoundaries-{iso}-ADM{level}-all.zip" + logger.info("Downloading %s ...", url) + + tmp_fd, tmp_name = tempfile.mkstemp(dir=dest.parent, suffix=".tmp") + os.close(tmp_fd) + tmp = Path(tmp_name) + try: + with httpx.stream("GET", url, follow_redirects=True, timeout=600) as r: + if r.status_code == 404: + raise HTTPException(404, f"geoBoundaries has no data for {iso!r} ADM{level}") + r.raise_for_status() + with tmp.open("wb") as f: + for chunk in r.iter_bytes(chunk_size=65_536): + f.write(chunk) + shutil.move(str(tmp), str(dest)) + except Exception: + tmp.unlink(missing_ok=True) + raise + + logger.info("Saved %s (%.1f MB)", dest, dest.stat().st_size / 1e6) + return dest + + +def _read(iso: str, level: int) -> gpd.GeoDataFrame: + zip_path = _download(iso, level) + shp_name = f"geoBoundaries-{iso}-ADM{level}.shp" + logger.info("Reading %s ...", shp_name) + + try: + gdf = gpd.read_file(f"zip://{zip_path}!{shp_name}", engine="pyogrio") + except Exception as exc: + raise HTTPException(404, f"Could not read {shp_name} from zip: {exc}") from exc + + gdf["nodeid"] = range(len(gdf)) + gdf["name"] = gdf["shapeName"] + return gdf[["nodeid", "name", "shapeID", "geometry"]].rename(columns={"shapeID": "gid"}) + + +app = FastAPI(title="geoboundaries-service") +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"]) + + +@app.get("/health") +def health(): + return {"status": "ok"} + + +@app.get("/boundaries/{iso}/{level}") +def boundaries(iso: str, level: int): + if not 0 <= level <= 5: + raise HTTPException(400, f"admin_level must be 0–5, got {level}") + gdf = _read(iso.upper(), level) + return JSONResponse(content=gdf.to_crs("EPSG:4326").__geo_interface__) diff --git a/services/geoboundaries/requirements.txt b/services/geoboundaries/requirements.txt new file mode 100644 index 0000000..1e48c37 --- /dev/null +++ b/services/geoboundaries/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.110 +uvicorn[standard]>=0.27 +httpx>=0.27 +geopandas>=0.14 +pyogrio>=0.7 diff --git a/services/geoboundaries/test_client.py b/services/geoboundaries/test_client.py new file mode 100644 index 0000000..b3aeac4 --- /dev/null +++ b/services/geoboundaries/test_client.py @@ -0,0 +1,115 @@ +""" +Test client for geoboundaries-service. + +Usage: + python test_client.py [--url http://localhost:8102] + +Uses Luxembourg (LUX) for download tests — small file, fast. +Same structural checks as gadm-service test client; property names +differ (gid comes from shapeID, not GID_N). +""" + +import argparse +import sys +import time + +import httpx + +BASE = "http://localhost:8102" +TEST_ISO = "LUX" + + +def get(path: str, timeout: float = 120) -> httpx.Response: + return httpx.get(f"{BASE}{path}", timeout=timeout) + + +def check(label: str, cond: bool, detail: str = "") -> None: + status = "PASS" if cond else "FAIL" + print(f" [{status}] {label}" + (f" — {detail}" if detail else "")) + if not cond: + sys.exit(1) + + +def test_health(): + print("── /health ─────────────────────────────────────────────") + r = get("/health") + check("HTTP 200", r.status_code == 200) + check("status == ok", r.json().get("status") == "ok") + + +def test_boundaries_level0(): + print(f"── GET /boundaries/{TEST_ISO}/0 (country outline) ─────") + r = get(f"/boundaries/{TEST_ISO}/0") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + fc = r.json() + check("type == FeatureCollection", fc.get("type") == "FeatureCollection") + features = fc.get("features", []) + check("exactly 1 feature at level 0", len(features) == 1, f"got {len(features)}") + props = features[0].get("properties", {}) + check("nodeid present", "nodeid" in props) + check("name present", "name" in props) + check("gid present", "gid" in props) + geom = features[0].get("geometry", {}) + check("geometry present", bool(geom)) + check("geometry is Polygon/MultiPolygon", geom.get("type") in ("Polygon", "MultiPolygon")) + + +def test_boundaries_level1(): + print(f"── GET /boundaries/{TEST_ISO}/1 ────────────────────────") + r = get(f"/boundaries/{TEST_ISO}/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + features = r.json().get("features", []) + check("at least 1 feature", len(features) >= 1, f"got {len(features)}") + props = features[0].get("properties", {}) + check("nodeid present", "nodeid" in props) + check("name present", "name" in props) + + +def test_lowercase_iso(): + print(f"── GET /boundaries/lux/1 (lowercase ISO) ───────────────") + r = get("/boundaries/lux/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + + +def test_cache_speed(): + print(f"── Second request uses cache (fast) ────────────────────") + t0 = time.monotonic() + r = get(f"/boundaries/{TEST_ISO}/1") + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200) + check("responded in < 10s (cache hit)", elapsed < 10, f"{elapsed:.2f}s") + + +def test_invalid_level(): + print("── GET /boundaries/LUX/9 (invalid level) ───────────────") + r = get("/boundaries/LUX/9") + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def test_unknown_iso(): + print("── GET /boundaries/ZZZ/1 (unknown ISO) ─────────────────") + r = get("/boundaries/ZZZ/1") + check("HTTP 404", r.status_code == 404, f"got {r.status_code}") + + +def main(): + global BASE + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=BASE) + args = parser.parse_args() + BASE = args.url.rstrip("/") + + print(f"\nRunning geoboundaries-service tests against {BASE}\n") + print(f"Note: first run downloads {TEST_ISO} shapefiles — may take a moment.\n") + test_health() + test_boundaries_level0() + test_boundaries_level1() + test_lowercase_iso() + test_cache_speed() + test_invalid_level() + test_unknown_iso() + print("\nAll tests passed.") + + +if __name__ == "__main__": + main() diff --git a/services/unocha/Dockerfile b/services/unocha/Dockerfile new file mode 100644 index 0000000..377fddd --- /dev/null +++ b/services/unocha/Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY main.py . + +ENV CACHE_DIR=/cache + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/unocha/README.md b/services/unocha/README.md new file mode 100644 index 0000000..a319d28 --- /dev/null +++ b/services/unocha/README.md @@ -0,0 +1,109 @@ +# unocha-service + +A microservice that serves [UNOCHA](https://www.unocha.org) global administrative boundary GeoJSON from the [Humanitarian Data Exchange (HDX)](https://data.humdata.org). It is the **default shape source** for `laser-generate` because OCHA boundaries are the standard reference for humanitarian response operations. + +## Why it exists + +UNOCHA's Common Operational Dataset (COD) administrative boundaries are the authoritative reference used by humanitarian organisations worldwide. They are designed to be consistent across UN agencies, NGOs, and government partners, making them the natural default for disease modelling work that feeds into response planning. This service provides: + +- A one-time download and extraction of the full global GDB (~1–2 GB) at startup. +- Per-(ISO, level) in-memory caching so each country/level combination is read from disk only once. +- The same GeoJSON `FeatureCollection` response format as the other shape services. + +## API + +### `GET /health` + +Returns `{"status": "ok", "gdb_ready": true}` once startup is complete, or `{"status": "warming", "gdb_ready": false}` while the GDB is still being downloaded or extracted. + +### `GET /boundaries/{ISO}/{level}` + +Returns a GeoJSON `FeatureCollection` of administrative boundaries. + +| Parameter | Description | +|-----------|-------------| +| `ISO` | ISO 3166-1 alpha-3 country code (e.g. `NGA`) | +| `level` | Admin level 0–3 (UNOCHA does not provide levels 4–5) | + +**Feature properties:** + +| Property | Description | +|----------|-------------| +| `nodeid` | Zero-based integer index | +| `name` | Admin unit name (from `adm{level}_name` column) | +| `gid` | P-code (from `adm{level}_pcode` column) — consistent with OCHA COD-AB | + +**Error responses:** + +| Status | Condition | +|--------|-----------| +| 400 | `level` outside 0–3 | +| 404 | No UNOCHA data for the requested ISO / level | +| 503 | GDB not yet ready — startup download/extraction still in progress | + +## How it differs from the other shape services + +| | unocha-service | gadm-service | geoboundaries-service | +|---|---|---|---| +| Source | UNOCHA HDX global GDB | GADM 4.1 | geoBoundaries v6.0.0 | +| License | CC-BY-IGO | Non-commercial | CC-BY | +| Levels | 0–3 | 0–5 | 0–3 (varies) | +| Cache unit | Single global GDB | Per-country zip | Per-(ISO, level) zip | +| Startup download | Yes (~1–2 GB at boot) | No (on-demand) | No (on-demand) | +| In-memory cache | Yes, per (ISO, level) | No | No | +| P-codes | Yes (COD-AB) | No (GADM GIDs) | Yes (shapeID) | +| Default in laser-generate | **Yes** | No | No | + +UNOCHA boundaries use P-codes that align with other humanitarian datasets (population figures, health facility lists, incident reports), which is why they are the default. + +## Startup behaviour + +On first boot the service: +1. Downloads the global GDB zip from HDX (~1–2 GB, 5–15 minutes depending on connectivity). +2. Extracts the `.gdb` directory from the zip. +3. Sets `_gdb_ready = True` and begins accepting requests. + +With a warm cache (zip already in `$CACHE_DIR`) extraction takes ~30 seconds. Kubernetes readiness probe on `/health` prevents traffic from reaching the pod until `gdb_ready` is `true`. + +## In-memory layer cache + +The first request for any `(ISO, level)` pair reads the relevant GDB layer, filters to the requested country, and stores the result in a process-level dictionary. All subsequent requests for that pair are served directly from memory with no disk I/O. + +## Running locally + +```bash +pip install -r requirements.txt + +CACHE_DIR=~/.laser/cache/unocha uvicorn main:app --host 0.0.0.0 --port 8103 +``` + +Wait for `UNOCHA GDB ready — accepting requests.` in the log before sending requests. Test: + +```bash +curl http://localhost:8103/boundaries/NGA/1 +``` + +## Docker + +```bash +docker build -t laser-unocha-service . +docker run -p 8103:8000 -v unocha-cache:/cache laser-unocha-service +``` + +## AKS deployment + +Deployed as part of `services/AKS/geodata-services.yaml`. Uses a 5 Gi PVC (`laser-unocha-cache`) for the GDB, and 2 CPU / 3 Gi memory limit. The `LoadBalancer` service exposes port 80. + +```bash +kubectl apply -f services/AKS/geodata-services.yaml --kubeconfig services/AKS/kube.conf + +# Watch startup progress (GDB download takes several minutes on a cold pod) +kubectl logs -f deployment/laser-unocha-svc -n laser-ai --kubeconfig services/AKS/kube.conf +``` + +## Known limitations + +- **Startup latency**: A cold pod is unavailable for several minutes during the HDX download. Plan for this after initial deployment or after pod restarts. +- **Levels 0–3 only**: UNOCHA does not provide sub-district (level 4+) boundaries globally. Use `gadm-service` for finer administrative detail. +- **HDX URL stability**: The download URL is a direct link to a specific HDX resource ID. If UNOCHA publishes a new version, the URL in `main.py` must be updated. +- **Single replica**: The PVC is `ReadWriteOnce`; scaling requires a shared file store or an alternative caching strategy. diff --git a/services/unocha/main.py b/services/unocha/main.py new file mode 100644 index 0000000..8acdf6e --- /dev/null +++ b/services/unocha/main.py @@ -0,0 +1,147 @@ +""" +unocha-service — serves UNOCHA global admin boundary GeoJSON. + +GET /health +GET /boundaries/{ISO}/{admin_level} → GeoJSON FeatureCollection + +On startup, downloads the single ~1-2 GB global GDB zip from HDX (if not +already cached) and extracts it. All country/level requests are served by +reading the appropriate GDB layer, filtering by iso3, and caching the result +in memory so each (ISO, level) pair is only loaded from disk once. +""" + +import logging +import os +import shutil +import tempfile +import warnings +import zipfile +from contextlib import asynccontextmanager +from pathlib import Path + +import geopandas as gpd +import httpx +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path.home() / ".laser" / "cache" / "unocha")) + +_ZIP_NAME = "global_admin_boundaries_matched_latest.gdb.zip" +_GDB_NAME = "global_admin_boundaries_matched_latest.gdb" +_HDX_URL = ( + "https://data.humdata.org/dataset/70f1cb54-a30c-43b2-a751-44e77d8f5ade" + "/resource/733a9d4c-4e70-4f67-a5af-4138922cf43f/download/" + + _ZIP_NAME +) + +# In-memory cache: (iso, level) → GeoDataFrame +_cache: dict[tuple[str, int], gpd.GeoDataFrame] = {} +_gdb_ready = False + + +def _zip_path() -> Path: + return CACHE_DIR / _ZIP_NAME + + +def _gdb_path() -> Path: + return CACHE_DIR / _GDB_NAME + + +def _download_zip() -> None: + dest = _zip_path() + if dest.exists(): + logger.info("Cache hit: %s", dest) + return + CACHE_DIR.mkdir(parents=True, exist_ok=True) + logger.info("Downloading UNOCHA global GDB (~1-2 GB) — this takes a few minutes ...") + tmp_fd, tmp_name = tempfile.mkstemp(dir=CACHE_DIR, suffix=".tmp") + os.close(tmp_fd) + tmp = Path(tmp_name) + try: + with httpx.stream("GET", _HDX_URL, follow_redirects=True, timeout=1800) as r: + r.raise_for_status() + with tmp.open("wb") as f: + downloaded = 0 + for chunk in r.iter_bytes(chunk_size=1_048_576): # 1 MB chunks + f.write(chunk) + downloaded += len(chunk) + if downloaded % (100 * 1_048_576) == 0: + logger.info(" ... %.0f MB downloaded", downloaded / 1e6) + shutil.move(str(tmp), str(dest)) + except Exception: + tmp.unlink(missing_ok=True) + raise + logger.info("Saved %s (%.0f MB)", dest, dest.stat().st_size / 1e6) + + +def _extract_gdb() -> None: + gdb = _gdb_path() + if gdb.exists(): + logger.info("GDB already extracted: %s", gdb) + return + logger.info("Extracting GDB from zip (may take a minute) ...") + with zipfile.ZipFile(_zip_path(), "r") as zf: + zf.extractall(CACHE_DIR) + logger.info("Extraction complete: %s", gdb) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global _gdb_ready + CACHE_DIR.mkdir(parents=True, exist_ok=True) + _download_zip() + _extract_gdb() + _gdb_ready = True + logger.info("UNOCHA GDB ready — accepting requests.") + yield + + +def _read(iso: str, level: int) -> gpd.GeoDataFrame: + key = (iso, level) + if key in _cache: + return _cache[key] + + layer = f"admin{level}" + logger.info("Reading GDB layer %r for %s ...", layer, iso) + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", category=RuntimeWarning) + full = gpd.read_file(_gdb_path(), layer=layer, engine="pyogrio") + + country = full[full["iso3"] == iso].copy() + if country.empty: + raise HTTPException(404, f"No UNOCHA data for ISO {iso!r} at admin level {level}") + + pcode_col = f"adm{level}_pcode" + name_col = f"adm{level}_name" + + country["nodeid"] = range(len(country)) + country["name"] = country[name_col] if level < 4 else country["adm3_name"] + country["adm4_name"] + country = country[["nodeid", "name", pcode_col, "geometry"]].rename( + columns={pcode_col: "gid"} + ) + _cache[key] = country + logger.info("Cached %s / admin%d — %d features", iso, level, len(country)) + return country + + +app = FastAPI(title="unocha-service", lifespan=lifespan) +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"]) + + +@app.get("/health") +def health(): + return {"status": "ok" if _gdb_ready else "warming", "gdb_ready": _gdb_ready} + + +@app.get("/boundaries/{iso}/{level}") +def boundaries(iso: str, level: int): + if not _gdb_ready: + raise HTTPException(503, "GDB not ready yet — startup download/extraction in progress") + if not 0 <= level <= 3: + raise HTTPException(400, f"UNOCHA supports admin levels 0–3, got {level}") + gdf = _read(iso.upper(), level) + return JSONResponse(content=gdf.to_crs("EPSG:4326").__geo_interface__) diff --git a/services/unocha/requirements.txt b/services/unocha/requirements.txt new file mode 100644 index 0000000..1e48c37 --- /dev/null +++ b/services/unocha/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.110 +uvicorn[standard]>=0.27 +httpx>=0.27 +geopandas>=0.14 +pyogrio>=0.7 diff --git a/services/unocha/test_client.py b/services/unocha/test_client.py new file mode 100644 index 0000000..29170f1 --- /dev/null +++ b/services/unocha/test_client.py @@ -0,0 +1,150 @@ +""" +Test client for unocha-service. + +Usage: + python test_client.py [--url http://localhost:8103] [--wait] + +The service downloads a 1-2 GB global GDB on first startup. Use --wait to +poll /health until gdb_ready=true before running tests (useful in CI or +after a fresh container start). Without --wait, tests run immediately and +will get 503s if the GDB isn't ready yet. +""" + +import argparse +import sys +import time + +import httpx + +BASE = "http://localhost:8103" +TEST_ISO = "NGA" # Nigeria — well-covered in UNOCHA data + + +def get(path: str, timeout: float = 120) -> httpx.Response: + return httpx.get(f"{BASE}{path}", timeout=timeout) + + +def check(label: str, cond: bool, detail: str = "") -> None: + status = "PASS" if cond else "FAIL" + print(f" [{status}] {label}" + (f" — {detail}" if detail else "")) + if not cond: + sys.exit(1) + + +def wait_for_ready(timeout_s: int = 1800) -> None: + print(f"Waiting for GDB to be ready (up to {timeout_s}s) ...") + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + r = httpx.get(f"{BASE}/health", timeout=5) + body = r.json() + if body.get("gdb_ready"): + print(" GDB ready.\n") + return + status = body.get("status", "?") + print(f" status={status} — waiting ...") + except Exception: + print(" service not yet reachable — waiting ...") + time.sleep(10) + print(" TIMED OUT waiting for GDB ready") + sys.exit(1) + + +def test_health(): + print("── /health ─────────────────────────────────────────────") + r = get("/health") + check("HTTP 200", r.status_code == 200) + body = r.json() + check("gdb_ready == true", body.get("gdb_ready") is True, str(body)) + + +def test_boundaries_level0(): + print(f"── GET /boundaries/{TEST_ISO}/0 ────────────────────────") + r = get(f"/boundaries/{TEST_ISO}/0") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + fc = r.json() + check("type == FeatureCollection", fc.get("type") == "FeatureCollection") + features = fc.get("features", []) + check("exactly 1 feature at level 0", len(features) == 1, f"got {len(features)}") + props = features[0].get("properties", {}) + check("nodeid present", "nodeid" in props) + check("name present", "name" in props) + check("gid present", "gid" in props) + geom = features[0].get("geometry", {}) + check("geometry present", bool(geom)) + check("geometry is Polygon/MultiPolygon", geom.get("type") in ("Polygon", "MultiPolygon")) + + +def test_boundaries_level1(): + print(f"── GET /boundaries/{TEST_ISO}/1 ────────────────────────") + r = get(f"/boundaries/{TEST_ISO}/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + features = r.json().get("features", []) + check("multiple features at level 1", len(features) > 1, f"got {len(features)}") + props = features[0].get("properties", {}) + check("nodeid present", "nodeid" in props) + check("name present", "name" in props) + + +def test_cache_speed(): + print(f"── Second request uses in-memory cache (fast) ──────────") + t0 = time.monotonic() + r = get(f"/boundaries/{TEST_ISO}/1") + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200) + check("responded in < 5s (memory cache)", elapsed < 5, f"{elapsed:.2f}s") + + +def test_lowercase_iso(): + print(f"── GET /boundaries/nga/1 (lowercase) ───────────────────") + r = get("/boundaries/nga/1") + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + + +def test_invalid_level(): + print("── GET /boundaries/NGA/5 (unsupported level) ───────────") + r = get("/boundaries/NGA/5") + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def test_unknown_iso(): + print("── GET /boundaries/ZZZ/1 (unknown ISO) ─────────────────") + r = get("/boundaries/ZZZ/1") + check("HTTP 404", r.status_code == 404, f"got {r.status_code}") + + +def test_503_before_ready(): + print("── /health returns warming status when not ready ────────") + # Can't easily test 503 on a live ready service; just confirm health + # reports gdb_ready correctly + r = get("/health") + check("HTTP 200", r.status_code == 200) + check("gdb_ready key present", "gdb_ready" in r.json()) + + +def main(): + global BASE + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=BASE) + parser.add_argument("--wait", action="store_true", + help="Poll /health until gdb_ready before running tests") + args = parser.parse_args() + BASE = args.url.rstrip("/") + + print(f"\nRunning unocha-service tests against {BASE}\n") + if args.wait: + wait_for_ready() + + test_health() + test_boundaries_level0() + test_boundaries_level1() + test_cache_speed() + test_lowercase_iso() + test_invalid_level() + test_unknown_iso() + test_503_before_ready() + print("\nAll tests passed.") + + +if __name__ == "__main__": + main() diff --git a/services/unwpp/Dockerfile b/services/unwpp/Dockerfile new file mode 100644 index 0000000..377fddd --- /dev/null +++ b/services/unwpp/Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY main.py . + +ENV CACHE_DIR=/cache + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/unwpp/README.md b/services/unwpp/README.md new file mode 100644 index 0000000..3d9a62d --- /dev/null +++ b/services/unwpp/README.md @@ -0,0 +1,112 @@ +# unwpp-service + +A microservice that serves demographic data from the [UN World Population Prospects 2024](https://population.un.org/wpp/) for use by `laser-generate`. + +## Why it exists + +LASER models require three demographic inputs per country: crude birth and death rates over time (CBR/CDR), the age distribution at the start of the simulation, and a life expectancy curve. The UN WPP is the authoritative global source for all three, but the raw dataset files are large (hundreds of MB each, gzip-compressed CSV). Loading and filtering them on the user's machine on every run is slow and requires a pandas/numpy environment. Running it once in a shared service means: + +- The four WPP CSV files are downloaded once at startup and held entirely in memory. +- All subsequent requests are served from in-memory DataFrames — typical response time is under 100 ms. +- `laser-generate` fetches all three demographic tables with a single HTTP call. + +## API + +### `GET /health` + +Returns `{"status": "ok", "loaded": ["age_dist", "indicators", "life_1950", "life_2024"]}` once startup is complete. The `loaded` list shows which datasets are in memory. + +### `GET /demographics/{ISO}?start_year=YYYY&end_year=YYYY` + +Returns CBR/CDR, age distribution, and life expectancy for the requested country and year range. + +| Parameter | Description | +|-----------|-------------| +| `ISO` | ISO 3166-1 alpha-3 country code (e.g. `NGA`) | +| `start_year` | First year of simulation (1950–2100) | +| `end_year` | Last year of simulation (1950–2100) | + +**Response:** + +```json +{ + "iso": "NGA", + "start_year": 2020, + "end_year": 2020, + "cxr": [ + { "year": 2020, "CBR": 37.2, "CDR": 10.4 } + ], + "age_dist": [ + { "age_start": 0, "pop_total": 18234.5 }, + { "age_start": 5, "pop_total": 16102.3 }, + ... + ], + "life_exp": [ + { "age": 0, "cumulative_deaths": 4823.0 }, + { "age": 1, "cumulative_deaths": 5201.0 }, + ... + ] +} +``` + +- `cxr`: one row per year between `start_year` and `end_year` +- `age_dist`: 5-year age groups at `start_year` (from WPP medium variant) +- `life_exp`: 101 rows (ages 0–100), cumulative deaths per 100 000 births at `start_year`; uses the 1950–2023 life table for years ≤ 2023 and the 2024–2100 projection for later years + +**Error responses:** + +| Status | Condition | +|--------|-----------| +| 400 | `start_year > end_year` | +| 404 | ISO code not found in WPP dataset | + +## Data source + +Four files downloaded at startup from the UN WPP 2024 CSV release: + +| Internal key | File | +|---|---| +| `age_dist` | `WPP2024_Population1JanuaryByAge5GroupSex_Medium.csv.gz` | +| `indicators` | `WPP2024_Demographic_Indicators_Medium.csv.gz` | +| `life_1950` | `WPP2024_Life_Table_Complete_Medium_Both_1950-2023.csv.gz` | +| `life_2024` | `WPP2024_Life_Table_Complete_Medium_Both_2024-2100.csv.gz` | + +## Startup behaviour + +The service downloads and loads all four files before accepting requests. On a cold start with an empty cache this takes 2–5 minutes (download + CSV parse). With a warm cache (files already in `$CACHE_DIR`) it takes ~30 seconds for the in-memory load. The `/health` endpoint is available throughout startup but returns `"loaded": []` until all datasets are ready. + +## Running locally + +```bash +pip install -r requirements.txt + +CACHE_DIR=~/.laser/cache/unwpp uvicorn main:app --host 0.0.0.0 --port 8100 +``` + +Wait for the log line `Pre-warm complete — ready to serve.` before sending requests. Test: + +```bash +curl "http://localhost:8100/demographics/NGA?start_year=2020&end_year=2020" +``` + +## Docker + +```bash +docker build -t laser-unwpp-service . +docker run -p 8100:8000 -v unwpp-cache:/cache laser-unwpp-service +``` + +## AKS deployment + +Deployed as part of `services/AKS/geodata-services.yaml`. Uses a 2 Gi PVC (`laser-unwpp-cache`) for the WPP CSV files, 2 CPU / 6 Gi memory limit (the four in-memory DataFrames total ~2–3 GB). + +```bash +kubectl apply -f services/AKS/geodata-services.yaml --kubeconfig services/AKS/kube.conf +kubectl logs -f deployment/laser-unwpp-svc -n laser-ai --kubeconfig services/AKS/kube.conf +``` + +## Known limitations + +- **Startup latency**: A cold-start pod is unavailable for several minutes. Kubernetes readiness probe on `/health` prevents traffic until all datasets are loaded. +- **Memory footprint**: ~2–3 GB resident — size the pod accordingly. +- **WPP 2024 only**: The 2022 and earlier WPP releases are not supported. Country coverage and historical estimates reflect WPP 2024 methodology. diff --git a/services/unwpp/main.py b/services/unwpp/main.py new file mode 100644 index 0000000..570f443 --- /dev/null +++ b/services/unwpp/main.py @@ -0,0 +1,143 @@ +""" +unwpp-service — serves filtered UN World Population Prospects demographic data. + +GET /health +GET /demographics/{ISO}?start_year=YYYY&end_year=YYYY + +On startup, downloads all four WPP dataset files to CACHE_DIR if not already +present (pre-warm), then loads them into memory. Requests are served entirely +from in-memory DataFrames. +""" + +import logging +import os +from contextlib import asynccontextmanager +from pathlib import Path + +import httpx +import numpy as np +import pandas as pd +from fastapi import FastAPI, HTTPException, Query +from fastapi.middleware.cors import CORSMiddleware + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path.home() / ".laser" / "cache" / "unwpp")) + +_WPP_BASE = ( + "https://population.un.org/wpp/assets/Excel%20Files/" + "1_Indicator%20(Standard)/CSV_FILES" +) + +_FILES = { + "age_dist": "WPP2024_Population1JanuaryByAge5GroupSex_Medium.csv.gz", + "indicators": "WPP2024_Demographic_Indicators_Medium.csv.gz", + "life_1950": "WPP2024_Life_Table_Complete_Medium_Both_1950-2023.csv.gz", + "life_2024": "WPP2024_Life_Table_Complete_Medium_Both_2024-2100.csv.gz", +} + +_CSV_DTYPES = {"Notes": str, "ISO3_code": str, "ISO2_code": str, "LocTypeName": str} + +# In-memory DataFrames populated at startup +_data: dict[str, pd.DataFrame] = {} + + +def _download(key: str) -> Path: + filename = _FILES[key] + dest = CACHE_DIR / filename + if dest.exists(): + logger.info("Cache hit: %s", dest) + return dest + CACHE_DIR.mkdir(parents=True, exist_ok=True) + url = f"{_WPP_BASE}/{filename}" + logger.info("Downloading %s ...", url) + with httpx.stream("GET", url, follow_redirects=True, timeout=600) as r: + r.raise_for_status() + with dest.open("wb") as f: + for chunk in r.iter_bytes(chunk_size=65_536): + f.write(chunk) + logger.info("Saved %s (%.1f MB)", dest, dest.stat().st_size / 1e6) + return dest + + +def _load(key: str) -> pd.DataFrame: + path = _download(key) + logger.info("Loading %s ...", path.name) + df = pd.read_csv(path, compression="gzip", dtype=_CSV_DTYPES) + logger.info("Loaded %s — %d rows", path.name, len(df)) + return df + + +@asynccontextmanager +async def lifespan(app: FastAPI): + logger.info("Pre-warming: downloading and loading all WPP datasets ...") + for key in _FILES: + _data[key] = _load(key) + logger.info("Pre-warm complete — ready to serve.") + yield + + +app = FastAPI(title="unwpp-service", lifespan=lifespan) +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"]) + + +@app.get("/health") +def health(): + return {"status": "ok", "loaded": list(_data.keys())} + + +@app.get("/demographics/{iso}") +def demographics( + iso: str, + start_year: int = Query(..., ge=1950, le=2100), + end_year: int = Query(..., ge=1950, le=2100), +): + if start_year > end_year: + raise HTTPException(400, "start_year must be <= end_year") + iso = iso.upper() + + # ── CBR / CDR ────────────────────────────────────────────────────────────── + country_demo = _data["indicators"][_data["indicators"]["ISO3_code"] == iso] + if country_demo.empty: + raise HTTPException(404, f"ISO code not found: {iso!r}") + + cxr = ( + country_demo[country_demo["Time"].between(start_year, end_year)] + .sort_values("Time")[["Time", "CBR", "CDR"]] + .rename(columns={"Time": "year"}) + .to_dict(orient="records") + ) + + # ── Age distribution at start_year ───────────────────────────────────────── + country_pop = _data["age_dist"][ + (_data["age_dist"]["ISO3_code"] == iso) & (_data["age_dist"]["Time"] == start_year) + ] + age_dist = ( + country_pop.sort_values("AgeGrpStart")[["AgeGrpStart", "PopTotal"]] + .rename(columns={"AgeGrpStart": "age_start", "PopTotal": "pop_total"}) + .to_dict(orient="records") + ) + + # ── Life expectancy (cumulative deaths) at start_year ────────────────────── + life_key = "life_1950" if start_year <= 2023 else "life_2024" + country_life = _data[life_key][ + (_data[life_key]["ISO3_code"] == iso) & (_data[life_key]["Time"] == start_year) + ].sort_values("AgeGrpStart").reset_index(drop=True) + + survival = country_life["lx"].to_numpy() + cumulative = 100_000 - np.round(survival) + cumulative = np.append(cumulative[1:], 100_000) + life_exp = [ + {"age": int(age), "cumulative_deaths": float(cd)} + for age, cd in zip(country_life["AgeGrpStart"].tolist(), cumulative) + ] + + return { + "iso": iso, + "start_year": start_year, + "end_year": end_year, + "cxr": cxr, + "age_dist": age_dist, + "life_exp": life_exp, + } diff --git a/services/unwpp/requirements.txt b/services/unwpp/requirements.txt new file mode 100644 index 0000000..1365e5f --- /dev/null +++ b/services/unwpp/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.110 +uvicorn[standard]>=0.27 +httpx>=0.27 +pandas>=2.0 +numpy>=1.26 diff --git a/services/unwpp/test_client.py b/services/unwpp/test_client.py new file mode 100644 index 0000000..b9870ca --- /dev/null +++ b/services/unwpp/test_client.py @@ -0,0 +1,124 @@ +""" +Test client for unwpp-service. + +Usage: + python test_client.py [--url http://localhost:8100] + +Runs a series of checks against a live unwpp-service instance: + - /health responds OK + - /demographics returns expected shape and value ranges + - Edge cases: unknown ISO, inverted year range, boundary years +""" + +import argparse +import sys + +import httpx + +BASE = "http://localhost:8100" + + +def get(path: str, **params) -> httpx.Response: + return httpx.get(f"{BASE}{path}", params=params, timeout=30) + + +def check(label: str, cond: bool, detail: str = "") -> None: + status = "PASS" if cond else "FAIL" + print(f" [{status}] {label}" + (f" — {detail}" if detail else "")) + if not cond: + sys.exit(1) + + +def test_health(): + print("── /health ─────────────────────────────────────────────") + r = get("/health") + check("HTTP 200", r.status_code == 200) + body = r.json() + check("status == ok", body.get("status") == "ok") + loaded = body.get("loaded", []) + for key in ("age_dist", "indicators", "life_1950", "life_2024"): + check(f"dataset loaded: {key}", key in loaded) + + +def test_demographics_basic(): + print("── GET /demographics/NGA (2000–2025) ───────────────────") + r = get("/demographics/NGA", start_year=2000, end_year=2025) + check("HTTP 200", r.status_code == 200) + body = r.json() + check("iso == NGA", body["iso"] == "NGA") + + cxr = body["cxr"] + check("cxr has 26 rows", len(cxr) == 26, f"got {len(cxr)}") + check("cxr first year == 2000", cxr[0]["year"] == 2000) + check("cxr last year == 2025", cxr[-1]["year"] == 2025) + check("CBR > 0", all(row["CBR"] > 0 for row in cxr)) + check("CDR > 0", all(row["CDR"] > 0 for row in cxr)) + + age_dist = body["age_dist"] + check("age_dist non-empty", len(age_dist) > 0, f"got {len(age_dist)}") + check("age_dist has age_start=0", any(r["age_start"] == 0 for r in age_dist)) + check("age_dist pop_total > 0", all(r["pop_total"] > 0 for r in age_dist)) + + life_exp = body["life_exp"] + check("life_exp non-empty", len(life_exp) > 0, f"got {len(life_exp)}") + check("life_exp has age=0", any(r["age"] == 0 for r in life_exp)) + check("life_exp cumulative_deaths >= 0", all(r["cumulative_deaths"] >= 0 for r in life_exp)) + + +def test_demographics_lowercase_iso(): + print("── GET /demographics/nga (lowercase ISO) ───────────────") + r = get("/demographics/nga", start_year=2010, end_year=2010) + check("HTTP 200 (normalised to uppercase)", r.status_code == 200) + check("iso == NGA", r.json()["iso"] == "NGA") + + +def test_single_year(): + print("── GET /demographics/ETH (start_year == end_year) ──────") + r = get("/demographics/ETH", start_year=2015, end_year=2015) + check("HTTP 200", r.status_code == 200) + cxr = r.json()["cxr"] + check("cxr has exactly 1 row", len(cxr) == 1, f"got {len(cxr)}") + + +def test_future_years(): + print("── GET /demographics/IND (2030–2050) ───────────────────") + r = get("/demographics/IND", start_year=2030, end_year=2050) + check("HTTP 200", r.status_code == 200) + cxr = r.json()["cxr"] + check("cxr has 21 rows", len(cxr) == 21, f"got {len(cxr)}") + life_exp = r.json()["life_exp"] + check("life_exp uses 2024 file", len(life_exp) > 0) + + +def test_unknown_iso(): + print("── GET /demographics/ZZZ (unknown ISO) ─────────────────") + r = get("/demographics/ZZZ", start_year=2000, end_year=2010) + check("HTTP 404", r.status_code == 404, f"got {r.status_code}") + + +def test_inverted_years(): + print("── GET /demographics/NGA (start > end) ─────────────────") + r = get("/demographics/NGA", start_year=2025, end_year=2000) + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def main(): + global BASE + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=BASE) + args = parser.parse_args() + BASE = args.url.rstrip("/") + + print(f"\nRunning unwpp-service tests against {BASE}\n") + test_health() + test_demographics_basic() + test_demographics_lowercase_iso() + test_single_year() + test_future_years() + test_unknown_iso() + test_inverted_years() + print("\nAll tests passed.") + + +if __name__ == "__main__": + main() diff --git a/services/worldpop/Dockerfile b/services/worldpop/Dockerfile new file mode 100644 index 0000000..cd2b40c --- /dev/null +++ b/services/worldpop/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY requirements.txt . +RUN apt-get update && apt-get install -y --no-install-recommends libexpat1 && rm -rf /var/lib/apt/lists/* +RUN pip install --no-cache-dir -r requirements.txt + +COPY main.py prewarm_countries.py ./ + +ENV CACHE_DIR=/cache + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/worldpop/README.md b/services/worldpop/README.md new file mode 100644 index 0000000..b71a979 --- /dev/null +++ b/services/worldpop/README.md @@ -0,0 +1,122 @@ +# worldpop-service + +A microservice that aggregates [WorldPop](https://www.worldpop.org) UN-adjusted population rasters into per-polygon counts for use by `laser-generate` and the browser choropleth client. + +## Why it exists + +Population data for spatial epidemic models must be attributed to each administrative unit (node). WorldPop provides globally consistent ~1 km rasters, but summing pixel values within arbitrary polygon boundaries requires GDAL/rasterio, significant RAM, and for large countries many minutes of computation. Running this once in a shared service means: + +- The raster is downloaded and cached **once** per country/year — subsequent requests are served from the local GeoTIFF. +- No rasterio/GDAL installation is required on the user's machine. +- The browser choropleth client can aggregate population with a single `fetch()` call. + +## API + +### `GET /health` + +Returns `{"status": "ok"}`. + +### `POST /aggregate/{ISO}?year=YYYY` + +Aggregates population for each polygon in the request body and returns a map of `nodeid → population`. + +| Parameter | Description | +|-----------|-------------| +| `ISO` | ISO 3166-1 alpha-3 country code (e.g. `NGA`) | +| `year` | Raster year (2000–2020, default 2020) | + +**Request body:** GeoJSON `FeatureCollection`. Each feature must have a `nodeid` integer in its `properties`. + +**Response:** JSON object mapping `nodeid` (string key) to population count (float). + +```json +{ "0": 4521302.0, "1": 1203847.0, "2": 892100.0, ... } +``` + +The response is **streamed** — the first bytes arrive quickly and results accumulate as each polygon is processed. This keeps the Azure Load Balancer's 4-minute idle TCP timeout from killing long-running aggregation requests. + +### `POST /prewarm/{ISO}?year=YYYY` + +Downloads and caches the raster for `ISO`/`year` without performing any aggregation. Use this to pre-populate the cache for large countries (BRA, IND, RUS, …) before the first `/aggregate` call. + +Returns `{"status": "ok", "iso": "...", "year": ...}` once the raster is on disk. + +**Error responses:** + +| Status | Condition | +|--------|-----------| +| 400 | Body is not a GeoJSON FeatureCollection, or has no features | +| 404 | No WorldPop data for the requested ISO / year | +| 422 | Raster exceeds 2 GB and has not been pre-warmed (use `/prewarm` first) | + +## Raster source + +WorldPop UN-adjusted constrained dataset: +``` +https://data.worldpop.org/GIS/Population/Global_2000_2020/{year}/{ISO}/{iso}_ppp_{year}_UNadj.tif +``` + +Only years 2000–2020 are available. `laser-generate` automatically clamps `start_year` to 2020 for the raster year. + +## Memory and performance + +| Raster size | Mode | Notes | +|-------------|------|-------| +| < 2 GB uncompressed | In-memory | Entire raster loaded once; polygon reads are fast array slices | +| ≥ 2 GB uncompressed | Windowed strips | 8 M pixels (~64 MB) read per strip per polygon; bounded memory regardless of country size | + +In windowed mode, features are sorted by raster scan order (row then column) before aggregation, which maximises GDAL tile-cache reuse across adjacent polygons and is critical for level-2 queries with hundreds of small features (e.g. TZA/2 with 186 districts). + +GDAL tile cache is set to 512 MB (`GDAL_CACHEMAX=512`) at startup. + +## Caching + +Rasters are cached at `$CACHE_DIR/{iso_lower}_ppp_{year}_UNadj.tif`. A per-(ISO, year) mutex prevents duplicate concurrent downloads. The cache directory defaults to `/cache` in the container (backed by a 100 Gi Kubernetes PVC in AKS) and `~/.laser/cache/worldpop` locally. + +## Running locally + +```bash +pip install -r requirements.txt + +CACHE_DIR=~/.laser/cache/worldpop uvicorn main:app --host 0.0.0.0 --port 8104 +``` + +Test with a small country first — NGA (~200 MB raster) is a reasonable smoke-test: + +```bash +python test_client.py --url http://localhost:8104 +``` + +For large countries (BRA, IND), pre-warm before aggregating: + +```bash +curl -X POST "http://localhost:8104/prewarm/BRA?year=2020" +``` + +## Docker + +```bash +docker build -t laser-worldpop-service . +docker run -p 8104:8000 -v worldpop-cache:/cache laser-worldpop-service +``` + +## AKS deployment + +Deployed as part of `services/AKS/geodata-services.yaml`. Uses a 100 Gi PVC (`laser-worldpop-cache`), 4 CPU / 4 Gi memory limit. A separate prewarm Job (`worldpop-prewarm-job.yaml`) populates the cache for the default humanitarian country list after initial deployment. + +```bash +# Deploy service +kubectl apply -f services/AKS/geodata-services.yaml --kubeconfig services/AKS/kube.conf + +# Run prewarm job (once, after service is Ready) +kubectl apply -f services/AKS/worldpop-prewarm-job.yaml --kubeconfig services/AKS/kube.conf + +# Monitor prewarm progress +kubectl logs -f job/laser-worldpop-prewarm -n laser-ai --kubeconfig services/AKS/kube.conf +``` + +## Known limitations + +- **Year cap at 2020**: WorldPop's Global_2000_2020 dataset ends at 2020. There is no 2021+ raster. +- **Large country first-run latency**: BRA (~3.7 GB raster, ~30 min download), IND (~1.4 GB) benefit greatly from pre-warming. +- **Level-2 aggregation time**: Even with the raster cached, 500+ small polygons in windowed mode can take several minutes. The browser client warns and falls back to the CLI for known large-raster countries at high admin levels. diff --git a/services/worldpop/client.html b/services/worldpop/client.html new file mode 100644 index 0000000..9b3c80a --- /dev/null +++ b/services/worldpop/client.html @@ -0,0 +1,689 @@ + + + + + WorldPop Population Choropleth + + + + + + +
+ + + + + + + + + +
+ +
+
+
+
+
Loading…
+
+
+ +
+ Run locally to generate model files: + + +
+ + + + diff --git a/services/worldpop/main.py b/services/worldpop/main.py new file mode 100644 index 0000000..32fd3ed --- /dev/null +++ b/services/worldpop/main.py @@ -0,0 +1,300 @@ +""" +worldpop-service — aggregates WorldPop population raster into polygon areas. + +GET /health +POST /prewarm/{iso}?year=2020 → {"status": "ok", "iso": "...", "year": ...} +POST /aggregate/{iso}?year=2020 → {nodeid: population, ...} + body: GeoJSON FeatureCollection; each feature must have "nodeid" in properties + +Rasters are downloaded on demand and cached on disk. The WorldPop UN-adjusted +dataset (Global_2000_2020) is used. Per-(iso,year) locks prevent duplicate +concurrent downloads. +""" + +import logging +import os +import shutil +import tempfile +import threading +from contextlib import asynccontextmanager +from pathlib import Path + +import numpy as np +import geopandas as gpd +import httpx +import rasterio +import rasterio.features +import rasterio.windows +from fastapi import Body, FastAPI, HTTPException, Query +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import StreamingResponse +from shapely.geometry import mapping + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path.home() / ".laser" / "cache" / "worldpop")) +_YEAR_DEFAULT = 2020 +_MAX_RASTER_MB = 2000 # on-demand size limit; larger rasters must use generate.py + +# 512 MB GDAL block cache — default is 64 MB which is too small when reading many +# per-polygon windows from a large compressed GeoTIFF (constant cache thrash). +os.environ.setdefault("GDAL_CACHEMAX", "512") + +# Per-(iso, year) lock prevents duplicate concurrent downloads of the same raster. +_lock_registry_mu: threading.Lock = threading.Lock() +_download_locks: dict[tuple[str, int], threading.Lock] = {} + + +def _raster_path(iso: str, year: int) -> Path: + return CACHE_DIR / f"{iso.lower()}_ppp_{year}_UNadj.tif" + + +def _worldpop_url(iso: str, year: int) -> str: + return ( + f"https://data.worldpop.org/GIS/Population/" + f"Global_2000_2020/{year}/{iso.upper()}/" + f"{iso.lower()}_ppp_{year}_UNadj.tif" + ) + + +def _download_raster(iso: str, year: int, allow_oversize: bool = False) -> Path: + dest = _raster_path(iso, year) + if dest.exists(): + logger.info("Cache hit: %s", dest) + return dest + + # Acquire a per-(iso, year) lock so concurrent requests wait rather than + # each downloading the same raster simultaneously. + key = (iso, year) + with _lock_registry_mu: + if key not in _download_locks: + _download_locks[key] = threading.Lock() + lock = _download_locks[key] + + with lock: + if dest.exists(): # another thread finished while we waited + logger.info("Cache hit (post-lock): %s", dest) + return dest + + CACHE_DIR.mkdir(parents=True, exist_ok=True) + url = _worldpop_url(iso, year) + + # Pre-flight: check file size before committing to a long download. + # If Content-Length is unavailable, proceed and let the download run. + try: + head = httpx.head(url, follow_redirects=True, timeout=30) + if head.status_code == 404: + raise HTTPException(404, f"No WorldPop data for ISO {iso!r} year={year}") + cl = head.headers.get("content-length") + if cl: + mb = int(cl) // (1024 * 1024) + if mb > _MAX_RASTER_MB and not allow_oversize: + raise HTTPException( + 422, + f"{iso} raster is {mb} MB — too large for on-demand download " + f"(limit {_MAX_RASTER_MB} MB). " + f"Pre-warm it first: POST /prewarm/{iso}?year={year} " + f"(runs in the background; may take 30+ min for large countries)." + ) + logger.info("Pre-flight: %s year=%d is %d MB — proceeding", iso, year, mb) + except HTTPException: + raise + except Exception as exc: + logger.warning("Pre-flight HEAD failed (%s) — proceeding with download", exc) + + logger.info("Downloading WorldPop raster for %s year=%d ...", iso, year) + tmp_fd, tmp_name = tempfile.mkstemp(dir=CACHE_DIR, suffix=".tmp") + os.close(tmp_fd) + tmp = Path(tmp_name) + try: + with httpx.stream("GET", url, follow_redirects=True, timeout=600) as r: + if r.status_code == 404: + raise HTTPException(404, f"No WorldPop data for ISO {iso!r} year={year}") + r.raise_for_status() + with tmp.open("wb") as f: + downloaded = 0 + for chunk in r.iter_bytes(chunk_size=1_048_576): + f.write(chunk) + downloaded += len(chunk) + if downloaded % (50 * 1_048_576) == 0: + logger.info(" ... %.0f MB downloaded", downloaded / 1e6) + shutil.move(str(tmp), str(dest)) + except Exception: + tmp.unlink(missing_ok=True) + raise + logger.info("Saved %s (%.0f MB)", dest, dest.stat().st_size / 1e6) + return dest + + +@asynccontextmanager +async def lifespan(app: FastAPI): + CACHE_DIR.mkdir(parents=True, exist_ok=True) + logger.info("worldpop-service ready — rasters downloaded on demand via /prewarm or /aggregate.") + yield + + +app = FastAPI(title="worldpop-service", lifespan=lifespan) +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["GET", "POST"], + allow_headers=["*"], +) + + +@app.get("/health") +def health(): + return {"status": "ok"} + + +@app.post("/prewarm/{iso}") +def prewarm( + iso: str, + year: int = Query(_YEAR_DEFAULT, ge=2000, le=2020), +): + # Prewarm bypasses the _MAX_RASTER_MB guard that /aggregate enforces. + # Rationale: the guard exists to prevent Azure LB idle-timeout on /aggregate + # requests. Prewarm runs from the K8s job (prewarm_countries.py) which + # tolerates minutes-long downloads. Users blocked by /aggregate's 422 are + # told to POST /prewarm first, so that path must succeed for large rasters. + _download_raster(iso.upper(), year, allow_oversize=True) + return {"status": "ok", "iso": iso.upper(), "year": year} + + +# Maximum pixels read from disk per strip in windowed mode. Each strip is one +# horizontal band of a polygon's bounding-box window. Keeping this at 8 M pixels +# caps per-strip peak memory at ~64 MB (float32 data + bool mask). +_MAX_STRIP_PIXELS = 8_000_000 + + +def _sum_polygon_windowed(src, win, transform, geom, nodata) -> float: + """Sum raster pixels inside geom using horizontal strips to bound memory. + + Used when the polygon's bounding-box window is too large to read at once + (e.g. Amazonian states in BRA). Reads _MAX_STRIP_PIXELS pixels at a time. + """ + height = int(win.height) + width = int(win.width) + strip_rows = max(1, _MAX_STRIP_PIXELS // max(width, 1)) + total = 0.0 + for row_start in range(0, height, strip_rows): + strip_h = min(strip_rows, height - row_start) + strip_win = rasterio.windows.Window( + int(win.col_off), int(win.row_off) + row_start, width, strip_h + ) + strip_data = src.read(1, window=strip_win) + strip_transform = rasterio.windows.transform(strip_win, transform) + strip_mask = rasterio.features.geometry_mask( + [mapping(geom)], + out_shape=(strip_h, width), + transform=strip_transform, + invert=True, + ) + vals = strip_data[strip_mask].astype(np.float64) + if nodata is not None: + vals = vals[vals != nodata] + total += float(np.sum(vals)) + return total + + +def _aggregate_stream(iso: str, year: int, features: list, gdf): + """Generator that yields a JSON object one key:value at a time. + + Streaming keeps data flowing to the client throughout computation so the + Azure Load Balancer (4-min idle timeout) never sees a silent connection, + even when aggregating large-country rasters like BRA or IDN. + """ + raster_path = _raster_path(iso, year) + _2GB = 2 * 1024 ** 3 + + with rasterio.open(str(raster_path)) as src: + nodata = src.nodata + transform = src.transform + src_win = rasterio.windows.Window(0, 0, src.width, src.height) + uncompressed = src.width * src.height * 4 # float32 bytes + + if uncompressed < _2GB: + data = src.read(1) + logger.info("In-memory mode (%.0f MB uncompressed)", uncompressed / 1e6) + ordered = list(zip(features, gdf.geometry)) + else: + data = None + logger.info("Windowed mode (%.0f MB uncompressed — too large for in-memory)", + uncompressed / 1e6) + # Sort by raster row then column so consecutive polygons share GDAL + # tile-cache entries — critical for level-2 queries with hundreds of + # small polygons spread across a large compressed raster. + def _scan_key(feat_geom): + _, geom = feat_geom + win = rasterio.windows.from_bounds( + *geom.bounds, transform=transform + ).round_offsets() + return (int(win.row_off), int(win.col_off)) + ordered = sorted(zip(features, gdf.geometry), key=_scan_key) + logger.info("Sorted %d features by raster scan order", len(ordered)) + + first = True + for feat, geom in ordered: + nodeid = int(feat["properties"]["nodeid"]) + try: + win = rasterio.windows.from_bounds( + *geom.bounds, transform=transform + ).round_lengths().round_offsets() + win = win.intersection(src_win) + + height = int(win.height) + width = int(win.width) + if height <= 0 or width <= 0: + pop = 0.0 + elif data is not None: + row_off = int(win.row_off) + col_off = int(win.col_off) + window_data = data[row_off:row_off + height, col_off:col_off + width] + window_transform = rasterio.windows.transform(win, transform) + geom_mask = rasterio.features.geometry_mask( + [mapping(geom)], + out_shape=(height, width), + transform=window_transform, + invert=True, + ) + vals = window_data[geom_mask].astype(np.float64) + if nodata is not None: + vals = vals[vals != nodata] + pop = float(np.sum(vals)) + else: + # Windowed mode: read in strips to bound peak memory. + pop = _sum_polygon_windowed(src, win, transform, geom, nodata) + except Exception: + logger.exception("Error aggregating nodeid=%d for %s", nodeid, iso) + pop = 0.0 + + prefix = b"{" if first else b"," + first = False + yield prefix + f'"{nodeid}":{pop}'.encode() + + yield b"}" if not first else b"{}" + logger.info("Aggregated %d features for %s/%d", len(features), iso, year) + + +@app.post("/aggregate/{iso}") +def aggregate( + iso: str, + year: int = Query(_YEAR_DEFAULT, ge=2000, le=2020), + body: dict = Body(...), +): + if body.get("type") != "FeatureCollection": + raise HTTPException(400, "Body must be a GeoJSON FeatureCollection") + features = body.get("features", []) + if not features: + raise HTTPException(400, "FeatureCollection has no features") + + _download_raster(iso.upper(), year) + + gdf = gpd.GeoDataFrame.from_features(features, crs="EPSG:4326") + logger.info("Aggregating %s/%d over %d features ...", iso.upper(), year, len(gdf)) + + return StreamingResponse( + _aggregate_stream(iso.upper(), year, features, gdf), + media_type="application/json", + ) diff --git a/services/worldpop/prewarm_countries.py b/services/worldpop/prewarm_countries.py new file mode 100644 index 0000000..14fcf08 --- /dev/null +++ b/services/worldpop/prewarm_countries.py @@ -0,0 +1,110 @@ +""" +Prewarm worldpop-service raster cache for a list of countries. + +Usage: + python prewarm_countries.py [--url http://localhost:8104] [--year 2020] + [--workers 4] [--countries FILE] + +Designed to run as a Kubernetes init Job that pre-populates the PVC before +worldpop-service pods start serving traffic. Each ISO triggers a +POST /prewarm/{iso} call; the service downloads and caches the raster. + +Default country list covers common LASER / humanitarian contexts (~50 ISOs). +Pass --countries to override with a newline-delimited file of ISO3 codes. +""" + +import argparse +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +import httpx + +# fmt: off +DEFAULT_COUNTRIES = [ + # Sub-Saharan Africa + "NGA", "ETH", "COD", "TZA", "KEN", "UGA", "MOZ", "GHA", "MDG", "CMR", + "CIV", "NER", "BFA", "MLI", "SEN", "ZMB", "ZWE", "SOM", "SSD", "SDN", + "AGO", "RWA", "BDI", "TCD", "GIN", "SLE", "LBR", "MWI", "NAM", "BWA", + # North Africa / Middle East + "EGY", "DZA", "MAR", "TUN", "LBY", "SYR", "IRQ", "YEM", "AFG", "PAK", + # South / Southeast Asia + "IND", "BGD", "NPL", "MMR", "KHM", "LAO", "PHL", "IDN", + # Latin America + "BRA", "COL", "PER", "BOL", "HTI", + # Western Europe — small rasters, useful for evaluator ground-truthing + "GBR", "DEU", "FRA", +] +# fmt: on + +_MAX_RETRIES = 15 # up to 15 min of retries per country +_RETRY_WAIT = 60 # seconds between retries + + +def prewarm_one(url: str, iso: str, year: int) -> tuple[str, bool, str]: + # Retry on connection reset — Azure LB 4-min idle timeout fires while a + # large raster is downloading server-side; the download continues and the + # next attempt hits the cache. + for attempt in range(1, _MAX_RETRIES + 1): + try: + r = httpx.post(f"{url}/prewarm/{iso}?year={year}", timeout=1800) + if r.status_code == 200: + return iso, True, "" + return iso, False, f"HTTP {r.status_code}: {r.text[:120]}" + except (httpx.RemoteProtocolError, httpx.ReadError, httpx.ConnectError) as exc: + if attempt == _MAX_RETRIES: + return iso, False, f"gave up after {attempt} retries: {exc}" + print(f" [WAIT] {iso} — connection reset, retry {attempt}/{_MAX_RETRIES} in {_RETRY_WAIT}s", + flush=True) + time.sleep(_RETRY_WAIT) + except Exception as exc: + return iso, False, str(exc) + return iso, False, "unreachable" + + +def main() -> None: + parser = argparse.ArgumentParser(description="Pre-warm WorldPop raster cache.") + parser.add_argument("--url", default="http://localhost:8104", + help="worldpop-service base URL") + parser.add_argument("--year", type=int, default=2020, + help="WorldPop year (2000–2020, default 2020)") + parser.add_argument("--workers", type=int, default=4, + help="Parallel download workers (default 4)") + parser.add_argument("--countries", metavar="FILE", + help="Newline-delimited file of ISO3 codes (overrides default list)") + args = parser.parse_args() + + base_url = args.url.rstrip("/") + + if args.countries: + with open(args.countries) as f: + isos = [ln.strip().upper() for ln in f if ln.strip() and not ln.startswith("#")] + else: + isos = DEFAULT_COUNTRIES + + print(f"Pre-warming {len(isos)} countries at {base_url} (year={args.year}, workers={args.workers})\n", + flush=True) + + t0 = time.monotonic() + ok, failed = [], [] + + with ThreadPoolExecutor(max_workers=args.workers) as pool: + futures = {pool.submit(prewarm_one, base_url, iso, args.year): iso for iso in isos} + for fut in as_completed(futures): + iso, success, msg = fut.result() + if success: + ok.append(iso) + print(f" [OK] {iso}", flush=True) + else: + failed.append(iso) + print(f" [FAIL] {iso} — {msg}", flush=True) + + elapsed = time.monotonic() - t0 + print(f"\n{len(ok)}/{len(isos)} countries cached in {elapsed:.0f}s", flush=True) + if failed: + print(f"Failed: {', '.join(failed)}", flush=True) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/services/worldpop/requirements.txt b/services/worldpop/requirements.txt new file mode 100644 index 0000000..63fcbb2 --- /dev/null +++ b/services/worldpop/requirements.txt @@ -0,0 +1,7 @@ +fastapi>=0.110 +uvicorn[standard]>=0.27 +httpx>=0.27 +geopandas>=0.14 +pyogrio>=0.7 +rasterio>=1.3 +numpy>=1.24 diff --git a/services/worldpop/test_client.py b/services/worldpop/test_client.py new file mode 100644 index 0000000..2f1261d --- /dev/null +++ b/services/worldpop/test_client.py @@ -0,0 +1,260 @@ +""" +Test client for worldpop-service. + +Usage: + python test_client.py [--url http://localhost:8104] [--wait] + python test_client.py --url http://4.155.140.158 --large-country BRA + +The service downloads WorldPop rasters on demand. Use --wait to poll /health +before running tests. The first run for a new ISO downloads the raster. + +Default ISO is LUX (Luxembourg, ~1 MB raster — fast, always downloaded on demand). +Pass --large-country to also test a large pre-warmed country (e.g. BRA, NGA). +""" + +import argparse +import sys +import time + +import httpx + +BASE = "http://localhost:8104" +TEST_ISO = "LUX" + +# Bounding box polygon covering Luxembourg (WGS84) +_LUX_RING = [[5.7, 49.4], [6.5, 49.4], [6.5, 50.2], [5.7, 50.2], [5.7, 49.4]] +TEST_GEOJSON = { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": {"nodeid": 0, "name": "Luxembourg"}, + "geometry": {"type": "Polygon", "coordinates": [_LUX_RING]}, + } + ], +} + +# Simple 4-polygon GeoJSON for NGA that exercises the multi-polygon path +# (bounding boxes of 4 rough quadrants of Nigeria, WGS84) +_NGA_QUADS = [ + {"nodeid": 0, "name": "NW", "ring": [[3.0,9.5],[9.5,9.5],[9.5,14.0],[3.0,14.0],[3.0,9.5]]}, + {"nodeid": 1, "name": "NE", "ring": [[9.5,9.5],[15.0,9.5],[15.0,14.0],[9.5,14.0],[9.5,9.5]]}, + {"nodeid": 2, "name": "SW", "ring": [[3.0,4.0],[9.5,4.0],[9.5,9.5],[3.0,9.5],[3.0,4.0]]}, + {"nodeid": 3, "name": "SE", "ring": [[9.5,4.0],[15.0,4.0],[15.0,9.5],[9.5,9.5],[9.5,4.0]]}, +] +NGA_GEOJSON = { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": {"nodeid": q["nodeid"], "name": q["name"]}, + "geometry": {"type": "Polygon", "coordinates": [q["ring"]]}, + } + for q in _NGA_QUADS + ], +} + + +def get(path: str, timeout: float = 30) -> httpx.Response: + return httpx.get(f"{BASE}{path}", timeout=timeout) + + +def post(path: str, body: dict, timeout: float = 300) -> httpx.Response: + return httpx.post(f"{BASE}{path}", json=body, timeout=timeout) + + +def check(label: str, cond: bool, detail: str = "") -> None: + status = "PASS" if cond else "FAIL" + print(f" [{status}] {label}" + (f" — {detail}" if detail else "")) + if not cond: + sys.exit(1) + + +def wait_for_ready(timeout_s: int = 120) -> None: + print(f"Waiting for service to be ready (up to {timeout_s}s) ...") + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + r = httpx.get(f"{BASE}/health", timeout=5) + if r.status_code == 200: + print(" Service ready.\n") + return + except Exception: + print(" Service not yet reachable — waiting ...") + time.sleep(5) + print(" TIMED OUT waiting for service") + sys.exit(1) + + +def test_health(): + print("── /health ─────────────────────────────────────────────") + r = get("/health") + check("HTTP 200", r.status_code == 200) + body = r.json() + check("status == ok", body.get("status") == "ok", str(body)) + + +def test_prewarm(): + print(f"── POST /prewarm/{TEST_ISO} (downloads raster) ─────────") + r = post(f"/prewarm/{TEST_ISO}", {}, timeout=600) + check("HTTP 200", r.status_code == 200, f"got {r.status_code}") + body = r.json() + check("status == ok", body.get("status") == "ok", str(body)) + check("iso present", "iso" in body) + check("year present", "year" in body) + + +def test_aggregate(): + print(f"── POST /aggregate/{TEST_ISO} ─────────────────────────") + r = post(f"/aggregate/{TEST_ISO}", TEST_GEOJSON, timeout=60) + check("HTTP 200", r.status_code == 200, f"got {r.status_code}: {r.text[:200]}") + result = r.json() + check("nodeid 0 in result", "0" in result, str(list(result.keys())[:5])) + pop = result["0"] + check("population > 100_000", pop > 100_000, f"got {pop:.0f}") + check("population < 1_500_000", pop < 1_500_000, f"got {pop:.0f}") + + +def test_cache_speed(): + print(f"── Second /aggregate/{TEST_ISO} uses cached raster ─────") + t0 = time.monotonic() + r = post(f"/aggregate/{TEST_ISO}", TEST_GEOJSON, timeout=60) + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200) + check("responded in < 10s (cached raster)", elapsed < 10, f"{elapsed:.2f}s") + + +def test_nga_multipolygon(): + print("── POST /aggregate/NGA (4 quads, multi-polygon) ─────────") + t0 = time.monotonic() + r = post("/aggregate/NGA", NGA_GEOJSON, timeout=120) + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200, f"got {r.status_code}: {r.text[:200]}") + result = r.json() + check("4 nodeids returned", len(result) == 4, f"got {len(result)}: {list(result.keys())}") + total = sum(result.values()) + check("total NGA pop plausible (150M–230M)", 150_000_000 < total < 230_000_000, + f"got {total:.0f}") + check("each quad > 0", all(v > 0 for v in result.values()), + str({k: f"{v:.0f}" for k, v in result.items()})) + print(f" Quads: { {k: f'{v/1e6:.1f}M' for k, v in result.items()} }") + print(f" Total: {total/1e6:.1f}M ({elapsed:.1f}s)") + + +def test_large_country(iso: str, shapes_url: str): + """Fetch real admin-1 shapes from GADM and aggregate with worldpop. + + This is the critical test: large countries (BRA, IDN) have states whose + raster windows exceed 8 M pixels and must be downsampled via Resampling.sum. + """ + print(f"── POST /aggregate/{iso} (real admin-1 shapes from GADM) ──") + + # Fetch shapes + print(f" Fetching {iso}/1 from {shapes_url} ...") + try: + rs = httpx.get(f"{shapes_url}/boundaries/{iso}/1", timeout=120) + except Exception as exc: + print(f" [SKIP] Could not reach GADM at {shapes_url}: {exc}") + return + check(f"GADM HTTP 200 for {iso}", rs.status_code == 200, + f"got {rs.status_code}") + fc = rs.json() + n_features = len(fc["features"]) + print(f" {n_features} features, raw body {len(rs.content)/1024:.0f} KB") + + # Truncate coordinates (same as client.html) + def trunc(coords, dp=3): + if isinstance(coords[0], (int, float)): + return [round(v, dp) for v in coords] + return [trunc(c, dp) for c in coords] + + fc_slim = { + "type": "FeatureCollection", + "features": [ + {**f, "geometry": {**f["geometry"], + "coordinates": trunc(f["geometry"]["coordinates"])}} + for f in fc["features"] + ], + } + import json + body_bytes = json.dumps(fc_slim).encode() + print(f" Slim body: {len(body_bytes)/1024:.0f} KB") + + # Aggregate + print(f" POSTing to {BASE}/aggregate/{iso}?year=2020 ...") + t0 = time.monotonic() + try: + r = httpx.post( + f"{BASE}/aggregate/{iso}?year=2020", + content=body_bytes, + headers={"Content-Type": "application/json"}, + timeout=600, + ) + except Exception as exc: + elapsed = time.monotonic() - t0 + print(f" [FAIL] Exception after {elapsed:.1f}s: {type(exc).__name__}: {exc}") + sys.exit(1) + + elapsed = time.monotonic() - t0 + check("HTTP 200", r.status_code == 200, + f"got {r.status_code}: {r.text[:300]}") + result = r.json() + check(f"{n_features} nodeids returned", len(result) == n_features, + f"got {len(result)}") + total = sum(result.values()) + print(f" {len(result)} regions, total pop {total/1e6:.1f}M ({elapsed:.1f}s)") + check("all nodeids present", len(result) == n_features) + check("total population > 0", total > 0, f"got {total:.0f}") + + +def test_unknown_iso(): + print("── POST /aggregate/ZZZ (unknown ISO) ────────────────────") + r = post("/aggregate/ZZZ", TEST_GEOJSON, timeout=60) + check("HTTP 404", r.status_code == 404, f"got {r.status_code}") + + +def test_bad_body(): + print("── POST /aggregate with non-FeatureCollection body ──────") + r = post(f"/aggregate/{TEST_ISO}", {"type": "Point", "coordinates": [6.1, 49.8]}, timeout=30) + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def test_empty_features(): + print("── POST /aggregate with empty features array ─────────────") + r = post(f"/aggregate/{TEST_ISO}", {"type": "FeatureCollection", "features": []}, timeout=30) + check("HTTP 400", r.status_code == 400, f"got {r.status_code}") + + +def main(): + global BASE + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=BASE, help="worldpop-service base URL") + parser.add_argument("--wait", action="store_true", + help="Wait for service to be ready before running tests") + parser.add_argument("--large-country", metavar="ISO", + help="Also test a large pre-warmed country (e.g. BRA). " + "Requires --gadm-url.") + parser.add_argument("--gadm-url", default="http://48.200.52.126", + help="GADM service base URL (for --large-country shapes)") + args = parser.parse_args() + BASE = args.url.rstrip("/") + + print(f"\nRunning worldpop-service tests against {BASE}\n") + if args.wait: + wait_for_ready() + + test_health() + test_prewarm() + test_aggregate() + test_cache_speed() + test_nga_multipolygon() + if args.large_country: + test_large_country(args.large_country.upper(), args.gadm_url) + test_unknown_iso() + test_bad_body() + test_empty_features() + print("\nAll tests passed.") + + +if __name__ == "__main__": + main() diff --git a/src/laser/init/generate.py b/src/laser/init/generate.py new file mode 100644 index 0000000..f654f94 --- /dev/null +++ b/src/laser/init/generate.py @@ -0,0 +1,369 @@ +""" +generate.py — service-based equivalent of the laser-init Extract + Transform phases. + +Calls the geodata microservices to produce the same data files as `laser-init`, +without downloading anything directly. The Load phase (config.yaml, model scripts, +validation plots) can then be run by passing the output directory to the existing +laser-init CLI once it supports a --data-dir option, or by running the loader +directly. + +Produces (equivalent to laser-init output): + {ISO}_admin{level}.gpkg — GeoPackage (nodeid, name, population, geometry) + cxr.csv — Crude birth/death rates (Time, CBR, CDR) + age_dist.csv — Age distribution (AgeGrpStart, PopTotal) + life_exp.csv — Life expectancy curve (cumulative_deaths) + config.yaml — Model configuration (default parameters) + provenance.json — Data sources and timestamps + +Usage: + python generate.py ETH 2 2010 2020 --shape-source gadm + python generate.py NGA 1 2015 2025 --shape-source unocha --output-dir NGA/2015 + +Service URLs (in precedence order): + 1. --shapes-url / --worldpop-url / --unwpp-url CLI flags + 2. laser_config.yaml keys: gadm_url, geoboundaries_url, unocha_url, worldpop_url, unwpp_url + 3. localhost fallbacks (8101/8102/8103/8104/8100) +""" + +import argparse +import json +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +import geopandas as gpd +import httpx +import pandas as pd +import yaml +from shapely.geometry import shape + +from laser.init.config import configuration as _cfg + +# Localhost fallbacks (used only when neither laser_config.yaml nor --*-url flags supply a URL). +_SHAPE_SOURCE_LOCALHOST = { + "gadm": "http://127.0.0.1:8101", + "geoboundaries": "http://127.0.0.1:8102", + "unocha": "http://127.0.0.1:8103", +} + +# Config-file keys for each shape source. +_SHAPE_SOURCE_CFG_KEY = { + "gadm": "gadm_url", + "geoboundaries": "geoboundaries_url", + "unocha": "unocha_url", +} + + +def _default_shapes_url(source: str) -> str: + return _cfg.get(_SHAPE_SOURCE_CFG_KEY[source], _SHAPE_SOURCE_LOCALHOST[source]) + + +# ── HTTP helpers ─────────────────────────────────────────────────────────────── + +def _get(url: str, timeout: float = 60) -> dict: + r = httpx.get(url, timeout=timeout) + r.raise_for_status() + return r.json() + + +def _post(url: str, body: dict, timeout: float = 1800) -> dict: + r = httpx.post(url, json=body, timeout=timeout) + r.raise_for_status() + return r.json() + + +# ── Service calls ────────────────────────────────────────────────────────────── + +def fetch_shapes(shapes_url: str, iso: str, level: int) -> dict: + url = f"{shapes_url}/boundaries/{iso}/{level}" + print(f" shapes ← {url}") + fc = _get(url) + print(f" {len(fc['features'])} features") + return fc + + +def _truncate_coords(coords, dp: int): + """Recursively round GeoJSON coordinates to `dp` decimal places. + + 3 dp ≈ 100 m at the equator; WorldPop rasters are ~1 km resolution, so + this loses nothing meaningful but cuts payload size 5–10× for detailed + borders (e.g. BRA/1 dropped from ~38 MB to a few MB). Matches the + behaviour of services/worldpop/client.html. + """ + if not coords: + return coords + if isinstance(coords[0], (int, float)): + return [round(v, dp) for v in coords] + return [_truncate_coords(c, dp) for c in coords] + + +def fetch_population(wp_url: str, iso: str, year: int, fc: dict) -> dict: + url = f"{wp_url}/aggregate/{iso}?year={year}" + print(f" worldpop← {url} (first run downloads raster — may take minutes)") + # Truncate coordinates before POSTing (see _truncate_coords). + slim_fc = { + "type": "FeatureCollection", + "features": [ + { + **f, + "geometry": { + **f["geometry"], + "coordinates": _truncate_coords(f["geometry"]["coordinates"], 3), + }, + } + for f in fc["features"] + ], + } + # The Azure LB has a 4-min idle timeout; a slow raster download causes a + # connection reset mid-request. Retry: the download continues server-side + # and subsequent calls hit the cache quickly. + for attempt in range(1, 11): + try: + pop = _post(url, slim_fc) + break + except (httpx.RemoteProtocolError, httpx.ReadError, httpx.ConnectError): + if attempt == 10: + raise + wait = 60 + print(f" worldpop connection reset (raster download in progress) " + f"— retry {attempt}/10 in {wait}s ...") + time.sleep(wait) + total = sum(pop.values()) + print(f" {len(pop)} features, total pop {total:,.0f}") + return pop + + +def fetch_demographics(unwpp_url: str, iso: str, start_year: int, end_year: int) -> dict: + url = f"{unwpp_url}/demographics/{iso}?start_year={start_year}&end_year={end_year}" + print(f" unwpp ← {url}") + demo = _get(url) + print(f" {len(demo['cxr'])} years CBR/CDR, " + f"{len(demo['age_dist'])} age groups, " + f"{len(demo['life_exp'])} life-exp rows") + return demo + + +# ── Build outputs ────────────────────────────────────────────────────────────── + +def build_gdf(fc: dict, pop: dict) -> gpd.GeoDataFrame: + rows = [] + for f in fc["features"]: + p = f["properties"] + nodeid = int(p["nodeid"]) + rows.append({ + "nodeid": nodeid, + "name": p.get("name", ""), + "population": round(float(pop.get(str(nodeid), 0.0))), + "geometry": shape(f["geometry"]), + }) + gdf = gpd.GeoDataFrame(rows, crs="EPSG:4326") + return gdf.sort_values("nodeid").reset_index(drop=True) + + +def write_gpkg(gdf: gpd.GeoDataFrame, output_dir: Path, iso: str, level: int) -> Path: + dest = output_dir / f"{iso}_admin{level}.gpkg" + gdf.to_file(dest, driver="GPKG") + print(f" → {dest.name} ({len(gdf)} features, total pop {gdf.population.sum():,.0f})") + return dest + + +def write_cxr(demo: dict, output_dir: Path) -> Path: + dest = output_dir / "cxr.csv" + pd.DataFrame( + [{"Time": r["year"], "CBR": r["CBR"], "CDR": r["CDR"]} for r in demo["cxr"]] + ).to_csv(dest, index=False) + print(f" → {dest.name} ({len(demo['cxr'])} rows)") + return dest + + +def write_age_dist(demo: dict, output_dir: Path) -> Path: + dest = output_dir / "age_dist.csv" + pd.DataFrame( + [{"AgeGrpStart": r["age_start"], "PopTotal": r["pop_total"]} for r in demo["age_dist"]] + ).to_csv(dest, index=False) + print(f" → {dest.name} ({len(demo['age_dist'])} rows)") + return dest + + +def write_life_exp(demo: dict, output_dir: Path) -> Path: + dest = output_dir / "life_exp.csv" + pd.DataFrame( + [{"cumulative_deaths": r["cumulative_deaths"]} for r in demo["life_exp"]] + ).to_csv(dest, index=False) + print(f" → {dest.name} ({len(demo['life_exp'])} rows)") + return dest + + +def write_config(output_dir: Path, iso: str, level: int) -> Path: + """Write a standalone config.yaml (used when --emit-scripts is not set).""" + dest = output_dir / "config.yaml" + # Key names match what AbmLoader and the model scripts expect (underscores). + cfg = { + "data_dir": str(output_dir.resolve()), + "datafiles": { + "shape_data": f"{iso}_admin{level}.gpkg", + "cxr_data": "cxr.csv", + "pop_data": "age_dist.csv", + "exp_data": "life_exp.csv", + }, + "simulation": { + "nyears": 10, + "r0": 2.5, + "exposed_duration_shape": 4.5, + "exposed_duration_scale": 1.0, + "infectious_duration_mean": 7.0, + "naive_population": True, + "gravity_k": 500, + "gravity_a": 1, + "gravity_b": 1, + "gravity_c": 2, + }, + } + with dest.open("w") as f: + yaml.safe_dump(cfg, f, default_flow_style=False, sort_keys=False) + print(f" → {dest.name}") + return dest + + +def _emit_scripts( + output_dir: Path, + iso: str, + level: int, + gpkg: Path, + cxr: Path, + age_dist: Path, + life_exp: Path, + model: str, +) -> None: + """Call the laser-init Load phase to emit model scripts and validation plots.""" + from laser.init.cli import write_plots + from laser.init.loaders.abm import AbmLoader + + print("\nEmitting model scripts via laser-init Load phase ...") + AbmLoader().emit_script( + mode="ABM", + model=model, + shape_filename=gpkg, + cxr_filename=cxr, + pop_filename=age_dist, + exp_filename=life_exp, + output_dir=output_dir, + ) + print(f" → config.yaml, {model.lower()}.py, plot.py") + + print("Writing validation plots ...") + write_plots(gpkg, cxr, age_dist, life_exp, output_dir) + print(" → choropleth.png, cbr_cdr.png, age_distribution.png, life_expectancy.png, report.pdf") + + +def write_provenance( + output_dir: Path, iso: str, level: int, + shapes_url: str, wp_url: str, unwpp_url: str, + shape_source: str, raster_year: int, +) -> Path: + dest = output_dir / "provenance.json" + ts = datetime.now(timezone.utc).isoformat() + prov = { + f"{iso}_admin{level}.gpkg": { + "shape_source": shape_source, + "shapes_service": shapes_url, + "worldpop_service": wp_url, + "worldpop_year": raster_year, + "timestamp": ts, + }, + "cxr.csv": {"unwpp_service": unwpp_url, "timestamp": ts}, + "age_dist.csv": {"unwpp_service": unwpp_url, "timestamp": ts}, + "life_exp.csv": {"unwpp_service": unwpp_url, "timestamp": ts}, + } + dest.write_text(json.dumps(prov, indent=2)) + print(f" → {dest.name}") + return dest + + +# ── CLI ──────────────────────────────────────────────────────────────────────── + +def main() -> None: + parser = argparse.ArgumentParser( + description="Generate laser-init data files via geodata microservices.", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("country", help="ISO-3 country code (e.g. ETH)") + parser.add_argument("level", type=int, help="Admin level (0–3)") + parser.add_argument("start_year", type=int, help="Start year (e.g. 2010)") + parser.add_argument("end_year", type=int, help="End year (e.g. 2020)") + parser.add_argument("--shape-source", + choices=["gadm", "geoboundaries", "unocha"], default="unocha", + help="Boundary data source (default: unocha)") + parser.add_argument("--output-dir", type=Path, default=None, + help="Output directory (default: ./{ISO}/{start_year})") + parser.add_argument("--shapes-url", default=None, + help="Shape service base URL (auto-selected from --shape-source; " + "falls back to laser_config.yaml then localhost)") + parser.add_argument("--worldpop-url", + default=_cfg.get("worldpop_url", "http://127.0.0.1:8104")) + parser.add_argument("--unwpp-url", + default=_cfg.get("unwpp_url", "http://127.0.0.1:8100")) + parser.add_argument("--raster-year", type=int, default=None, + help="WorldPop raster year (default: start_year clamped to 2020)") + parser.add_argument("--model", choices=["SI", "SIR", "SEIR"], default="SEIR", + help="Model type for emitted script (default: SEIR)") + parser.add_argument("--emit-scripts", action="store_true", + help="Also emit model scripts and validation plots via laser-init " + "Load phase (requires laser-init to be installed)") + args = parser.parse_args() + + iso = args.country.upper() + level = args.level + start_year = args.start_year + end_year = args.end_year + shape_src = args.shape_source + raster_year = args.raster_year or min(start_year, 2020) + output_dir = args.output_dir or (Path(iso) / str(start_year)) + + shapes_url = (args.shapes_url or _default_shapes_url(shape_src)).rstrip("/") + wp_url = args.worldpop_url.rstrip("/") + unwpp_url = args.unwpp_url.rstrip("/") + + output_dir.mkdir(parents=True, exist_ok=True) + + print(f"\nlaser-init (services) — {iso} admin{level} {start_year}–{end_year}" + f" shape-source={shape_src}") + print() + + try: + fc = fetch_shapes(shapes_url, iso, level) + pop = fetch_population(wp_url, iso, raster_year, fc) + demo = fetch_demographics(unwpp_url, iso, start_year, end_year) + except httpx.HTTPStatusError as exc: + print(f"\nHTTP {exc.response.status_code} from {exc.request.url}") + print(exc.response.text[:300]) + sys.exit(1) + except Exception as exc: + print(f"\nError: {exc}") + sys.exit(1) + + print() + gdf = build_gdf(fc, pop) + gpkg = write_gpkg(gdf, output_dir, iso, level) + cxr = write_cxr(demo, output_dir) + age_dist = write_age_dist(demo, output_dir) + life_exp = write_life_exp(demo, output_dir) + write_provenance(output_dir, iso, level, + shapes_url, wp_url, unwpp_url, shape_src, raster_year) + + if args.emit_scripts: + _emit_scripts(output_dir, iso, level, gpkg, cxr, age_dist, life_exp, args.model) + else: + write_config(output_dir, iso, level) + print(f"\nDone — {output_dir}/") + print(f"Run with --emit-scripts to also generate {args.model.lower()}.py, " + f"plot.py, config.yaml, and validation plots.") + return + + print(f"\nDone — {output_dir}/") + print(f"To run the model:\n cd {output_dir} && python {args.model.lower()}.py") + + +if __name__ == "__main__": + main() diff --git a/tests/test_generate.py b/tests/test_generate.py new file mode 100644 index 0000000..d70da94 --- /dev/null +++ b/tests/test_generate.py @@ -0,0 +1,269 @@ +"""Tests for laser.init.generate module (laser-generate entry point). + +Covers module/entry-point existence, argument parsing, data-building helpers, +and file writers. All tests are offline — network calls are mocked. +""" + +import json +import sys +from pathlib import Path +from unittest.mock import patch + +import geopandas as gpd +import pandas as pd +import pytest + +from laser.init import generate as gen + +# ── Sample data shared across tests ─────────────────────────────────────────── + +SAMPLE_FC = { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": {"nodeid": 0, "name": "North"}, + "geometry": {"type": "Polygon", "coordinates": [ + [[3.0, 7.0], [3.0, 14.0], [15.0, 14.0], [15.0, 7.0], [3.0, 7.0]] + ]}, + }, + { + "type": "Feature", + "properties": {"nodeid": 1, "name": "South"}, + "geometry": {"type": "Polygon", "coordinates": [ + [[3.0, 4.0], [3.0, 7.0], [15.0, 7.0], [15.0, 4.0], [3.0, 4.0]] + ]}, + }, + ], +} + +SAMPLE_POP = {"0": 100_000.0, "1": 50_000.0} + +SAMPLE_DEMO = { + "cxr": [{"year": 2020, "CBR": 37.0, "CDR": 13.0}], + "age_dist": [ + {"age_start": 0, "pop_total": 20000}, + {"age_start": 5, "pop_total": 18000}, + ], + "life_exp": [{"cumulative_deaths": i * 100} for i in range(101)], +} + + +# ── Entry point existence ────────────────────────────────────────────────────── + +class TestEntryPoint: + def test_main_callable(self): + """main() must exist and be callable — it is the laser-generate entry point.""" + assert callable(gen.main) + + def test_entry_point_registered(self): + """laser-generate must be declared in pyproject.toml [project.scripts].""" + pyproject = Path(__file__).parent.parent / "pyproject.toml" + content = pyproject.read_text() + assert 'laser-generate = "laser.init.generate:main"' in content + + def test_module_importable(self): + """laser.init.generate must be importable (i.e. it lives in the package tree).""" + import laser.init.generate # noqa: F401 + + +# ── Argument parsing ─────────────────────────────────────────────────────────── + +class TestArgParsing: + def _parse(self, argv): + """Run main() with mocked HTTP and sys.argv; return exit code.""" + with patch.object(sys, "argv", ["laser-generate"] + argv): + with patch.object(gen, "fetch_shapes", return_value=SAMPLE_FC): + with patch.object(gen, "fetch_population", return_value=SAMPLE_POP): + with patch.object(gen, "fetch_demographics", return_value=SAMPLE_DEMO): + try: + gen.main() + return 0 + except SystemExit as exc: + return exc.code + + def test_help_exits_zero(self): + """--help must print usage and exit 0.""" + with patch.object(sys, "argv", ["laser-generate", "--help"]): + with pytest.raises(SystemExit) as exc_info: + gen.main() + assert exc_info.value.code == 0 + + def test_missing_args_exits_nonzero(self): + """Omitting required positional args must exit non-zero.""" + with patch.object(sys, "argv", ["laser-generate"]): + with pytest.raises(SystemExit) as exc_info: + gen.main() + assert exc_info.value.code != 0 + + def test_default_shape_source_is_unocha(self, tmp_path): + """--shape-source defaults to 'unocha' when not specified. + + Patches _default_shapes_url to return a known sentinel per source, so + the assertion is independent of any laser_config.yaml on the dev's box. + """ + called_urls = [] + + def capture_shapes(url, iso, level): + called_urls.append(url) + return SAMPLE_FC + + def fake_default(source): + return f"http://test-{source}.example" + + with patch.object(sys, "argv", ["laser-generate", "NGA", "1", "2020", "2020", + "--output-dir", str(tmp_path)]): + with patch.object(gen, "_default_shapes_url", side_effect=fake_default): + with patch.object(gen, "fetch_shapes", side_effect=capture_shapes): + with patch.object(gen, "fetch_population", return_value=SAMPLE_POP): + with patch.object(gen, "fetch_demographics", return_value=SAMPLE_DEMO): + gen.main() + + assert len(called_urls) == 1 + assert "test-unocha" in called_urls[0] + + def test_gadm_shape_source_uses_correct_port(self, tmp_path): + """--shape-source gadm must route to the gadm URL, not unocha's.""" + called_urls = [] + + def capture_shapes(url, iso, level): + called_urls.append(url) + return SAMPLE_FC + + def fake_default(source): + return f"http://test-{source}.example" + + with patch.object(sys, "argv", ["laser-generate", "NGA", "1", "2020", "2020", + "--shape-source", "gadm", + "--output-dir", str(tmp_path)]): + with patch.object(gen, "_default_shapes_url", side_effect=fake_default): + with patch.object(gen, "fetch_shapes", side_effect=capture_shapes): + with patch.object(gen, "fetch_population", return_value=SAMPLE_POP): + with patch.object(gen, "fetch_demographics", return_value=SAMPLE_DEMO): + gen.main() + + assert "test-gadm" in called_urls[0] + + +# ── Data-building helpers ────────────────────────────────────────────────────── + +class TestBuildGdf: + def test_returns_geodataframe(self): + """build_gdf must return a GeoDataFrame.""" + gdf = gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + assert isinstance(gdf, gpd.GeoDataFrame) + + def test_row_count_matches_features(self): + """One row per feature.""" + gdf = gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + assert len(gdf) == len(SAMPLE_FC["features"]) + + def test_population_values_correct(self): + """Population column must reflect the pop dict values, rounded.""" + gdf = gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + assert gdf.loc[gdf.nodeid == 0, "population"].iloc[0] == 100_000 + assert gdf.loc[gdf.nodeid == 1, "population"].iloc[0] == 50_000 + + def test_sorted_by_nodeid(self): + """Rows must be sorted by nodeid ascending.""" + gdf = gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + assert list(gdf.nodeid) == sorted(gdf.nodeid.tolist()) + + def test_crs_is_wgs84(self): + """GeoDataFrame must have WGS-84 CRS.""" + gdf = gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + assert gdf.crs is not None + assert gdf.crs.to_epsg() == 4326 + + def test_missing_pop_defaults_to_zero(self): + """Features absent from pop dict get population=0.""" + gdf = gen.build_gdf(SAMPLE_FC, {"0": 999.0}) # nodeid 1 missing + assert gdf.loc[gdf.nodeid == 1, "population"].iloc[0] == 0 + + +# ── File writers ─────────────────────────────────────────────────────────────── + +class TestFileWriters: + @pytest.fixture + def gdf(self): + return gen.build_gdf(SAMPLE_FC, SAMPLE_POP) + + def test_write_gpkg_creates_file(self, tmp_path, gdf): + dest = gen.write_gpkg(gdf, tmp_path, "NGA", 1) + assert dest.exists() + assert dest.name == "NGA_admin1.gpkg" + + def test_write_gpkg_readable(self, tmp_path, gdf): + dest = gen.write_gpkg(gdf, tmp_path, "NGA", 1) + result = gpd.read_file(dest) + assert len(result) == len(gdf) + + def test_write_cxr_creates_file(self, tmp_path): + dest = gen.write_cxr(SAMPLE_DEMO, tmp_path) + assert dest.exists() + df = pd.read_csv(dest) + assert list(df.columns) == ["Time", "CBR", "CDR"] + assert len(df) == 1 + assert df.iloc[0]["CBR"] == pytest.approx(37.0) + + def test_write_age_dist_creates_file(self, tmp_path): + dest = gen.write_age_dist(SAMPLE_DEMO, tmp_path) + assert dest.exists() + df = pd.read_csv(dest) + assert list(df.columns) == ["AgeGrpStart", "PopTotal"] + assert len(df) == 2 + + def test_write_life_exp_creates_file(self, tmp_path): + dest = gen.write_life_exp(SAMPLE_DEMO, tmp_path) + assert dest.exists() + df = pd.read_csv(dest) + assert "cumulative_deaths" in df.columns + assert len(df) == 101 + + def test_write_provenance_creates_json(self, tmp_path): + dest = gen.write_provenance( + tmp_path, "NGA", 1, + "http://localhost:8101", "http://localhost:8104", "http://localhost:8100", + "gadm", 2020, + ) + assert dest.exists() + prov = json.loads(dest.read_text()) + assert "NGA_admin1.gpkg" in prov + assert prov["NGA_admin1.gpkg"]["worldpop_year"] == 2020 + + +# ── End-to-end (offline, mocked services) ───────────────────────────────────── + +class TestMainEndToEnd: + def test_produces_expected_files(self, tmp_path): + """Running main() with mocked services must produce all core output files.""" + with patch.object(sys, "argv", ["laser-generate", "NGA", "1", "2020", "2020", + "--output-dir", str(tmp_path)]): + with patch.object(gen, "fetch_shapes", return_value=SAMPLE_FC): + with patch.object(gen, "fetch_population", return_value=SAMPLE_POP): + with patch.object(gen, "fetch_demographics", return_value=SAMPLE_DEMO): + gen.main() + + assert (tmp_path / "NGA_admin1.gpkg").exists() + assert (tmp_path / "cxr.csv").exists() + assert (tmp_path / "age_dist.csv").exists() + assert (tmp_path / "life_exp.csv").exists() + assert (tmp_path / "provenance.json").exists() + assert (tmp_path / "config.yaml").exists() + + def test_raster_year_clamped_to_2020(self, tmp_path): + """start_year > 2020 must clamp raster_year to 2020 (WorldPop cap).""" + pop_years = [] + + def capture_pop(url, iso, year, fc): + pop_years.append(year) + return SAMPLE_POP + + with patch.object(sys, "argv", ["laser-generate", "NGA", "1", "2025", "2025", + "--output-dir", str(tmp_path)]): + with patch.object(gen, "fetch_shapes", return_value=SAMPLE_FC): + with patch.object(gen, "fetch_population", side_effect=capture_pop): + with patch.object(gen, "fetch_demographics", return_value=SAMPLE_DEMO): + gen.main() + + assert pop_years == [2020]