Validation dataset manifests

Hebog identifies validation data through strict, versioned JSON manifests under config/datasets/. hebog.validation is repository tooling, not installed from the wheel; use it from a source checkout after uv sync --all-groups. Loading a manifest validates metadata only and never downloads or generates an image:

from pathlib import Path

from hebog.validation.datasets import load_dataset_manifest

manifest = load_dataset_manifest(
    Path("config/datasets/phase-0-development.json")
)

Roles and governance

Every dataset has exactly one role: development for routine red-green-refactor work, regression to preserve a reviewed defect or decision, or qualification, frozen before the algorithm work and held out from tuning. Each entry records purpose, provenance, redistribution status, restoring beam, WCS, expected image statistics, the complete synthetic recipe and the recipe's canonical SHA-256. Angles are degrees, pixel coordinates are explicit (x, y), shapes are (y, x) and flux densities are Jy/beam. The schema rejects unknown fields, duplicate identifiers, invalid geometry, stale checksums and inconsistent statistics.

A recipe checksum protects the generation inputs, not a materialised FITS file; the frozen PyBDSF reference products record artifact checksums, tool revisions and configuration in their own manifest under config/baselines/. The 30,000² regression and 100,000² qualification entries are logical recipes: tests generate bounded windows and never allocate the plane.

validation_strata name possibly overlapping sets of source indices (SNR, shape, blend, edge) so that qualification code can prove sample counts before looking at results. Manifest schema 2 adds association_truth_groups: every emitter belongs to one group, individually-resolvable or unresolved-blend, with a frozen centroid and summed brightness that validation recomputes, and association_group_strata keep unresolved members out of individual completeness and flux denominators. New populations must be seed-disjoint from every existing manifest.

Generator versions

Version	Adds	Use
1	Analytic Gaussian sources, uniform RMS and deterministic per-pixel noise	Frozen recipes only
2	An affine, globally addressed RMS multiplier; half-open invalid rectangles materialised as NaN; unequal pixel scales and WCS rotation, written as signed CDELT with an explicit PC matrix, through which the beam ellipse is also transformed	Frozen recipes only
3	An elliptical Gaussian noise-correlation function in pixel coordinates (the compact datasets use the restoring-beam covariance) and noise_realization_seeds, which define one Monte Carlo campaign over a shared truth recipe	Frozen recipes only
4	An independent noise realization for every seed, correlated or not; the checksum covers the whole recipe	Every new population

Earlier versions remain so that their frozen recipes rebuild exactly; their checksums never change. iter_dataset_recipes expands a campaign's seeds.

Independent realizations

Every pixel's noise is a hash of the seed and the pixel address. Versions 1 and 2 combine the row-major address with the seed by XOR before hashing, so two seeds that differ by d give one noise field with pixel a moved to a ^ d: seeds 1000 and 1001 swap neighbouring pixels, and seeds whose XOR is below 16 share every aligned 16-pixel block sum. Such images are not independent samples, so a version 1 or 2 record that lists noise realization seeds is refused when it is validated. Version 3 hashes row and column separately, and version 4 hashes the seed as well, so nearby seeds share no noise value and their block sums are uncorrelated, which unit tests check. The 23 September flux-calibration population was drawn under version 2 with seeds 2 apart; m1-flux-calibration.json is its replacement under version 4.

Deterministic generation

Synthetic noise depends only on the generator version, seed and global pixel address, never on call order, tile shape or worker assignment, so independently generated windows stitch exactly into the one-window image:

from hebog.validation.datasets import generate_synthetic_window

dataset = manifest.datasets[0]
window = generate_synthetic_window(
    dataset.recipe,
    y_start=0,
    y_stop=32,
    x_start=0,
    x_stop=32,
)

generate_synthetic_image is a convenience for bounded test images and refuses a complete allocation above its safety limit; use generate_synthetic_window for large planes. Tests never regenerate frozen expected products implicitly.

hebog.validation.datasets

Versioned validation-dataset manifests and synthetic image generation.

The generator is deliberately stateless at the pixel level. Generating any window produces the same values as slicing a plane generated in one call, so large validation images do not depend on a particular partition layout.

Every pixel's noise is a hash of the seed and the pixel address. Versions 1 and 2 combine the seed with the plane address by XOR before hashing, so two seeds give one noise field with its pixels rearranged; they remain only so their frozen recipes rebuild exactly, and refuse noise realization seeds. Version 3 hashes the row and the column before the seed enters, and version 4 hashes the seed too, so each of their seeds gives an independent realization.

AssociationGroupValidationStratum

Bases: _ManifestModel

One named subset of explicit truth associations.

Source code in src/hebog/validation/datasets.py

class AssociationGroupValidationStratum(_ManifestModel):
    """One named subset of explicit truth associations."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    group_identifiers: tuple[str, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_group_identifiers(self) -> Self:
        """Require canonical unique group identities."""
        if (
            tuple(sorted(set(self.group_identifiers)))
            != self.group_identifiers
        ):
            raise ValueError(
                "validation stratum group identifiers must be unique and "
                "sorted"
            )
        return self

validate_group_identifiers()

Require canonical unique group identities.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_group_identifiers(self) -> Self:
    """Require canonical unique group identities."""
    if (
        tuple(sorted(set(self.group_identifiers)))
        != self.group_identifiers
    ):
        raise ValueError(
            "validation stratum group identifiers must be unique and "
            "sorted"
        )
    return self

AssociationTruthGroup

Bases: _ManifestModel

One explicitly identified observable association in injected truth.

Source code in src/hebog/validation/datasets.py

class AssociationTruthGroup(_ManifestModel):
    """One explicitly identified observable association in injected truth."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    source_indices: tuple[int, ...] = Field(min_length=1)
    resolution_class: Literal["individually-resolvable", "unresolved-blend"]
    reference_position_xy: tuple[float, float]
    reference_integrated_brightness_jy_pixels_per_beam: float = Field(
        gt=0,
        allow_inf_nan=False,
    )

    @model_validator(mode="after")
    def validate_group(self) -> Self:
        """Require canonical membership and resolution cardinality."""
        if tuple(sorted(set(self.source_indices))) != self.source_indices:
            raise ValueError(
                "truth-group source indices must be unique and sorted"
            )
        if any(index < 0 for index in self.source_indices):
            raise ValueError("truth-group source indices must be non-negative")
        if not all(np.isfinite(value) for value in self.reference_position_xy):
            raise ValueError("truth-group reference position must be finite")
        if (
            self.resolution_class == "individually-resolvable"
            and len(self.source_indices) != 1
        ):
            raise ValueError(
                "individually resolvable truth group must contain one source"
            )
        if (
            self.resolution_class == "unresolved-blend"
            and len(self.source_indices) < _MINIMUM_UNRESOLVED_GROUP_SOURCES
        ):
            raise ValueError(
                "unresolved blend must contain at least two sources"
            )
        return self

validate_group()

Require canonical membership and resolution cardinality.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_group(self) -> Self:
    """Require canonical membership and resolution cardinality."""
    if tuple(sorted(set(self.source_indices))) != self.source_indices:
        raise ValueError(
            "truth-group source indices must be unique and sorted"
        )
    if any(index < 0 for index in self.source_indices):
        raise ValueError("truth-group source indices must be non-negative")
    if not all(np.isfinite(value) for value in self.reference_position_xy):
        raise ValueError("truth-group reference position must be finite")
    if (
        self.resolution_class == "individually-resolvable"
        and len(self.source_indices) != 1
    ):
        raise ValueError(
            "individually resolvable truth group must contain one source"
        )
    if (
        self.resolution_class == "unresolved-blend"
        and len(self.source_indices) < _MINIMUM_UNRESOLVED_GROUP_SOURCES
    ):
        raise ValueError(
            "unresolved blend must contain at least two sources"
        )
    return self

BeamMetadata

Bases: _ManifestModel

Synthetic restoring-beam metadata in image-pixel units.

Source code in src/hebog/validation/datasets.py

class BeamMetadata(_ManifestModel):
    """Synthetic restoring-beam metadata in image-pixel units."""

    major_fwhm_pixels: float = Field(gt=0, allow_inf_nan=False)
    minor_fwhm_pixels: float = Field(gt=0, allow_inf_nan=False)
    position_angle_degrees: float = Field(allow_inf_nan=False)

    @model_validator(mode="after")
    def validate_axis_order(self) -> Self:
        """Require the named major axis to be at least the minor axis."""
        if self.minor_fwhm_pixels > self.major_fwhm_pixels:
            raise ValueError("minor beam axis cannot exceed major beam axis")
        return self

validate_axis_order()

Require the named major axis to be at least the minor axis.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_axis_order(self) -> Self:
    """Require the named major axis to be at least the minor axis."""
    if self.minor_fwhm_pixels > self.major_fwhm_pixels:
        raise ValueError("minor beam axis cannot exceed major beam axis")
    return self

DatasetManifest

Bases: _ManifestModel

Versioned collection of uniquely identified validation datasets.

Source code in src/hebog/validation/datasets.py

class DatasetManifest(_ManifestModel):
    """Versioned collection of uniquely identified validation datasets."""

    schema_version: Literal[1, 2, 3]
    manifest_id: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    datasets: tuple[DatasetRecord, ...]

    @model_validator(mode="after")
    def validate_unique_identifiers(self) -> Self:
        """Prevent ambiguous test-data lookup."""
        identifiers = [dataset.identifier for dataset in self.datasets]
        if len(set(identifiers)) != len(identifiers):
            raise ValueError("dataset identifiers must be unique")
        if self.schema_version == 1 and any(
            dataset.association_truth_groups for dataset in self.datasets
        ):
            raise ValueError(
                "association truth groups require manifest schema 2"
            )
        if self.schema_version < _MULTISCALE_MANIFEST_SCHEMA_VERSION and any(
            dataset.multiscale_truth_groups for dataset in self.datasets
        ):
            raise ValueError(
                "multiscale truth groups require manifest schema 3"
            )
        if self.schema_version == _MULTISCALE_MANIFEST_SCHEMA_VERSION and any(
            not dataset.multiscale_truth_groups for dataset in self.datasets
        ):
            raise ValueError(
                "manifest schema 3 requires multiscale truth groups"
            )
        return self

validate_unique_identifiers()

Prevent ambiguous test-data lookup.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_unique_identifiers(self) -> Self:
    """Prevent ambiguous test-data lookup."""
    identifiers = [dataset.identifier for dataset in self.datasets]
    if len(set(identifiers)) != len(identifiers):
        raise ValueError("dataset identifiers must be unique")
    if self.schema_version == 1 and any(
        dataset.association_truth_groups for dataset in self.datasets
    ):
        raise ValueError(
            "association truth groups require manifest schema 2"
        )
    if self.schema_version < _MULTISCALE_MANIFEST_SCHEMA_VERSION and any(
        dataset.multiscale_truth_groups for dataset in self.datasets
    ):
        raise ValueError(
            "multiscale truth groups require manifest schema 3"
        )
    if self.schema_version == _MULTISCALE_MANIFEST_SCHEMA_VERSION and any(
        not dataset.multiscale_truth_groups for dataset in self.datasets
    ):
        raise ValueError(
            "manifest schema 3 requires multiscale truth groups"
        )
    return self

DatasetRecord

Bases: _ManifestModel

One governed validation dataset and its generation provenance.

Source code in src/hebog/validation/datasets.py

class DatasetRecord(_ManifestModel):
    """One governed validation dataset and its generation provenance."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    role: DatasetRole
    purpose: str = Field(min_length=1)
    provenance: str = Field(min_length=1)
    redistribution: RedistributionStatus
    beam: BeamMetadata
    wcs: WcsMetadata
    expected_statistics: ExpectedImageStatistics
    recipe: SyntheticRecipe
    recipe_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    noise_realization_seeds: tuple[int, ...] = ()
    validation_strata: tuple[SourceValidationStratum, ...] = ()
    classification_strata: tuple[SourceValidationStratum, ...] = ()
    association_truth_groups: tuple[AssociationTruthGroup, ...] = ()
    association_group_strata: tuple[
        AssociationGroupValidationStratum, ...
    ] = ()
    multiscale_truth_groups: tuple[MultiscaleTruthGroup, ...] = Field(
        default=(),
        exclude_if=lambda value: not value,
    )
    multiscale_group_strata: tuple[MultiscaleGroupValidationStratum, ...] = (
        Field(
            default=(),
            exclude_if=lambda value: not value,
        )
    )

    @model_validator(mode="after")
    def validate_recipe_checksum(self) -> Self:
        """Detect stale provenance whenever generation inputs change."""
        expected = recipe_sha256(self.recipe)
        if self.recipe_sha256 != expected:
            raise ValueError(
                "recipe_sha256 does not match the canonical recipe"
            )
        if (
            self.recipe.generator_version == 1
            and self.wcs.rotation_degrees_counterclockwise != 0.0
        ):
            raise ValueError("generator version 1 cannot use rotated WCS")
        if (
            self.recipe.generator_version < _CORRELATED_NOISE_GENERATOR_VERSION
            and self.noise_realization_seeds
        ):
            raise ValueError(
                "generator versions 1 and 2 cannot record noise realization "
                "seeds: their seeds rearrange one noise field rather than "
                "drawing independent ones; use generator version 4"
            )
        if len(set(self.noise_realization_seeds)) != len(
            self.noise_realization_seeds
        ):
            raise ValueError("noise realization seeds must be unique")
        if self.recipe.seed in self.noise_realization_seeds:
            raise ValueError(
                "noise realization seeds must not repeat the base seed"
            )
        if any(
            seed < 0 or seed > _UINT64_LIMIT
            for seed in self.noise_realization_seeds
        ):
            raise ValueError("noise realization seeds must fit uint64")
        self._validate_source_strata()
        self._validate_association_truth()
        self._validate_multiscale_truth()
        if (
            self.expected_statistics.background_jy_per_beam
            != self.recipe.background
            or self.expected_statistics.noise_rms_jy_per_beam
            != self.recipe.noise_rms
        ):
            raise ValueError(
                "expected statistics must match the synthetic recipe"
            )
        invalid_pixels = sum(
            (rectangle.y_stop - rectangle.y_start)
            * (rectangle.x_stop - rectangle.x_start)
            for rectangle in self.recipe.invalid_rectangles
        )
        height, width = self.recipe.shape_yx
        expected_finite_fraction = 1.0 - invalid_pixels / (height * width)
        if not np.isclose(
            self.expected_statistics.finite_fraction,
            expected_finite_fraction,
            rtol=0.0,
            atol=np.finfo(np.float64).eps,
        ):
            raise ValueError(
                "expected finite fraction must match invalid rectangles"
            )
        return self

    def _validate_source_strata(self) -> None:
        """Require unique strata that refer only to declared source truth."""
        all_strata: tuple[SourceValidationStratum, ...] = ()
        for strata, label in (
            (self.validation_strata, "validation"),
            (self.classification_strata, "classification"),
        ):
            identifiers = tuple(stratum.identifier for stratum in strata)
            if len(set(identifiers)) != len(identifiers):
                raise ValueError(f"{label} stratum identifiers must be unique")
            all_strata += strata
        classification_indices = tuple(
            source_index
            for stratum in self.classification_strata
            for source_index in stratum.source_indices
        )
        if len(set(classification_indices)) != len(classification_indices):
            raise ValueError("classification strata must not overlap")
        if any(
            source_index >= len(self.recipe.sources)
            for stratum in all_strata
            for source_index in stratum.source_indices
        ):
            raise ValueError("source stratum index must identify recipe truth")

    def canonical_source_strata(self) -> tuple[SourceValidationStratum, ...]:
        """Return one authoritative definition for every source stratum.

        Classification strata are governed disjoint populations. They replace
        an older validation stratum with the same identifier instead of being
        unioned with it. Distinct validation strata such as SNR and edge
        populations remain available as independent diagnostic dimensions.
        """
        classification_identifiers = {
            stratum.identifier for stratum in self.classification_strata
        }
        canonical = (
            *(
                stratum
                for stratum in self.validation_strata
                if stratum.identifier not in classification_identifiers
            ),
            *self.classification_strata,
        )
        return tuple(sorted(canonical, key=lambda item: item.identifier))

    def _validate_association_truth(self) -> None:
        """Bind explicit observable groups and strata to analytic sources."""
        if not self.association_truth_groups:
            if self.association_group_strata:
                raise ValueError(
                    "association group strata require association truth groups"
                )
            return
        identifiers = tuple(
            group.identifier for group in self.association_truth_groups
        )
        if len(set(identifiers)) != len(identifiers):
            raise ValueError(
                "association truth-group identifiers must be unique"
            )
        memberships = tuple(
            index
            for group in self.association_truth_groups
            for index in group.source_indices
        )
        if sorted(memberships) != list(range(len(self.recipe.sources))):
            raise ValueError(
                "association truth groups must partition recipe sources"
            )
        for group in self.association_truth_groups:
            self._validate_association_group_quantities(group)
        stratum_identifiers = tuple(
            stratum.identifier for stratum in self.association_group_strata
        )
        if len(set(stratum_identifiers)) != len(stratum_identifiers):
            raise ValueError(
                "association group stratum identifiers must be unique"
            )
        known_identifiers = set(identifiers)
        if any(
            identifier not in known_identifiers
            for stratum in self.association_group_strata
            for identifier in stratum.group_identifiers
        ):
            raise ValueError(
                "association group stratum must identify governed truth"
            )

    def _validate_association_group_quantities(
        self,
        group: AssociationTruthGroup,
    ) -> None:
        """Require stored group position and brightness to match emitters."""
        self._validate_group_quantities(
            source_indices=group.source_indices,
            reference_position_xy=group.reference_position_xy,
            reference_integrated_brightness=(
                group.reference_integrated_brightness_jy_pixels_per_beam
            ),
            group_kind="truth-group",
        )

    def _validate_multiscale_truth(self) -> None:
        """Bind morphology and scale strata to complete analytic truth."""
        if not self.multiscale_truth_groups:
            if self.multiscale_group_strata:
                raise ValueError(
                    "multiscale strata require multiscale truth groups"
                )
            return
        identifiers = tuple(
            group.identifier for group in self.multiscale_truth_groups
        )
        if len(set(identifiers)) != len(identifiers):
            raise ValueError("multiscale truth identifiers must be unique")
        memberships = tuple(
            source_index
            for group in self.multiscale_truth_groups
            for source_index in group.source_indices
        )
        if sorted(memberships) != list(range(len(self.recipe.sources))):
            raise ValueError(
                "multiscale truth groups must partition recipe sources"
            )
        for group in self.multiscale_truth_groups:
            self._validate_group_quantities(
                source_indices=group.source_indices,
                reference_position_xy=group.reference_position_xy,
                reference_integrated_brightness=(
                    group.reference_integrated_brightness_jy_pixels_per_beam
                ),
                group_kind="multiscale truth-group",
            )
        stratum_identifiers = tuple(
            stratum.identifier for stratum in self.multiscale_group_strata
        )
        if len(set(stratum_identifiers)) != len(stratum_identifiers):
            raise ValueError("multiscale stratum identifiers must be unique")
        known_identifiers = set(identifiers)
        referenced_identifiers = {
            identifier
            for stratum in self.multiscale_group_strata
            for identifier in stratum.group_identifiers
        }
        if not referenced_identifiers <= known_identifiers:
            raise ValueError("multiscale stratum must identify governed truth")
        if referenced_identifiers != known_identifiers:
            raise ValueError("every multiscale truth group requires a stratum")

    def _validate_group_quantities(
        self,
        *,
        source_indices: tuple[int, ...],
        reference_position_xy: tuple[float, float],
        reference_integrated_brightness: float,
        group_kind: str,
    ) -> None:
        """Match stored position and brightness to analytic emitters."""
        sources = tuple(self.recipe.sources[index] for index in source_indices)
        brightnesses = np.asarray(
            [
                source.peak_flux_jy_per_beam
                * 2.0
                * np.pi
                * source.major_sigma_pixels
                * source.minor_sigma_pixels
                for source in sources
            ],
            dtype=np.float64,
        )
        total = float(np.sum(brightnesses))
        reference_position = (
            float(
                np.dot(
                    brightnesses,
                    [source.x_pixel for source in sources],
                )
                / total
            ),
            float(
                np.dot(
                    brightnesses,
                    [source.y_pixel for source in sources],
                )
                / total
            ),
        )
        if not np.allclose(
            reference_position_xy,
            reference_position,
            rtol=1e-12,
            atol=1e-12,
        ):
            raise ValueError(
                f"{group_kind} reference position does not match source truth"
            )
        if not np.isclose(
            reference_integrated_brightness,
            total,
            rtol=1e-12,
            atol=0.0,
        ):
            raise ValueError(
                f"{group_kind} reference integrated brightness does not "
                "match source truth"
            )

canonical_source_strata()

Return one authoritative definition for every source stratum.

Classification strata are governed disjoint populations. They replace an older validation stratum with the same identifier instead of being unioned with it. Distinct validation strata such as SNR and edge populations remain available as independent diagnostic dimensions.

Source code in src/hebog/validation/datasets.py

def canonical_source_strata(self) -> tuple[SourceValidationStratum, ...]:
    """Return one authoritative definition for every source stratum.

    Classification strata are governed disjoint populations. They replace
    an older validation stratum with the same identifier instead of being
    unioned with it. Distinct validation strata such as SNR and edge
    populations remain available as independent diagnostic dimensions.
    """
    classification_identifiers = {
        stratum.identifier for stratum in self.classification_strata
    }
    canonical = (
        *(
            stratum
            for stratum in self.validation_strata
            if stratum.identifier not in classification_identifiers
        ),
        *self.classification_strata,
    )
    return tuple(sorted(canonical, key=lambda item: item.identifier))

validate_recipe_checksum()

Detect stale provenance whenever generation inputs change.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_recipe_checksum(self) -> Self:
    """Detect stale provenance whenever generation inputs change."""
    expected = recipe_sha256(self.recipe)
    if self.recipe_sha256 != expected:
        raise ValueError(
            "recipe_sha256 does not match the canonical recipe"
        )
    if (
        self.recipe.generator_version == 1
        and self.wcs.rotation_degrees_counterclockwise != 0.0
    ):
        raise ValueError("generator version 1 cannot use rotated WCS")
    if (
        self.recipe.generator_version < _CORRELATED_NOISE_GENERATOR_VERSION
        and self.noise_realization_seeds
    ):
        raise ValueError(
            "generator versions 1 and 2 cannot record noise realization "
            "seeds: their seeds rearrange one noise field rather than "
            "drawing independent ones; use generator version 4"
        )
    if len(set(self.noise_realization_seeds)) != len(
        self.noise_realization_seeds
    ):
        raise ValueError("noise realization seeds must be unique")
    if self.recipe.seed in self.noise_realization_seeds:
        raise ValueError(
            "noise realization seeds must not repeat the base seed"
        )
    if any(
        seed < 0 or seed > _UINT64_LIMIT
        for seed in self.noise_realization_seeds
    ):
        raise ValueError("noise realization seeds must fit uint64")
    self._validate_source_strata()
    self._validate_association_truth()
    self._validate_multiscale_truth()
    if (
        self.expected_statistics.background_jy_per_beam
        != self.recipe.background
        or self.expected_statistics.noise_rms_jy_per_beam
        != self.recipe.noise_rms
    ):
        raise ValueError(
            "expected statistics must match the synthetic recipe"
        )
    invalid_pixels = sum(
        (rectangle.y_stop - rectangle.y_start)
        * (rectangle.x_stop - rectangle.x_start)
        for rectangle in self.recipe.invalid_rectangles
    )
    height, width = self.recipe.shape_yx
    expected_finite_fraction = 1.0 - invalid_pixels / (height * width)
    if not np.isclose(
        self.expected_statistics.finite_fraction,
        expected_finite_fraction,
        rtol=0.0,
        atol=np.finfo(np.float64).eps,
    ):
        raise ValueError(
            "expected finite fraction must match invalid rectangles"
        )
    return self

DatasetRole

Bases: str, Enum

The one test lane allowed to consume a dataset.

Source code in src/hebog/validation/datasets.py

class DatasetRole(str, Enum):
    """The one test lane allowed to consume a dataset."""

    DEVELOPMENT = "development"
    REGRESSION = "regression"
    QUALIFICATION = "qualification"

ExpectedImageStatistics

Bases: _ManifestModel

Generator parameters used as expected image statistics.

Source code in src/hebog/validation/datasets.py

class ExpectedImageStatistics(_ManifestModel):
    """Generator parameters used as expected image statistics."""

    background_jy_per_beam: float = Field(allow_inf_nan=False)
    noise_rms_jy_per_beam: float = Field(ge=0, allow_inf_nan=False)
    finite_fraction: float = Field(ge=0, le=1, allow_inf_nan=False)

MultiscaleGroupValidationStratum

Bases: _ManifestModel

One named subset of governed multiscale truth objects.

Source code in src/hebog/validation/datasets.py

class MultiscaleGroupValidationStratum(_ManifestModel):
    """One named subset of governed multiscale truth objects."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    group_identifiers: tuple[str, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_group_identifiers(self) -> Self:
        """Require canonical unique multiscale truth identities."""
        if self.group_identifiers != tuple(
            sorted(set(self.group_identifiers))
        ):
            raise ValueError(
                "multiscale stratum group identifiers must be unique and "
                "sorted"
            )
        return self

validate_group_identifiers()

Require canonical unique multiscale truth identities.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_group_identifiers(self) -> Self:
    """Require canonical unique multiscale truth identities."""
    if self.group_identifiers != tuple(
        sorted(set(self.group_identifiers))
    ):
        raise ValueError(
            "multiscale stratum group identifiers must be unique and "
            "sorted"
        )
    return self

MultiscaleTruthGroup

Bases: _ManifestModel

One governed extended object built from analytic Gaussian emitters.

Source code in src/hebog/validation/datasets.py

class MultiscaleTruthGroup(_ManifestModel):
    """One governed extended object built from analytic Gaussian emitters."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    source_indices: tuple[int, ...] = Field(min_length=1)
    morphology: Literal[
        "artifact",
        "curved-filament",
        "diffuse",
        "filament",
        "mixed-compact-extended",
        "shell",
    ]
    catalogue_role: Literal["astronomical-source", "artifact"]
    reference_position_xy: tuple[float, float]
    reference_integrated_brightness_jy_pixels_per_beam: float = Field(
        gt=0,
        allow_inf_nan=False,
    )
    major_extent_beams: float = Field(gt=0, allow_inf_nan=False)
    minor_extent_beams: float = Field(gt=0, allow_inf_nan=False)
    governed_scale_orders: tuple[int, ...] = Field(min_length=1)
    crosses_tile_boundary: bool
    crosses_tile_corner: bool = Field(
        default=False,
        exclude_if=lambda value: not value,
    )
    compact_deblend_disposition: Literal[
        "compact-eligible",
        "deferred-extended",
    ] = Field(
        default="compact-eligible",
        exclude_if=lambda value: value == "compact-eligible",
    )
    touches_image_edge: bool

    @model_validator(mode="after")
    def validate_group(self) -> Self:
        """Require canonical membership, scale order, and truth role."""
        if tuple(sorted(set(self.source_indices))) != self.source_indices:
            raise ValueError(
                "multiscale truth source indices must be unique and sorted"
            )
        if any(index < 0 for index in self.source_indices):
            raise ValueError(
                "multiscale truth source indices must be non-negative"
            )
        if not all(np.isfinite(value) for value in self.reference_position_xy):
            raise ValueError(
                "multiscale truth reference position must be finite"
            )
        if self.minor_extent_beams > self.major_extent_beams:
            raise ValueError("multiscale minor extent cannot exceed major")
        if self.governed_scale_orders != tuple(
            sorted(set(self.governed_scale_orders))
        ) or any(
            order not in {1, 2, 3} for order in self.governed_scale_orders
        ):
            raise ValueError("multiscale truth scale orders must be canonical")
        expected_role = (
            "artifact"
            if self.morphology == "artifact"
            else "astronomical-source"
        )
        if self.catalogue_role != expected_role:
            raise ValueError(
                "multiscale morphology and catalogue role must agree"
            )
        if self.crosses_tile_corner and not self.crosses_tile_boundary:
            raise ValueError(
                "tile-corner truth must also cross a tile boundary"
            )
        return self

validate_group()

Require canonical membership, scale order, and truth role.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_group(self) -> Self:
    """Require canonical membership, scale order, and truth role."""
    if tuple(sorted(set(self.source_indices))) != self.source_indices:
        raise ValueError(
            "multiscale truth source indices must be unique and sorted"
        )
    if any(index < 0 for index in self.source_indices):
        raise ValueError(
            "multiscale truth source indices must be non-negative"
        )
    if not all(np.isfinite(value) for value in self.reference_position_xy):
        raise ValueError(
            "multiscale truth reference position must be finite"
        )
    if self.minor_extent_beams > self.major_extent_beams:
        raise ValueError("multiscale minor extent cannot exceed major")
    if self.governed_scale_orders != tuple(
        sorted(set(self.governed_scale_orders))
    ) or any(
        order not in {1, 2, 3} for order in self.governed_scale_orders
    ):
        raise ValueError("multiscale truth scale orders must be canonical")
    expected_role = (
        "artifact"
        if self.morphology == "artifact"
        else "astronomical-source"
    )
    if self.catalogue_role != expected_role:
        raise ValueError(
            "multiscale morphology and catalogue role must agree"
        )
    if self.crosses_tile_corner and not self.crosses_tile_boundary:
        raise ValueError(
            "tile-corner truth must also cross a tile boundary"
        )
    return self

RedistributionStatus

Bases: str, Enum

How a validation dataset may be obtained and shared.

Source code in src/hebog/validation/datasets.py

class RedistributionStatus(str, Enum):
    """How a validation dataset may be obtained and shared."""

    GENERATED_LOCALLY = "generated-locally"
    REDISTRIBUTABLE = "redistributable"
    RESTRICTED = "restricted"

SourceValidationStratum

Bases: _ManifestModel

One named, possibly overlapping subset of analytic source truth.

Source code in src/hebog/validation/datasets.py

class SourceValidationStratum(_ManifestModel):
    """One named, possibly overlapping subset of analytic source truth."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    source_indices: tuple[int, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_source_indices(self) -> Self:
        """Require canonical non-negative indices without double counting."""
        if any(index < 0 for index in self.source_indices):
            raise ValueError(
                "validation stratum source indices must be non-negative"
            )
        if tuple(sorted(set(self.source_indices))) != self.source_indices:
            raise ValueError(
                "validation stratum source indices must be unique and sorted"
            )
        return self

validate_source_indices()

Require canonical non-negative indices without double counting.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_source_indices(self) -> Self:
    """Require canonical non-negative indices without double counting."""
    if any(index < 0 for index in self.source_indices):
        raise ValueError(
            "validation stratum source indices must be non-negative"
        )
    if tuple(sorted(set(self.source_indices))) != self.source_indices:
        raise ValueError(
            "validation stratum source indices must be unique and sorted"
        )
    return self

SyntheticInvalidRectangle

Bases: _ManifestModel

One half-open rectangular region materialised as invalid pixels.

Source code in src/hebog/validation/datasets.py

class SyntheticInvalidRectangle(_ManifestModel):
    """One half-open rectangular region materialised as invalid pixels."""

    y_start: int = Field(ge=0)
    y_stop: int = Field(ge=1)
    x_start: int = Field(ge=0)
    x_stop: int = Field(ge=1)

    @model_validator(mode="after")
    def validate_order(self) -> Self:
        """Require a non-empty half-open rectangle."""
        if self.y_start >= self.y_stop or self.x_start >= self.x_stop:
            raise ValueError("invalid rectangle bounds must be increasing")
        return self

validate_order()

Require a non-empty half-open rectangle.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_order(self) -> Self:
    """Require a non-empty half-open rectangle."""
    if self.y_start >= self.y_stop or self.x_start >= self.x_stop:
        raise ValueError("invalid rectangle bounds must be increasing")
    return self

SyntheticNoiseCorrelation

Bases: _ManifestModel

Gaussian noise-correlation function in image-pixel coordinates.

Source code in src/hebog/validation/datasets.py

class SyntheticNoiseCorrelation(_ManifestModel):
    """Gaussian noise-correlation function in image-pixel coordinates."""

    major_fwhm_pixels: float = Field(gt=0, allow_inf_nan=False)
    minor_fwhm_pixels: float = Field(gt=0, allow_inf_nan=False)
    position_angle_degrees: float = Field(allow_inf_nan=False)
    truncation_sigma: float = Field(default=4.0, ge=3.0, le=8.0)

    @model_validator(mode="after")
    def validate_axis_order(self) -> Self:
        """Require the correlation major width to contain the minor width."""
        if self.minor_fwhm_pixels > self.major_fwhm_pixels:
            raise ValueError(
                "noise-correlation minor axis cannot exceed major axis"
            )
        return self

validate_axis_order()

Require the correlation major width to contain the minor width.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_axis_order(self) -> Self:
    """Require the correlation major width to contain the minor width."""
    if self.minor_fwhm_pixels > self.major_fwhm_pixels:
        raise ValueError(
            "noise-correlation minor axis cannot exceed major axis"
        )
    return self

SyntheticRecipe

Bases: _ManifestModel

Complete inputs to one version of the synthetic image generator.

Source code in src/hebog/validation/datasets.py

class SyntheticRecipe(_ManifestModel):
    """Complete inputs to one version of the synthetic image generator."""

    generator: Literal["hebog.synthetic.gaussian-noise"]
    generator_version: Literal[1, 2, 3, 4]
    seed: int = Field(ge=0, le=_UINT64_LIMIT)
    shape_yx: tuple[int, int]
    background: float = Field(allow_inf_nan=False)
    noise_rms: float = Field(ge=0, allow_inf_nan=False)
    sources: tuple[SyntheticSource, ...] = ()
    noise_rms_fractional_gradient_xy: tuple[float, float] = (0.0, 0.0)
    invalid_rectangles: tuple[SyntheticInvalidRectangle, ...] = ()
    noise_correlation: SyntheticNoiseCorrelation | None = None

    def _validate_sources(self, *, height: int, width: int) -> None:
        """Require every analytic source centre to be inside the plane."""
        for source in self.sources:
            if source.x_pixel >= width or source.y_pixel >= height:
                raise ValueError("source centre must be inside shape_yx")

    def _validate_noise_policy(self) -> None:
        """Require finite, positive, version-compatible RMS variation."""
        if not all(
            np.isfinite(value)
            for value in self.noise_rms_fractional_gradient_xy
        ):
            raise ValueError("noise RMS gradient must be finite")
        if self.generator_version == 1 and (
            self.noise_rms_fractional_gradient_xy != (0.0, 0.0)
            or self.invalid_rectangles
        ):
            raise ValueError("generator version 1 cannot use version 2 fields")
        if (
            self.generator_version < _CORRELATED_NOISE_GENERATOR_VERSION
            and self.noise_correlation is not None
        ):
            raise ValueError(
                "noise correlation requires generator version 3 or later"
            )
        if (
            self.generator_version == _CORRELATED_NOISE_GENERATOR_VERSION
            and self.noise_correlation is None
        ):
            raise ValueError("generator version 3 requires noise correlation")
        gradient_x, gradient_y = self.noise_rms_fractional_gradient_xy
        if 1.0 - 0.5 * (abs(gradient_x) + abs(gradient_y)) <= 0:
            raise ValueError("noise RMS gradient must remain positive")

    def _validate_invalid_rectangles(
        self,
        *,
        height: int,
        width: int,
    ) -> None:
        """Require non-overlapping invalid regions inside the plane."""
        for rectangle in self.invalid_rectangles:
            if rectangle.y_stop > height or rectangle.x_stop > width:
                raise ValueError("invalid rectangle must be inside shape_yx")
        for index, left in enumerate(self.invalid_rectangles):
            for right in self.invalid_rectangles[index + 1 :]:
                overlaps = (
                    left.y_start < right.y_stop
                    and right.y_start < left.y_stop
                    and left.x_start < right.x_stop
                    and right.x_start < left.x_stop
                )
                if overlaps:
                    raise ValueError("invalid rectangles must not overlap")

    @model_validator(mode="after")
    def validate_geometry(self) -> Self:
        """Keep dimensions positive and every source centre in the plane."""
        height, width = self.shape_yx
        if height <= 0 or width <= 0:
            raise ValueError("shape_yx dimensions must be positive")
        self._validate_sources(height=height, width=width)
        self._validate_noise_policy()
        self._validate_invalid_rectangles(height=height, width=width)
        return self

validate_geometry()

Keep dimensions positive and every source centre in the plane.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_geometry(self) -> Self:
    """Keep dimensions positive and every source centre in the plane."""
    height, width = self.shape_yx
    if height <= 0 or width <= 0:
        raise ValueError("shape_yx dimensions must be positive")
    self._validate_sources(height=height, width=width)
    self._validate_noise_policy()
    self._validate_invalid_rectangles(height=height, width=width)
    return self

SyntheticSource

Bases: _ManifestModel

Analytic elliptical Gaussian source in global pixel coordinates.

Source code in src/hebog/validation/datasets.py

class SyntheticSource(_ManifestModel):
    """Analytic elliptical Gaussian source in global pixel coordinates."""

    x_pixel: float = Field(ge=0, allow_inf_nan=False)
    y_pixel: float = Field(ge=0, allow_inf_nan=False)
    peak_flux_jy_per_beam: float = Field(gt=0, allow_inf_nan=False)
    major_sigma_pixels: float = Field(gt=0, allow_inf_nan=False)
    minor_sigma_pixels: float = Field(gt=0, allow_inf_nan=False)
    rotation_degrees_counterclockwise_from_x: float = Field(
        default=0.0,
        allow_inf_nan=False,
    )

    @model_validator(mode="after")
    def validate_axis_order(self) -> Self:
        """Require the analytic major axis to be the larger axis."""
        if self.minor_sigma_pixels > self.major_sigma_pixels:
            raise ValueError(
                "minor source axis cannot exceed major source axis"
            )
        return self

validate_axis_order()

Require the analytic major axis to be the larger axis.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_axis_order(self) -> Self:
    """Require the analytic major axis to be the larger axis."""
    if self.minor_sigma_pixels > self.major_sigma_pixels:
        raise ValueError(
            "minor source axis cannot exceed major source axis"
        )
    return self

WcsMetadata

Bases: _ManifestModel

Minimal celestial WCS provenance in canonical manifest units.

Source code in src/hebog/validation/datasets.py

class WcsMetadata(_ManifestModel):
    """Minimal celestial WCS provenance in canonical manifest units."""

    frame: Literal["icrs"] = "icrs"
    reference_pixel_xy: tuple[float, float]
    reference_sky_degrees: tuple[float, float]
    pixel_scale_degrees_xy: tuple[float, float]
    rotation_degrees_counterclockwise: float = Field(
        default=0.0,
        allow_inf_nan=False,
    )

    @model_validator(mode="after")
    def validate_wcs(self) -> Self:
        """Reject non-finite or degenerate WCS metadata."""
        values = (
            *self.reference_pixel_xy,
            *self.reference_sky_degrees,
            *self.pixel_scale_degrees_xy,
            self.rotation_degrees_counterclockwise,
        )
        if not all(np.isfinite(value) for value in values):
            raise ValueError("WCS values must be finite")
        _, declination_degrees = self.reference_sky_degrees
        if not (
            _MINIMUM_DECLINATION_DEGREES
            <= declination_degrees
            <= _MAXIMUM_DECLINATION_DEGREES
        ):
            raise ValueError("reference declination must be within [-90, 90]")
        if any(scale == 0 for scale in self.pixel_scale_degrees_xy):
            raise ValueError("WCS pixel scales must be non-zero")
        return self

validate_wcs()

Reject non-finite or degenerate WCS metadata.

Source code in src/hebog/validation/datasets.py

@model_validator(mode="after")
def validate_wcs(self) -> Self:
    """Reject non-finite or degenerate WCS metadata."""
    values = (
        *self.reference_pixel_xy,
        *self.reference_sky_degrees,
        *self.pixel_scale_degrees_xy,
        self.rotation_degrees_counterclockwise,
    )
    if not all(np.isfinite(value) for value in values):
        raise ValueError("WCS values must be finite")
    _, declination_degrees = self.reference_sky_degrees
    if not (
        _MINIMUM_DECLINATION_DEGREES
        <= declination_degrees
        <= _MAXIMUM_DECLINATION_DEGREES
    ):
        raise ValueError("reference declination must be within [-90, 90]")
    if any(scale == 0 for scale in self.pixel_scale_degrees_xy):
        raise ValueError("WCS pixel scales must be non-zero")
    return self

generate_synthetic_emission(recipe)

Return the injected emission alone: no background, no noise.

Invalid rectangles are NaN, as in the image. The plane is bounded by the same in-memory limit as :func:generate_synthetic_image.

Source code in src/hebog/validation/datasets.py

def generate_synthetic_emission(
    recipe: SyntheticRecipe,
) -> npt.NDArray[np.float64]:
    """Return the injected emission alone: no background, no noise.

    Invalid rectangles are NaN, as in the image. The plane is bounded by the
    same in-memory limit as :func:`generate_synthetic_image`.
    """
    height, width = recipe.shape_yx
    if height * width > _DEFAULT_MAXIMUM_IN_MEMORY_PIXELS:
        raise ValueError(
            "complete synthetic emission exceeds the in-memory limit"
        )
    emission = recipe.model_copy(update={"background": 0.0, "noise_rms": 0.0})
    return generate_synthetic_window(
        emission, y_start=0, y_stop=height, x_start=0, x_stop=width
    )

generate_synthetic_image(recipe, *, maximum_pixels=_DEFAULT_MAXIMUM_IN_MEMORY_PIXELS)

Generate a complete in-memory image for a bounded recipe.

Source code in src/hebog/validation/datasets.py

def generate_synthetic_image(
    recipe: SyntheticRecipe,
    *,
    maximum_pixels: int = _DEFAULT_MAXIMUM_IN_MEMORY_PIXELS,
) -> npt.NDArray[np.float64]:
    """Generate a complete in-memory image for a bounded recipe."""
    height, width = recipe.shape_yx
    if maximum_pixels <= 0:
        raise ValueError("maximum_pixels must be positive")
    if height * width > maximum_pixels:
        raise ValueError(
            "complete synthetic image exceeds maximum_pixels; use "
            "generate_synthetic_window for bounded generation"
        )
    return generate_synthetic_window(
        recipe,
        y_start=0,
        y_stop=height,
        x_start=0,
        x_stop=width,
    )

generate_synthetic_noise_rms(recipe)

Return the injected noise RMS at every pixel, gradient included.

Source code in src/hebog/validation/datasets.py

def generate_synthetic_noise_rms(
    recipe: SyntheticRecipe,
) -> npt.NDArray[np.float64]:
    """Return the injected noise RMS at every pixel, gradient included."""
    height, width = recipe.shape_yx
    if height * width > _DEFAULT_MAXIMUM_IN_MEMORY_PIXELS:
        raise ValueError(
            "complete synthetic noise RMS exceeds the in-memory limit"
        )
    return np.asarray(
        recipe.noise_rms
        * _noise_rms_scale(
            recipe, y_start=0, y_stop=height, x_start=0, x_stop=width
        ),
        dtype=np.float64,
    )

generate_synthetic_window(recipe, *, y_start, y_stop, x_start, x_stop)

Generate one half-open window without materialising the full plane.

Source code in src/hebog/validation/datasets.py

def generate_synthetic_window(
    recipe: SyntheticRecipe,
    *,
    y_start: int,
    y_stop: int,
    x_start: int,
    x_stop: int,
) -> npt.NDArray[np.float64]:
    """Generate one half-open window without materialising the full plane."""
    _validate_window(
        recipe,
        y_start=y_start,
        y_stop=y_stop,
        x_start=x_start,
        x_stop=x_stop,
    )
    standard_noise = _normal_noise(
        recipe,
        y_start=y_start,
        y_stop=y_stop,
        x_start=x_start,
        x_stop=x_stop,
    )
    rms_scale = _noise_rms_scale(
        recipe,
        y_start=y_start,
        y_stop=y_stop,
        x_start=x_start,
        x_stop=x_stop,
    )
    image = recipe.background + recipe.noise_rms * rms_scale * standard_noise
    y_grid = np.arange(y_start, y_stop, dtype=np.float64)[:, np.newaxis]
    x_grid = np.arange(x_start, x_stop, dtype=np.float64)[np.newaxis, :]

    for source in recipe.sources:
        rotation_radians = np.deg2rad(
            source.rotation_degrees_counterclockwise_from_x
        )
        cosine = np.cos(rotation_radians)
        sine = np.sin(rotation_radians)
        x_offset = x_grid - source.x_pixel
        y_offset = y_grid - source.y_pixel
        major_offset = cosine * x_offset + sine * y_offset
        minor_offset = -sine * x_offset + cosine * y_offset
        exponent = -0.5 * (
            np.square(major_offset / source.major_sigma_pixels)
            + np.square(minor_offset / source.minor_sigma_pixels)
        )
        image += source.peak_flux_jy_per_beam * np.exp(exponent)

    _apply_invalid_rectangles(
        image,
        recipe,
        (y_start, y_stop, x_start, x_stop),
    )

    return np.asarray(image, dtype=np.float64)

iter_dataset_recipes(dataset)

Expand one governed truth recipe across recorded noise realizations.

Source code in src/hebog/validation/datasets.py

def iter_dataset_recipes(
    dataset: DatasetRecord,
) -> tuple[SyntheticRecipe, ...]:
    """Expand one governed truth recipe across recorded noise realizations."""
    return (
        dataset.recipe,
        *(
            dataset.recipe.model_copy(update={"seed": seed})
            for seed in dataset.noise_realization_seeds
        ),
    )

load_dataset_manifest(path)

Load and validate a dataset manifest without resolving its data.

Source code in src/hebog/validation/datasets.py

def load_dataset_manifest(path: Path) -> DatasetManifest:
    """Load and validate a dataset manifest without resolving its data."""
    document = path.read_text(encoding="utf-8")
    return DatasetManifest.model_validate_json(document)

recipe_sha256(recipe)

Return the canonical SHA-256 provenance digest for a recipe.

Source code in src/hebog/validation/datasets.py

def recipe_sha256(recipe: SyntheticRecipe) -> str:
    """Return the canonical SHA-256 provenance digest for a recipe."""
    document = recipe.model_dump(mode="json")
    if recipe.generator_version == 1:
        document.pop("noise_rms_fractional_gradient_xy")
        document.pop("invalid_rectangles")
    if recipe.generator_version < _CORRELATED_NOISE_GENERATOR_VERSION:
        document.pop("noise_correlation")
    canonical_json = json.dumps(
        document,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return hashlib.sha256(canonical_json).hexdigest()