API Reference

Core Contracts

Configuration loading and defaults.

class euclid_dsps.config.Paths(catalog: 'Path', ssp: 'Path')

Parameters:

• catalog (Path)

• ssp (Path)

catalog: Path

ssp: Path

exception euclid_dsps.config.ConfigValidationError

Raised when a run configuration is internally inconsistent.

euclid_dsps.config.load_config(path)

Load a YAML config file.

Parameters:

path (str | Path)

Return type:

dict[str, Any]

euclid_dsps.config.normalize_config(config)

Fill lightweight defaults without hiding required paths.

Parameters:

config (dict[str, Any])

Return type:

dict[str, Any]

euclid_dsps.config.validate_config(config)

Validate the normalized runtime configuration.

Validation intentionally checks structure and scalar contracts only. It does not require local data files to exist, so CI can validate configs without shipping the private or large FS2 parquet files.

Parameters:

config (dict[str, Any])

Return type:

None

euclid_dsps.config.validate_catalog_columns(config, available_columns)

Validate that every configured catalog column exists in a data source.

Parameters:

• config (dict[str, Any])

• available_columns (set[str] | list[str] | tuple[str, ...])

Return type:

None

euclid_dsps.config.resolve_path(path, base_dir=None)

Resolve paths relative to the current working directory or config dir.

Parameters:

• path (str | Path)

• base_dir (str | Path | None)

Return type:

Path

Column metadata for the CosmoHub catalog used by this project.

class euclid_dsps.columns.CatalogColumn(name, source_name, group, dtype, unit, description, notes='')

Human-readable metadata for one catalog column.

Parameters:

• name (str)

• source_name (str)

• group (str)

• dtype (str)

• unit (str)

• description (str)

• notes (str)

name: str

source_name: str

group: str

dtype: str

unit: str

description: str

notes: str = ''

euclid_dsps.columns.known_catalog_columns()

Return the canonical local column names documented for catalog 353.

Return type:

set[str]

Catalog, observation, and artifact I/O.

class euclid_dsps.io.BandObservation(name: 'str', column: 'str', flux_fnu_cgs: 'float', mag_ab: 'float', sigma_mag: 'float', error_column: 'str | None' = None, flux_error_fnu_cgs: 'float | None' = None)

Parameters:

• name (str)

• column (str)

• flux_fnu_cgs (float)

• mag_ab (float)

• sigma_mag (float)

• error_column (str | None)

• flux_error_fnu_cgs (float | None)

name: str

column: str

flux_fnu_cgs: float

mag_ab: float

sigma_mag: float

error_column: str | None = None

flux_error_fnu_cgs: float | None = None

class euclid_dsps.io.GalaxyObservation(row_index: 'int', row: 'dict[str, Any]', bands: 'list[BandObservation]')

Parameters:

• row_index (int)

• row (dict[str, Any])

• bands (list[BandObservation])

row_index: int

row: dict[str, Any]

bands: list[BandObservation]

euclid_dsps.io.ensure_dir(path)

Parameters:

path (str | Path)

Return type:

Path

euclid_dsps.io.write_json(path, payload)

Parameters:

• path (str | Path)

• payload (Any)

Return type:

None

euclid_dsps.io.configured_output_formats(config)

Return configured tabular output formats.

Parameters:

config (dict[str, Any])

Return type:

list[str]

euclid_dsps.io.write_dataframe_outputs(frame, out_dir, stem, config, index=False)

Write a dataframe in configured formats and return filenames.

Parameters:

• frame (DataFrame)

• out_dir (str | Path)

• stem (str)

• config (dict[str, Any])

• index (bool)

Return type:

list[str]

euclid_dsps.io.to_jsonable(value)

Parameters:

value (Any)

Return type:

Any

euclid_dsps.io.dataclass_is_instance(value)

Parameters:

value (Any)

Return type:

bool

euclid_dsps.io.read_catalog(path, columns=None, nrows=None)

Read a parquet catalog into memory, optionally truncating rows.

Parameters:

• path (str | Path)

• columns (list[str] | None)

• nrows (int | None)

Return type:

DataFrame

euclid_dsps.io.truth_column_from_spec(spec)

Return the catalog column named by a truth-column config entry.

Parameters:

spec (Any)

Return type:

str | None

euclid_dsps.io.truth_value_from_spec(row, spec)

Read and optionally transform a truth value from a catalog row.

Parameters:

• row (dict[str, Any])

• spec (Any)

Return type:

float | None

euclid_dsps.io.iter_catalog_batches(path, columns=None, batch_size=10000, limit=None, row_indices=None, start_index=0)

Yield catalog batches without loading the full parquet into memory.

Parameters:

• path (str | Path)

• columns (list[str] | None)

• batch_size (int)

• limit (int | None)

• row_indices (set[int] | None)

• start_index (int)

Return type:

Iterable[DataFrame]

euclid_dsps.io.load_row_indices(path)

Load row indices from a one-column text or CSV file.

Parameters:

path (str | Path)

Return type:

list[int]

euclid_dsps.io.flux_fnu_cgs_to_abmag(flux)

Convert F_nu in erg/s/cm^2/Hz to AB magnitude.

Parameters:

flux (float)

Return type:

float

euclid_dsps.io.abmag_to_flux_fnu_cgs(mag)

Convert AB magnitude to F_nu in erg/s/cm^2/Hz.

Parameters:

mag (float)

Return type:

float

euclid_dsps.io.microjy_to_flux_fnu_cgs(flux_microjy)

Convert microJansky to F_nu in erg/s/cm^2/Hz.

Parameters:

flux_microjy (float)

Return type:

float

euclid_dsps.io.microjy_to_abmag(flux_microjy)

Convert microJansky to AB magnitude.

Parameters:

flux_microjy (float)

Return type:

float

euclid_dsps.io.flux_error_to_sigma_mag(flux_fnu_cgs, flux_error_fnu_cgs, floor=None, ceiling=None)

Convert a flux-density uncertainty into a local AB-mag uncertainty.

Parameters:

• flux_fnu_cgs (float)

• flux_error_fnu_cgs (float)

• floor (float | None)

• ceiling (float | None)

Return type:

float

euclid_dsps.io.build_observation(row_index, row, band_configs)

Build one photometric observation from a catalog row.

When a band declares error_column, the catalog flux-density error is converted to a local AB-magnitude uncertainty and used by the likelihood. The configured sigma_mag remains the fallback for bands without usable per-object errors.

Parameters:

• row_index (int)

• row (Series)

• band_configs (list[dict[str, Any]])

Return type:

GalaxyObservation

euclid_dsps.io.required_catalog_columns(config)

Parameters:

config (dict[str, Any])

Return type:

list[str]

Filter loading helpers.

class euclid_dsps.filters.FilterCurve(name: 'str', wave: 'np.ndarray', transmission: 'np.ndarray', source: 'str')

Parameters:

• name (str)

• wave (ndarray)

• transmission (ndarray)

• source (str)

name: str

wave: ndarray

transmission: ndarray

source: str

property effective_wavelength: float

Transmission-weighted central wavelength in Angstrom.

euclid_dsps.filters.load_filters(band_configs)

Load all filters declared in the config.

Parameters:

band_configs (list[dict[str, Any]])

Return type:

dict[str, FilterCurve]

euclid_dsps.filters.load_filter(name, filter_config)

Load an exact HDF5/FITS/DAT curve or build an approximate top-hat.

Parameters:

• name (str)

• filter_config (dict[str, Any])

Return type:

FilterCurve

euclid_dsps.filters.load_hdf5_group_filter(name, path, filter_config)

Load grouped HDF5 filter curves with <group>/wave datasets.

Parameters:

• name (str)

• path (Path)

• filter_config (dict[str, Any])

Return type:

FilterCurve

euclid_dsps.filters.load_ascii_filter(name, path, filter_config)

Load two-column ASCII throughput files.

Parameters:

• name (str)

• path (Path)

• filter_config (dict[str, Any])

Return type:

FilterCurve

euclid_dsps.filters.load_fits_filter(name, path, filter_config)

Load Euclid throughput FITS files and convert wavelength to Angstrom.

Parameters:

• name (str)

• path (Path)

• filter_config (dict[str, Any])

Return type:

FilterCurve

Native DSPS model wrapper.

class euclid_dsps.model.DspsContext(ssp: 'Any', filters: 'dict[str, FilterCurve]', n_sfh_bins: 'int' = 96, z_sun: 'float' = 0.0134, model_config: 'dict[str, Any] | None' = None, cosmos_dust_k_by_code: 'np.ndarray | None' = None, cosmos_dust_curve_names: 'tuple[str, ...]' = (), ssp_wave_jax: 'Any | None' = None, ssp_lgmet_jax: 'Any | None' = None, ssp_lg_age_gyr_jax: 'Any | None' = None, ssp_flux_jax: 'Any | None' = None, compressed_ssp_basis_jax: 'Any | None' = None, compressed_ssp_coeff_jax: 'Any | None' = None, compressed_ssp_scale_jax: 'Any | None' = None, ssp_surviving_mstar_jax: 'Any | None' = None, gas_lgmet_grid_jax: 'Any | None' = None, gas_lgu_grid_jax: 'Any | None' = None, ssp_flux_gas_grid_jax: 'Any | None' = None, compressed_gas_basis_jax: 'Any | None' = None, compressed_gas_coeff_jax: 'Any | None' = None, compressed_gas_scale_jax: 'Any | None' = None, agn_wave_jax: 'Any | None' = None, agn_fagn_grid_jax: 'Any | None' = None, agn_tau_grid_jax: 'Any | None' = None, agn_tage_grid_jax: 'Any | None' = None, agn_logzsol_grid_jax: 'Any | None' = None, agn_template_grid_jax: 'Any | None' = None, agn_component_lgmet_jax: 'Any | None' = None, agn_component_lg_age_gyr_jax: 'Any | None' = None, agn_component_grid_jax: 'Any | None' = None, compressed_agn_basis_jax: 'Any | None' = None, compressed_agn_coeff_jax: 'Any | None' = None, compressed_agn_scale_jax: 'Any | None' = None, ssp_emline_luminosity: 'np.ndarray | None' = None, ssp_emline_wave: 'np.ndarray | None' = None, ssp_emline_name: 'tuple[str, ...]' = (), nebular_emission_mode: 'str' = 'ssp_flux', jax_filters: 'tuple[tuple[Any, Any], ...]' = (), cosmos_dust_k_by_code_jax: 'Any | None' = None)

Parameters:

• ssp (Any)

• filters (dict[str, FilterCurve])

• n_sfh_bins (int)

• z_sun (float)

• model_config (dict[str, Any] | None)

• cosmos_dust_k_by_code (ndarray | None)

• cosmos_dust_curve_names (tuple[str, ...])

• ssp_wave_jax (Any | None)

• ssp_lgmet_jax (Any | None)

• ssp_lg_age_gyr_jax (Any | None)

• ssp_flux_jax (Any | None)

• compressed_ssp_basis_jax (Any | None)

• compressed_ssp_coeff_jax (Any | None)

• compressed_ssp_scale_jax (Any | None)

• ssp_surviving_mstar_jax (Any | None)

• gas_lgmet_grid_jax (Any | None)

• gas_lgu_grid_jax (Any | None)

• ssp_flux_gas_grid_jax (Any | None)

• compressed_gas_basis_jax (Any | None)

• compressed_gas_coeff_jax (Any | None)

• compressed_gas_scale_jax (Any | None)

• agn_wave_jax (Any | None)

• agn_fagn_grid_jax (Any | None)

• agn_tau_grid_jax (Any | None)

• agn_tage_grid_jax (Any | None)

• agn_logzsol_grid_jax (Any | None)

• agn_template_grid_jax (Any | None)

• agn_component_lgmet_jax (Any | None)

• agn_component_lg_age_gyr_jax (Any | None)

• agn_component_grid_jax (Any | None)

• compressed_agn_basis_jax (Any | None)

• compressed_agn_coeff_jax (Any | None)

• compressed_agn_scale_jax (Any | None)

• ssp_emline_luminosity (ndarray | None)

• ssp_emline_wave (ndarray | None)

• ssp_emline_name (tuple[str, ...])

• nebular_emission_mode (str)

• jax_filters (tuple[tuple[Any, Any], ...])

• cosmos_dust_k_by_code_jax (Any | None)

ssp: Any

filters: dict[str, FilterCurve]

n_sfh_bins: int = 96

z_sun: float = 0.0134

model_config: dict[str, Any] | None = None

cosmos_dust_k_by_code: ndarray | None = None

cosmos_dust_curve_names: tuple[str, ...] = ()

ssp_wave_jax: Any | None = None

ssp_lgmet_jax: Any | None = None

ssp_lg_age_gyr_jax: Any | None = None

ssp_flux_jax: Any | None = None

compressed_ssp_basis_jax: Any | None = None

compressed_ssp_coeff_jax: Any | None = None

compressed_ssp_scale_jax: Any | None = None

ssp_surviving_mstar_jax: Any | None = None

gas_lgmet_grid_jax: Any | None = None

gas_lgu_grid_jax: Any | None = None

ssp_flux_gas_grid_jax: Any | None = None

compressed_gas_basis_jax: Any | None = None

compressed_gas_coeff_jax: Any | None = None

compressed_gas_scale_jax: Any | None = None

agn_wave_jax: Any | None = None

agn_fagn_grid_jax: Any | None = None

agn_tau_grid_jax: Any | None = None

agn_tage_grid_jax: Any | None = None

agn_logzsol_grid_jax: Any | None = None

agn_template_grid_jax: Any | None = None

agn_component_lgmet_jax: Any | None = None

agn_component_lg_age_gyr_jax: Any | None = None

agn_component_grid_jax: Any | None = None

compressed_agn_basis_jax: Any | None = None

compressed_agn_coeff_jax: Any | None = None

compressed_agn_scale_jax: Any | None = None

ssp_emline_luminosity: ndarray | None = None

ssp_emline_wave: ndarray | None = None

ssp_emline_name: tuple[str, ...] = ()

nebular_emission_mode: str = 'ssp_flux'

jax_filters: tuple[tuple[Any, Any], ...] = ()

cosmos_dust_k_by_code_jax: Any | None = None

class euclid_dsps.model.ModelResult(parameters: 'dict[str, float]', derived: 'dict[str, float]', wave: 'np.ndarray', rest_sed: 'np.ndarray', dusted_rest_sed: 'np.ndarray', photometry: 'dict[str, dict[str, float]]')

Parameters:

• parameters (dict[str, float])

• derived (dict[str, float])

• wave (ndarray)

• rest_sed (ndarray)

• dusted_rest_sed (ndarray)

• photometry (dict[str, dict[str, float]])

parameters: dict[str, float]

derived: dict[str, float]

wave: ndarray

rest_sed: ndarray

dusted_rest_sed: ndarray

photometry: dict[str, dict[str, float]]

class euclid_dsps.model.JaxModelResult(wave: 'jnp.ndarray', rest_sed: 'jnp.ndarray', dusted_rest_sed: 'jnp.ndarray', model_mags: 'jnp.ndarray', t_obs_gyr: 'jnp.ndarray', formed_mass_msun: 'jnp.ndarray', surviving_stellar_mass_msun: 'jnp.ndarray', sfr_at_obs_msun_per_yr: 'jnp.ndarray', sfr_bins_msun_per_yr: 'jnp.ndarray', lookback_bin_edges_gyr: 'jnp.ndarray', stellar_intrinsic_sed: 'jnp.ndarray | None' = None, stellar_dusted_sed: 'jnp.ndarray | None' = None, gas_sed: 'jnp.ndarray | None' = None, agn_sed: 'jnp.ndarray | None' = None, pre_igm_sed: 'jnp.ndarray | None' = None, post_igm_sed: 'jnp.ndarray | None' = None)

Parameters:

• wave (jnp.ndarray)

• rest_sed (jnp.ndarray)

• dusted_rest_sed (jnp.ndarray)

• model_mags (jnp.ndarray)

• t_obs_gyr (jnp.ndarray)

• formed_mass_msun (jnp.ndarray)

• surviving_stellar_mass_msun (jnp.ndarray)

• sfr_at_obs_msun_per_yr (jnp.ndarray)

• sfr_bins_msun_per_yr (jnp.ndarray)

• lookback_bin_edges_gyr (jnp.ndarray)

• stellar_intrinsic_sed (jnp.ndarray | None)

• stellar_dusted_sed (jnp.ndarray | None)

• gas_sed (jnp.ndarray | None)

• agn_sed (jnp.ndarray | None)

• pre_igm_sed (jnp.ndarray | None)

• post_igm_sed (jnp.ndarray | None)

wave: jnp.ndarray

rest_sed: jnp.ndarray

dusted_rest_sed: jnp.ndarray

model_mags: jnp.ndarray

t_obs_gyr: jnp.ndarray

formed_mass_msun: jnp.ndarray

surviving_stellar_mass_msun: jnp.ndarray

sfr_at_obs_msun_per_yr: jnp.ndarray

sfr_bins_msun_per_yr: jnp.ndarray

lookback_bin_edges_gyr: jnp.ndarray

stellar_intrinsic_sed: jnp.ndarray | None = None

stellar_dusted_sed: jnp.ndarray | None = None

gas_sed: jnp.ndarray | None = None

agn_sed: jnp.ndarray | None = None

pre_igm_sed: jnp.ndarray | None = None

post_igm_sed: jnp.ndarray | None = None

class euclid_dsps.model.BatchSedResult(parameter_names, parameter_matrix, wave, rest_sed, dusted_rest_sed, model_mags, derived)

Batch DSPS SEDs and photometry from one JAX-vmapped call.

Parameters:

• parameter_names (list[str])

• parameter_matrix (ndarray)

• wave (ndarray)

• rest_sed (ndarray)

• dusted_rest_sed (ndarray)

• model_mags (ndarray)

• derived (dict[str, ndarray])

parameter_names: list[str]

parameter_matrix: ndarray

wave: ndarray

rest_sed: ndarray

dusted_rest_sed: ndarray

model_mags: ndarray

derived: dict[str, ndarray]

euclid_dsps.model.load_context(ssp_path, filters, n_sfh_bins=96, cosmos_config=None, nebular_emission='ssp_flux', model_config=None)

Parameters:

• ssp_path (str)

• filters (dict[str, FilterCurve])

• n_sfh_bins (int)

• cosmos_config (dict[str, Any] | None)

• nebular_emission (str)

• model_config (dict[str, Any] | None)

Return type:

DspsContext

euclid_dsps.model.parameters_for_row(base, parameter_columns, row, redshift_config=None)

Merge fixed config parameters with optional per-row catalog overrides.

Parameters:

• base (dict[str, Any])

• parameter_columns (dict[str, str])

• row (dict[str, Any])

• redshift_config (dict[str, Any] | None)

Return type:

dict[str, float]

euclid_dsps.model.resolve_redshift(params, row, redshift_config)

Resolve DSPS redshift from configured initializer.

Parameters:

• params (dict[str, float])

• row (dict[str, Any])

• redshift_config (dict[str, Any])

Return type:

float

euclid_dsps.model.redshift_prior_parameters(z_value, row, redshift_config)

Return row-level redshift prior metadata consumed by fit priors.

Parameters:

• z_value (float)

• row (dict[str, Any])

• redshift_config (dict[str, Any])

Return type:

dict[str, float]

euclid_dsps.model.run_dsps_model(context, params)

Run DSPS from simple SFH/metallicity parameters to SED and photometry.

Parameters:

• context (DspsContext)

• params (dict[str, float])

Return type:

ModelResult

euclid_dsps.model.run_dsps_model_jax(context, params)

Pure-JAX DSPS forward model used by gradient-based fits.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.run_dsps_model_mags_jax(context, params)

Return model magnitudes without constructing diagnostic SED outputs.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.run_lognormal_model_jax(context, params)

Legacy lognormal/MDF DSPS forward model.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.run_popcosmos_binned_model_jax(context, params)

PopCosmos-like seven-bin SFH forward model with age-dependent dust.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.run_popcosmos_binned_model_mags_jax(context, params)

Likelihood-only PopCosmos binned forward path.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.run_diffstar_reduced6_model_jax(context, params)

PopCosmos-like forward model using a six-free-parameter Diffstar SFH.

The catalog/config interface does not currently provide halo assembly parameters, so Diffstar is evaluated with DEFAULT_MAH_PARAMS from diffmah. Gas, dust, IGM, AGN, metallicity, and GPU memory behavior follow the current PopCosmos FSPS-grid path.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.run_diffstar_reduced6_model_mags_jax(context, params)

Likelihood-only Diffstar reduced6 forward path.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.run_diffsky_basic_model_jax(context, params)

Diffsky-aligned forward model using Diffstar + Diffmah object parameters.

This path is intentionally narrower than the PopCosmos-like full model: HLTDS exposes Diffstar, Diffmah, and dust latents, but not object-level gas metallicity/ionization or AGN latents. The first fitting target therefore keeps nebular gas and AGN out of the free parameterization instead of pretending those quantities are recoverable truths.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.run_diffsky_basic_model_mags_jax(context, params)

Likelihood-only Diffsky basic forward path.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_popcosmos_lookback_bin_edges_jax(t_obs)

Return PopCosmos-like seven-bin lookback edges in Gyr.

For normal galaxy ages the first three edges are fixed at 0, 0.03, and 0.10 Gyr, followed by three log-spaced intermediate edges, 0.85*t_obs, and t_obs. Very young universes switch to fixed fractions of t_obs so edges stay finite and strictly increasing.

Parameters:

t_obs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.logsfr_ratios_to_sfr_bins_jax(dlog10_sfr)

Convert six PopCosmos SFR log-ratios into seven relative SFR bins.

The convention is dlog10_sfr_i = log10(SFR_i / SFR_{i+1}), ordered from youngest to oldest lookback bin. Thus positive dlog10_sfr_1 makes the youngest bin higher than the second bin.

Parameters:

dlog10_sfr (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_popcosmos_sfh_table_jax(gal_t_table, t_obs, params)

Build a seven-bin constant-SFR table on cosmic-time samples.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_popcosmos_sfh_time_grid_jax(t_obs, params, n_sfh_bins, model_config=None)

Build the PopCosmos SFH table using the configured time-grid convention.

Parameters:

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

• n_sfh_bins (int)

• model_config (dict[str, Any] | None)

Return type:

tuple[jax.numpy.ndarray, jax.numpy.ndarray]

euclid_dsps.model.build_popcosmos_prospector_step_sfh_table_jax(t_obs, params)

Return a Prospector/FastStepBasis-like step SFH table.

Prospector converts log-age bins into two tabular-SFH samples around each bin edge using a small epsilon. Mirroring that convention removes the age-bin integration error from the DSPS benchmark without changing the PopCosmos seven-bin parameterization.

Parameters:

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

tuple[jax.numpy.ndarray, jax.numpy.ndarray]

euclid_dsps.model.popcosmos_age_weights_jax(t_obs, params, ssp_lg_age_gyr)

Integrate PopCosmos age bins directly onto the SSP age grid.

DSPS’ generic SFH-table age weighting interpolates a cumulative-mass table in log cosmic time. That is appropriate for smooth SFHs, but it distorts the sharp seven-bin PopCosmos step SFH. This direct overlap integral mirrors the mass-in-age-bin convention used by Prospector/FastStepBasis.

Parameters:

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

• ssp_lg_age_gyr (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.ssp_lg_age_bin_edges_jax(ssp_lg_age_gyr)

Return log10(age/Gyr) bin edges around sorted SSP age centers.

Parameters:

ssp_lg_age_gyr (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_diffstar_sfh_table_jax(gal_t_table, t_obs, params)

Evaluate a Diffstar SFH table using default Diffmah MAH parameters.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_diffsky_basic_sfh_table_jax(gal_t_table, t_obs, params)

Evaluate Diffstar using per-object Diffstar and Diffmah parameters.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• t_obs (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.diffsky_basic_dust_params_jax(params)

Map HLTDS dust latents to the existing DSPS attenuation kernel.

HLTDS exposes av and delta. The current DSPS path expects optical depth at V band, a slope-like dust index, and a birth-cloud ratio. We use A_V = 1.086 * tau_V and set the birth-cloud component to zero by default, because no object-level birth-cloud latent is present in the prepared HLTDS table.

Parameters:

params (dict[str, Any])

Return type:

tuple[jax.numpy.ndarray, jax.numpy.ndarray, jax.numpy.ndarray]

euclid_dsps.model.project_sfh_to_popcosmos_sfr_bins_jax(gal_t_table, gal_sfr_table, t_obs)

Project any SFH table into PopCosmos lookback-bin average SFRs.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• gal_sfr_table (jax.numpy.ndarray)

• t_obs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.project_sfh_to_popcosmos_dlogsfr_jax(gal_t_table, gal_sfr_table, t_obs)

Project any SFH to six adjacent PopCosmos log-SFR ratios.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• gal_sfr_table (jax.numpy.ndarray)

• t_obs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.normalize_sfh_to_stellar_mass_jax(gal_t_table, gal_sfr_table, ssp_lg_age_gyr, t_obs, log10_stellar_mass, frac_surviving_by_age=None)

Scale an SFH so the DSPS surviving mass matches log10_stellar_mass.

Parameters:

• gal_t_table (jnp.ndarray)

• gal_sfr_table (jnp.ndarray)

• ssp_lg_age_gyr (jnp.ndarray)

• t_obs (jnp.ndarray)

• log10_stellar_mass (jnp.ndarray)

• frac_surviving_by_age (jnp.ndarray | None)

Return type:

tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]

euclid_dsps.model.normalize_sfh_to_stellar_mass_with_age_weights_jax(gal_t_table, gal_sfr_table, ssp_lg_age_gyr, age_weights, log10_stellar_mass, frac_surviving_by_age=None)

Scale an SFH using precomputed SSP age weights.

Parameters:

• gal_t_table (jnp.ndarray)

• gal_sfr_table (jnp.ndarray)

• ssp_lg_age_gyr (jnp.ndarray)

• age_weights (jnp.ndarray)

• log10_stellar_mass (jnp.ndarray)

• frac_surviving_by_age (jnp.ndarray | None)

Return type:

tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]

euclid_dsps.model.interpolate_ssp_stellar_metallicity_jax(ssp_lgmet, ssp_flux, lgmet_abs)

Interpolate SSP flux to a single absolute log10 stellar metallicity.

Parameters:

• ssp_lgmet (jax.numpy.ndarray)

• ssp_flux (jax.numpy.ndarray)

• lgmet_abs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.interpolate_context_ssp_stellar_metallicity_jax(context, lgmet_abs)

Interpolate the base SSP, using compressed coefficients when configured.

Parameters:

• context (DspsContext)

• lgmet_abs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.lognormal_mdf_lgmet_weights_jax(ssp_lgmet, lgmet_abs_median, scatter_dex)

Return DSPS lognormal-MDF weights over the SSP metallicity grid.

Parameters:

• ssp_lgmet (jax.numpy.ndarray)

• lgmet_abs_median (jax.numpy.ndarray)

• scatter_dex (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.context_ssp_lognormal_mdf_jax(context, lgmet_abs_median, scatter_dex)

Return age-by-wave SSP flux integrated over a lognormal metallicity MDF.

Parameters:

• context (DspsContext)

• lgmet_abs_median (jax.numpy.ndarray)

• scatter_dex (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.diffsky_basic_ssp_flux_by_age_jax(context, model_config, lgmet_abs)

Return Diffsky-basic SSP flux by age for the configured metallicity model.

Parameters:

• context (DspsContext)

• model_config (dict[str, Any])

• lgmet_abs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.interpolate_popcosmos_ssp_stellar_metallicity_jax(context, params, model_config, lgmet_abs)

Return the SSP/gas grid at one stellar metallicity for PopCosmos paths.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

• model_config (dict[str, Any])

• lgmet_abs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.log10_stellar_metallicity_to_absolute_jax(log10_stellar_metallicity, z_sun)

Convert log10(Zstar/Zsun) to absolute log10(Zstar).

Parameters:

• log10_stellar_metallicity (jax.numpy.ndarray)

• z_sun (float)

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_popcosmos_dust_by_age_jax(wave, ssp_lg_age_gyr, sed_by_age, tau2, dust_index_n, tau1_over_tau2, model_config)

Apply the configured PopCosmos-like age-dependent attenuation model.

Parameters:

• wave (jax.numpy.ndarray)

• ssp_lg_age_gyr (jax.numpy.ndarray)

• sed_by_age (jax.numpy.ndarray)

• tau2 (jax.numpy.ndarray)

• dust_index_n (jax.numpy.ndarray)

• tau1_over_tau2 (jax.numpy.ndarray)

• model_config (dict[str, Any] | None)

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_charlot_fall_by_age_jax(wave, ssp_lg_age_gyr, sed_by_age, tau2, dust_index_n, tau1_over_tau2, birth_cloud_slope=-1.0)

Apply Charlot-Fall diffuse and birth-cloud attenuation by SSP age.

Parameters:

• wave (jax.numpy.ndarray)

• ssp_lg_age_gyr (jax.numpy.ndarray)

• sed_by_age (jax.numpy.ndarray)

• tau2 (jax.numpy.ndarray)

• dust_index_n (jax.numpy.ndarray)

• tau1_over_tau2 (jax.numpy.ndarray)

• birth_cloud_slope (float)

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_prospector_fsps_dust_by_age_jax(wave, ssp_lg_age_gyr, sed_by_age, tau2, dust_index_n, tau1_over_tau2, dust_tesc_logyr=7.0, dust1_index=-1.0)

Approximate Prospector/FSPS dust_type=4 plus birth-cloud attenuation.

The diffuse component follows a Calzetti/Kriek-Conroy-like curve via DSPS’ Noll+09 kernel, normalized so tau2 is the V-band optical depth. The birth-cloud component is a V-band optical depth with FSPS-style power-law index dust1_index and applies only below dust_tesc_logyr.

Parameters:

• wave (jax.numpy.ndarray)

• ssp_lg_age_gyr (jax.numpy.ndarray)

• sed_by_age (jax.numpy.ndarray)

• tau2 (jax.numpy.ndarray)

• dust_index_n (jax.numpy.ndarray)

• tau1_over_tau2 (jax.numpy.ndarray)

• dust_tesc_logyr (float)

• dust1_index (float)

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_igm_transmission_jax(wave_rest, rest_sed, z_obs, model_config)

Apply the configured IGM transmission model.

Parameters:

• wave_rest (jax.numpy.ndarray)

• rest_sed (jax.numpy.ndarray)

• z_obs (jax.numpy.ndarray)

• model_config (dict[str, Any] | None)

Return type:

jax.numpy.ndarray

euclid_dsps.model.fsps_madau95_igm_transmission_jax(wave_rest, z_obs, factor=1.0)

JAX port of FSPS igm_absorb.f90 Madau95 transmission.

Parameters:

• wave_rest (jax.numpy.ndarray)

• z_obs (jax.numpy.ndarray)

• factor (float)

Return type:

jax.numpy.ndarray

euclid_dsps.model.interpolate_gas_ssp_grid_jax(context, log10_gas_metallicity, log10_gas_ionization)

Interpolate a gas SSP grid to gas metallicity and ionization.

Parameters:

• context (DspsContext)

• log10_gas_metallicity (jax.numpy.ndarray)

• log10_gas_ionization (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.interpolate_compressed_gas_ssp_grid_jax(context, log10_gas_metallicity, log10_gas_ionization)

Interpolate a low-rank gas SSP grid and reconstruct only the selected grid.

Parameters:

• context (DspsContext)

• log10_gas_metallicity (jax.numpy.ndarray)

• log10_gas_ionization (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.interpolate_compressed_gas_ssp_stellar_metallicity_jax(context, log10_gas_metallicity, log10_gas_ionization, lgmet_abs)

Interpolate compressed gas and stellar-metallicity axes before expansion.

This is algebraically equivalent to reconstructing the gas-selected [stellar_lgmet, age, wave] grid and then interpolating the stellar metallicity axis, but it avoids materializing that extra metallicity axis in vmapped galaxy batches.

Parameters:

• context (DspsContext)

• log10_gas_metallicity (jax.numpy.ndarray)

• log10_gas_ionization (jax.numpy.ndarray)

• lgmet_abs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.gas_metallicity_constraint_penalty_jax(params, model_config, penalty=1e+30)

Hard PopCosmos-like constraint enforcing log10(Zgas/Zsun) >= log10(Zstar/Zsun).

Parameters:

• params (dict[str, Any])

• model_config (dict[str, Any] | None)

• penalty (float)

Return type:

jax.numpy.ndarray

euclid_dsps.model.add_agn_component_jax(context, wave, intrinsic_stellar_sed, dusted_sed, params, model_config)

Add an AGN template grid component.

The AGN amplitude uses a stable bolometric integral over the intrinsic stellar Lnu SED. This is an approximation until the exact FSPS/CLUMPY convention is audited.

Parameters:

• context (DspsContext)

• wave (jax.numpy.ndarray)

• intrinsic_stellar_sed (jax.numpy.ndarray)

• dusted_sed (jax.numpy.ndarray)

• params (dict[str, Any])

• model_config (dict[str, Any] | None)

Return type:

jax.numpy.ndarray

euclid_dsps.model.agn_component_jax(context, wave, intrinsic_stellar_sed, params, model_config, *, age_weights=None, formed_mass=None, stellar_lgmet_abs=None, template_tage_gyr=None, stellar_logzsol=None)

Return the DSPS AGN rest-frame Lnu component before IGM attenuation.

Parameters:

• context (DspsContext)

• wave (jnp.ndarray)

• intrinsic_stellar_sed (jnp.ndarray)

• params (dict[str, Any])

• model_config (dict[str, Any] | None)

• age_weights (jnp.ndarray | None)

• formed_mass (jnp.ndarray | float | None)

• stellar_lgmet_abs (jnp.ndarray | float | None)

• template_tage_gyr (jnp.ndarray | float | None)

• stellar_logzsol (jnp.ndarray | float | None)

Return type:

jnp.ndarray

euclid_dsps.model.agn_component_from_ssp_grid_jax(context, wave, params, model_config, *, age_weights, formed_mass, stellar_lgmet_abs)

Convolve an FSPS-native AGN component SSP grid with the SFH weights.

Parameters:

• context (DspsContext)

• wave (jnp.ndarray)

• params (dict[str, Any])

• model_config (dict[str, Any])

• age_weights (jnp.ndarray | None)

• formed_mass (jnp.ndarray | float | None)

• stellar_lgmet_abs (jnp.ndarray | float | None)

Return type:

jnp.ndarray

euclid_dsps.model.agn_component_from_compressed_grid_jax(context, wave, params, model_config, *, age_weights, formed_mass, stellar_lgmet_abs)

Convolve a compressed FSPS-native AGN component grid with the SFH.

Parameters:

• context (DspsContext)

• wave (jnp.ndarray)

• params (dict[str, Any])

• model_config (dict[str, Any])

• age_weights (jnp.ndarray | None)

• formed_mass (jnp.ndarray | float | None)

• stellar_lgmet_abs (jnp.ndarray | float | None)

Return type:

jnp.ndarray

euclid_dsps.model.apply_agn_host_attenuation_jax(wave, agn_lnu, params, model_config)

Apply optional host attenuation to the AGN component for audit experiments.

Parameters:

• wave (jax.numpy.ndarray)

• agn_lnu (jax.numpy.ndarray)

• params (dict[str, Any])

• model_config (dict[str, Any] | None)

Return type:

jax.numpy.ndarray

euclid_dsps.model.combine_agn_and_igm_jax(wave, dusted_sed, agn_sed, z_obs, model_config)

Combine stellar/gas and AGN spectra with the configured IGM ordering.

Parameters:

• wave (jax.numpy.ndarray)

• dusted_sed (jax.numpy.ndarray)

• agn_sed (jax.numpy.ndarray)

• z_obs (jax.numpy.ndarray)

• model_config (dict[str, Any] | None)

Return type:

tuple[jax.numpy.ndarray, jax.numpy.ndarray]

euclid_dsps.model.predict_mags_jax(context, wave, dusted_sed, z_obs)

Predict configured apparent AB magnitudes with DSPS photometry kernels.

Parameters:

• context (DspsContext)

• wave (jax.numpy.ndarray)

• dusted_sed (jax.numpy.ndarray)

• z_obs (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.model_mags_jax(context, params)

Return only model magnitudes for likelihood/gradient code.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.dynamic_model_args(context)

Return JAX arrays that should be dynamic inputs to jitted model calls.

Parameters:

context (DspsContext)

Return type:

tuple[Any, …]

euclid_dsps.model.bind_dynamic_model_args(context, args)

Return a shallow context copy with large JAX arrays supplied as arguments.

Parameters:

• context (DspsContext)

• args (tuple[Any, ...] | list[Any])

Return type:

DspsContext

euclid_dsps.model.model_mags_jax_dynamic(context, args, params)

Model magnitudes with large context arrays passed as dynamic JAX args.

Parameters:

• context (DspsContext)

• args (tuple[Any, ...] | list[Any])

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.run_dsps_model_jax_dynamic(context, args, params)

Full model result with large context arrays passed as dynamic JAX args.

Parameters:

• context (DspsContext)

• args (tuple[Any, ...] | list[Any])

• params (dict[str, Any])

Return type:

JaxModelResult

euclid_dsps.model.predict_batch_mags(context, parameter_names, parameter_matrix)

Predict magnitudes for many parameter rows with one JAX-vmapped call.

Parameters:

• context (DspsContext)

• parameter_names (list[str])

• parameter_matrix (ndarray)

Return type:

ndarray

euclid_dsps.model.derived_quantities_jax(context, params)

Return derived quantities needed for scientifically comparable reports.

Parameters:

• context (DspsContext)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.predict_batch_derived(context, parameter_names, parameter_matrix)

Compute derived quantities for many fitted parameter rows.

Parameters:

• context (DspsContext)

• parameter_names (list[str])

• parameter_matrix (ndarray)

Return type:

dict[str, ndarray]

euclid_dsps.model.predict_batch_seds(context, parameter_names, parameter_matrix)

Predict rest SEDs, dusted rest SEDs, magnitudes, and derived quantities.

This is the batch/GPU path used by COSMOS-template comparisons after MAP or population fits. It avoids one Python DSPS call per galaxy.

Parameters:

• context (DspsContext)

• parameter_names (list[str])

• parameter_matrix (ndarray)

Return type:

BatchSedResult

euclid_dsps.model.build_lognormal_sfh(gal_t_table, log10_sfr, sfh_t_peak, sfh_tau)

Build a positive SFH in Msun/yr on cosmic-time bins.

Parameters:

• gal_t_table (ndarray)

• log10_sfr (float)

• sfh_t_peak (float)

• sfh_tau (float)

Return type:

ndarray

euclid_dsps.model.build_sfh_table_jax(gal_t_table, params)

Build the simple production SFH table without leaving JAX.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.build_lognormal_sfh_jax(gal_t_table, log10_sfr, sfh_t_peak, sfh_tau)

JAX lognormal SFH used by production fits.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• log10_sfr (jax.numpy.ndarray)

• sfh_t_peak (jax.numpy.ndarray)

• sfh_tau (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.normalize_sfh_mass_jax(gal_t_table, gal_sfr_table, params)

Optionally scale an SFH to a configured formed stellar mass.

Without log10_formed_mass_msun this preserves the historical behavior, where log10_sfr is the SFH amplitude. With it, log10_sfr only sets the pre-normalization shape scale and the luminosity amplitude is controlled by the formed-mass parameter.

Parameters:

• gal_t_table (jax.numpy.ndarray)

• gal_sfr_table (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

tuple[jax.numpy.ndarray, jax.numpy.ndarray]

euclid_dsps.model.apply_dust(wave_angstrom, rest_sed, params)

Apply the configured attenuation model.

Parameters:

• wave_angstrom (ndarray)

• rest_sed (ndarray)

• params (dict[str, float])

Return type:

ndarray

euclid_dsps.model.apply_dust_jax(wave_angstrom, rest_sed, params, cosmos_dust_k_by_code=None)

Apply COSMOS two-component dust when available, else DSPS Salim dust.

Parameters:

• wave_angstrom (jax.numpy.ndarray)

• rest_sed (jax.numpy.ndarray)

• params (dict[str, Any])

• cosmos_dust_k_by_code (ndarray | None)

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_salim_dust_jax(wave_angstrom, rest_sed, params)

Apply DSPS Salim+2018-style attenuation without leaving JAX.

Parameters:

• wave_angstrom (jax.numpy.ndarray)

• rest_sed (jax.numpy.ndarray)

• params (dict[str, Any])

Return type:

jax.numpy.ndarray

euclid_dsps.model.apply_cosmos_two_component_dust_jax(rest_sed, params, cosmos_dust_k_by_code)

Apply the two COSMOS dust curves as a differentiable mixture.

Parameters:

• rest_sed (jax.numpy.ndarray)

• params (dict[str, Any])

• cosmos_dust_k_by_code (ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.model.comparison_rows(observation, result)

Parameters:

• observation (GalaxyObservation)

• result (ModelResult)

Return type:

list[dict[str, float | str]]

Public helpers for JAX parameter-vector model calls.

euclid_dsps.parameter_vectors.parameter_names_from_config(config)

Return the configured free-parameter order as an immutable tuple.

Parameters:

config (dict[str, Any])

Return type:

tuple[str, …]

euclid_dsps.parameter_vectors.free_parameter_bounds_from_config(config, parameter_names)

Return lower and upper bounds for parameter_names from a config.

Parameters:

• config (dict[str, Any])

• parameter_names (tuple[str, ...])

Return type:

tuple[ndarray, ndarray]

euclid_dsps.parameter_vectors.theta_vector_to_param_dict(theta, parameter_names)

Convert one physical parameter vector to a JAX parameter dictionary.

Parameters:

• theta (jax.numpy.ndarray)

• parameter_names (tuple[str, ...])

Return type:

dict[str, jax.numpy.ndarray]

euclid_dsps.parameter_vectors.theta_vector_to_model_param_dict(theta, parameter_names, model_config)

Convert free parameters to a full model-parameter dictionary.

Amortized configs often expose only a compact latent subset while the DSPS forward model still expects additional nuisance/SFH parameters. Those values come from model.fixed_parameters and are overwritten by the free latent parameters when names overlap.

Parameters:

• theta (jax.numpy.ndarray)

• parameter_names (tuple[str, ...])

• model_config (dict[str, Any] | None)

Return type:

dict[str, jax.numpy.ndarray]

euclid_dsps.parameter_vectors.theta_matrix_to_param_dicts_or_pytree(theta, parameter_names)

Return a JAX pytree mapping parameter names to leading-shaped arrays.

Parameters:

• theta (jax.numpy.ndarray)

• parameter_names (tuple[str, ...])

euclid_dsps.parameter_vectors.model_mags_from_theta_matrix_jax(context, model_args, theta, parameter_names)

Evaluate model magnitudes for theta vectors while preserving gradients.

theta may be shaped [D], [N,D], or [K,N,D]. The returned magnitudes preserve the leading dimensions and append the configured band dimension.

Parameters:

• theta (jax.numpy.ndarray)

• parameter_names (tuple[str, ...])

Return type:

jax.numpy.ndarray

euclid_dsps.parameter_vectors.derived_from_theta_matrix_jax(context, model_args, theta, parameter_names)

Evaluate DSPS derived quantities for compact free-parameter vectors.

Parameters:

• theta (jax.numpy.ndarray)

• parameter_names (tuple[str, ...])

Return type:

jax.numpy.ndarray

Inference

Single-galaxy fitting helpers.

class euclid_dsps.fit.FitResult(success: 'bool', message: 'str', best_parameters: 'dict[str, float]', chi2: 'float', photometric_objective: 'float', n_bands: 'int', trace: 'list[dict[str, float]]', model_result: 'ModelResult', gradient_norm: 'float')

Parameters:

• success (bool)

• message (str)

• best_parameters (dict[str, float])

• chi2 (float)

• photometric_objective (float)

• n_bands (int)

• trace (list[dict[str, float]])

• model_result (ModelResult)

• gradient_norm (float)

success: bool

message: str

best_parameters: dict[str, float]

chi2: float

photometric_objective: float

n_bands: int

trace: list[dict[str, float]]

model_result: ModelResult

gradient_norm: float

class euclid_dsps.fit.BatchFitResult(success: 'np.ndarray', message: 'str', parameter_names: 'list[str]', free_parameter_names: 'list[str]', best_parameter_matrix: 'np.ndarray', chi2: 'np.ndarray', photometric_objective: 'np.ndarray', gradient_norm: 'np.ndarray', model_mags: 'np.ndarray', trace: 'list[dict[str, float]]', device: 'str')

Parameters:

• success (ndarray)

• message (str)

• parameter_names (list[str])

• free_parameter_names (list[str])

• best_parameter_matrix (ndarray)

• chi2 (ndarray)

• photometric_objective (ndarray)

• gradient_norm (ndarray)

• model_mags (ndarray)

• trace (list[dict[str, float]])

• device (str)

success: ndarray

message: str

parameter_names: list[str]

free_parameter_names: list[str]

best_parameter_matrix: ndarray

chi2: ndarray

photometric_objective: ndarray

gradient_norm: ndarray

model_mags: ndarray

trace: list[dict[str, float]]

device: str

class euclid_dsps.fit.PopulationFitResult(batch: 'BatchFitResult', hyper_mu: 'dict[str, float]', hyper_sigma: 'dict[str, float]', hyper_relations: 'list[dict[str, Any]]', loss: 'float')

Parameters:

• batch (BatchFitResult)

• hyper_mu (dict[str, float])

• hyper_sigma (dict[str, float])

• hyper_relations (list[dict[str, Any]])

• loss (float)

batch: BatchFitResult

hyper_mu: dict[str, float]

hyper_sigma: dict[str, float]

hyper_relations: list[dict[str, Any]]

loss: float

euclid_dsps.fit.fit_one_galaxy(context, observation, base_params, fit_config)

Fit configured DSPS parameters with pure-JAX gradients.

Parameters:

• context (DspsContext)

• observation (GalaxyObservation)

• base_params (dict[str, float])

• fit_config (dict[str, Any])

Return type:

FitResult

euclid_dsps.fit.fit_galaxy_batch_adam(context, base_params_rows, observed_mag, sigma_mag, fit_config, truth_theta=None, observed_flux=None, flux_error=None, initial_theta=None)

Fit many independent galaxies in one JAX-vmapped Adam run.

Parameters:

• context (DspsContext)

• base_params_rows (list[dict[str, float]])

• observed_mag (ndarray)

• sigma_mag (ndarray)

• fit_config (dict[str, Any])

• truth_theta (ndarray | None)

• observed_flux (ndarray | None)

• flux_error (ndarray | None)

• initial_theta (ndarray | None)

Return type:

BatchFitResult

euclid_dsps.fit.fit_population_batch_adam(context, base_params_rows, observed_mag, sigma_mag, fit_config, initial_theta=None, truth_theta=None, observed_flux=None, flux_error=None)

Joint MAP fit with a Gaussian population prior over free parameters.

Parameters:

• context (DspsContext)

• base_params_rows (list[dict[str, float]])

• observed_mag (ndarray)

• sigma_mag (ndarray)

• fit_config (dict[str, Any])

• initial_theta (ndarray | None)

• truth_theta (ndarray | None)

• observed_flux (ndarray | None)

• flux_error (ndarray | None)

Return type:

PopulationFitResult

MCMC posterior sampling for selected galaxies.

class euclid_dsps.mcmc.MCMCResult(samples: 'dict[str, np.ndarray]', derived_samples: 'dict[str, np.ndarray]', summary: 'list[dict[str, float | str]]', posterior_model_mags: 'np.ndarray', observed_mag: 'np.ndarray', sigma_mag: 'np.ndarray', observed_flux_fnu_cgs: 'np.ndarray', flux_error_fnu_cgs: 'np.ndarray', band_names: 'list[str]', diagnostics: 'dict[str, Any]', chain_ids: 'np.ndarray | None' = None)

Parameters:

• samples (dict[str, ndarray])

• derived_samples (dict[str, ndarray])

• summary (list[dict[str, float | str]])

• posterior_model_mags (ndarray)

• observed_mag (ndarray)

• sigma_mag (ndarray)

• observed_flux_fnu_cgs (ndarray)

• flux_error_fnu_cgs (ndarray)

• band_names (list[str])

• diagnostics (dict[str, Any])

• chain_ids (ndarray | None)

samples: dict[str, ndarray]

derived_samples: dict[str, ndarray]

summary: list[dict[str, float | str]]

posterior_model_mags: ndarray

observed_mag: ndarray

sigma_mag: ndarray

observed_flux_fnu_cgs: ndarray

flux_error_fnu_cgs: ndarray

band_names: list[str]

diagnostics: dict[str, Any]

chain_ids: ndarray | None = None

euclid_dsps.mcmc.sample_one_galaxy(context, observation, base_params, fit_config, sample_config, initial_params=None)

Sample posterior over configured free parameters.

Parameters:

• context (DspsContext)

• observation (GalaxyObservation)

• base_params (dict[str, float])

• fit_config (dict[str, Any])

• sample_config (dict[str, Any])

• initial_params (dict[str, float] | None)

Return type:

MCMCResult

euclid_dsps.mcmc.sample_galaxy_batch_mclmc(context, observations, base_params_rows, fit_config, sample_config, initial_params_rows=None)

Sample multiple independent galaxy posteriors in one joint MCLMC state.

The target factorizes over galaxies, but BlackJAX advances a single joint state with shape (n_galaxies, n_free_parameters). This keeps the output contract identical to one result per galaxy while letting each transition evaluate the DSPS forward model with JAX batching.

Parameters:

• context (DspsContext)

• observations (list[GalaxyObservation])

• base_params_rows (list[dict[str, float]])

• fit_config (dict[str, Any])

• sample_config (dict[str, Any])

• initial_params_rows (list[dict[str, float] | None] | None)

Return type:

list[MCMCResult]

Pure-JAX posterior targets for non-NumPyro samplers.

class euclid_dsps.posterior_target.BoundedParameterTransform(names, lower, upper, gas_metallicity_constraint=None)

Logistic transform between unconstrained sampler space and fit bounds.

Parameters:

• names (tuple[str, ...])

• lower (jax.numpy.ndarray)

• upper (jax.numpy.ndarray)

• gas_metallicity_constraint (tuple[int, int] | None)

names: tuple[str, ...]

lower: jax.numpy.ndarray

upper: jax.numpy.ndarray

gas_metallicity_constraint: tuple[int, int] | None = None

to_bounded(y)

Parameters:

y (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

to_unconstrained(theta)

Parameters:

theta (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

log_abs_det_jacobian(y)

Parameters:

y (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

class euclid_dsps.posterior_target.PosteriorTarget(context, model_args, base_params, transform, prior_specs, observed, sigma, finite_mask, fit_config, likelihood_space, photometric_likelihood, student_t_dof, band_offsets)

A BlackJAX-friendly posterior over an unconstrained parameter vector.

Parameters:

• context (DspsContext)

• model_args (tuple[Any, ...])

• base_params (dict[str, float])

• transform (BoundedParameterTransform)

• prior_specs (tuple[dict[str, float | str], ...])

• observed (jax.numpy.ndarray)

• sigma (jax.numpy.ndarray)

• finite_mask (jax.numpy.ndarray)

• fit_config (dict[str, Any])

• likelihood_space (str)

• photometric_likelihood (str)

• student_t_dof (float)

• band_offsets (jax.numpy.ndarray)

context: DspsContext

model_args: tuple[Any, ...]

base_params: dict[str, float]

transform: BoundedParameterTransform

prior_specs: tuple[dict[str, float | str], ...]

observed: jax.numpy.ndarray

sigma: jax.numpy.ndarray

finite_mask: jax.numpy.ndarray

fit_config: dict[str, Any]

likelihood_space: str

photometric_likelihood: str

student_t_dof: float

band_offsets: jax.numpy.ndarray

property free_names: list[str]

physical_from_unconstrained(y)

Parameters:

y (jax.numpy.ndarray)

Return type:

dict[str, Any]

theta_from_unconstrained(y)

Parameters:

y (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

unconstrained_from_theta(theta)

Parameters:

theta (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

logdensity(y)

Parameters:

y (jax.numpy.ndarray)

Return type:

jax.numpy.ndarray

euclid_dsps.posterior_target.build_posterior_target(context, model_args, base_params, fit_config, sample_config, observed_mag, sigma_mag, observed_flux, flux_error)

Build the unconstrained-space posterior target used by BlackJAX.

Parameters:

• context (DspsContext)

• model_args (tuple[Any, ...])

• base_params (dict[str, float])

• fit_config (dict[str, Any])

• sample_config (dict[str, Any])

• observed_mag (ndarray)

• sigma_mag (ndarray)

• observed_flux (ndarray)

• flux_error (ndarray)

Return type:

PosteriorTarget

euclid_dsps.posterior_target.initial_unconstrained_position(target, initial_params, fit_config)

Return a finite unconstrained initial vector from MAP or config initial values.

Parameters:

• target (PosteriorTarget)

• initial_params (dict[str, float] | None)

• fit_config (dict[str, Any])

Return type:

jax.numpy.ndarray

Active Workflows

Workflow orchestration entry points.

euclid_dsps.workflows.fit_batch(config, out_dir, limit=25, batch_size=1000, row_indices_file=None, start_index=0)

Fit the configured free parameters for many rows.

The default path optimizes each parquet chunk with one JAX-vmapped Adam run.

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

• limit (int | None)

• batch_size (int)

• row_indices_file (str | None)

• start_index (int)

Return type:

None

euclid_dsps.workflows.fit_one(config, out_dir)

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

Return type:

None

euclid_dsps.workflows.prepare_one(config)

Parameters:

config (dict[str, Any])

euclid_dsps.workflows.run_batch(config, out_dir, limit=None, batch_size=10000, row_indices_file=None, start_index=0)

Run the same configured DSPS model over many catalog rows.

This is intentionally conservative: it writes a flat comparison table and supports per-row physical parameters through model.parameter_columns.

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

• limit (int | None)

• batch_size (int)

• row_indices_file (str | None)

• start_index (int)

Return type:

None

euclid_dsps.workflows.run_eda(config, out_dir)

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

Return type:

None

euclid_dsps.workflows.run_one(config, out_dir)

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

Return type:

DataFrame

euclid_dsps.workflows.sample_batch(config, out_dir, limit=5, batch_size=1, row_indices_file=None, start_index=0)

Sample independent galaxy posteriors.

NumPyro HMC/NUTS stays one object at a time. BlackJAX MCLMC can use a batched-galaxy joint target when batch_size > 1.

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

• limit (int | None)

• batch_size (int)

• row_indices_file (str | None)

• start_index (int)

Return type:

None

euclid_dsps.workflows.sample_one(config, out_dir)

Parameters:

• config (dict[str, Any])

• out_dir (str | Path)

Return type:

None

EDA and run reporting.

euclid_dsps.reporting.core.configure_plot_style()

Apply project-wide, publication-style matplotlib defaults.

Return type:

None

euclid_dsps.reporting.core.write_eda_outputs(df, band_configs, out_dir, redshift_config=None)

Parameters:

• df (DataFrame)

• band_configs (list[dict[str, Any]])

• out_dir (str | Path)

• redshift_config (dict[str, Any] | None)

Return type:

None

euclid_dsps.reporting.core.write_run_outputs(observation, result, out_dir, *, ground_truth_sed=None, include_filters=True)

Parameters:

• observation (GalaxyObservation)

• result (ModelResult)

• out_dir (str | Path)

• ground_truth_sed (Any | None)

• include_filters (bool)

Return type:

DataFrame

euclid_dsps.reporting.core.write_sed_diagnostic_outputs(observation, result, out_dir, *, stem, ground_truth_sed=None, include_filters=True)

Write one rich SED diagnostic: DSPS SED, optional COSMOS proxy, filters, photometry.

Parameters:

• observation (GalaxyObservation)

• result (ModelResult)

• out_dir (str | Path)

• stem (str)

• ground_truth_sed (Any | None)

• include_filters (bool)

Return type:

dict[str, Any]

euclid_dsps.reporting.core.write_fit_outputs(fit_result, out_dir)

Parameters:

• fit_result (Any)

• out_dir (str | Path)

Return type:

None

euclid_dsps.reporting.core.write_mcmc_outputs(mcmc_result, out_dir, truth_values=None)

Parameters:

• mcmc_result (Any)

• out_dir (str | Path)

• truth_values (dict[str, Any] | None)

Return type:

None

euclid_dsps.reporting.core.write_mcmc_batch_outputs(summary, predictive, diagnostics, out_dir)

Parameters:

• summary (DataFrame)

• predictive (DataFrame)

• diagnostics (DataFrame)

• out_dir (str | Path)

Return type:

None

euclid_dsps.reporting.core.write_population_corner_outputs(fits, free_parameters, out_dir, config=None)

Write population-level MAP point-estimate distributions.

Parameters:

• fits (DataFrame)

• free_parameters (list[str])

• out_dir (str | Path)

• config (dict[str, Any] | None)

Return type:

None

euclid_dsps.reporting.core.plot_population_parameter_histograms(params, path, truth=None)

Parameters:

• params (DataFrame)

• path (str | Path)

• truth (DataFrame | None)

Return type:

None

euclid_dsps.reporting.core.parameter_truth_metrics(frame, config=None)

Summarize paired inferred-vs-truth parameter errors.

Parameters:

• frame (DataFrame)

• config (dict[str, Any] | None)

Return type:

DataFrame

euclid_dsps.reporting.core.write_trace_truth_outputs(trace, out_dir, label, make_plots=True)

Parameters:

• trace (DataFrame)

• out_dir (str | Path)

• label (str)

• make_plots (bool)

Return type:

None

euclid_dsps.reporting.core.trace_truth_summary(trace)

Parameters:

trace (DataFrame)

Return type:

DataFrame

euclid_dsps.reporting.core.plot_trace_truth_metrics(trace, path)

Parameters:

• trace (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.write_batch_outputs(comparison, out_dir, label='batch', reporting_level='full', config=None)

Write aggregate tables and plots for multi-galaxy runs.

Parameters:

• comparison (DataFrame)

• out_dir (str | Path)

• label (str)

• reporting_level (str)

• config (dict[str, Any] | None)

Return type:

None

euclid_dsps.reporting.core.write_fit_diagnostic_outputs(fits, comparison, config, out_dir, label='batch_fit', hyperparameters=None)

Write fit audit tables that protect scientific interpretation.

Parameters:

• fits (DataFrame)

• comparison (DataFrame)

• config (dict[str, Any])

• out_dir (str | Path)

• label (str)

• hyperparameters (DataFrame | None)

Return type:

None

euclid_dsps.reporting.core.plot_fit_free_parameter_distributions(fits, config, path)

Plot MAP distributions for configured free parameters with fit bounds.

Parameters:

• fits (DataFrame)

• config (dict[str, Any])

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.fit_parameter_audit(fits, config)

Summarize whether reported fit columns were truly inferred.

Parameters:

• fits (DataFrame)

• config (dict[str, Any])

Return type:

DataFrame

euclid_dsps.reporting.core.fit_objective_components(fits, comparison, config, hyperparameters=None)

Post-hoc objective decomposition from saved MAP rows.

Parameters:

• fits (DataFrame)

• comparison (DataFrame)

• config (dict[str, Any])

• hyperparameters (DataFrame | None)

Return type:

DataFrame

euclid_dsps.reporting.core.plot_population_bias_heatmap(by_row, path)

Plot a heatmap of mode-aware fit quality in the Redshift-Mass plane.

Parameters:

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_color_redshift_diagnostics(comparison, by_row, path)

Plot broad-band color-redshift diagnostics, POP-COSMOS style.

Parameters:

• comparison (DataFrame)

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_physical_population_diagnostics(by_row, path)

Plot fitted/proxy physical relations in redshift bins.

Parameters:

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.summarize_by_band(valid)

Parameters:

valid (DataFrame)

Return type:

DataFrame

euclid_dsps.reporting.core.summarize_by_row(valid)

Parameters:

valid (DataFrame)

Return type:

DataFrame

euclid_dsps.reporting.core.residuals_by_property(by_row)

Summarize row residuals against catalog/fit properties.

Parameters:

by_row (DataFrame)

Return type:

DataFrame

euclid_dsps.reporting.core.redshift_attractor_summary(by_row, bin_width=0.05, min_count=5, max_modes=30)

Summarize repeated fitted-redshift modes from MAP output.

Parameters:

• by_row (DataFrame)

• bin_width (float)

• min_count (int)

• max_modes (int)

Return type:

DataFrame

euclid_dsps.reporting.core.plot_flux_distributions(df, columns, path)

Parameters:

• df (DataFrame)

• columns (list[str])

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_color_distributions(df, columns, path)

Parameters:

• df (DataFrame)

• columns (list[str])

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_redshift_distributions(df, redshift_config, path)

Parameters:

• df (DataFrame)

• redshift_config (dict[str, Any])

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_physical_parameters_distributions(df, path)

Parameters:

• df (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_sed(result, path)

Parameters:

• result (ModelResult)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_sed_diagnostic(result, path, *, observation=None, ground_truth_sed=None, include_filters=True)

Plot DSPS SED, COSMOS proxy SED, and model-anchored photometry residuals.

Parameters:

• result (ModelResult)

• path (str | Path)

• observation (GalaxyObservation | None)

• ground_truth_sed (Any | None)

• include_filters (bool)

Return type:

None

euclid_dsps.reporting.core.plot_photometry_comparison(comparison, path)

Parameters:

• comparison (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_fit_trace(trace, path)

Parameters:

• trace (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.write_posterior_predictive(mcmc_result, path)

Parameters:

• mcmc_result (Any)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_mcmc_traces(samples, path, chain_ids=None)

Parameters:

• samples (DataFrame)

• path (str | Path)

• chain_ids (ndarray | None)

Return type:

None

euclid_dsps.reporting.core.plot_corner(samples, path)

Parameters:

• samples (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.posterior_comparable_frame(samples, derived, truth_values)

Build posterior columns that have like-for-like truth/proxy values.

Parameters:

• samples (DataFrame)

• derived (DataFrame)

• truth_values (dict[str, Any])

Return type:

DataFrame

euclid_dsps.reporting.core.plot_corner_with_truth(samples, truth_values, path)

Parameters:

• samples (DataFrame)

• truth_values (dict[str, Any])

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_posterior_predictive(mcmc_result, path)

Parameters:

• mcmc_result (Any)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_posterior_intervals(summary, path)

Parameters:

• summary (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_posterior_predictive(predictive, path)

Parameters:

• predictive (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_mcmc_diagnostics(diagnostics, path)

Parameters:

• diagnostics (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.write_workflow_comparison(map_fits, population_fits, hmc_summary, hmc_diagnostics, free_parameters, out_dir, hmc_samples=None)

Parameters:

• map_fits (DataFrame)

• population_fits (DataFrame)

• hmc_summary (DataFrame)

• hmc_diagnostics (DataFrame)

• free_parameters (list[str])

• out_dir (str | Path)

• hmc_samples (DataFrame | None)

Return type:

None

euclid_dsps.reporting.core.workflow_parameter_comparison(map_fits, population_fits, free_parameters)

Parameters:

• map_fits (DataFrame)

• population_fits (DataFrame)

• free_parameters (list[str])

Return type:

DataFrame

euclid_dsps.reporting.core.workflow_fit_comparison(map_fits, population_fits)

Parameters:

• map_fits (DataFrame)

• population_fits (DataFrame)

Return type:

DataFrame

euclid_dsps.reporting.core.workflow_hmc_comparison(map_fits, population_fits, hmc_summary, free_parameters)

Parameters:

• map_fits (DataFrame)

• population_fits (DataFrame)

• hmc_summary (DataFrame)

• free_parameters (list[str])

Return type:

DataFrame

euclid_dsps.reporting.core.plot_workflow_parameter_corners(map_fits, population_fits, hmc_summary, hmc_samples, free_parameters, out)

Parameters:

• map_fits (DataFrame)

• population_fits (DataFrame)

• hmc_summary (DataFrame)

• hmc_samples (DataFrame | None)

• free_parameters (list[str])

• out (Path)

Return type:

None

euclid_dsps.reporting.core.paired_fit_truth_frames(fits, config=None)

Return aligned frames for parameters with both fit_ and truth_ columns.

Parameters:

• fits (DataFrame)

• config (dict[str, Any] | None)

Return type:

tuple[DataFrame, DataFrame]

euclid_dsps.reporting.core.plot_corner_overlay(inferred, truth, path)

Parameters:

• inferred (DataFrame)

• truth (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_map_population_parameters(comparison, path)

Parameters:

• comparison (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_map_population_chi2(comparison, path)

Parameters:

• comparison (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_hmc_map_population(comparison, path)

Parameters:

• comparison (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_dashboard(valid, by_row, path)

Parameters:

• valid (DataFrame)

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_residuals_by_band(valid, path)

Parameters:

• valid (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_observed_vs_model(valid, path)

Parameters:

• valid (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_redshift_truth(by_row, path)

Parameters:

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_redshift_attractors(by_row, attractors, path)

Parameters:

• by_row (DataFrame)

• attractors (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_batch_parameter_truth(by_row, path, config=None)

Parameters:

• by_row (DataFrame)

• path (str | Path)

• config (dict[str, Any] | None)

Return type:

None

euclid_dsps.reporting.core.plot_residuals_by_property(by_row, path)

Parameters:

• by_row (DataFrame)

• path (str | Path)

Return type:

None

euclid_dsps.reporting.core.plot_residual_boxplot(valid, ax)

Parameters:

• valid (DataFrame)

• ax (matplotlib.pyplot.Axes)

Return type:

None

euclid_dsps.reporting.core.plot_observed_model_scatter(valid, ax)

Parameters:

• valid (DataFrame)

• ax (matplotlib.pyplot.Axes)

Return type:

None

euclid_dsps.reporting.core.plot_redshift_scatter(by_row, ax)

Parameters:

• by_row (DataFrame)

• ax (matplotlib.pyplot.Axes)

Return type:

None

euclid_dsps.reporting.core.ordered_bands(df)

Parameters:

df (DataFrame)

Return type:

list[str]

CLI helpers for Diffsky/OpenCosmo dataset investigation.

euclid_dsps.diffsky_data.cli.add_diffsky_subcommands(sub)

Parameters:

sub (_SubParsersAction)

Return type:

None

euclid_dsps.diffsky_data.cli.run_diffsky_command(args)

Parameters:

args (Namespace)

Return type:

None

euclid_dsps.diffsky_data.cli.write_diffsky_fit_report(*, run_dir, config_path, label, reporting_level)

Regenerate aggregate tables and write a compact Diffsky recovery report.

Parameters:

• run_dir (Path)

• config_path (Path | None)

• label (str)

• reporting_level (str)

Return type:

Path

Diffsky/FENIKS proposal generation and DSPS closure utilities.

class euclid_dsps.synthetic_diffsky.SyntheticDiffskyConfig(output_dir: 'Path', proposal_backend: 'str', calibration_dir: 'str', calibration_name: 'str', diffsky_commit: 'str | None', n_host_halos_per_shard: 'int', max_shards: 'int', jax_batch_size: 'int', z_min: 'float', z_max: 'float', lgmp_min: 'float', lgmp_max: 'float', sky_area_degsq: 'float', mc_merge: 'int', z_phot_table_size: 'int', logmp_cutoff: 'float', weighted_lc_photdata_kwargs: 'dict[str, Any]', mc_lc_phot_kwargs: 'dict[str, Any]', metallicity_grid_policy: 'str', stellar_metallicity_scatter_dex: 'float', max_duplication_fraction: 'float', duplication_gate: 'str', min_ess_fraction: 'float', pool_size_factor: 'float', selection: 'dict[str, Any]', output_layers: 'dict[str, Any]', flux_error_model: 'dict[str, Any]', splits: 'dict[str, SplitGenerationConfig]', config_hash: 'str | None' = None)

Parameters:

• output_dir (Path)

• proposal_backend (str)

• calibration_dir (str)

• calibration_name (str)

• diffsky_commit (str | None)

• n_host_halos_per_shard (int)

• max_shards (int)

• jax_batch_size (int)

• z_min (float)

• z_max (float)

• lgmp_min (float)

• lgmp_max (float)

• sky_area_degsq (float)

• mc_merge (int)

• z_phot_table_size (int)

• logmp_cutoff (float)

• weighted_lc_photdata_kwargs (dict[str, Any])

• mc_lc_phot_kwargs (dict[str, Any])

• metallicity_grid_policy (str)

• stellar_metallicity_scatter_dex (float)

• max_duplication_fraction (float)

• duplication_gate (str)

• min_ess_fraction (float)

• pool_size_factor (float)

• selection (dict[str, Any])

• output_layers (dict[str, Any])

• flux_error_model (dict[str, Any])

• splits (dict[str, SplitGenerationConfig])

• config_hash (str | None)

output_dir: Path

proposal_backend: str

calibration_dir: str

calibration_name: str

diffsky_commit: str | None

n_host_halos_per_shard: int

max_shards: int

jax_batch_size: int

z_min: float

z_max: float

lgmp_min: float

lgmp_max: float

sky_area_degsq: float

mc_merge: int

z_phot_table_size: int

logmp_cutoff: float

weighted_lc_photdata_kwargs: dict[str, Any]

mc_lc_phot_kwargs: dict[str, Any]

metallicity_grid_policy: str

stellar_metallicity_scatter_dex: float

max_duplication_fraction: float

duplication_gate: str

min_ess_fraction: float

pool_size_factor: float

selection: dict[str, Any]

output_layers: dict[str, Any]

flux_error_model: dict[str, Any]

splits: dict[str, SplitGenerationConfig]

config_hash: str | None = None

euclid_dsps.synthetic_diffsky.load_synthetic_diffsky_config(config, *, smoke=False, max_galaxies=None)

Resolve the synthetic Diffsky generation block from a normalized config.

Parameters:

• config (dict[str, Any])

• smoke (bool)

• max_galaxies (int | None)

Return type:

SyntheticDiffskyConfig

Canonical theta construction from synthetic FENIKS truth columns.

euclid_dsps.synthetic_diffsky.truth_theta.build_trueparam_theta(frame, config, *, allow_partial_truth=False, truth_schema=None)

Build the canonical Diffsky-basic theta matrix from truth columns.

Parameters:

• frame (DataFrame)

• config (dict[str, Any])

• allow_partial_truth (bool)

• truth_schema (str | None)

Return type:

tuple[ndarray, DataFrame]

Supervised population-prior learning from Diffsky truth parameters.

class euclid_dsps.prior_learning.ParameterSpec(name, column, semantic, lower=None, upper=None)

One supervised-prior parameter and its source truth column.

Parameters:

• name (str)

• column (str)

• semantic (str)

• lower (float | None)

• upper (float | None)

name: str

column: str

semantic: str

lower: float | None = None

upper: float | None = None

to_dict()

Return type:

dict[str, Any]

class euclid_dsps.prior_learning.PriorWorkflowPlan(dataset_dir, schema, parameters, artifacts, stages, warnings, blockers)

Resolved workflow contract, preflight status, and launch recipes.

Parameters:

• dataset_dir (str)

• schema (str)

• parameters (tuple[str, ...])

• artifacts (tuple[WorkflowArtifact, ...])

• stages (tuple[WorkflowStage, ...])

• warnings (tuple[str, ...])

• blockers (tuple[str, ...])

dataset_dir: str

schema: str

parameters: tuple[str, ...]

artifacts: tuple[WorkflowArtifact, ...]

stages: tuple[WorkflowStage, ...]

warnings: tuple[str, ...]

blockers: tuple[str, ...]

property dataset_contract_ready: bool

Whether required FENIKS splits/schema are present and usable.

property ready: bool

Backward-compatible alias for the dataset contract readiness gate.

to_dict()

Return type:

dict[str, Any]

class euclid_dsps.prior_learning.TruthSchema(name, parameters, missing_columns, reduced)

Resolved schema after applying available-column and missing policies.

Parameters:

• name (str)

• parameters (tuple[ParameterSpec, ...])

• missing_columns (tuple[str, ...])

• reduced (bool)

name: str

parameters: tuple[ParameterSpec, ...]

missing_columns: tuple[str, ...]

reduced: bool

to_dict()

Return type:

dict[str, Any]

class euclid_dsps.prior_learning.WorkflowArtifact(name, path, kind, required_for, exists, detail='')

One file or directory used by the workflow.

Parameters:

• name (str)

• path (str)

• kind (str)

• required_for (str)

• exists (bool)

• detail (str)

name: str

path: str

kind: str

required_for: str

exists: bool

detail: str = ''

class euclid_dsps.prior_learning.WorkflowStage(name, purpose, ready, command, sbatch=None, notes=())

One executable step in the FENIKS prior workflow.

Parameters:

• name (str)

• purpose (str)

• ready (bool)

• command (str)

• sbatch (str | None)

• notes (tuple[str, ...])

name: str

purpose: str

ready: bool

command: str

sbatch: str | None = None

notes: tuple[str, ...] = ()

euclid_dsps.prior_learning.build_truth_schema(available_columns, *, schema_name, missing_policy='reduce')

Resolve a supervised-prior truth schema against available columns.

Parameters:

• available_columns (Sequence[str])

• schema_name (str)

• missing_policy (str)

Return type:

TruthSchema

euclid_dsps.prior_learning.build_feniks_prior_workflow_plan(generation_config, *, validation_config=None, prior_config, amortized_config, generation_config_path='configs/diffsky_synthetic_feniks_260617_50k.yaml', validation_config_path='configs/diffsky_synthetic_feniks_260617_50k.yaml', prior_config_path='configs/prior_diffsky_synthetic_feniks_full_realnvp.yaml', amortized_config_path='configs/amortized_diffsky_synthetic_feniks_full_gpu.yaml', prior_out='outputs/runs/prior_diffsky_synthetic_feniks_full_realnvp', amortized_out='outputs/runs/amortized_diffsky_synthetic_feniks_full', inference_out='outputs/runs/amortized_diffsky_synthetic_feniks_full_test_infer', map_out='outputs/runs/map_diffsky_synthetic_feniks_under_prior', mclmc_out='outputs/runs/mclmc_diffsky_synthetic_feniks_flat', inferred_prior_out='outputs/runs/prior_diffsky_synthetic_feniks_from_inferred', repo_root='.')

Build a cheap, file-aware plan for the FENIKS prior-learning ladder.

Parameters:

• generation_config (dict[str, Any])

• validation_config (dict[str, Any] | None)

• prior_config (dict[str, Any])

• amortized_config (dict[str, Any])

• generation_config_path (str)

• validation_config_path (str)

• prior_config_path (str)

• amortized_config_path (str)

• prior_out (str)

• amortized_out (str)

• inference_out (str)

• map_out (str)

• mclmc_out (str)

• inferred_prior_out (str)

• repo_root (str | Path)

Return type:

PriorWorkflowPlan

Amortized NN+DSPS+NF inference components.

Photo-z and posterior-vs-truth metrics for amortized FENIKS inference.

euclid_dsps.amortized.redshift_metrics.redshift_metrics_from_samples(posterior_samples, truth, *, z_parameter='z_obs', truth_column='redshift_true')

Compute per-object and aggregate redshift posterior metrics.

Parameters:

• posterior_samples (DataFrame)

• truth (DataFrame)

• z_parameter (str)

• truth_column (str)

Return type:

tuple[DataFrame, dict[str, float | int]]

euclid_dsps.amortized.redshift_metrics.summarize_redshift_metrics(frame)

Parameters:

frame (DataFrame)

Return type:

dict[str, float | int]

euclid_dsps.amortized.redshift_metrics.redshift_metrics_by_truth_bin(frame, *, bins=(0.0, 0.2, 0.4, 0.6, 0.8, 1.0, 1.2, 1.5, 2.0, 3.0, 6.0))

Summarize photo-z metrics in fixed bins of true redshift.

Parameters:

• frame (DataFrame)

• bins (tuple[float, ...])

Return type:

DataFrame

euclid_dsps.amortized.redshift_metrics.posterior_truth_metrics(posterior_samples, truth, parameter_pairs)

Summarize posterior median recovery for configured parameter/truth pairs.

Parameters:

• posterior_samples (DataFrame)

• truth (DataFrame)

• parameter_pairs (tuple[tuple[str, str], ...])

Return type:

DataFrame

euclid_dsps.amortized.redshift_metrics.write_redshift_metrics_for_run(*, dataset_path, run_dir, out_dir=None, label='run')

Compute redshift and posterior-vs-truth metrics for one inference run.

Parameters:

• dataset_path (str | Path)

• run_dir (str | Path)

• out_dir (str | Path | None)

• label (str)

Return type:

dict[str, Path]