Temporal aggregation infrastructure (Steps 1-3)

[espg]

Summary

Adds the infrastructure for temporal aggregation pipelines, enabling magg to support use cases beyond spatial binning (e.g., computing storm summary statistics from reanalysis data). Implements Steps 1–3 from the implementation plan on #12.

Motivated by the aggregation patterns in antarctic_AR_dataset, which does identical orchestration/dispatch but for temporal reduction of gridded fields under spatial masks.

What's done

• pipeline.type config field — configs can now declare themselves as spatial (default, backward compatible), temporal, or event. Validation branches per type: spatial pipelines validate grid/source/function; temporal pipelines validate collections, spatial_func, and temporal_reducer references.

• Shared orchestration (orchestrate.py) — extracted the Lambda dispatch machinery (ThreadPoolExecutor + boto3 invoke + retry + log parsing + cost estimation) that was duplicated between magg and antarctic_AR_dataset. invoke_lambda.py refactored to use it.

• Generalized auth — get_s3_credentials(daac=) accepts any DAAC (NSIDC, GES_DISC, etc.); get_nsidc_s3_credentials kept as alias.

• Temporal building blocks (temporal.py) — ported from antarctic_AR_dataset:

  • 5 streaming accumulators: Max, Min, Sum, WeightedMean, FirstLandfallCapture
  • 6 per-timestep spatial functions: max, min, weighted_sum, weighted_mean, max_gradient, min_over_levels
  • specs_from_config() bridge to convert YAML config to internal spec dicts

• merra2_storm.yaml built-in config — 15 storm metrics matching the antarctic_AR_dataset registry, fully validated by the temporal config path.

• Tests — 187 pass (88 new). Covers accumulators, spatial functions with synthetic xarray grids, config roundtrips, temporal validation, orchestrate pure functions.

What remains

- process_storm() — the main temporal processing function (analogous to process_morton_cell() for spatial). Reads MERRA-2 from S3 via xarray, applies spatial funcs per timestep, feeds streaming accumulators, returns scalar dict. To be ported/adapted from artools/cloud/worker.py.
- Temporal Lambda handler — deployment/aws/temporal_handler.py wrapping process_storm() with serialization/deserialization of event payloads (base64 storm masks, etc.).
- Temporal orchestrator CLI — a main() equivalent that loads storm catalogs, builds granule indices, and dispatches temporal Lambda invocations using orchestrate.dispatch_lambda().

The above was the original plan, written before #13 landed. Now that runner.py provides the composable agg() API with pluggable backends, the remaining work is:

• process_event() — the temporal equivalent of process_morton_cell(). Reads gridded data from S3 via xarray, applies spatial funcs per timestep, feeds streaming accumulators, returns scalar dict per event. Ported/adapted from artools/cloud/worker.py. Named process_event() (not process_storm()) since it generalizes beyond storms.
• Extend agg() in runner.py — branch on get_pipeline_type(config) == "temporal" to call _run_temporal_local() / _run_temporal_lambda(), which iterate over events (not morton cells) and call process_event(). The existing CLI (invoke_lambda.py) would work for temporal pipelines with no changes — just pass --config merra2_storm.yaml.
• DRY up runner.py Lambda dispatch — _run_lambda() / _invoke_lambda_cell() duplicates retry + ThreadPoolExecutor logic that orchestrate.dispatch_lambda() now provides. Should refactor to use it.
• Derived variable convention — how to handle computed variables like _rainfall = PRECCU + PRECLS in config (currently special-cased in artools worker).
• Event pipeline type (pipeline.type: event) — dynamic geometry, lifecycle metadata. Deferred until temporal works end-to-end.

Closes #12 when all remaining items are complete.

Test plan

• All 187 existing + new tests pass (pytest tests/)
• Backward compatibility: default_config("atl06") works unchanged, defaults to spatial
• merra2_storm.yaml loads and validates as temporal pipeline
• Accumulator edge cases: empty, NaN, None inputs
• Spatial functions tested with synthetic 3×3 xarray grids
• Orchestrate cost estimation and log parsing verified with synthetic data
• Integration test with actual MERRA-2 data (deferred — requires S3 access)

🤖 Generated with Claude Code

[espg]

I'd like to pick this back up -- but main has diverged significantly over the past three weeks. Before we can advance this, we'll need to reconcile the conflicts with main. please do that.

[espg]

🤖 from Claude

Investigated the reconciliation. This isn't a merge-conflict problem — it's an unrelated-histories problem, and the diff between the two architectures is much larger than what "fix conflicts" implies. Need a directional call from you before pushing anything.

What I found

origin/magg_temporal and origin/main have no common merge-base: git merge-base origin/magg_temporal origin/main returns empty. Forcing a merge with --allow-unrelated-histories produces:

• 57 files auto-added from main (didn't exist in magg_temporal — almost the entire current src/zagg/ tree, docs site, CI workflows, deployment scripts that didn't exist when this branch forked).
• 37 files with add/add conflicts where both sides independently created the same path with different content (CI workflows, all docs/api/*, pyproject.toml, deployment/aws/{lambda_handler.py,invoke_lambda.py,deploy.sh}, every notebook, every test in tests/).
• The branch is on the pre-rename src/magg/ layout: it has src/magg/orchestrate.py and src/magg/temporal.py; main has neither (no temporal.py, no orchestrate.py anywhere in src/zagg/) and lives at src/zagg/. PR setting up publishing, library rename #16 (project rename magg→zagg) merged 2026-04-20 — this branch forked before that.

So a literal "merge main into magg_temporal and resolve conflicts" doesn't produce something coherent — main has a redesigned grids module, the catalog API (ShardMap/Catalog/CMRSource), the process_shard/processing.py rewrite from PR #23, the concurrency probe from #41, and the full hierarchical-read seam from #43/#47 that this branch knows nothing about. Forcing a merge would leave the temporal code calling APIs (get_nsidc_s3_credentials, the old process_morton_cell, the old _read_group signature) that no longer exist in their referenced form.

Three honest paths forward

(A) Port the temporal feature onto main (recommended). Open a fresh PR on claude/12-temporal-aggregation based on current main. Bring across the substance — temporal.py's accumulators + spatial functions + specs_from_config, the merra2_storm.yaml template, the pipeline.type config field — and re-implement the orchestration on top of the current runner.agg() API rather than the pre-rename magg.runner shape. PR #15 stays a historical reference; close it pointing at the new PR.

• Pro: the result is a PR that actually applies to today's codebase, reviewable as a focused temporal feature without dragging 90 commits of pre-rename history. Builds on the current process_shard / Arrow handoff / concurrency probe, all of which the temporal path wants anyway.
• Pro: the unfinished items in Temporal aggregation infrastructure (Steps 1-3) #15's body ("process_event() port, extend agg() to branch on temporal pipeline type, DRY up Lambda dispatch using orchestrate.dispatch_lambda(), derived-variable convention") are easier on the new architecture — the Lambda dispatch DRY is now the dispatcher abstraction discussed in Dispatcher abstraction as shared infrastructure (#20, #12) #63.
• Con: ~6 months of pre-rename commit history on magg_temporal doesn't carry over (it lands as a fresh re-implementation, not a merge).

(B) Force --allow-unrelated-histories merge and hand-resolve all 94 paths. Mechanically possible; the result would be a working tree, but only after I rewrite the temporal code against the new src/zagg/ APIs as part of conflict resolution. Functionally identical to (A), just with an uglier git history (the merge commit shows 90 unrelated commits joining main mid-stream).

• Pro: preserves linear continuity with PR Temporal aggregation infrastructure (Steps 1-3) #15.
• Con: the "merge commit" is misleading — the actual code change is a rewrite, not a merge.

(C) Defer. Land the dispatcher abstraction (#63) and the registry (#64) first — both are explicit prerequisites called out in #12. Then port temporal onto an architecture that already has the dispatch + registry seams the temporal engine wants to plug into. PR #15 stays parked until then.

• Pro: avoids re-doing the dispatch DRY twice (once during port, once during Dispatcher abstraction as shared infrastructure (#20, #12) #63).
• Con: parks Temporal aggregation infrastructure (Steps 1-3) #15 longer.

My lean

(A) with a (C)-flavored qualifier: open the port PR now (claude/12-temporal-aggregation), but scope the first milestone to just landing temporal.py + merra2_storm.yaml + pipeline.type config validation — i.e. the building blocks that don't yet need a dispatcher. Defer wiring agg() to branch on pipeline.type == "temporal" until #63 lands so the dispatcher does the fan-out for free. That captures most of the value from PR #15 without doubling the dispatch work.

Action requested

Tell me which path — (A), (B), or (C) — and I'll move. Not pushing anything to magg_temporal in the meantime: any push here would either be a force-overwrite (forbidden) or a merge commit that misrepresents the change shape. Keeping the blocked label cleared and applying waiting so this is on your queue rather than mine.

---

Generated by Claude Code

[espg]

superseded by #70