AWS Builder Center

Builder Center

PlumeTrace: 72-Hour Source-Attributed PM2.5 Forecast and Action Engine on AWS (Amazon Bedrock, Step Functions, Lambda)

PlumeTrace traces Delhi-NCR air pollution to source crop fires using a serverless forecast engine and an Amazon Bedrock Copilot that drafts district reports, farmer alerts, and gig-rider plans. Deep dive into Step Functions, GRIB decoding, and OR-Tools. | Amazon Bedrock | AWS Step Functions | AWS Lambda | Amazon DynamoDB | Amazon S3 | Amazon EventBridge | Amazon API Gateway | Amazon Cognito | Amazon Translate | Amazon Polly | Amazon Location Service | AWS CDK |

Building PlumeTrace: 72-Hour Source-Attributed PM2.5 Forecast and Action Engine on AWS

Every autumn, Delhi-NCR's air quality collapses, and the public conversation collapses with it into the same argument: how much of this is crop-residue burning in Punjab and Haryana, and how much is local? Most dashboards show you what the PM2.5 is. We wanted to build something that answers where it came from — with an uncertainty band, not a verdict — and then does something useful with that answer.
PlumeTrace is one forecasting engine with two products on top of it, built for the WeMakeDevs × AWS Environmental Hacks (Air track) over four days. Every few hours it combines satellite fire detections, weather-model winds and ground sensors, traces the air backwards from each monitoring station, and estimates how much of the next 72 hours of PM2.5 is coming from crop fires and which districts they're in. On top of that sits an Amazon Bedrock-powered Copilot that turns forecasts into targeted actions: district evidence reports, supportive farmer alerts with audio, and gig-rider shift plans that cap pollution exposure.
This is a technical write-up of the engineering that powers it — the distributed orchestration, data decoding, generative AI workflows with Bedrock, and the architectural decisions we made along the way.


The shape of the system

The whole thing is contract-first, event-driven, and serverless. Modules never call each other's internals; they talk through frozen schemas and EventBridge events.


The stack :

Area
Stack
Engine / science
Python 3.12 · xarray/cfgrib · NumPy/SciPy · LightGBM · h3
API + Copilot
Bun + Express 5 on Lambda (Lambda Web Adapter, response streaming) · Amazon Bedrock (Claude) ConverseStream + Guardrails · SSE
Fleet solver
OR-Tools CP-SAT (+ greedy fallback) · Amazon Location route matrix
Data / async
DynamoDB (on-demand) · S3 · EventBridge · Step Functions
Web
React 19 + Vite · MapLibre + deck.gl · Cognito · Amplify Hosting
IaC
AWS CDK (TypeScript)


Decision 1: Contracts frozen in hour two, so nobody blocks anyone

Five people, four days, one pipeline where every stage depends on the previous one's output. The thing that saved us was freezing the data contracts before anyone wrote real logic.  zod  is the source of truth; it's exported to JSON Schema and mirrored by hand in Pydantic, and contract tests check both against 11 realistic mock JSON files.
The payoff is that everything has a mock mode. The API serves  contracts/mocks  under  MOCK_MODE=1 ; the Python engine runs offline against  ./.local-s3  under  PT_LOCAL=1 . The frontend built against a deployed mock API URL on day 1 and never waited for the real engine. When the real data landed, it dropped into the same shapes.


Decision 2: The agent runs on live data and produces the deliverables

The Copilot is an intelligent tool-use loop (max 10 turns) on Amazon Bedrock, running inside the API rather than as a chatbot bolted on the side. It isn't answering from training data — it executes a real tool loop against live state: it calls the engine and the REST APIs for the current forecast, source attribution and fleet status, and from that live data it builds finished deliverables — the district evidence report, the Punjabi farmer alert, and the gig-rider shift plan.
Those tools return real engine/API payloads, which the model works on turn after turn until the report is complete. The one step the agent leaves to a person is the final send: a human reviews the generated report/alert/plan and approves it, and that approval publishes the  action.approved  event that delivers it. That keeps a human accountable for anything that reaches a real phone, and doubles as the prompt-injection boundary.
JavaScript
// api/src/agent/loop.js — each turn runs tools against live engine/API data,

// feeds the results back, and continues until the deliverable is built.

const msg = await stream.finalMessage();

messages.push({ role: 'assistant', content: msg.content });

if (msg.stop_reason !== 'tool_use') {

  onEvent({ type: 'done', stop_reason: msg.stop_reason });

  return;

}

const toolUses = msg.content.filter((b) => b.type === 'tool_use');

for (const tu of toolUses) {

  onEvent({ type: 'tool_call', tool_use_id: tu.id, tool_name: tu.name, input: tu.input });

  const { output, actionType, actionId } = await runTool(tu.name, tu.input, { user });

  // runTool hits the real engine/API; its result feeds the next turn.

}

Guardrails and streaming

The agent runs on Amazon Bedrock (Claude) with ConverseStream and Bedrock Guardrails, so the deny topics — individual-farmer PII, certainty beyond the model, blaming language — along with the grounding and supportive-tone rules are enforced on every turn. The model transport is kept pluggable behind the loop, so the same zod-typed tools and every  ChatStreamEvent  stay identical no matter what sits underneath. SSE still streams through Lambda because the container runs the Lambda Web Adapter with  AWS_LWA_INVOKE_MODE=response_stream .
The lesson: Keep the model transport pluggable behind the agent loop, and the loop, its tools and its events stay stable and testable.


The engine: Orchestrating a scientific pipeline on Step Functions

 EngineRun  is a Standard workflow. Each state is a Lambda container image (the scientific stack — cfgrib/eccodes, LightGBM, h3 — packaged cleanly in ECR), with a 15-minute per-function timeout and a 30-minute workflow timeout.
text
ResolveRun

  → Parallel(

      IngestFIRMS  (catch → resilient fallback, pipeline continues),

      Map IngestGFSHour over f000..f072,

      IngestOpenAQ

    )

  → Trajectories → Forecast → Gridding → Summarize → Publish
The handlers are deliberately thin — one per state, no heavy science logic inside handlers. They enforce a single execution envelope,  {run_id, degraded[]} , threading through the whole run:
Python
# engine/plumetrace_engine/handlers.py

@handler("ingest_openaq")

def ingest_openaq(payload: dict) -> dict:

    payload = _env(payload)

    # OpenAQ supplies observations for verification, not the forecast itself.

    # Transient API rate-limits must NOT crash the pipeline — flag and continue.

    try:

        payload["openaq"] = openaq.ingest(_secret("plumetrace/openaq_key", "OPENAQ_API_KEY"))

    except Exception as exc:

        payload.setdefault("degraded", []).append("openaq")

        payload["openaq"] = {"degraded": True}

    return payload

Resilience and data transparency

This envelope architecture ensures high reliability and data provenance. If an upstream external sensor provider is temporarily rate-limited, the pipeline doesn't crash or hide gaps — it completes successfully, transparently flags which input feeds were impacted, and conveys that metadata directly to the UI banner and published summary. For an environmental intelligence system, honest data provenance and transparency build lasting trust.


The science, briefly

Back-trajectories

We build a  WindField  from the GFS run — an xarray stack sampled bilinearly in space and linearly in time. The transport wind is the mean of the 925 hPa and 10 m winds, dropping to 10 m only where the boundary layer is shallow ( hpbl  below threshold), because a shallow layer decouples 925 hPa from the surface:
Python
u = 0.5 * (s["u925"] + s["u10"])

v = 0.5 * (s["v925"] + s["v10"])

shallow = s["hpbl"] < self.cfg.hpbl_shallow_m

u = np.where(shallow, s["u10"], u)   # surface-only transport in a shallow PBL

v = np.where(shallow, s["v10"], v)

The forecast, with a model you can swap

The forecast layer is model-agnostic. At each cold start the engine loads a trained LightGBM p10/p50/p90 quantile model if its artifact is present in the model dir, and otherwise falls back to a persistence baseline — the same  predict_quantiles(df) → (p10, p50, p90)  interface and the same output contract either way. Promoting the trained model to the live path is dropping  q10/q50/q90.txt + metadata.json  into the model dir; nothing in the pipeline changes.
Python
# engine/.../model/infer.py — one interface, trained model or safe fallback

def load_model(model_dir=None, cfg=CONFIG):

    if model_dir and Path(model_dir).exists():

        try:

            boosters = {a: lgb.Booster(model_file=f"{model_dir}/{n}.txt")

                        for a, n in zip(QUANTILES, ("q10", "q50", "q90"))}

            return LightGBMQuantileModel(boosters, meta)  # log-target, expm1, de-crossed

        except Exception:

            pass                        # a bad artifact never blocks a run

    return PersistenceModel()           # persistence-v0 baseline
The LightGBM path trains on a log target, inverts with  expm1 , and fixes quantile crossing by sorting each row. The training side is a full offline pipeline —  build_training_set.py → train.py → backtest.py  over ERA5/FIRMS/OpenAQ backfills, with a SageMaker job option — and the Forecast-Skill page scores live skill separately from backtest, so the two are never conflated. Running the persistence baseline while the trained model is promoted through that skill gate is a deliberate staging choice: the counterfactual attribution, the live fire and wind inputs, and the p10–p90 bands are real regardless of which model sits behind the interface.

Attribution is a counterfactual, not a label

For each station/lead we predict PM2.5 with the real fire load, then re-predict with fire features zeroed, and take the fire share as the clipped ratio of the difference — for every quantile:
Python
share = np.clip((full - nofire) / np.maximum(full, EPS), 0.0, 1.0)

share.sort(axis=1)  # keep p10 <= p50 <= p90 after the ratio
District shares then split that fire share by each district's fire load $FL_d / \sum FL_d$. We always publish the p10–p90 band. Reports never name individuals.

Gridding

Gridding is IDW (power 2, up to 6 nearest stations, cut off at 25 km, null beyond) onto an H3 res-7 grid over the NCR box — the shape  deck.gl  renders directly.


Scaling compute and distributed execution

Orchestrating dozens of forecast hours and scientific libraries in a serverless environment required deliberate concurrency and resource profiling:
  • Step Functions Map concurrency tuning: The GFS ingest maps over 73 forecast hours (f000–f072). We tune  maxConcurrency  directly in CDK to throttle parallel downloads cleanly, matching network throughput without overloading downstream workers.
  • Optimized memory profiling: Scientific containers handling atmospheric datasets were profiled to operate efficiently within allocated serverless memory envelopes, balancing cold-start speed and numerical performance.
  • Resilient retry strategies: Upstream telemetry APIs have rate limits. We implement exponential backoff and jitter directly in Step Functions task definitions rather than relying on brittle client loops.
TypeScript
// infra/lib/engine-stack.ts — Step Functions Map with calibrated concurrency

const gfsMap = new sfn.Map(this, "IngestGFS", {

  maxConcurrency: 3, // calibrated for optimal parallel throughput

  itemSelector: { 

    "run_id.$": "$.run_id", 

    "fff.$": "$$.Map.Item.Value",

    "degraded.$": "$.degraded" 

  },

});

// Resilient retry with exponential backoff on throttling

t.addRetry({ /* exponential backoff on Lambda.TooManyRequestsException */ });


Other sharp edges worth naming

  • GRIB decoding in Lambda: GFS GRIB2 files carry conflicting level coordinates that break an xarray merge. The fix was to drop the conflicting level-coords before merging and to store  valid_time  explicitly as  numpy.datetime64 . Cryptic to diagnose, one line to fix.
  • Partial GFS downloads: Full GFS files are ~500 MB. We fetch byte ranges using the  .idx  sidecar and crop to the AOI + 1°, which keeps each S3 object around 10 MB.
  • DynamoDB GSI sort keys are single attributes: We needed to query "the map for this run at this valid hour," i.e. sort on  valid_hour#h3 . GSIs don't take composite sort keys, so each Forecast item carries a dedicated  gsi1sk = "<valid_hour>#<h3>"  and the API reads  run_id = :r AND begins_with(gsi1sk, :valid_hour) .
  • Multilingual audio delivery: Farmer alerts go out as Punjabi text (via Amazon Translate) paired with regional audio delivery via Amazon Polly. Correctness and accessibility over assumptions.
  • Cloud builds, no local Docker dependency: The scientific image builds in CodeBuild and is pushed to ECR; CDK points the Lambdas at the new tag. Nobody's laptop RAM is on the critical path, and iteration loops remain smooth.


Limitations we state up front

The inputs are real — live satellite fires and live weather-model winds — and we'd rather name the limits than oversell. Two matter most scientifically:
  • Satellite overpass bias: VIIRS passes are ~13:30 and ~01:30 local, and cloud/haze hides fires, so detected fire radiative power is a lower bound and a relative index, never a fire count.
  • Fire-persistence assumption: for hours beyond the latest detection we assume detected fires persist (scaled down per day). This is the single biggest forecast uncertainty, and it's printed on every report.
The trajectory model is deliberately kinematic — not WRF-Chem or HYSPLIT-grade dispersion.
Scope notes: ground observations feed verification and the forecast-skill scoring (live vs backtest) and are gap-filled where sensors are sparse, rather than driving the forecast directly; fleet data is synthetic and labelled SIMULATED, with rider health confined to a KMS-encrypted table that's never exposed over the API.


The fleet solver: Capping exposure with CP-SAT

The gig-rider product re-plans delivery shifts to cap each rider's inhaled pollution dose. It's an OR-Tools CP-SAT model over 30-minute slots: boolean decision variables per  (order, rider, slot) , dose coefficients precomputed from the forecast grid, a soft 100% budget constraint with a large penalty, and an objective that minimizes the worst rider's exposure first:
Python
model.Minimize(1000 * max_pct + total_dose + 2 * total_extra

               + OVER_PENALTY * sum(over_terms))
A greedy solver is built first and always returns a feasible plan, so CP-SAT infeasibility or a timeout never breaks the demo. The target was 50 riders solved end-to-end in under 60 seconds; a 30-second solve limit across 8 workers hits it comfortably.


What we'd take to the next project

  • Contract-first with real mocks is worth the two hours. It's what let five people build a tightly-coupled pipeline in parallel without stepping on each other.
  • Build for graceful degradation. An envelope that tracks upstream data availability turns a fragile pipeline into a resilient system with full data lineage.
  • Separate the swappable parts. Because the agent logic didn't know or care which transport sat behind it, the model layer stayed clean and modular.
  • Calibrate concurrency and resources early. Building explicit concurrency tuning and retry policies into Step Functions ensures your system stays rock-solid under high volume.
PlumeTrace won't settle the stubble-burning debate. But it shows a path: real public data, an estimate with its uncertainty shown honestly, and an intelligent Bedrock Copilot with human approval before anything reaches a real person.
# amazon-bedrock# serverless# amazon-sagemaker# aws-lambda# amazon-dynamodb
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article