Event interchange + catalog for the temporal pipeline (antarctic_AR_dataset integration)

[espg]

🤖 from Claude

Tracking issue for making jbbutler/antarctic_AR_dataset consume zagg's temporal pipeline for its heavy per-storm aggregations. The engine itself landed via #70 (with #73 registries and #76 dispatch); this issue covers the remaining work: the event interchange/catalog layer plus a handful of capability gaps found by auditing every attribute_utils.py kernel against zagg.temporal built-ins. Supersedes the remaining scope of #12. The prior downstream attempt, jbbutler/antarctic_AR_dataset#1, was closed unmerged — its aggregation_registry.py design already lives in src/zagg/temporal.py, and its catalog/auth ideas are folded into the plan below.

Attribute coverage audit

All 14 field-based attributes (the ones that stream MERRA-2) expressed as temporal specs, vs. what current main supports. The ~10 mask-geometry attributes (max/mean area, duration, start/end dates, southward extent, cumulative space-time, landfall region, trajectory, elevation gradient) need no MERRA-2 reads and stay client-side in artools (decided; trajectory is per-timestep and doesn't fit the tabular row model anyway).

output	variable / collection	spatial_func × reducer	mask	needs
max_T2m_ais	T2M / slv	max × max	ais	works today
max_T2M_anomaly_ais	T2M / slv	max × max (anomaly)	ais	works today
max_IWV_ais	TQV / slv	max × max	ais	works today
max_IWV_anomaly_ais	TQV / slv	max × max (anomaly)	ais	works today
avg_vIVT_ais	VFLXQV / flx	weighted_mean × weighted_mean (negate)	ais	gap 3
max_vIVT_ais	VFLXQV / flx	max × max (negate)	ais	gap 3
max_vIVT	VFLXQV / flx	max × max (negate)	full	gap 3
min_SLP	SLP / slv	min × first_landfall	ocean	gap 2
max_SLP_gradient	SLP / slv	max_gradient × first_landfall	ocean	gap 2
max_landfalling_v850hPa	V850 / v850	max × first_landfall (negate)	ocean	gaps 2, 3
avg_landfalling_v850hPa	V850 / v850	weighted_mean × first_landfall (negate)	ocean	gaps 2, 3
avg_landfalling_minomega	OMEGA / omega	min_over_levels × first_landfall	ais	gaps 2, 3
cumulative_rainfall_ais	PRECCU+PRECLS / flx	weighted_sum × sum	ais	gaps 3, 4, 5
cumulative_snowfall_ais	PRECSN / flx	weighted_sum × sum	ais	gaps 3, 4, 5

4 of 14 run on current main unchanged; the rest need the gaps below. No artools-side zagg plugin appears necessary — every gap is generic enough to be a zagg built-in or config feature.

Capability gaps

Gap 1 — event interchange + catalog (the core of this issue). The temporal pipeline requires callers to pass events= directly (runner.py _run_lambda_events; masks by URI on Lambda). The storm catalogs are monolithic pandas-HDF5 files (one per ST-DBSCAN hyperparameter setting, data_array column of masks) — pytables, not byte-range addressable, unreadable by zagg.temporal.open_dataset. Needed:

• an exporter that explodes one monolithic catalog into per-storm NetCDF event files — writing both the raw mask and the 24-h-lookahead augmented mask (augment_storm_da semantics: extend 24 h past storm end + 9-step rolling max) — plus a catalog table (parquet) with event_key, time range, bbox, mask URIs, and hyperparameter provenance;
• an event catalog builder: bulk CMR/earthaccess query per collection DOI × the catalog's full time span → {collection: {date: uri}} index (the approach from [WIP] Lambda orchestrator  jbbutler/antarctic_AR_dataset#1's catalog.py), then per-event granule URI lists (precip collections resolve against the augmented time range);
• events_from_catalog() emitting local tuples (local backend) or URI dicts (Lambda).

Open sub-question: exporter/catalog placement — zagg.catalog.events (generic: "monolithic mask catalog → event files" is domain-neutral) vs. artools (it knows the pandas-HDF5 schema). Proposal: split — generic event-catalog + events_from_catalog() in zagg, the pytables-specific parser in artools.

Gap 2 — first_landfall event trigger. EVENT_TRIGGERS (registry.py:182) has no built-ins; process_event supports trigger: gating but nothing registers one. Without it, the first_landfall reducer captures the first timestep of the storm, not the first landfall — wrong for 5 specs. Add a built-in generic first_intersection trigger (first timestep where event ∩ a named static mask is non-empty; default ais_mask), registered also under the alias first_landfall. Storms that never intersect → no triggered timesteps → NaN (matches the existing is_landfalling filter semantics; test explicitly).

Gap 3 — per-collection time alignment + multi-granule collections. Two reader-level features:

• collection_uris is {name: single_uri} (lambda_handler.py:85, runner.py:2031) but a multi-day storm spans several MERRA-2 daily granules — accept {name: [uris]} and concat along time in read_temporal_inputs.
• MERRA-2 FLX/V850 collections are 1-hourly with half-hour timestamps; the storm masks are 3-hourly on the hour, so process_event's .sel(time=batch_times) KeyErrors. Add per-collection time_offset (e.g. -30min). This requires data_source.collections to become a mapping with metadata (currently a list of names in merra2_storm.yaml) — keep the list form working.

Gap 4 — precip processing branch. resample: {freq: 3h, how: sum, scale: 3600} (1-hourly kg m⁻² s⁻¹ rates → 3-hourly totals) and derived: variables (rainfall: PRECCU + PRECLS) as per-collection reader config. Modeled as a second logical collection sharing the FLX DOI so vIVT (no resample) and precip (resample) don't collide.

Gap 5 — per-spec event mask selection. Precip specs aggregate under the augmented mask, whose time axis extends past the primary mask's. Event payload gains named alternates (event_masks: {augmented: uri}) and specs a event_mask: key (default primary). Simplest process_event shape: group specs by event mask and run the streaming loop per group (the augmented group re-reads only the precip collection for the extra timesteps); avoids a union-time-axis loop.

Gap 6 — GES DISC credentials. _run_local hardwires NSIDC credentials (runner.py:1183); MERRA-2 direct-S3 needs the GES DISC creds endpoint. Add a gesdisc credential provider via the existing credential_provider registry and a config key selecting the provider. Note this is a parameterization, not new machinery — earthaccess/EDL already speaks to GES DISC today (the downstream repo already streams MERRA-2 through it, and jbbutler/antarctic_AR_dataset#1 fetched GES DISC S3 creds directly); the announced 2026 GES DISC→Earthdata web consolidation changes portals, not the CMR/S3 access path. GES DISC S3 is us-west-2, same region as the Lambda deployment.

Draft downstream config

ar_attributes.yaml (keys marked [NEW] are the gaps above)

pipeline:
  type: temporal

data_source:
  reader: xarray_s3
  collections:                        # [NEW] mapping form with metadata
    merra2_slv:                       # T2M, TQV, SLP (3-hourly, on the hour)
      doi: "10.5067/3Z173KIE2TPD"
    merra2_flx:                       # VFLXQV (1-hourly, half-hour stamps)
      doi: "10.5067/Q5GVUVUIVGO7"
      time_offset: "-30min"           # [NEW]
    merra2_precip:                    # same DOI, precip processing branch
      doi: "10.5067/Q5GVUVUIVGO7"
      time_offset: "-30min"           # [NEW]
      resample: {freq: "3h", how: sum, scale: 3600}   # [NEW]
      derived:                        # [NEW]
        rainfall: "PRECCU + PRECLS"
    merra2_v850:
      doi: "10.5067/VJAFPLI1CSIV"
      time_offset: "-30min"           # [NEW]
    merra2_omega:
      doi: "10.5067/QBZ6MG944HW0"
  static_data:
    ais_mask: ../antarctic_AR_catalogs/static/ais_mask.nc
    cell_areas: ../antarctic_AR_catalogs/static/cell_areas.nc
    climatology: ../antarctic_AR_catalogs/static/merra2_monthly_climatology.nc

aggregation:
  variables:
    max_T2m_ais:
      {variable: T2M, collection: merra2_slv, spatial_func: max,
       temporal_reducer: max, mask: ais}
    max_T2M_anomaly_ais:
      {variable: T2M, collection: merra2_slv, spatial_func: max,
       temporal_reducer: max, mask: ais, anomaly: true}
    max_IWV_ais:
      {variable: TQV, collection: merra2_slv, spatial_func: max,
       temporal_reducer: max, mask: ais}
    max_IWV_anomaly_ais:
      {variable: TQV, collection: merra2_slv, spatial_func: max,
       temporal_reducer: max, mask: ais, anomaly: true}
    avg_vIVT_ais:
      {variable: VFLXQV, collection: merra2_flx, spatial_func: weighted_mean,
       temporal_reducer: weighted_mean, mask: ais, negate: true}
    max_vIVT_ais:
      {variable: VFLXQV, collection: merra2_flx, spatial_func: max,
       temporal_reducer: max, mask: ais, negate: true}
    max_vIVT:
      {variable: VFLXQV, collection: merra2_flx, spatial_func: max,
       temporal_reducer: max, mask: full, negate: true}
    min_SLP:
      {variable: SLP, collection: merra2_slv, spatial_func: min,
       temporal_reducer: first_landfall, mask: ocean, trigger: first_landfall}
    max_SLP_gradient:
      {variable: SLP, collection: merra2_slv, spatial_func: max_gradient,
       temporal_reducer: first_landfall, mask: ocean, trigger: first_landfall}
    max_landfalling_v850hPa:
      {variable: V850, collection: merra2_v850, spatial_func: max,
       temporal_reducer: first_landfall, mask: ocean, negate: true,
       trigger: first_landfall}
    avg_landfalling_v850hPa:
      {variable: V850, collection: merra2_v850, spatial_func: weighted_mean,
       temporal_reducer: first_landfall, mask: ocean, negate: true,
       trigger: first_landfall}
    avg_landfalling_minomega:
      {variable: OMEGA, collection: merra2_omega, spatial_func: min_over_levels,
       temporal_reducer: first_landfall, mask: ais, trigger: first_landfall}
    cumulative_rainfall_ais:
      {variable: rainfall, collection: merra2_precip, spatial_func: weighted_sum,
       temporal_reducer: sum, mask: ais, event_mask: augmented}
    cumulative_snowfall_ais:
      {variable: PRECSN, collection: merra2_precip, spatial_func: weighted_sum,
       temporal_reducer: sum, mask: ais, event_mask: augmented}

output:
  format: parquet
  store: ./output/ar_storm_attributes.parquet

Plan

zagg PR (this repo, phased):

• Phase 1 — first_landfall/first_intersection built-in event trigger + never-landfall NaN semantics (gap 2). Small, unblocks 5 specs.
• Phase 2 — collections-as-mapping config + multi-URI collections with time-concat + time_offset (gap 3). Update merra2_storm.yaml to the mapping form; keep list form valid.
• Phase 3 — resample + derived per-collection reader features (gap 4).
• Phase 4 — named event masks (event_masks payload alternates, spec event_mask: key, per-group streaming loop) (gap 5).
• Phase 5 — event catalog: granule index builder (DOI × time span via earthaccess), event catalog table format, events_from_catalog() for local/Lambda (gap 1, zagg half).
• Phase 6 — gesdisc credential provider + provider selection in config (gap 6).

Downstream PR (jbbutler/antarctic_AR_dataset, after the above): monolithic-H5 → per-storm NetCDF exporter (raw + augmented masks, catalog parquet with hyperparameter provenance; source data assumed at ../antarctic_AR_catalogs/); ar_attributes.yaml; thin driver; parity test vs attribute_utils.py on a storm subset; demo notebook (local backend, ~5 storms); geometry attributes remain in artools; close jbbutler/antarctic_AR_dataset#1 with a pointer. Lambda-scale validation run is operator-executed.

Decisions made (maintainer)

• Exporter ships both raw and augmented masks per storm (no in-worker mask transform seam).
• Mask-geometry attributes stay client-side in artools.
• Temporal aggregation #12 closed in favor of this issue.
• Storm catalogs parsed locally from ../antarctic_AR_catalogs/ monolithic HDF5 (HF dataset is monolithic per hyperparameter setting; no per-storm layout exists upstream).

Questions for review / parity risks

1. Weighted-mean semantics differ. attribute_utils.compute_average computes a per-timestep area-weighted ratio, then an unweighted mean over timesteps (mean-of-ratios). zagg's WeightedMeanAccumulator accumulates global sum(weighted)/sum(weights) (ratio-of-sums, i.e. area-time weighting). These diverge whenever storm area varies over time — which is always. Affects avg_vIVT_ais. Options: (a) add a mean_of_ratios reducer for bit-parity, or (b) accept ratio-of-sums as the (arguably better-defined) estimator and document the delta. Needs a call before the parity test is written.
2. NaN-aware weights. compute_avg_landfalling_wind excludes NaN cells from the weight sum (fillna + notnull weights); spatial_weighted_mean_parts does not. Matters only where V850 is NaN under the mask (mask is ocean, so likely negligible — verify in parity).
3. Exporter/catalog placement (gap 1 sub-question): generic halves in zagg, pytables parser in artools — confirm or relocate.
4. spatial_max_gradient masks the sin(lat)=0 equator row that the downstream kernel lets blow up to inf — no-op for lat ∈ [−86, −39], noted for completeness.

[espg]

🤖 from Claude

Design resolution (maintainer decision): no third-party code on Lambda workers. Arbitrary user-code upload to the fleet is off the table; the existing contract stands — the dispatch payload is pure data (capability names + composable numpy expressions + URIs), and names resolve only against what ships in the zagg layer. Consequence: the capability registries are a local-backend extensibility feature and an upstreaming funnel, not a deployment path. Prototype a custom trigger/reducer as a local plugin; upstream it as a built-in (or express it as a numpy composition in config) to make it Lambda-eligible. This deletes the "plugin delivery to Lambda" problem raised earlier in this thread rather than solving it.

Re-cut of the plan under that rule — the downstream repo becomes a pure declarative consumer (config + data prep + client-side post-processing; no plugin code at all for this integration):

In zagg (all generic, all built-in):

• first_intersection event trigger (aliased first_landfall): first timestep where event ∩ named static mask is non-empty; no-intersection → NaN. Worker-side, so downstream is not an option under the no-user-code rule.
• mean_of_ratios streaming reducer — resolves parity question (1): the downstream estimator is just another composable aggregation; add it rather than argue semantics.
• Reader features as declarative options on the built-in xarray_s3 reader, not a MERRA-2 class: collections-as-mapping, multi-URI concat along time, time_offset, resample: {freq, how, scale}, derived: variables as numpy expressions (same trust boundary as the spatial path's expression:).
• gesdisc credential provider built-in (orchestrator-side; symmetric with NSIDC).
• Wire the currently-unconsumed reader:/credential-provider registry resolution into the execution path — built-ins and local plugins share one code path; the Lambda whitelist is simply "what's in the layer".
• Document the event payload contract (event_key, mask URI, per-collection URIs, static URIs) as the stable consumer interface.

Dropped from zagg scope:

• Named event masks (gap 5): keep "an event = one mask". Precip attributes run as a second event set (augmented masks, FLX-only read) and the two Parquet outputs join on event_key client-side.
• events_from_catalog/granule-index builder in zagg: defer (rule of two). The earthaccess DOI × time-range indexer starts downstream; it runs orchestrator-side where user code is fine.

Downstream (jbbutler/antarctic_AR_dataset), unchanged from before: monolithic-H5 → per-storm NetCDF exporter (raw + augmented masks, catalog parquet with hyperparameter provenance), granule indexer + event payload builder, ar_attributes.yaml, geometry attributes client-side, parity test, demo notebook, two-run join.

Revised zagg PR phases:

• Phase 1 — first_intersection/first_landfall trigger + NaN semantics
• Phase 2 — mean_of_ratios reducer
• Phase 3 — reader: collections-as-mapping, multi-URI concat, time_offset, resample, derived expressions; resolve reader: through the registry
• Phase 4 — gesdisc credential provider + config-selected provider via registry
• Phase 5 — event payload contract docs + spec params: pass-through (local plugins still receive full specs)

Open questions (2)–(4) from the issue body: (2) NaN-aware weights — verify in parity, likely negligible under the ocean mask; (3) resolved as above (indexer downstream, generic halves deferred); (4) no action needed.

[espg]

🤖 from Claude

zagg side complete — #214 is ready for review (all six phases, reviews folded, CI green). Downstream work starts now; per maintainer direction it is tracked here on this issue, with the branch pushed to espg/antarctic_AR_dataset (not the englacial fork) and the PR raised from there.

Downstream phase plan (antarctic_AR_dataset repo):

• Phase A — exporter. artools/zagg_events.py: one monolithic pandas-HDF5 catalog (local clone of the HF dataset at ../antarctic_AR_catalogs/, one file per ST-DBSCAN hyperparameter setting, e.g. epsspace0.5_epstime12_minpts5_nreppts10_seed12345.h5) → per-storm NetCDF event files under ../antarctic_AR_catalogs/derived/<hyperparam-tag>/ — raw mask and 24-h augmented mask (via the existing augment_storm_da) as separate single-variable files, matching zagg's event contract — plus a catalog parquet (event_key, raw/augmented time ranges, bbox, mask paths, hyperparameter provenance parsed from the filename). Reads through the same pd.read_hdf path as loading_utils.load_catalog so schema knowledge stays in one place. Unit-tested against a synthetic in-test catalog; validated against one real file.
• Phase B — granule index + payload builder. Bulk earthaccess query per collection DOI over the catalog span → JSON-cached {collection: {date: uri}} index (the [WIP] Lambda orchestrator  jbbutler/antarctic_AR_dataset#1 catalog.py design); events_for_zagg() emitting local tuples / Lambda URI dicts per the zagg event contract (precip events resolve granules against the augmented time range); S3 mirror step for fleet runs.
• Phase C — configs + statics. ar_attributes.yaml (12 field specs, raw masks) + ar_precip.yaml (2 precip specs, augmented masks, merra2_precip collection with time_offset/resample/derived: rainfall), both credentials_provider: gesdisc; statics prep (cell areas + AIS mask already in the HF dataset; monthly climatology builder).
• Phase D — parity harness. Runnable script (needs Earthdata login, so not CI): ~10 storms through zagg local backend vs attribute_utils.py, per-attribute tolerance table. Settles the two known wrinkles: avg_vIVT_ais via the new mean_of_ratios reducer, and the NaN-weight delta in avg_landfalling_wind.
• Phase E — geometry attributes + join. Client-side driver for the ~10 mask-only attributes (existing attribute_utils kernels) merged with the two zagg parquet outputs on event_key into the final storm table.
• Phase F — demo notebook + dependency wiring. notebooks/zagg_attributes.ipynb (exporter → 5 storms → local agg() → parity table → joined table; Lambda cell marked operator-run); zagg as an optional extra in pyproject.toml; close [WIP] Lambda orchestrator  jbbutler/antarctic_AR_dataset#1 with a pointer when the cross-fork PR opens.
• Phase G — scale validation. Operator-executed Lambda run over all 8,714 storms; deliverable is the runnable script + mirrored inputs, pre-validated on a small slice.

Same per-phase adversarial-review discipline as #214. The working draft PR opens against the espg fork's main so review threads and CI run without notifying the upstream repo on every push; the final cross-fork PR to jbbutler is a maintainer call once phases land.