diff --git a/src/zagg/configs/gedi01b_waveform_healpix_hive.yaml b/src/zagg/configs/gedi01b_waveform_healpix_hive.yaml index 15db397cb..c9b724de4 100644 --- a/src/zagg/configs/gedi01b_waveform_healpix_hive.yaml +++ b/src/zagg/configs/gedi01b_waveform_healpix_hive.yaml @@ -186,6 +186,21 @@ aggregation: fill_value: 0 # weights: flux # PR #431 rebase: the section-2.0 weights # # declaration lands there; enable on rebase. + # # Load-bearing for the pyramid below, not just + # # for the leaf: `zagg.pyramid.declared_fields` + # # keys `weights` (and the `gain` beside it) + # # into the manifest entry ONLY when it is set, + # # and `sweep_overview._overview_config` carries + # # that pair into every overview level's + # # template on the same condition. Commented + # # out, leaf and overview agree by ABSENCE -- + # # which spec section 2.0 makes readers take as + # # `counts`, i.e. integer observation counts, + # # while these weights are (count - noise_mean) + # # * gain floats. Enable this before building a + # # GEDI ladder anyone will analyze; until then + # # every level is mis-declared the same way the + # # leaf already is. params: delta: 8192 # ratified cross-sensor floor (issue #422, decision 7) counts: rxwaveform # params-as-column: per-sample ADC values @@ -267,7 +282,12 @@ output: sharded: true child_order: 18 # leaf cell resolution (~20 m, ~GEDI footprint) store_layout: hive - pyramid: false # composability none initially (issue #422) + pyramid: false # opt-in (runbook: CA builds pyramids-off on + # 0.49); rx_flux classifies approximate since + # issue #508, so pyramid: {} builds the ladder + # -- but read rx_flux's `weights: flux` note + # first: until PR #431 enables it, every level + # of that ladder declares itself `counts`. worker: memory: 4096 # optimized probe (run 5d081b68, 4/4 green): diff --git a/src/zagg/processing/streaming.py b/src/zagg/processing/streaming.py index f21efc99a..65821f1cf 100644 --- a/src/zagg/processing/streaming.py +++ b/src/zagg/processing/streaming.py @@ -61,6 +61,32 @@ _TDIGEST_WHERE_FUNCTION = "zagg.stats.tdigest.build_tdigest_where" _TDIGEST_SPILL_FUNCTIONS = (*_TDIGEST_FUNCTIONS, _TDIGEST_WHERE_FUNCTION) +#: The digest FAMILY (issue #508): reducers whose STORED payloads are spec-§2 +#: weights-sorted ``(k, 2)`` centroid arrays and therefore fold by the +#: order-independent k-way t-digest law (``merge_tdigests_kway``) wherever a +#: fold operates on stored payloads rather than raw rows — the D24 +#: composability arm (:func:`zagg.semantics.field_composability`) and BOTH +#: stored-payload fold sites it licenses: the sweep's overview fold +#: (:func:`zagg.sweep_overview.fold_digests`) and the worker-side leaf column +#: (:func:`zagg.column.fold_column`, gated by +#: :func:`zagg.column.leaf_column_plan`, which filters the same declaration +#: through :func:`zagg.column.composable_fields`). Membership here is +#: therefore not free at build time — see ``build_waveform_digest``'s +#: docstring for why a GEDI-scale store declares its ladder sweep-only. +#: The k-way law is weight-agnostic — flux weights fold like counts (spec +#: §2.0, issue #431) — which is what admits ``build_waveform_digest``: its +#: build is waveform-specific, but its stored payload is a standard §2 +#: centroid array whose companions ride the same channel overloads. +#: Deliberately NOT consumed by :func:`validate_streaming` / +#: :func:`validate_spill_fold`: those gates re-run BUILDERS over raw rows (per +#: flush / per block), and the waveform builder's noise-model columns are not +#: threaded there — the tuples above keep their exact members (the issue #508 +#: phase-1 characterization pins that posture). ``build_tdigest_where`` is +#: likewise NOT here: its D24 class stays ``none`` pending the gate-drift +#: ruling raised on issue #508. +_WAVEFORM_DIGEST_FUNCTION = "zagg.stats.waveform.build_waveform_digest" +_DIGEST_FAMILY_FUNCTIONS = (*_TDIGEST_FUNCTIONS, _WAVEFORM_DIGEST_FUNCTION) + #: The packed composition reducer (issue #321): the SPILL fold collapses its #: per-block ``(word, n_signal)`` pairs in one pass via #: ``merge_composition_kway`` — issue #370 option (a), accepting the documented diff --git a/src/zagg/semantics.py b/src/zagg/semantics.py index 136bab21a..c08753b33 100644 --- a/src/zagg/semantics.py +++ b/src/zagg/semantics.py @@ -373,9 +373,12 @@ def field_composability(meta: dict) -> str: ``validate_streaming`` accepts, widened by the exact sum/min/max laws): - ``exact`` — scalar ``function`` in :data:`EXACT_MERGE_LAWS`; - - ``approximate`` — a ragged t-digest field with the standard ``(2,)`` - centroid inner shape, **located or not** (merge is order-dependent; - ``np.isclose`` equality class, cf. D24); + - ``approximate`` — a ragged digest-family field with the standard + ``(2,)`` centroid inner shape, **located or not** (merge is + order-dependent; ``np.isclose`` equality class, cf. D24). The family is + :data:`zagg.processing.streaming._DIGEST_FAMILY_FUNCTIONS` — the + t-digest builders plus the waveform flux digest (issue #508), every + reducer whose stored payload folds by the k-way law; - ``none`` — everything else: expressions, vector fields, chunk-resolution companions, a ``temporal: per-cell`` dense companion (see below), and any scalar reducer without an exact law (mean, std, median, quantiles, ...). @@ -403,8 +406,8 @@ def field_composability(meta: dict) -> str: word grammar's join over a cell group, not its own reducer, so classifying it by ``function`` would fold e.g. ``nanmax`` over toc words and emit a word whose conservative-envelope claim is false. Wiring a dense toc law is not - what the ruling asked for, and no shipped config needs it (the GEDI template - declares ``pyramid: false``), so it is left out rather than guessed at. + what the ruling asked for, and no shipped config pairs a per-cell + companion with a pyramid, so it is left out rather than guessed at. """ from zagg.config import get_output_signature from zagg.time_axis import TOC_SHAPE_PER_CELL @@ -416,9 +419,9 @@ def field_composability(meta: dict) -> str: return "none" function = _fold_function_name(meta.get("function")) if sig["kind"] == "ragged": - from zagg.processing.streaming import _TDIGEST_FUNCTIONS + from zagg.processing.streaming import _DIGEST_FAMILY_FUNCTIONS - if meta.get("function") in _TDIGEST_FUNCTIONS and tuple(sig["inner_shape"]) == (2,): + if meta.get("function") in _DIGEST_FAMILY_FUNCTIONS and tuple(sig["inner_shape"]) == (2,): return "approximate" return "none" if sig["kind"] == "scalar" and function in EXACT_MERGE_LAWS: diff --git a/src/zagg/stats/waveform.py b/src/zagg/stats/waveform.py index 2bc11c22a..5983d3ead 100644 --- a/src/zagg/stats/waveform.py +++ b/src/zagg/stats/waveform.py @@ -203,13 +203,27 @@ def build_waveform_digest( the samples that survived into it (:func:`zagg.stats.tdigest._centroid_envelopes`). Given, the return is a ``(digest, words)`` pair. The per-centroid **shape** is the one the espg - ruling of 2026-08-17 makes universal, identical to the located channel's; - the ruling's *at every level* half does not describe this reducer's stores, - because ``build_waveform_digest`` is absent from - ``zagg.processing.streaming._TDIGEST_FUNCTIONS`` — so a waveform field is - D24 class ``none`` (issue #422, and the GEDI template's ``pyramid: false``) - and exists at native resolution only. There is no waveform overview to carry - a companion, and a reader must not expect one. + ruling of 2026-08-17 makes universal, identical to the located channel's, + and its *at every level* half describes this reducer's stores too: + ``build_waveform_digest`` is a member of the shared digest family + (``zagg.processing.streaming._DIGEST_FAMILY_FUNCTIONS``, issue #508), so a + waveform field is D24 class ``approximate`` and folds through the overview + pyramid when one is declared — each overview level carries the + per-centroid companion beside the folded payload. The build-time GATES are + unchanged (the merge/spill folds still refuse the builder, so a waveform + shard aggregates pooled or single-block-spill only), but the build-time + WORK is not: ``leaf_column_plan`` filters the declaration through the same + composable classes, so a pyramid-ON waveform config also folds this column + worker-side (:func:`zagg.column.fold_column` at the tail of + ``hive.process_and_write_hive``) — a node-order k-way merge over every + resident digest, whose measured envelope + (:func:`zagg.column.write_leaf_column`'s memory note: ~2.0 GB at ~17.6M + centroids, on top of a loaded 4 GB heap) is NOT validated at GEDI's + 21-33M-kept-rows-per-shard scale. So the ruled deployment path for a + GEDI-scale store is pyramids-OFF aggregation, then + :func:`zagg.sweep_overview.declare_pyramid` plus a sweep-only overview + pass (the runbook contingency) — which builds the ladder without ever + putting the column fold on the aggregating worker. What the words say is ruling 2's honesty property: a waveform record's samples share one instant, so a single-shot cell's centroids all carry that diff --git a/tests/test_column.py b/tests/test_column.py index 8a56a8d31..628bb15ef 100644 --- a/tests/test_column.py +++ b/tests/test_column.py @@ -151,6 +151,49 @@ def test_all_none_fields_mean_no_column_at_all(self): assert leaf_column_plan(cfg, grid) is None +class TestWaveformEntersTheColumn: + """Issue #508 review pin: the D24 class flip is not confined to the sweep. + + ``leaf_column_plan`` filters the declaration through the same + ``composable_fields``, so an ``approximate`` rx_flux now enters the + WORKER-side fold (:func:`zagg.column.fold_column` at the tail of + ``hive.process_and_write_hive``) the moment the shipped GEDI template + declares a pyramid — a consequence recorded here rather than discovered on + a fleet run. The memory envelope of that fold at GEDI scale is unvalidated + (see ``write_leaf_column``'s memory note), which is why the runbook builds + CA pyramids-off and retrofits the ladder sweep-only (``declare_pyramid``). + """ + + def _cfg(self): + from zagg.config import default_config + + return default_config("gedi01b_waveform_healpix_hive") + + def test_shipped_template_declares_no_column(self): + from zagg.column import leaf_column_plan + from zagg.grids import from_config + + # ``pyramid: false`` as shipped: no declaration, so no column at all. + cfg = self._cfg() + assert leaf_column_plan(cfg, from_config(cfg)) is None + + def test_pyramid_knob_folds_rx_flux_worker_side(self): + from zagg.column import leaf_column_plan + from zagg.grids import from_config + + cfg = self._cfg() + cfg.output["pyramid"] = {} # the /2 default flip at chunk_inner 12 + plan = leaf_column_plan(cfg, from_config(cfg)) + assert plan is not None + resolutions, fields = plan + assert resolutions == [12, 11, 10, 9] + # Pre-#508 this was ``{"count"}`` — the digest-family membership is + # what adds the second entry, and rx_flux is the ragged one. + assert set(fields) == {"count", "rx_flux"} + assert fields["rx_flux"]["class"] == "approximate" + assert fields["rx_flux"]["temporal"] == "per-centroid" + + class TestLeafSlabs: def test_staged_refs_pass_through(self): slabs = _cell_slabs({0: [1.0, 2.0]}) diff --git a/tests/test_spill_crossblock.py b/tests/test_spill_crossblock.py index 34dab14fa..2b13cba92 100644 --- a/tests/test_spill_crossblock.py +++ b/tests/test_spill_crossblock.py @@ -347,6 +347,160 @@ def _pooled_build(meta, n=8): return resolve_function(meta["function"])(cell_data["h_ph"], **params) +_WAVEFORM_FUNCTION = "zagg.stats.waveform.build_waveform_digest" + + +def _waveform_variables(temporal=True, delta=64): + """A ``rx_flux``-shaped field: the gedi01b template's reducer + channel. + + Sources ``h_ph`` and reads the noise-model columns via the params-as-column + path, exactly as the shipped template wires ``rxwaveform``/``noise_mean``/ + ``noise_stddev`` (issue #508 phase 1 harness). + """ + field = { + "kind": "ragged", + "function": _WAVEFORM_FUNCTION, + "source": "h_ph", + "inner_shape": [2], + "dtype": "float32", + "fill_value": 0, + "params": { + "delta": delta, + "counts": "wf_counts", + "noise_mean": "wf_noise_mean", + "noise_stddev": "wf_noise_stddev", + "gain": 1.0, + "false_positive_rate": 1.0e-3, + "samples_per_record": 60, + }, + } + if temporal: + field["temporal"] = "per-centroid" + return { + "count": {"function": "len", "source": "h_ph", "dtype": "int32", "fill_value": 0}, + "rx_flux": field, + } + + +def _with_waveform_columns(dfs, seed=0): + """Add the noise-model columns ``_waveform_variables`` declares. + + Counts sit well above the zero-mean noise floor, so every row survives the + clip and ``sum(weights)`` over a cell is the sum of its rows' counts. + """ + rng = np.random.default_rng(seed) + for df in dfs: + n = len(df) + df["wf_counts"] = rng.uniform(5.0, 50.0, n) + df["wf_noise_mean"] = np.zeros(n, dtype=np.float64) + df["wf_noise_stddev"] = np.full(n, 0.1, dtype=np.float64) + return dfs + + +class TestWaveformSpillBaseline: + """How the spill path treats ``build_waveform_digest`` TODAY (issue #508). + + Phase 1 characterization, recorded before any registry changes: the SERC + GEDI fleet runs (0.47–0.49) completed under ``{mode: spill}``, and the + mechanism that carried them is the NON-mergeable single-block regime. + ``validate_spill_fold`` refuses the builder (it is outside + ``_TDIGEST_SPILL_FUNCTIONS``) — and on the SHIPPED template the refusal is + OVER-determined, naming the per-shot ``single_shot_value``/``shot_*`` + scalar companions alongside it — so ``SpillAggregator`` records the + verdict, replays the pooled machinery byte-identically while the shard + fits in one block, and raises ``SpillOverflowError`` naming the field on + the first block close. The synthetic below declares only ``count`` + + ``rx_flux`` so the builder is isolated as the cause; the last test pins + the deployed config, where it is not the only one. Issue #508 changes + NONE of this: the shared digest-family registry feeds the D24 pyramid + classification (stored-payload folds), not the build-time spill fold, + which would need the noise-model columns re-threaded per block. + """ + + def test_probe_refuses_the_waveform_builder(self): + # The per-centroid channel passes the temporal arm (issue #477); the + # refusal is the ragged function arm — no cross-block fold law. + cfg = _config(_waveform_variables(), streaming=_SPILL) + with pytest.raises(ValueError, match="'rx_flux'.*build_waveform_digest.*fold law"): + validate_spill_fold(cfg) + + def test_spill_accepts_the_config_as_non_mergeable(self): + # Accepted — single-block exact — with the probe's verdict recorded so + # a block close can name the field (issue #474 message discipline). + cfg = _config(_waveform_variables(), streaming=_SPILL) + agg = SpillAggregator(cfg, _grid(cfg), "pandas", 1) + assert not agg._mergeable + assert agg._digest_fields == {} + assert "rx_flux" in agg._fold_problems and "fold law" in agg._fold_problems + agg.close() + + def test_single_block_regime_is_byte_identical_to_pooled(self, monkeypatch): + # The SERC mechanism: one block -> the pooled replay, payload AND the + # per-centroid temporal channel byte-identical to the pooled path. + key = _shard_key() + pooled_cfg = _config(_waveform_variables()) + spill_cfg = _config(_waveform_variables(), streaming=_SPILL) + grid = _grid(pooled_cfg) + dfs = _with_waveform_columns( + _granule_dfs(grid, key, _CELL_LISTS[:3], obs_per_cell=40, seed=6, times=True) + ) + df_p, ragged_p, _ = _run(monkeypatch, pooled_cfg, grid, key, list(dfs)) + df_s, ragged_s, _ = _run(monkeypatch, spill_cfg, _grid(spill_cfg), key, list(dfs)) + pd.testing.assert_series_equal(df_p["count"], df_s["count"]) + assert set(ragged_p) == set(ragged_s) == {"rx_flux"} + pay_p, idx_p, locs_p, times_p = _channels_of(ragged_p["rx_flux"]) + pay_s, idx_s, locs_s, times_s = _channels_of(ragged_s["rx_flux"]) + assert idx_p == idx_s and len(pay_p) > 0 + assert locs_p is None and locs_s is None + for a, b in zip(pay_p, pay_s, strict=True): + np.testing.assert_array_equal(a, b) + for a, b in zip(times_p, times_s, strict=True): + np.testing.assert_array_equal(a, b) + + def test_block_close_raises_overflow_naming_the_field(self, monkeypatch): + # The recorded boundary of the mechanism above: past one block there is + # no fold law, and the overflow splices the probe's verdict. + from zagg.processing.spill import SpillOverflowError + + _force_tiny_blocks(monkeypatch) + key = _shard_key() + cfg = _config(_waveform_variables(), streaming=_SPILL) + grid = _grid(cfg) + dfs = _with_waveform_columns( + _granule_dfs(grid, key, _CELL_LISTS[:2], obs_per_cell=10, seed=1, times=True) + ) + with pytest.raises(SpillOverflowError, match="'rx_flux'.*build_waveform_digest.*fold law"): + _run(monkeypatch, cfg, grid, key, dfs) + + def test_the_shipped_template_is_refused_by_more_than_the_builder(self): + # The synthetic above isolates the builder; the DEPLOYED config is + # over-determined. Seven of its eight fields are per-shot scalars with + # no cross-block fold either, so even if issue #508 admitted + # ``build_waveform_digest`` to a shared digest-family registry the GEDI + # store would stay non-mergeable -- ``spill_blocks_closed: 0`` remains a + # hard precondition of the shipped template. Its own + # ``worker.extra_disk`` comment says the same: "validate_spill_fold + # rejects it (build_waveform_digest and the single_shot_value + # companions have no cross-block fold law)". + from zagg.config import default_config + + with pytest.raises(ValueError) as exc: + validate_spill_fold(default_config("gedi01b_waveform_healpix_hive")) + message = str(exc.value) + assert "'rx_flux'" in message and "build_waveform_digest" in message + for name in ( + "shot_count", + "shot_number", + "noise_mean", + "noise_stddev", + "rx_energy", + "elevation_bin0", + "elevation_lastbin", + ): + assert f"'{name}'" in message + assert "zagg.stats.waveform.single_shot_value" in message + + class TestSpillFoldProbe: """The spill mergeability probe (validate_spill_fold) vs merge mode.""" diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 8f7a6b234..e7e4adeaa 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -177,6 +177,29 @@ def test_non_tdigest_ragged_rejected(self): with pytest.raises(ValueError, match="merge law"): validate_streaming(cfg) + def test_waveform_digest_ragged_rejected(self): + # Issue #508 phase 1 baseline: build_waveform_digest is outside + # _TDIGEST_FUNCTIONS, so merge mode refuses it through the same + # no-merge-law arm as any other non-tdigest ragged reducer. + cfg = _config() + cfg.aggregation["variables"]["h_tdigest"]["function"] = ( + "zagg.stats.waveform.build_waveform_digest" + ) + with pytest.raises(ValueError, match="h_tdigest.*build_waveform_digest.*merge law"): + validate_streaming(cfg) + + def test_waveform_temporal_field_refused_via_the_temporal_arm(self): + # The shipped template shape (rx_flux carries temporal: per-centroid): + # the companion refusal fires before the function arm ever sees the + # builder, and routes to mode: spill (issue #508 phase 1 baseline). + cfg = _config() + cfg.aggregation["variables"]["h_tdigest"]["function"] = ( + "zagg.stats.waveform.build_waveform_digest" + ) + cfg.aggregation["variables"]["h_tdigest"]["temporal"] = "per-centroid" + with pytest.raises(ValueError, match="temporal companions.*cannot stream.*mode: spill"): + validate_streaming(cfg) + def test_pairwise_tdigest_reducer_is_streamable(self): # build_tdigest_pairwise carries the pairwise merge law (issue #279), # so it must validate as a mergeable ragged reducer just like the diff --git a/tests/test_sweep_overview.py b/tests/test_sweep_overview.py index a901d1183..c464d46a2 100644 --- a/tests/test_sweep_overview.py +++ b/tests/test_sweep_overview.py @@ -237,13 +237,13 @@ def test_temporal_per_cell_stays_none(self): } assert field_composability(meta) == "none" - def test_a_temporal_waveform_field_is_still_none(self): - # The per-centroid SHAPE does not put a field in the pyramid: the class - # is decided by the reducer, and ``build_waveform_digest`` is outside - # ``_TDIGEST_FUNCTIONS`` by the issue #422 ruling (GEDI declares - # ``pyramid: false``). So a waveform field carrying the channel stays - # class ``none`` and never appears above native resolution — there is no - # ``rx_flux_times`` on any overview (review finding). + def test_a_temporal_waveform_field_is_approximate(self): + # Issue #508 (superseding the #422-era exclusion this test used to + # pin): ``build_waveform_digest`` is a member of the shared digest + # family — its stored payload is a §2 centroid array and the k-way law + # is weight-agnostic (flux weights fold like counts, issue #431 §2) — + # so a waveform field classifies ``approximate`` and folds through the + # pyramid, per-centroid companion at every level. meta = { "kind": "ragged", "function": "zagg.stats.waveform.build_waveform_digest", @@ -251,12 +251,76 @@ def test_a_temporal_waveform_field_is_still_none(self): "temporal": "per-centroid", "dtype": "float32", } - assert field_composability(meta) == "none" - # ... and the SAME declaration under the standard reducer does fold, so - # the class turns on the function, not on the channel. + assert field_composability(meta) == "approximate" + # The inner-shape guard still applies to the family's new member. + assert field_composability({**meta, "inner_shape": [3]}) == "none" + + def test_gedi_waveform_template_classifies_approximate(self): + # Issue #508: the SHIPPED template's rx_flux — build_waveform_digest + # with a per-centroid clock — classifies ``approximate`` via the + # shared digest-family registry, which is what lets the SERC probe's + # ``pyramid = {}`` declare a ladder instead of ``{"class": "none"}``. + # The meta-level pin above records the mechanism; this one records + # that the template hits it. + from zagg.config import default_config + + classes = composability_classes(default_config("gedi01b_waveform_healpix_hive")) + assert classes["rx_flux"] == "approximate" + + def test_digest_family_registry_members_pinned_by_value(self): + # The issue #508 contract: ONE new registry, existing tuples keep + # their exact members. Pinned by value so a drive-by addition to any + # of the three is a loud diff, not a silent gate widening (the mortie + # #194 lesson; the where-gate question is standing on issue #508). + from zagg.processing.streaming import ( + _DIGEST_FAMILY_FUNCTIONS, + _TDIGEST_FUNCTIONS, + _TDIGEST_SPILL_FUNCTIONS, + ) + + assert _TDIGEST_FUNCTIONS == ( + "zagg.stats.tdigest.build_tdigest", + "zagg.stats.tdigest.build_tdigest_pairwise", + ) + assert _TDIGEST_SPILL_FUNCTIONS == ( + *_TDIGEST_FUNCTIONS, + "zagg.stats.tdigest.build_tdigest_where", + ) + assert _DIGEST_FAMILY_FUNCTIONS == ( + *_TDIGEST_FUNCTIONS, + "zagg.stats.waveform.build_waveform_digest", + ) + + def test_where_strata_template_classifies_none_today(self): + # Issue #508 phase 1 baseline for the OTHER gate pair, and a pin the + # phase-2 registry must not flip. ``build_tdigest_where`` is already + # inconsistent across the two gates: ``_TDIGEST_SPILL_FUNCTIONS`` + # admits it ("It folds k-way, like the ``build_tdigest`` it delegates + # to"), yet D24 tests ``_TDIGEST_FUNCTIONS`` and so classifies a + # where-stratum field ``none`` even though its stored payload is an + # ordinary (k, 2) centroid array. Per the plan (issue #508), a gate + # found already inconsistent is RAISED rather than silently unified: + # the divergence is on the issue, and the phase-2 shared registry is + # scoped to ``build_waveform_digest`` only -- it deliberately does NOT + # admit ``build_tdigest_where``, so these stay ``none`` until espg + # rules otherwise. + from zagg.config import default_config + + classes = composability_classes(default_config("atl03_tdigest_strata_healpix")) + assert classes["h_tdigest_signal"] == "none" + assert classes["h_tdigest_noise"] == "none" + # The meta-level mechanism behind that template result. assert ( - field_composability({**meta, "function": "zagg.stats.tdigest.build_tdigest"}) - == "approximate" + field_composability( + { + "kind": "ragged", + "function": "zagg.stats.tdigest.build_tdigest_where", + "inner_shape": [2], + "params": {"where": "h_ph > 0"}, + "dtype": "float32", + } + ) + == "none" ) def test_located_declaration_rides_the_manifest_entry(self): @@ -653,6 +717,127 @@ def test_overview_template_stamps_the_flux_declaration(self, tmp_path): check_weights_match(dict(group["counts_d"].attrs), fields["counts_d"], "counts_d") +class TestWaveformPyramidDeclaration: + """Template-time classification for the waveform digest (issue #508). + + The SERC probe shape: ``gedi01b_waveform_healpix_hive`` + ``output.pyramid + = {}`` + ``rx_flux.overview_delta = 512`` under the probe's uniform-δ4096 + override (the packaged template ships δ8192). Classification runs + worker-side at template time, so this IS the deployed surface the 0.50 + fleet re-runs the probe against. + """ + + def _probe_cfg(self): + from zagg.config import default_config + + cfg = default_config("gedi01b_waveform_healpix_hive") + cfg.output["pyramid"] = {} + rx = cfg.aggregation["variables"]["rx_flux"] + # 512 is the SERC probe's literal declaration, and it coincides with + # the OVERVIEW_DELTA_CAP fallback under δ4096 — so this value alone + # cannot prove the declaration is read. The δ128 variant below does. + rx["overview_delta"] = 512 + rx["params"]["delta"] = 4096 # the probe's uniform-δ override + return cfg + + def test_probe_config_declares_the_ladder(self): + from zagg.pyramid import declared_fields + + fields, excluded = declared_fields(self._probe_cfg()) + # The exact manifest entry the 0.49 probe SHOULD have produced instead + # of {"class": "none"} — delta as the test config declares it (#424 + # records the split budget resolved), companion shape included. + assert fields["rx_flux"] == { + "class": "approximate", + "method": "tdigest_kway", + "dtype": "float32", + "inner_shape": [2], + "delta": 4096, + "overview_delta": 512, + "temporal": "per-centroid", + } + assert "rx_flux" not in excluded + # Non-vacuous pin on the SAME probe config: the capped fallback can + # only ever yield min(4096, 512) = 512, so a budget the cap cannot + # produce shows the DECLARED value wins here too (the generic law is + # test_declared_fields_records_the_resolved_budget). + capped = self._probe_cfg() + capped.aggregation["variables"]["rx_flux"]["overview_delta"] = 128 + assert declared_fields(capped)[0]["rx_flux"]["overview_delta"] == 128 + assert fields["count"]["class"] == "exact" + # The per-record companions keep the ruled option-A absence: scalar + # reducers without an exact law declare class only, at native + # resolution only (issue #201). + for name in ( + "shot_count", + "shot_number", + "noise_mean", + "noise_stddev", + "rx_energy", + "elevation_bin0", + "elevation_lastbin", + ): + assert fields[name] == {"class": "none"} + assert name in excluded + + def test_manifest_block_covers_every_level(self): + # pyramid = {} on the gedi grid (9 -> chunk 12 -> 18) takes the ruled + # /2 default: the leaf entry plus the fixed ladder to 0, ONE fields map + # covering every level — rx_flux is declared at each of them. + from zagg.sweep_overview import build_pyramid_block + + block = build_pyramid_block(self._probe_cfg(), 9, 12) + assert block["spec"] == "zagg-pyramid/2" + assert [e["node"] for e in block["overviews"]] == list(range(9, -1, -1)) + assert block["overviews"][0]["cells"] == [12] + # The manifest entry is what the /2 writers reconstruct the field + # from, so the block must carry the whole declaration — not just the + # class: the §8.3 companion shape and the resolved fold budget travel + # in it or the overview template loses the sibling. + rx = block["overview"]["fields"]["rx_flux"] + assert rx["class"] == "approximate" + assert rx["temporal"] == "per-centroid" + assert rx["overview_delta"] == 512 + + def test_overview_template_emits_the_times_sibling(self, tmp_path): + # The companion path at overview levels (issue #410, per-centroid at + # EVERY level): the manifest entry is the only description the + # overview writer has, and _overview_config must turn its temporal key + # into the ``rx_flux_times`` sibling + §8.3 declaration, exactly as a + # leaf template does. + from zagg.column import composable_fields + from zagg.grids.healpix import HealpixGrid + from zagg.sweep_overview import _overview_config, build_pyramid_block + from zagg.time_axis import temporal_declaration + + # Sourced from the MANIFEST block, not from declared_fields directly, + # because that is the seam production walks: the staged sweep lifts + # pyramid.overview.fields out of the manifest and hands it on. A + # regression that dropped the temporal key on the way into the block + # fails here. + fields = build_pyramid_block(self._probe_cfg(), 9, 12)["overview"]["fields"] + # The production writers template only the composable classes — on the + # /2 path, which is the one this store declares: run_stage_sweep + # filters the manifest map before _write_stage_overview, and the + # worker column filters in leaf_column_plan. The class-none companions + # never reach the template, which is what the last assertion pins. + grid = HealpixGrid(2, 4, config=_overview_config(composable_fields(fields)), sharded=True) + grid.emit_shard_template(open_store(str(tmp_path / "ov.zarr")), overwrite=True) + group = zarr.open_group( + open_store(str(tmp_path / "ov.zarr")), path="4", mode="r", zarr_format=3 + ) + assert "rx_flux_times" in group + assert temporal_declaration(dict(group["rx_flux_times"].attrs)) == { + "spec": "zagg-toc/1", + "shape": "per-centroid", + "grammar": "mortie-toc/1", + } + # The payload binds the sibling by name (§1.2). + assert dict(group["rx_flux"].attrs)["times"] == "rx_flux_times" + # And the excluded per-record companions are absent from the template. + assert "shot_number" not in group + + class TestPyramidBlock: """The manifest declaration (Phase C): template time + config grammar.""" @@ -1492,6 +1677,192 @@ def test_temporal_alone_folds_through_the_pyramid(self, tmp_path): assert _toc_contains(times[i], times_truth[0][lo:hi]) +#: The waveform-store declaration for the harness above (issue #508): the +#: digest family's new member, per-centroid companion, no located channel — +#: the gedi01b shape reduced to the fold-relevant keys. +WAVEFORM_FIELDS_DECL = { + "count": {"class": "exact", "method": "sum", "dtype": "int32", "fill_value": 0}, + "rx_flux": { + "class": "approximate", + "method": "tdigest_kway", + "dtype": "float32", + "inner_shape": [2], + "delta": 64, + "overview_delta": 64, + "temporal": "per-centroid", + }, +} + + +def _waveform_leaf_cfg(): + from zagg.config import PipelineConfig + + return PipelineConfig( + aggregation={ + "coordinates": {"morton": {"dtype": "uint64", "fill_value": 0}}, + "variables": { + "count": {"function": "len", "dtype": "int32", "fill_value": 0}, + "rx_flux": { + "kind": "ragged", + "function": "zagg.stats.waveform.build_waveform_digest", + "inner_shape": [2], + "temporal": "per-centroid", + "dtype": "float32", + "fill_value": 0, + }, + }, + } + ) + + +def _make_waveform_leaf(root, decimal, cells, *, seed=11): + """A committed leaf whose payloads come from ``build_waveform_digest``. + + ``cells`` maps leaf row -> sample count; every sample carries an INTEGER + photoelectron count over a zero noise floor, so each cell's flux (the sum + of its centroid weights) is exactly its integer count total — sums of + small ints are exact in float32, which is what lets the per-level parity + assertions below demand equality, not closeness. Returns ``(flux_total, + {row: (digest, words)})``. + """ + from conftest import TOC_BASE, toc_words + from mortie import generate_morton_children + + from zagg.grids.healpix import HealpixGrid + from zagg.stats.waveform import build_waveform_digest + + grid = HealpixGrid(SHARD_ORDER, CELL_ORDER, config=_waveform_leaf_cfg()) + word = morton_word(decimal) + store = open_store(shard_leaf_path(str(root), word)) + grid.emit_shard_template(store, overwrite=True) + group = zarr.open_group(store, path=str(CELL_ORDER), mode="r+", zarr_format=3) + n_cells = 4 ** (CELL_ORDER - SHARD_ORDER) + group["morton"][:] = np.asarray(generate_morton_children(word, CELL_ORDER), dtype=np.uint64) + rng = np.random.default_rng(seed) + count = np.zeros(n_cells, np.int32) + digest = np.full(n_cells, b"", dtype=object) + times = np.full(n_cells, b"", dtype=object) + flux_total = 0.0 + truth = {} + for i, n in cells.items(): + values = rng.normal(0.0, 10.0, n) + counts = rng.integers(2, 30, n).astype(np.float64) + base = str(np.datetime64(TOC_BASE, "ns") + np.timedelta64(3600 * (i + 1), "s")) + d, w = build_waveform_digest( + values, + 64, + counts=counts, + noise_mean=np.zeros(n), + noise_stddev=np.full(n, 0.25), + temporal=toc_words(n, base=base), + ) + # Zero noise floor + counts >= 2 clear the clip threshold (0.77 at the + # reducer's default operating point, n_σ = 3.09 against σ = 0.25; the + # margin holds at the shipped gedi01b point too, n_σ = 4.49 → 1.12), so + # nothing is dropped and the digest mass IS the count total. + assert float(d[:, 1].sum()) == float(counts.sum()) + count[i] = n + digest[i] = encode_digest(d, "float32") + times[i] = encode_digest(w, "uint64") + flux_total += float(counts.sum()) + truth[i] = (d, w) + group["count"][:] = count + group["rx_flux"][:] = digest + group["rx_flux_times"][:] = times + stamp_commit(store, cells_with_data=len(cells), granule_count=1) + return flux_total, truth + + +class TestWaveformOverviewFold: + """The runbook step-2 fold gates, in-repo (issue #508 phase 4). + + A synthetic waveform store — leaves built by ``build_waveform_digest`` + itself — swept through the production overview fold: per-level flux + ``weight_total`` equals the leaf total through the k-way law at EVERY + declared level (the weight-agnostic licensing fact, issue #431 §2), the + ``rx_flux_times`` sibling rides every level row-aligned, and the ragged + ``(2,)`` element round-trips. + """ + + def _sweep(self, tmp_path, cells, orders=(1, 0)): + _write_manifest(tmp_path, orders=orders, fields=WAVEFORM_FIELDS_DECL) + leaf_total, truth = _make_waveform_leaf(tmp_path, "-311", cells) + result = run_sweep(str(tmp_path), [(morton_word("-311"), None)], families=("overview",)) + assert result["families"]["overview"]["failed"] == 0 + return leaf_total, truth + + def test_weight_total_parity_and_times_at_every_level(self, tmp_path): + # 300 samples at δ64 force genuine k-way merges at the fine level and + # a fold-of-merges at the coarse one; the mass must survive both. + from zagg.stats.toc import cell_envelope + + leaf_total, truth = self._sweep(tmp_path, {0: 300, 3: 80, 7: 41}) + for node_rel, order, partitioned in (("-3/1", 3, True), ("-3", 2, False)): + g = _overview_group(tmp_path, node_rel, "all.zarr", order) + rows = [decode_digest(bytes(b), "float32") for b in g["rx_flux"][:]] + level_total = sum(float(d[:, 1].sum()) for d in rows if d.size) + assert level_total == leaf_total, f"mass lost/invented at {node_rel}" + words = [decode_digest(bytes(b), "uint64", ()) for b in g["rx_flux_times"][:]] + # Leaf rows fold into contiguous runs of the coarser level, so its + # row ``j`` is fed exactly by leaf rows ``j * per_row ...`` (the + # slots past the level's own cell count stay empty). Every + # ``_make_waveform_leaf`` row sits an hour off its neighbour, so the + # words below are pinned to the CONTRIBUTORS' instants rather than + # to some merely well-formed word: a column carried forward from + # another row widens the envelope out of the hour and fails here. + per_row = LEAF_CELLS // 4 ** (order - SHARD_ORDER) + populated = 0 + for j, (d, w) in enumerate(zip(rows, words, strict=True)): + assert w.shape == (d.shape[0],), "§1.1 row alignment" + members = [truth[i] for i in range(j * per_row, (j + 1) * per_row) if i in truth] + if not members: + assert w.size == 0, f"words in an unfed row of {node_rel}" + continue + take = np.argsort(np.concatenate([t[0][:, 0] for t in members]), kind="stable") + weights = np.concatenate([t[0][:, 1] for t in members])[take] + instants = np.concatenate([t[1] for t in members])[take] + if partitioned: + # The fine level merges the leaf centroids themselves, whole + # and value-ordered, so the leaf's own words index it: cut + # by cumulative FLUX (these weights are photoelectron + # counts, not observation counts) and each merged word must + # cover its contributors' instants — §8.3 over the §9.1 + # partition, as in ``TestBothChannelsOverviewFold``. + fed = np.cumsum(weights) + got = np.concatenate([[0.0], np.cumsum(d[:, 1].astype(np.float64))]) + for k, (lo, hi) in enumerate(zip(got[:-1], got[1:], strict=True)): + lo_i = int(np.searchsorted(fed, lo, side="right")) + hi_i = int(np.searchsorted(fed, hi, side="right")) + assert hi_i > lo_i, "a centroid folded no contributor whole" + assert _toc_contains(w[k], instants[lo_i:hi_i]) + # The coarse level folds the FINE level's centroids, whose + # values are weighted means and so no longer interleave in leaf + # order — the claim that survives the re-partition is the + # cell-level identity, exactly as ``test_cascade_folds_both_ + # siblings`` states it: nothing invented, nothing dropped. + assert int(cell_envelope(w)) == int(cell_envelope(instants)) + populated += len(w) + assert populated > 0 + + def test_single_contributor_row_round_trips_byte_identical(self, tmp_path): + # One populated leaf row under the coarse cell: whichever arm serves it + # (the k-way merge is byte-idempotent on one digest, so the two are + # indistinguishable here — ``TestBothChannelsOverviewFold`` parametrizes + # them apart), the overview element must be the leaf's EXACT bytes and + # words. The ragged (2,) element of a builder-origin payload survives + # the fold and re-encodes byte-identically. + _, truth = self._sweep(tmp_path, {0: 25}, orders=(1,)) + g = _overview_group(tmp_path, "-3/1", "all.zarr", 3) + raw = bytes(g["rx_flux"][:][0]) + d = decode_digest(raw, "float32") + assert d.ndim == 2 and d.shape[1] == 2 + np.testing.assert_array_equal(d, truth[0][0].astype(np.float32)) + assert encode_digest(d, "float32") == raw + np.testing.assert_array_equal( + decode_digest(bytes(g["rx_flux_times"][:][0]), "uint64", ()), truth[0][1] + ) + + class TestLocatedDeclarationGate: """The §9 declaration is checked at retrofit time AND at fold time.