diff --git a/papers/bmodes/config/config.yaml b/papers/bmodes/config/config.yaml index 1198d984..4bc70ac4 100644 --- a/papers/bmodes/config/config.yaml +++ b/papers/bmodes/config/config.yaml @@ -34,9 +34,10 @@ fiducial: B: SP_v1.4.6.3_B_leak_corr C: SP_v1.4.6.3_C_leak_corr - min_sep_int: 0.5 - max_sep_int: 300.0 - nbins_int: 1000 + # k=34, with outward-rounded 0.08–300 arcmin coverage. + min_sep_int: 0.07939053012074941 + max_sep_int: 301.3330747561176 + nbins_int: 1015 # COSEBIS parameters # Compute 20 modes, evaluate PTEs for both 6 and 20 diff --git a/papers/cosmo_val/config/config.yaml b/papers/cosmo_val/config/config.yaml index 9a31e0b8..0e8d6158 100644 --- a/papers/cosmo_val/config/config.yaml +++ b/papers/cosmo_val/config/config.yaml @@ -81,9 +81,11 @@ cosmo_val: # The fine ξ± grid both B-mode statistics (COSEBIs, pure-E/B) integrate over. integration: - min_sep: 0.08 - max_sep: 300 - nbins: 1000 + # k=34; reporting edges are fine edges 312, 346, ..., 992. + # Coverage 0.08–300 arcmin is rounded outward to whole fine bins. + min_sep: 0.07939053012074941 + max_sep: 301.3330747561176 + nbins: 1015 # COSEBIs on the integration grid: mode count and the scale cuts scanned. cosebis: @@ -103,9 +105,9 @@ fiducial: min_sep: 1.0 max_sep: 250.0 nbins: 20 - min_sep_int: 0.5 - max_sep_int: 300.0 - nbins_int: 1000 + min_sep_int: 0.07939053012074941 + max_sep_int: 301.3330747561176 + nbins_int: 1015 nmodes: 20 mock_version: SP_v1.4.6.3 fiducial_min_scale: 12 diff --git a/src/sp_validation/angular_binning.py b/src/sp_validation/angular_binning.py new file mode 100644 index 00000000..2a82c763 --- /dev/null +++ b/src/sp_validation/angular_binning.py @@ -0,0 +1,45 @@ +"""Logarithmic angular-grid contracts, usable by the host workflow.""" + +import math + + +def validate_nested_grids(reporting, integration): + """Require reporting bins to be exact unions of uniform fine bins. + + Both grids use ``min_sep``, ``max_sep`` and ``nbins`` in the same units. + Return the integer oversampling factor. The tolerance allows floating-point + round-off in logarithms, not snapping a requested edge onto a nearby one. + """ + + def grid(config): + lo, hi, n = ( + float(config["min_sep"]), + float(config["max_sep"]), + int(config["nbins"]), + ) + if config.get("bin_type", "Log") != "Log": + raise ValueError("nested ξ grids require logarithmic binning") + if not (0 < lo < hi and n > 0 and n == float(config["nbins"])): + raise ValueError( + "log grids require 0 < min_sep < max_sep and positive integer nbins" + ) + return lo, hi, n, math.log(hi / lo) / n + + rlo, _, rn, rb = grid(reporting) + flo, _, fn, fb = grid(integration) + factor = rb / fb + offset = math.log(rlo / flo) / fb + tol = 1e-10 + prefix = "integration grid must nest into reporting grid: " + if factor < 1 or not math.isclose(factor, round(factor), abs_tol=tol, rel_tol=0): + raise ValueError( + prefix + + f"reporting/fine bin_size ratio {factor:.12g} is not a positive integer" + ) + if not math.isclose(offset, round(offset), abs_tol=tol, rel_tol=0): + raise ValueError( + prefix + f"reporting min_sep lies at non-integer fine edge {offset:.12g}" + ) + if offset < -tol or offset + rn * factor > fn + tol: + raise ValueError(prefix + "the fine grid does not cover every reporting edge") + return int(round(factor)) diff --git a/src/sp_validation/correlation.py b/src/sp_validation/correlation.py index 131a81bc..842aa6fa 100644 --- a/src/sp_validation/correlation.py +++ b/src/sp_validation/correlation.py @@ -3,8 +3,12 @@ import json from pathlib import Path +import numpy as np import treecorr +from sp_validation.angular_binning import validate_nested_grids +from sp_validation.b_modes import _reporting_binning + def _measurement_configs(config): """Return the explicit means settings and default-tolerance patched settings.""" @@ -64,7 +68,7 @@ def measure_with_patches(measure, catalogs, config, means=None): its explicit ``bin_slop``. The patched pass omits ``bin_slop`` so TreeCorr uses its standard default. A cached ``means`` can be reused for repeated covariance draws of the same catalogue. With no patches, one means-config - measurement supplies both products. + measurement supplies both products unless supplied means are reused. """ configs = _measurement_configs(config) if all(cat.npatch == 1 for cat in catalogs.values()): @@ -84,13 +88,39 @@ def measure_with_patches(measure, catalogs, config, means=None): return means, patched -def process_gg(config, cat1, cat2=None): - """Measure GG means with ``config`` and patched resampling at TreeCorr's default. +def rebin_gg_means(fine, config): + """Pair-weighted fine ξ± means on exactly nested reporting edges. - The shared two-pass mechanism omits ``bin_slop`` only from the patched - configuration. Covariance estimates, including joint and derived-statistic - jackknives, retain TreeCorr's per-patch results. Published pair counts, - weights, separations and complex correlations come from the means pass. + The same operator used by pure-E/B averages the complex correlations and + separations. Pair counts and weights are sums over the corresponding fine + bins. This object supplies means only, not a covariance or patch results. + """ + reporting = treecorr.GGCorrelation({**config, "var_method": "shot"}) + if fine.sep_units != reporting.sep_units: + raise ValueError("fine and reporting ξ± must use the same separation units") + validate_nested_grids( + config, dict(min_sep=fine.min_sep, max_sep=fine.max_sep, nbins=fine.nbins) + ) + fine_edges = np.append(fine.left_edges, fine.right_edges[-1]) + report_edges = np.append(reporting.left_edges, reporting.right_edges[-1]) + operator, edges = _reporting_binning(fine.weight, fine_edges, report_edges) + for name in ("xip", "xim", "xip_im", "xim_im", "meanr", "meanlogr"): + getattr(reporting, name)[:] = operator @ getattr(fine, name) + indices = np.searchsorted(fine_edges, edges) + for i, (lo, hi) in enumerate(zip(indices[:-1], indices[1:])): + reporting.weight[i] = np.sum(fine.weight[lo:hi]) + reporting.npairs[i] = np.sum(fine.npairs[lo:hi]) + return reporting + + +def process_gg(config, cat1, cat2=None, means=None): + """Measure GG means and patched resampling products at distinct tolerances. + + The configured ``bin_slop`` applies to unpatched means; the shared + two-pass mechanism omits it from the patched configuration so TreeCorr's + default sets the covariance and resampling tolerance. Supplied ``means`` + replace the unpatched reporting pass, for example with pair-weighted fine-grid + means. The patched covariance is materialized before the means fields change. """ catalogs = {"cat1": cat1} if cat2 is not None: @@ -104,10 +134,9 @@ def measure(cats, measurement_config): gg.process(cats["cat1"], cat2=cats.get("cat2")) return gg - means, gg = measure_with_patches(measure, catalogs, config) + means, gg = measure_with_patches(measure, catalogs, config, means=means) if means is not gg: - # TreeCorr estimates covariance lazily from fields that the means pass - # replaces below, so freeze the patched-pass estimates first. + # Freeze the patched covariance and variances before replacing means. _ = gg.varxip, gg.varxim, gg.cov for name in ( "xip", diff --git a/src/sp_validation/cosmo_val/core.py b/src/sp_validation/cosmo_val/core.py index bd9137ae..7ac4d816 100644 --- a/src/sp_validation/cosmo_val/core.py +++ b/src/sp_validation/cosmo_val/core.py @@ -11,6 +11,7 @@ from shear_psf_leakage import leakage, run_object, run_scale from .. import io +from ..angular_binning import validate_nested_grids from ..b_modes import ( _get_pte_from_scale_cut, covariance_label, @@ -146,6 +147,10 @@ class CosmologyValidation( Maximum angular separation in arcminutes for correlation function binning. nbins : int, default 20 Number of angular bins for TreeCorr real-space correlation functions. + integration : dict, optional + Unpatched fine ξ± grid (min_sep, max_sep, nbins) supplying the published + reporting means. Its bins must nest into the constructor's reporting + grid. Without it, reporting means use an unpatched reporting pass. b_target : float, default 0.01 Maximum logarithmic-bin tolerance for full-sample means, with bin_slop capped at 1 per grid. Patched covariance passes use TreeCorr's default. @@ -315,6 +320,7 @@ def __init__( compute_tomography=False, force_run=False, b_target=0.01, + integration=None, ): self.b_target = b_target self.rho_tau_method = rho_tau_method @@ -388,6 +394,18 @@ def __init__( } self.treecorr_config = self._binning() + self.integration = None + if integration is not None: + if integration.get("npatch", 1) != 1: + raise ValueError( + "the integration grid supplying ξ± means must use npatch=1" + ) + self.integration = { + "min_sep": float(integration["min_sep"]), + "max_sep": float(integration["max_sep"]), + "nbins": int(integration["nbins"]), + } + validate_nested_grids(self.treecorr_config, self.integration) self.catalog_config_path = Path(catalog_config) with self.catalog_config_path.open("r") as file: @@ -624,6 +642,15 @@ def _binning(self, min_sep=None, max_sep=None, nbins=None, **extra): config["bin_slop"] = extra.get("bin_slop", min(1, self.b_target / bin_size)) return config + def _integration_binning(self, min_sep, max_sep, nbins, *, default): + """Configured fine grid, with a statistic's standalone fallback bounds.""" + grid = self.integration or dict(zip(("min_sep", "max_sep", "nbins"), default)) + return self._binning( + grid["min_sep"] if min_sep is None else min_sep, + grid["max_sep"] if max_sep is None else max_sep, + grid["nbins"] if nbins is None else nbins, + ) + def _read_shear_cols(self, ver, *keys): """Read shear-catalog columns by their config-key names. diff --git a/src/sp_validation/cosmo_val/cosebis.py b/src/sp_validation/cosmo_val/cosebis.py index 079720bf..f91676b5 100644 --- a/src/sp_validation/cosmo_val/cosebis.py +++ b/src/sp_validation/cosmo_val/cosebis.py @@ -21,9 +21,9 @@ class CosebisMixin: def calculate_cosebis( self, version, - min_sep_int=0.5, - max_sep_int=500, - nbins_int=1000, + min_sep_int=None, + max_sep_int=None, + nbins_int=None, npatch=None, nmodes=10, cov_path=None, @@ -48,13 +48,13 @@ def calculate_cosebis( The catalog version to compute the COSEBIs for. min_sep_int : float, optional Minimum separation for integration binning (fine binning for COSEBIs). - Defaults to 0.5 arcmin. + Defaults to the configured integration grid, or 0.5 arcmin. max_sep_int : float, optional Maximum separation for integration binning (fine binning for COSEBIs). - Defaults to 500 arcmin. + Defaults to the configured integration grid, or 500 arcmin. nbins_int : int, optional Number of bins for integration binning (fine binning for COSEBIs). - Defaults to 1000. + Defaults to the configured integration grid, or 1000. npatch : int, optional Number of patches for the jackknife resampling. Defaults to self.npatch. nmodes : int, optional @@ -102,7 +102,12 @@ def calculate_cosebis( npatch = npatch or self.npatch # Always use integration binning for COSEBIs calculation (fine binning) - treecorr_config = self._binning(min_sep_int, max_sep_int, nbins_int) + treecorr_config = self._integration_binning( + min_sep_int, max_sep_int, nbins_int, default=(0.5, 500, 1000) + ) + min_sep_int, max_sep_int, nbins_int = ( + treecorr_config[k] for k in ("min_sep", "max_sep", "nbins") + ) # Calculate single fine-binned correlation function for COSEBIs print( @@ -171,9 +176,9 @@ def plot_cosebis( self, version=None, output_dir=None, - min_sep_int=0.5, - max_sep_int=500, - nbins_int=1000, # Integration binning + min_sep_int=None, + max_sep_int=None, + nbins_int=None, # Integration binning npatch=None, nmodes=10, cov_path=None, @@ -201,7 +206,7 @@ def plot_cosebis( Output directory for plots. Defaults to self.cc['paths']['output']. min_sep_int, max_sep_int, nbins_int : float, float, int Integration binning parameters for correlation function - (default: 0.5, 500, 1000) + (the configured grid, otherwise 0.5, 500, 1000) npatch : int, optional Number of patches for jackknife covariance. Defaults to instance value. nmodes : int @@ -234,6 +239,13 @@ def plot_cosebis( # Determine variance method based on whether theoretical covariance is used var_method = "analytic" if cov_path is not None else "jackknife" + integration = self._integration_binning( + min_sep_int, max_sep_int, nbins_int, default=(0.5, 500, 1000) + ) + min_sep_int, max_sep_int, nbins_int = ( + integration[k] for k in ("min_sep", "max_sep", "nbins") + ) + # Create output filename with integration parameters to match Snakemake out_stub = ( f"{output_dir}/{version}_cosebis_minsep={min_sep_int}_" diff --git a/src/sp_validation/cosmo_val/pure_eb.py b/src/sp_validation/cosmo_val/pure_eb.py index 503b02e8..619ea021 100644 --- a/src/sp_validation/cosmo_val/pure_eb.py +++ b/src/sp_validation/cosmo_val/pure_eb.py @@ -7,6 +7,7 @@ import numpy as np +from ..angular_binning import validate_nested_grids from ..b_modes import ( calculate_eb_statistics, calculate_pure_eb_correlation, @@ -26,9 +27,9 @@ def calculate_pure_eb( min_sep=None, max_sep=None, nbins=None, - min_sep_int=0.08, - max_sep_int=300, - nbins_int=1000, + min_sep_int=None, + max_sep_int=None, + nbins_int=None, npatch=None, compute_tomography=False, cov_path_int=None, @@ -48,7 +49,8 @@ def calculate_pure_eb( min_sep, max_sep, nbins : float, float, int, optional Reporting binning. Default to the values in self.treecorr_config. min_sep_int, max_sep_int, nbins_int : float, float, int, optional - Integration binning (default: 0.08-300 arcmin, 1000 bins). It must + Integration binning (the configured grid, otherwise 0.08–300 arcmin, + 1000 bins). It must extend beyond the reporting range on both sides. npatch : int, optional Jackknife patch count. Defaults to self.npatch. @@ -83,11 +85,16 @@ def calculate_pure_eb( reporting_edges = np.geomspace( reporting["min_sep"], reporting["max_sep"], reporting["nbins"] + 1 ) + integration = self._integration_binning( + min_sep_int, max_sep_int, nbins_int, default=(0.08, 300, 1000) + ) + if self.integration is not None: + validate_nested_grids(reporting, integration) ggs_int = self.calculate_2pcf_version( version, npatch=npatch, compute_tomography=compute_tomography, - **self._binning(min_sep_int, max_sep_int, nbins_int), + **integration, ) results = {} @@ -119,9 +126,9 @@ def plot_pure_eb( min_sep=None, max_sep=None, nbins=None, - min_sep_int=0.08, - max_sep_int=300, - nbins_int=1000, + min_sep_int=None, + max_sep_int=None, + nbins_int=None, npatch=None, compute_tomography=False, cov_path_int=None, @@ -150,7 +157,7 @@ def plot_pure_eb( Binning parameters for reporting scale. Uses treecorr_config if None. min_sep_int, max_sep_int, nbins_int : float, float, int Binning parameters for integration scale - (default: 0.08-300 arcmin, 1000 bins) + (the configured grid, otherwise 0.08–300 arcmin, 1000 bins) npatch : int, optional Number of patches for jackknife covariance. Uses self.npatch if None. compute_tomography : bool, optional @@ -182,6 +189,12 @@ def plot_pure_eb( output_dir = output_dir or self.cc["paths"]["output"] npatch = npatch or self.npatch + integration = self._integration_binning( + min_sep_int, max_sep_int, nbins_int, default=(0.08, 300, 1000) + ) + min_sep_int, max_sep_int, nbins_int = ( + integration[k] for k in ("min_sep", "max_sep", "nbins") + ) var_method = "jackknife" if cov_path_int is None else "analytic" # Use treecorr_config defaults for reporting scale binning diff --git a/src/sp_validation/cosmo_val/real_space.py b/src/sp_validation/cosmo_val/real_space.py index a55c1106..7fe6a2a9 100644 --- a/src/sp_validation/cosmo_val/real_space.py +++ b/src/sp_validation/cosmo_val/real_space.py @@ -11,9 +11,11 @@ import numpy as np import treecorr +from sp_validation.angular_binning import validate_nested_grids from sp_validation.correlation import ( measurement_matches, process_gg, + rebin_gg_means, write_measurement_metadata, ) from sp_validation.statistics import jackknife_patch_centers @@ -26,6 +28,7 @@ def calculate_2pcf_version( npatch=None, compute_tomography=False, read_cached=False, + fine_correlations=None, **treecorr_config, ): """ @@ -52,6 +55,11 @@ def calculate_2pcf_version( including patched runs. They carry the published means and variances, but no patch results for dense or derived jackknife covariance. + fine_correlations (dict, optional): Unpatched fine-grid GG objects + keyed by bin pair, measured from the same catalogue selections. + With a configured integration grid these supply the reporting means; + otherwise the fine grid is measured once by this method. + **treecorr_config: Additional TreeCorr configuration parameters that will override the instance's default `treecorr_config`. For example, `min_sep=1`. @@ -66,8 +74,11 @@ def calculate_2pcf_version( dump with a configuration/mean-source JSON sidecar. Unpatched runs can reuse it; patched measurements remeasure to retain covariance. Plotting can opt into reading the saved columns. - - Full-sample means use ``bin_slop`` from the configured means pass. - The patched covariance pass omits it and uses TreeCorr's default. + - With ``integration`` configured, reporting means are pair-weighted + fine-grid means on exactly nested edges. Other measurements use + the configured unpatched means pass. + - ``b_target`` applies only to means. The patched covariance pass + omits ``bin_slop`` so TreeCorr uses its default. - Seeded patch centres are computed once from the full catalogue and shared by every tomographic bin pair. """ @@ -78,6 +89,20 @@ def calculate_2pcf_version( "var_method": "jackknife" if int(npatch) > 1 else "shot", } + integration = self.integration + use_fine = integration is not None and any( + treecorr_config[key] != integration[key] + for key in ("min_sep", "max_sep", "nbins") + ) + cache_config = dict(treecorr_config) + if use_fine: + validate_nested_grids(treecorr_config, integration) + cache_config["integration"] = integration + elif fine_correlations is not None: + raise ValueError( + "fine_correlations needs a configured reporting/integration grid pair" + ) + if compute_tomography: tomo_bin_ids, tomo_bin_pairs = self._get_tomo_bins(ver) if tomo_bin_ids is None or tomo_bin_pairs is None: @@ -97,7 +122,8 @@ def calculate_2pcf_version( if ( (int(npatch) == 1 or read_cached) and not self.force_run - and measurement_matches(out_fname, treecorr_config) + and fine_correlations is None + and measurement_matches(out_fname, cache_config) ): self.print_done(f"Skipping 2PCF calculation, {out_fname} exists") gg = treecorr.GGCorrelation(treecorr_config) @@ -107,6 +133,10 @@ def calculate_2pcf_version( to_compute.append((bin1, bin2)) if to_compute: + if use_fine and fine_correlations is None: + fine_correlations = self.calculate_2pcf_version( + ver, npatch=1, compute_tomography=compute_tomography, **integration + ) cols = self._shear_columns(ver, compute_tomography) patch_centers = self._patch_centers(cols, npatch) @@ -118,7 +148,25 @@ def calculate_2pcf_version( else None ) - gg = process_gg(treecorr_config, cat_gal1, cat_gal2) + means = None + if use_fine: + pair = f"tomo_bin_{bin1}_tomo_bin_{bin2}" + if pair not in fine_correlations: + raise ValueError(f"missing fine-grid correlation for {pair}") + fine = fine_correlations[pair] + if fine.npatch1 != 1 or fine.npatch2 != 1: + raise ValueError( + "fine-grid means must come from unpatched catalogues" + ) + actual = dict( + min_sep=fine.min_sep, max_sep=fine.max_sep, nbins=fine.nbins + ) + if actual != integration: + raise ValueError( + "fine correlation does not match the configured integration grid" + ) + means = rebin_gg_means(fine, treecorr_config) + gg = process_gg(treecorr_config, cat_gal1, cat_gal2, means=means) if (bin1, bin2) == ("all", "all"): # Columns only. The covariance matrix lives in the SACC part; @@ -132,7 +180,7 @@ def calculate_2pcf_version( precision=17, ) write_measurement_metadata( - self._xi_txt_path(ver, treecorr_config, npatch), treecorr_config + self._xi_txt_path(ver, treecorr_config, npatch), cache_config ) ggs[f"tomo_bin_{bin1}_tomo_bin_{bin2}"] = gg diff --git a/src/sp_validation/cosmo_val/sacc_writers.py b/src/sp_validation/cosmo_val/sacc_writers.py index b10bfe81..7eadadc0 100644 --- a/src/sp_validation/cosmo_val/sacc_writers.py +++ b/src/sp_validation/cosmo_val/sacc_writers.py @@ -38,6 +38,9 @@ def xi_to_sacc( theta_nom=None, npairs=None, weight=None, + meanlogr=None, + xip_im=None, + xim_im=None, variances=None, covariance=None, ): @@ -58,6 +61,9 @@ def xi_to_sacc( theta_nom=theta_nom, npairs=npairs, weight=weight, + meanlogr=meanlogr, + xip_im=xip_im, + xim_im=xim_im, ) if covariance is not None and variances is not None: raise ValueError("give xi_to_sacc a dense covariance or variances, not both") diff --git a/src/sp_validation/sacc_io.py b/src/sp_validation/sacc_io.py index ff11db0b..2d16e7b5 100644 --- a/src/sp_validation/sacc_io.py +++ b/src/sp_validation/sacc_io.py @@ -212,6 +212,9 @@ def add_xi( theta_nom=None, npairs=None, weight=None, + meanlogr=None, + xip_im=None, + xim_im=None, ): """Add a real-space shear 2PCF (ξ+ then ξ−) for one tracer pair. @@ -234,18 +237,27 @@ def add_xi( Nominal bin centres — TreeCorr ``rnom`` — stored as ``theta_nom``. npairs, weight : array_like, optional TreeCorr pair counts and weights, stored per point. + meanlogr, xip_im, xim_im : array_like, optional + Auxiliary separation and imaginary means for lossless fine-grid rebinning. """ _check_ascending("theta", theta) tracers = _pair(bins) - optional = {"theta_nom": theta_nom, "npairs": npairs, "weight": weight} + optional = { + "theta_nom": theta_nom, + "npairs": npairs, + "weight": weight, + "meanlogr": meanlogr, + } extras = {key: arr for key, arr in optional.items() if arr is not None} - for dtype, xi in ((XI_PLUS, xip), (XI_MINUS, xim)): + for dtype, xi, imaginary in ((XI_PLUS, xip, xip_im), (XI_MINUS, xim, xim_im)): for n, th in enumerate(theta): tags = { "theta": float(th), "grid": grid, **{key: float(arr[n]) for key, arr in extras.items()}, } + if imaginary is not None: + tags["xi_im"] = float(imaginary[n]) s.add_data_point(dtype, tracers, float(xi[n]), **tags) @@ -585,6 +597,18 @@ def get_xi_weight(s, bins, *, grid): return _tag(s, XI_PLUS, _pair(bins), "weight", grid=grid) +def get_xi_aux(s, bins, *, grid): + """Full-precision auxiliary means needed to rebin an integration ξ± part.""" + tracers = _pair(bins) + fields = { + name: _tag(s, XI_PLUS, tracers, name, grid=grid) + for name in ("theta_nom", "meanlogr", "weight", "npairs") + } + fields["xip_im"] = _tag(s, XI_PLUS, tracers, "xi_im", grid=grid) + fields["xim_im"] = _tag(s, XI_MINUS, tracers, "xi_im", grid=grid) + return fields + + def get_pseudo_cl(s, bins): """Return ``(ell_eff, cl_ee, cl_bb, cl_eb, window)`` for one tracer pair. diff --git a/src/sp_validation/tests/regression/test_tomo_pure_eb_mapping_keyerror.py b/src/sp_validation/tests/regression/test_tomo_pure_eb_mapping_keyerror.py index 2e6ce4bc..85db50ce 100644 --- a/src/sp_validation/tests/regression/test_tomo_pure_eb_mapping_keyerror.py +++ b/src/sp_validation/tests/regression/test_tomo_pure_eb_mapping_keyerror.py @@ -48,6 +48,7 @@ class _TomoCV(PureEBMixin): versions = ["v"] npatch = 32 + integration = None treecorr_config = {"min_sep": 2.0, "max_sep": 60.0, "nbins": 3} def __init__(self, output): @@ -64,6 +65,13 @@ def _binning(self, min_sep=None, max_sep=None, nbins=None): "nbins": nbins or self.treecorr_config["nbins"], } + def _integration_binning(self, min_sep, max_sep, nbins, *, default): + return self._binning( + default[0] if min_sep is None else min_sep, + default[1] if max_sep is None else max_sep, + default[2] if nbins is None else nbins, + ) + def calculate_2pcf_version(self, version, npatch, compute_tomography, **config): return { BIN_KEY: _synthetic_gg( diff --git a/src/sp_validation/tests/test_nested_means.py b/src/sp_validation/tests/test_nested_means.py new file mode 100644 index 00000000..1f50ce8e --- /dev/null +++ b/src/sp_validation/tests/test_nested_means.py @@ -0,0 +1,186 @@ +"""Exactly nested, pair-weighted reporting means and their measurement routes.""" + +import importlib.util +from pathlib import Path + +import numpy as np +import pytest +import treecorr + +from sp_validation import sacc_io +from sp_validation.angular_binning import validate_nested_grids +from sp_validation.correlation import rebin_gg_means +from sp_validation.cosmo_val import CosmologyValidation +from sp_validation.statistics import jackknife_patch_centers +from sp_validation.tests.test_cosmo_val import ( + TestCosmologyValidation as _CatalogFactory, +) + +REPORT = dict(min_sep=1.0, max_sep=250.0, nbins=20) +FINE = dict(min_sep=0.07939053012074941, max_sep=301.3330747561176, nbins=1015) +FIELDS = ("xip", "xim", "xip_im", "xim_im", "meanr", "meanlogr", "weight", "npairs") + + +def _small_grids(): + report = dict(min_sep=5.0, max_sep=200.0, nbins=8) + step = np.log(40) / (8 * 4) + fine = dict(min_sep=5 * np.exp(-5 * step), max_sep=200 * np.exp(5 * step), nbins=42) + return report, fine + + +def test_shipped_nested_grid_edges(): + assert validate_nested_grids(REPORT, FINE) == 34 + edges = np.geomspace(FINE["min_sep"], FINE["max_sep"], 1016) + np.testing.assert_allclose(edges[312:993:34], np.geomspace(1, 250, 21), rtol=1e-14) + assert FINE["min_sep"] < 0.08 and FINE["max_sep"] > 300 + + +@pytest.mark.parametrize( + "fine", + [ + dict(min_sep=0.08, max_sep=300, nbins=1000), + { + **FINE, + "min_sep": FINE["min_sep"] * np.exp(0.000811979546744448), + "max_sep": FINE["max_sep"] * np.exp(0.000811979546744448), + }, + dict(min_sep=np.exp(np.log(250) / (20 * 34)), max_sep=250, nbins=679), + ], +) +def test_non_nested_grids_fail_clearly(fine): + with pytest.raises(ValueError, match="must nest"): + validate_nested_grids(REPORT, fine) + + +def test_exact_weighted_rebin_matches_direct_treecorr(): + rng = np.random.default_rng(4) + cat = treecorr.Catalog( + x=rng.random(500), + y=rng.random(500), + g1=rng.normal(size=500), + g2=rng.normal(size=500), + w=np.exp(rng.normal(size=500)), + ) + # Both slops zero force leaf-pair counting, independent of cell aggregation. + report = dict( + min_sep=0.01, + max_sep=1.0, + nbins=8, + bin_slop=0, + angle_slop=0, + min_top=3, + num_threads=1, + ) + step = np.log(100) / (8 * 4) + fine_config = { + **report, + "min_sep": 0.01 * np.exp(-5 * step), + "max_sep": np.exp(5 * step), + "nbins": 42, + } + direct, fine = treecorr.GGCorrelation(report), treecorr.GGCorrelation(fine_config) + direct.process(cat) + fine.process(cat) + rebinned = rebin_gg_means(fine, report) + for name in FIELDS: + np.testing.assert_allclose( + getattr(rebinned, name), getattr(direct, name), rtol=1e-12, atol=1e-12 + ) + + +@pytest.mark.parametrize("tomography", [False, True]) +def test_reporting_path_uses_fine_means_and_only_a_patched_reporting_pass( + tmp_path, monkeypatch, tomography +): + report, fine_grid = _small_grids() + params, ver = _CatalogFactory._write_synthetic_catalogs( + tmp_path, n_gal=2000, with_tomography=tomography + ) + cv = CosmologyValidation( + [ver], + theta_min=report["min_sep"], + theta_max=report["max_sep"], + nbins=report["nbins"], + npatch=8, + integration=fine_grid, + **params, + ) + cv.treecorr_config.update(min_top=2, num_threads=1) + fine = cv.calculate_2pcf_version( + ver, npatch=1, compute_tomography=tomography, **fine_grid + ) + cols = cv._shear_columns(ver, tomography) + centers = jackknife_patch_centers(cv._bin_catalog(cols, "all", 1), 8) + angle = np.deg2rad(0.35) + rotation = np.array( + [ + [np.cos(angle), -np.sin(angle), 0], + [np.sin(angle), np.cos(angle), 0], + [0, 0, 1], + ] + ) + calls = [] + process = treecorr.GGCorrelation.process + + def recording(gg, cat, *args, **kwargs): + calls.append((gg.nbins, cat.npatch)) + return process(gg, cat, *args, **kwargs) + + monkeypatch.setattr(treecorr.GGCorrelation, "process", recording) + outputs = [] + for layout in (centers, centers @ rotation.T): + monkeypatch.setattr(cv, "_patch_centers", lambda *args, c=layout: c) + outputs.append( + cv.calculate_2pcf_version( + ver, compute_tomography=tomography, fine_correlations=fine + ) + ) + assert calls == [(8, 8)] * (6 if tomography else 2) + for pair in fine: + a, b = outputs[0][pair], outputs[1][pair] + expected = rebin_gg_means(fine[pair], cv._binning()) + for name in FIELDS: + np.testing.assert_array_equal(getattr(a, name), getattr(b, name)) + np.testing.assert_array_equal(getattr(a, name), getattr(expected, name)) + assert not np.array_equal(a.cov, b.cov) + + +def test_sacc_fine_product_drives_the_reporting_producer(tmp_path): + report, fine_grid = _small_grids() + params, ver = _CatalogFactory._write_synthetic_catalogs(tmp_path, n_gal=1200) + script = Path(__file__).resolve().parents[3] / "workflow/scripts/run_2pcf.py" + spec = importlib.util.spec_from_file_location("nested_run_2pcf", script) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + shared = dict( + ver=ver, cat_config=params["catalog_config"], output_dir=params["output_dir"] + ) + fine_path, report_path = tmp_path / "fine.sacc", tmp_path / "report.sacc" + fine = module.run_2pcf( + **shared, **fine_grid, npatch=1, grid="integration", sacc_out=fine_path + ) + measured = module.run_2pcf( + **shared, + **report, + npatch=8, + integration=fine_grid, + fine_xi=fine_path, + sacc_out=report_path, + ) + expected = rebin_gg_means(fine, {**report, "sep_units": "arcmin"}) + for name in FIELDS: + np.testing.assert_array_equal(getattr(measured, name), getattr(expected, name)) + part = sacc_io.load(str(report_path), allow_unblinded=True) + np.testing.assert_array_equal(part.covariance.dense, measured.cov) + np.testing.assert_array_equal( + sacc_io.get_xi_aux(part, (0, 0), grid="reporting")["meanlogr"], + measured.meanlogr, + ) + # A raw-catalogue covariance cannot be paired with a concealed mean input. + part = sacc_io.load(str(fine_path), allow_unblinded=True) + part.metadata["concealed"] = True + sacc_io.save(part, fine_path, type="data") + with pytest.raises(ValueError, match="pre-blinding"): + module.run_2pcf( + **shared, **report, npatch=8, integration=fine_grid, fine_xi=fine_path + ) diff --git a/src/sp_validation/tests/test_xi_grids.py b/src/sp_validation/tests/test_xi_grids.py index 60361d10..4299d341 100644 --- a/src/sp_validation/tests/test_xi_grids.py +++ b/src/sp_validation/tests/test_xi_grids.py @@ -33,7 +33,11 @@ def _load_common(): "theta_max": 250.0, "nbins": 20, "npatch": 100, - "integration": {"min_sep": 0.08, "max_sep": 300, "nbins": 1000}, + "integration": { + "min_sep": 0.07939053012074941, + "max_sep": 301.3330747561176, + "nbins": 1015, + }, "cosebis": {"nmodes": 20, "scale_cuts": [[12, 83]]}, } } @@ -42,9 +46,9 @@ def _load_common(): "max_sep": 250.0, "nbins": 20, "npatch": 1, - "min_sep_int": 0.5, - "max_sep_int": 300, - "nbins_int": 1000, + "min_sep_int": 0.07939053012074941, + "max_sep_int": 301.3330747561176, + "nbins_int": 1015, } @@ -55,11 +59,19 @@ def test_tag_is_built_from_canonical_values(): `max_sep: 300` that reached the tag as "300" would have the consumer ask for a path the producer never writes. """ - grids = common.xi_grids(CONFIG, FIDUCIAL) - # Counts stay integers, so no "nbins=1000.0" creeps into a name. + config = { + "cosmo_val": { + "theta_min": 1, + "theta_max": 256, + "nbins": 8, + "npatch": 100, + "integration": {"min_sep": 0.0625, "max_sep": 512, "nbins": 52}, + } + } + grids = common.xi_grids(config, FIDUCIAL) assert ( common.grid_binning(grids["integration"]) - == "minsep=0.08_maxsep=300.0_nbins=1000_npatch=1" + == "minsep=0.0625_maxsep=512.0_nbins=52_npatch=1" ) @@ -88,4 +100,15 @@ def test_workflow_without_cosmo_val_falls_back_to_fiducial(): """papers/bmodes carries no cosmo_val block; its grids come from FIDUCIAL.""" grids = common.xi_grids({}, FIDUCIAL) assert grids["reporting"]["npatch"] == 1 - assert grids["integration"]["min_sep"] == 0.5 + assert grids["integration"]["min_sep"] == FIDUCIAL["min_sep_int"] + + +def test_workflow_rejects_an_incompatible_integration_grid(): + config = { + "cosmo_val": { + **CONFIG["cosmo_val"], + "integration": {"min_sep": 0.08, "max_sep": 300, "nbins": 1000}, + } + } + with pytest.raises(ValueError, match="must nest"): + common.xi_grids(config, FIDUCIAL) diff --git a/workflow/common.py b/workflow/common.py index fb99cd4c..3552b820 100644 --- a/workflow/common.py +++ b/workflow/common.py @@ -48,6 +48,14 @@ def _plain(path): image_revision = _container.image_revision resolve_image = _container.resolve_image +_binning = importlib.util.module_from_spec( + importlib.util.spec_from_file_location( + "_spv_angular_binning", REPO_SRC / "sp_validation" / "angular_binning.py" + ) +) +_binning.__loader__.exec_module(_binning) +validate_nested_grids = _binning.validate_nested_grids + # Every job inherits this launch's environment (the slurm executor submits with # --export=ALL), and the Snakemake each job step starts keeps its source cache # under XDG_CACHE_HOME, which a login shell may point at node-local storage. @@ -320,6 +328,9 @@ def xi_grids(config, fiducial): grid[key] = float(grid[key]) for key in ("nbins", "npatch"): grid[key] = int(grid[key]) + validate_nested_grids(grids["reporting"], grids["integration"]) + if grids["integration"]["npatch"] != 1: + raise ValueError("the integration grid supplying ξ± means must use npatch=1") return grids @@ -418,6 +429,7 @@ def cv_cov_tau(version, fiducial=None): "n_cov", "n_sim_cov", "b_target", + "integration", "theta_min", "theta_max", "nbins", diff --git a/workflow/rules/cosmo_val.smk b/workflow/rules/cosmo_val.smk index f4eb0e2a..f3349bc4 100644 --- a/workflow/rules/cosmo_val.smk +++ b/workflow/rules/cosmo_val.smk @@ -6,7 +6,7 @@ # each diagnostic is a rule, and the rules are linked by the SACC parts and # products they write under COSMO_VAL (= cosmo_val/output): # -# catalogue ──→ xi (one job per grid: reporting, integration) +# catalogue ──→ xi integration ──→ xi reporting (means + patched covariance) # ├─ integration part ─┬──→ pure_eb (part, npz, figures) # │ └──→ cosebis (part, npz, figures) # └─ reporting .txt ──→ 2pcf plot, ratio_xi_sys_xi diff --git a/workflow/rules/twopoint.smk b/workflow/rules/twopoint.smk index 3c7b2890..c0976cf3 100644 --- a/workflow/rules/twopoint.smk +++ b/workflow/rules/twopoint.smk @@ -13,6 +13,18 @@ def xi_binning(grid): return grid_binning(XI_GRIDS[grid]) +def is_named_reporting(w): + """The configured reporting grid, independently of covariance patch count.""" + return all(float(w[key]) == XI_GRIDS["reporting"][key] for key in ("min_sep", "max_sep", "nbins")) + + +def xi_fine_input(w): + """Only the named reporting measurement consumes the named fine product.""" + if not is_named_reporting(w): + return [] + return [str(COSMO_VAL / f"{w.version}_xi_{xi_binning('integration')}.sacc")] + + rule xi: """TreeCorr ξ±(θ) for one version on one angular grid. @@ -21,6 +33,7 @@ rule xi: """ input: catalog=get_shear_catalog, + fine_xi=xi_fine_input, output: txt=str(COSMO_VAL / "xi_{version}_tomo_bin_all_minsep={min_sep}_maxsep={max_sep}_nbins={nbins}_npatch={npatch}.txt"), sacc=str(COSMO_VAL / "{version}_xi_minsep={min_sep}_maxsep={max_sep}_nbins={nbins}_npatch={npatch}.sacc"), @@ -35,6 +48,7 @@ rule xi: output_dir=str(COSMO_VAL), grid=lambda w: grid_of(XI_GRIDS, w), b_target=config.get("cosmo_val", {}).get("b_target", 0.01), + integration=lambda w: XI_GRIDS["integration"] if is_named_reporting(w) else None, resources: # The fine integration grid needs more memory and wall time than the # ~20-bin reporting one; scale on nbins rather than splitting the rule. diff --git a/workflow/scripts/run_2pcf.py b/workflow/scripts/run_2pcf.py index 911778cb..b153557a 100644 --- a/workflow/scripts/run_2pcf.py +++ b/workflow/scripts/run_2pcf.py @@ -12,8 +12,10 @@ --cat-config /path/to/cosmo_val/cat_config.yaml \ --out -The measurement is binning-agnostic: the reporting and the fine integration -grids are the same compute with different ``--min-sep/--max-sep/--nbins``. +The integration job measures unpatched ξ± once. A configured reporting job +consumes that SACC part, averages its means with pair weights on exactly nested +edges, and measures the reporting covariance. Without a fine input, a standalone +job uses unpatched reporting means. ``CosmologyValidation.calculate_2pcf_version`` measures the non-tomographic ``("all", "all")`` pair and writes its ``xi_{basename}.txt`` dump (a raw byproduct); the ξ± data product is born as SACC here, a *part* named by its @@ -29,6 +31,8 @@ import os import numpy as np +import sacc +import treecorr from sp_validation import sacc_io from sp_validation.cosmo_val import CosmologyValidation @@ -46,6 +50,8 @@ def run_2pcf( sacc_out=None, grid="reporting", b_target=0.01, + integration=None, + fine_xi=None, ): """Measure ξ±(θ) for ``ver`` and write its reporting SACC part. @@ -58,11 +64,19 @@ def run_2pcf( Snakemake-declared output); it defaults to a binning-derived name under the resolved output directory for the CLI path. + ``integration`` and ``fine_xi`` together select the fine-grid mean source; + this measurement producer consumes the unblinded intermediate before the + data products enter blinding. + Returns ------- treecorr.GGCorrelation The measured correlation object (also the source of the SACC part). """ + if (integration is None) != (fine_xi is None): + raise ValueError( + "supply the configured integration grid and its fine ξ± part together" + ) cv = CosmologyValidation( versions=[ver], catalog_config=cat_config, @@ -70,11 +84,45 @@ def run_2pcf( # so the SACC provenance metadata stamps the npatch actually measured npatch=npatch, b_target=b_target, + theta_min=min_sep, + theta_max=max_sep, + nbins=nbins, + integration=integration, ) + fine_correlations = None + if fine_xi is not None: + # This is a raw-catalogue measurement producer, before blinding. + # A concealed input cannot be mixed with its raw-catalogue covariance. + part = sacc.Sacc.load_fits(os.fspath(fine_xi)) + if part.metadata.get("concealed", False): + raise ValueError( + "reporting measurement requires a pre-blinding fine ξ± product" + ) + if ( + part.metadata.get("catalogue_version") != ver + or part.metadata.get("npatch") != 1 + ): + raise ValueError( + "fine ξ± must be the same catalogue version measured with npatch=1" + ) + fine = treecorr.GGCorrelation(cv._binning(**cv.integration, var_method="shot")) + theta, xip, xim = sacc_io.get_xi(part, (0, 0), grid="integration") + fields = sacc_io.get_xi_aux(part, (0, 0), grid="integration") + if fields["theta_nom"].shape != fine.rnom.shape or not np.allclose( + fields["theta_nom"], fine.rnom, rtol=1e-12, atol=0 + ): + raise ValueError( + "fine ξ± product does not match the configured integration grid" + ) + for name, values in {"meanr": theta, "xip": xip, "xim": xim, **fields}.items(): + if name != "theta_nom": + getattr(fine, name)[:] = values + fine_correlations = {"tomo_bin_all_tomo_bin_all": fine} gg = cv.calculate_2pcf_version( ver, npatch=npatch, compute_tomography=False, + fine_correlations=fine_correlations, min_sep=min_sep, max_sep=max_sep, nbins=nbins, @@ -92,6 +140,9 @@ def run_2pcf( theta_nom=gg.rnom, npairs=gg.npairs, weight=gg.weight, + meanlogr=gg.meanlogr, + xip_im=gg.xip_im, + xim_im=gg.xim_im, covariance=gg.cov if jackknife else None, variances=None if jackknife else np.concatenate([gg.varxip, gg.varxim]), ) @@ -116,6 +167,8 @@ def _from_snakemake(smk): output_dir=p["output_dir"], grid=p.get("grid", "reporting"), b_target=p.get("b_target", 0.01), + integration=p.get("integration"), + fine_xi=smk.input.fine_xi[0] if smk.input.fine_xi else None, # The SACC part goes exactly where the rule declares it; the .txt # byproduct still lands under the resolved output dir. sacc_out=smk.output["sacc"], @@ -149,7 +202,21 @@ def _from_cli(argv=None): "--grid", default="reporting", help="SACC grid tag for the measured points" ) ap.add_argument("--b-target", type=float, default=0.01) + ap.add_argument("--fine-xi", help="Unpatched integration-grid SACC part") + ap.add_argument("--min-sep-int", type=float) + ap.add_argument("--max-sep-int", type=float) + ap.add_argument("--nbins-int", type=int) a = ap.parse_args(argv) + integration_args = (a.min_sep_int, a.max_sep_int, a.nbins_int) + if a.fine_xi is not None and any(value is None for value in integration_args): + ap.error("--fine-xi requires --min-sep-int, --max-sep-int and --nbins-int") + if a.fine_xi is None and any(value is not None for value in integration_args): + ap.error("integration-grid arguments require --fine-xi") + integration = ( + dict(zip(("min_sep", "max_sep", "nbins"), integration_args)) + if a.fine_xi + else None + ) run_2pcf( ver=a.ver, min_sep=a.min_sep, @@ -160,6 +227,8 @@ def _from_cli(argv=None): output_dir=a.out, grid=a.grid, b_target=a.b_target, + integration=integration, + fine_xi=a.fine_xi, ) diff --git a/workflow/tests/conftest.py b/workflow/tests/conftest.py index 1dd691f4..c86c4c54 100644 --- a/workflow/tests/conftest.py +++ b/workflow/tests/conftest.py @@ -90,6 +90,7 @@ def _load_module(path, name, env): container = _load_module( REPO / "src" / "sp_validation" / "container.py", "spv_container", {} ) +paths = _load_module(REPO / "workflow" / "common.py", "test_workflow_paths", {}) @dataclasses.dataclass @@ -142,7 +143,7 @@ def _cat_config(catalogue): @pytest.fixture(scope="session") def toy(tmp_path_factory): - root = tmp_path_factory.mktemp("toy") + root = paths._plain(tmp_path_factory.mktemp("toy")) skip = shutil.ignore_patterns(".snakemake", "__pycache__", "tests") shutil.copytree(REPO / "workflow", root / "workflow", ignore=skip) shutil.copytree( diff --git a/workflow/tests/test_dag.py b/workflow/tests/test_dag.py index 438f91a1..6ecd252c 100644 --- a/workflow/tests/test_dag.py +++ b/workflow/tests/test_dag.py @@ -37,6 +37,22 @@ def test_assemble_resolves(toy): }, job.input +def test_reporting_xi_depends_on_the_fine_product(toy): + result = toy.snakemake("-n", "assemble_sacc_all") + assert result.returncode == 0, result.stdout + jobs = [job for job in parse_jobs(result.stdout) if job.rule == "xi"] + grids = toy.common.xi_grids(toy.config, toy.config["fiducial"]) + fine_tag = toy.common.grid_binning(grids["integration"]) + reporting_tag = toy.common.grid_binning(grids["reporting"]) + for version in VERSIONS: + fine_part = str(toy.cosmo_val / f"{version}_xi_{fine_tag}.sacc") + report_part = str(toy.cosmo_val / f"{version}_xi_{reporting_tag}.sacc") + reporting = next(job for job in jobs if report_part in job.output) + integration = next(job for job in jobs if fine_part in job.output) + assert fine_part in reporting.input + assert fine_part not in integration.input + + def test_one_integration_grid(toy): """COSEBIs and pure-E/B share the integration-grid part and its covariance.""" result = toy.snakemake("-n", "assemble_sacc_all") @@ -83,7 +99,7 @@ def test_outputs_stay_in_the_output_roots(toy, named): def test_every_spelling_of_an_output_root_declares_the_same_paths(toy, tmp_path, root): """Snakemake keys its persistence records by path string, so a symlinked spelling of an output root declares the tree's resolved paths.""" - tree = Path(toy.env[root]).resolve() + tree = toy.common._plain(toy.env[root]) link = tmp_path / "link" link.symlink_to(tree, target_is_directory=True) result = toy.snakemake("-n", "all", env={**toy.env, root: str(link)})