Python API

This is the supported public API. Start with the tutorial; see capability and status for current limits. The API is experimental and can change between 0.x releases.

Need	Use
Run the finder	hebog.find_sources, hebog.SourceFinderRequest, hebog.SourceFinderConfig
Choose where work runs	hebog.executors
Read products	hebog.io.read_catalogue_fits_product, hebog.io.read_diagnostics_product
Handle failures	hebog.SourceFinderError and its subclasses; hebog.SourceFinderStagingWarning for what a killed run left

Developers extending Hebog will also want the internal API.

Top-level package

hebog

A scalable source finder for radio continuum images.

InvalidSourceFinderInputError

Bases: SourceFinderError

The requested input cannot be read as a supported FITS image.

Source code in src/hebog/pipeline.py

class InvalidSourceFinderInputError(SourceFinderError):
    """The requested input cannot be read as a supported FITS image."""

SourceFinderConfig dataclass

Pipeline-neutral scientific thresholds for one image analysis.

Thresholds are explicit because a value appropriate for one survey, image product, or pipeline stage is not a universal scientific default. Island-size cuts are likewise explicit pixel counts; a compatibility adapter may derive them from reviewed beam metadata before constructing this scheduler-independent configuration. background is estimated, Hebog's spatial background, or zero, which subtracts nothing while the RMS is still estimated, as PyBDSF's mean_map="zero" does; which one a workflow uses is that workflow's adapter's choice, as are its filtering options. The residual multiscale segmentation policy has its own explicit configuration because its area and support rules use beam rather than pixel units. continuum is the general source-association profile; compact is an explicit component-level profile that is incomplete for extended emission.

A threshold or pixel count may be any real number or integer, a NumPy scalar included, and is held as the Python float or int it equals, so equal configurations serialize and hash alike.

SourceFinderConfig(5, 3, 7) SourceFinderConfig(detection_threshold_sigma=5.0, island_threshold_sigma=3.0, minimum_island_pixels=7, maximum_island_pixels=None, profile='continuum', background='estimated')

Source code in src/hebog/config.py

@dataclass(frozen=True, slots=True)
class SourceFinderConfig:
    """Pipeline-neutral scientific thresholds for one image analysis.

    Thresholds are explicit because a value appropriate for one survey,
    image product, or pipeline stage is not a universal scientific default.
    Island-size cuts are likewise explicit pixel counts; a compatibility
    adapter may derive them from reviewed beam metadata before constructing
    this scheduler-independent configuration.
    ``background`` is ``estimated``, Hebog's spatial background, or
    ``zero``, which subtracts nothing while the RMS is still estimated, as
    PyBDSF's ``mean_map="zero"`` does; which one a workflow uses is that
    workflow's adapter's choice, as are its filtering options. The residual
    multiscale segmentation policy has its own explicit configuration because
    its area and support rules use beam rather than pixel units.
    ``continuum`` is the general source-association profile; ``compact`` is an
    explicit component-level profile that is incomplete for extended emission.

    A threshold or pixel count may be any real number or integer, a NumPy
    scalar included, and is held as the Python ``float`` or ``int`` it
    equals, so equal configurations serialize and hash alike.

    >>> SourceFinderConfig(5, 3, 7)
    SourceFinderConfig(detection_threshold_sigma=5.0, \
island_threshold_sigma=3.0, minimum_island_pixels=7, \
maximum_island_pixels=None, profile='continuum', background='estimated')
    """

    detection_threshold_sigma: float
    island_threshold_sigma: float
    minimum_island_pixels: int
    maximum_island_pixels: int | None = None
    profile: Literal["continuum", "compact"] = "continuum"
    background: Literal["estimated", "zero"] = "estimated"

    def __post_init__(self) -> None:
        """Validate thresholds and pixel cuts, and hold plain values."""
        if self.profile not in {"continuum", "compact"}:
            raise ValueError(
                "source-finder profile must be 'continuum' or 'compact'"
            )
        if self.background not in {"estimated", "zero"}:
            raise ValueError(
                "source-finder background must be 'estimated' or 'zero'"
            )
        for name in ("detection_threshold_sigma", "island_threshold_sigma"):
            object.__setattr__(
                self, name, _sigma_threshold(name, getattr(self, name))
            )
        if self.island_threshold_sigma >= self.detection_threshold_sigma:
            raise ValueError(
                "island_threshold_sigma must be lower than "
                "detection_threshold_sigma"
            )
        if (
            isinstance(self.minimum_island_pixels, bool)
            or not isinstance(self.minimum_island_pixels, Integral)
            or self.minimum_island_pixels < 1
        ):
            raise ValueError(
                "minimum_island_pixels must be a positive integer"
            )
        maximum = self.maximum_island_pixels
        if maximum is not None and (
            isinstance(maximum, bool)
            or not isinstance(maximum, Integral)
            or maximum < self.minimum_island_pixels
        ):
            raise ValueError(
                "maximum_island_pixels must be an integer no smaller than "
                "minimum_island_pixels"
            )
        object.__setattr__(
            self, "minimum_island_pixels", int(self.minimum_island_pixels)
        )
        if maximum is not None:
            object.__setattr__(self, "maximum_island_pixels", int(maximum))

__post_init__()

Validate thresholds and pixel cuts, and hold plain values.

Source code in src/hebog/config.py

def __post_init__(self) -> None:
    """Validate thresholds and pixel cuts, and hold plain values."""
    if self.profile not in {"continuum", "compact"}:
        raise ValueError(
            "source-finder profile must be 'continuum' or 'compact'"
        )
    if self.background not in {"estimated", "zero"}:
        raise ValueError(
            "source-finder background must be 'estimated' or 'zero'"
        )
    for name in ("detection_threshold_sigma", "island_threshold_sigma"):
        object.__setattr__(
            self, name, _sigma_threshold(name, getattr(self, name))
        )
    if self.island_threshold_sigma >= self.detection_threshold_sigma:
        raise ValueError(
            "island_threshold_sigma must be lower than "
            "detection_threshold_sigma"
        )
    if (
        isinstance(self.minimum_island_pixels, bool)
        or not isinstance(self.minimum_island_pixels, Integral)
        or self.minimum_island_pixels < 1
    ):
        raise ValueError(
            "minimum_island_pixels must be a positive integer"
        )
    maximum = self.maximum_island_pixels
    if maximum is not None and (
        isinstance(maximum, bool)
        or not isinstance(maximum, Integral)
        or maximum < self.minimum_island_pixels
    ):
        raise ValueError(
            "maximum_island_pixels must be an integer no smaller than "
            "minimum_island_pixels"
        )
    object.__setattr__(
        self, "minimum_island_pixels", int(self.minimum_island_pixels)
    )
    if maximum is not None:
        object.__setattr__(self, "maximum_island_pixels", int(maximum))

SourceFinderError

Bases: Exception

Base class for failures at the public source-finding boundary.

Source code in src/hebog/pipeline.py

class SourceFinderError(Exception):
    """Base class for failures at the public source-finding boundary."""

SourceFinderImageTooLargeError

Bases: SourceFinderError

An input exceeds the public finder's bounded image-size envelope.

Source code in src/hebog/pipeline.py

class SourceFinderImageTooLargeError(SourceFinderError):
    """An input exceeds the public finder's bounded image-size envelope."""

SourceFinderOutputExistsError

Bases: SourceFinderError

The output destination exists without this request's products.

Source code in src/hebog/pipeline.py

class SourceFinderOutputExistsError(SourceFinderError):
    """The output destination exists without this request's products."""

SourceFinderRequest dataclass

Inputs for one independent source-finding analysis.

flat_noise_image_path names the primary-beam-uncorrected image of the same sector, on the same pixel grid; when it is given, the run also estimates that image's background and RMS and publishes its RMS map. supplied_metadata applies to both images.

Source code in src/hebog/data_models/source_finding.py

@dataclass(frozen=True, slots=True)
class SourceFinderRequest:
    """Inputs for one independent source-finding analysis.

    ``flat_noise_image_path`` names the primary-beam-uncorrected image of the
    same sector, on the same pixel grid; when it is given, the run also
    estimates that image's background and RMS and publishes its RMS map.
    ``supplied_metadata`` applies to both images.
    """

    image_path: Path
    output_directory: Path
    run_id: str
    schema_version: Literal[2] = 2
    supplied_metadata: SuppliedImageMetadata | None = None
    flat_noise_image_path: Path | None = None

    def __post_init__(self) -> None:
        """Reject unsupported schema versions and empty run identifiers."""
        if self.schema_version != _REQUEST_SCHEMA_VERSION:
            raise ValueError(
                "unsupported source-finder request schema version"
            )
        if not self.run_id:
            raise ValueError("run_id must not be empty")

__post_init__()

Reject unsupported schema versions and empty run identifiers.

Source code in src/hebog/data_models/source_finding.py

def __post_init__(self) -> None:
    """Reject unsupported schema versions and empty run identifiers."""
    if self.schema_version != _REQUEST_SCHEMA_VERSION:
        raise ValueError(
            "unsupported source-finder request schema version"
        )
    if not self.run_id:
        raise ValueError("run_id must not be empty")

SourceFinderResult

Bases: BaseModel

Versioned materialised products from one scientific image analysis.

Source code in src/hebog/data_models/source_finding.py

class SourceFinderResult(BaseModel):
    """Versioned materialised products from one scientific image analysis."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    run_id: str
    catalogue: MaterializedProduct
    rms: MaterializedProduct
    mask: MaterializedProduct
    diagnostics: MaterializedProduct
    source_count: int
    gaussian_component_count: int
    island_count: int
    wall_seconds: float
    flat_noise_rms: MaterializedProduct | None = None
    schema_version: Literal[3] = 3

    @model_validator(mode="after")
    def _validate_result(self) -> Self:
        """Require one exact, scientifically labelled product set."""
        self._validate_identity_and_counts()
        self._validate_product_roles()
        return self

    def _validate_identity_and_counts(self) -> None:
        """Validate the run, population counts, and elapsed time."""
        if not self.run_id:
            raise ValueError("source-finder result run ID must not be empty")
        _require_population_counts(
            source_count=self.source_count,
            gaussian_component_count=self.gaussian_component_count,
            island_count=self.island_count,
        )
        if not isfinite(self.wall_seconds) or self.wall_seconds < 0:
            raise ValueError("wall_seconds must be finite and non-negative")

    def _validate_product_roles(self) -> None:
        """Validate roles, distinct paths, and scientific availability."""
        expected_roles = (
            ("catalogue", self.catalogue, "source-catalogue"),
            ("rms", self.rms, "rms"),
            ("mask", self.mask, "source-filtering-mask"),
            ("diagnostics", self.diagnostics, "diagnostics"),
            *(
                ()
                if self.flat_noise_rms is None
                else (("flat_noise_rms", self.flat_noise_rms, "rms"),)
            ),
        )
        for name, product, expected_role in expected_roles:
            if product.product_role != expected_role:
                raise ValueError(
                    f"{name} product role must be {expected_role}"
                )
        paths = tuple(product.path for _, product, _ in expected_roles)
        if len(set(paths)) != len(paths):
            raise ValueError("materialized product paths must be distinct")
        required_products = (
            self.catalogue,
            self.mask,
            self.diagnostics,
        )
        if any(
            product.scientific_status != "valid"
            for product in required_products
        ):
            raise ValueError(
                "only RMS may be scientifically unavailable in a success"
            )

    @property
    def catalogue_path(self) -> Path:
        """Return the materialised catalogue path for workflow consumers."""
        return self.catalogue.path

    @property
    def rms_path(self) -> Path:
        """Return the materialised RMS image path."""
        return self.rms.path

    @property
    def mask_path(self) -> Path:
        """Return the materialised source-filtering mask path."""
        return self.mask.path

    @property
    def diagnostics_path(self) -> Path:
        """Return the materialised diagnostics path."""
        return self.diagnostics.path

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one serialized materialised result."""
        result = cls.model_validate_json(payload)
        if result.canonical_json_bytes() != payload:
            raise ValueError("source-finder result JSON must be canonical")
        return result

catalogue_path property

Return the materialised catalogue path for workflow consumers.

diagnostics_path property

Return the materialised diagnostics path.

mask_path property

Return the materialised source-filtering mask path.

rms_path property

Return the materialised RMS image path.

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/source_finding.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

from_json_bytes(payload) classmethod

Validate one serialized materialised result.

Source code in src/hebog/data_models/source_finding.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one serialized materialised result."""
    result = cls.model_validate_json(payload)
    if result.canonical_json_bytes() != payload:
        raise ValueError("source-finder result JSON must be canonical")
    return result

SourceFinderStagingWarning

Bases: UserWarning

An earlier run left a hidden staging directory beside the output.

A run that is killed cannot remove its staging directory. The next run to the same output removes one whose owner has provably stopped and leaves any other in place; it reports each with this warning.

Source code in src/hebog/pipeline.py

class SourceFinderStagingWarning(UserWarning):
    """An earlier run left a hidden staging directory beside the output.

    A run that is killed cannot remove its staging directory. The next run
    to the same output removes one whose owner has provably stopped and
    leaves any other in place; it reports each with this warning.
    """

SuppliedImageMetadata dataclass

Physical metadata a caller supplies for keywords a FITS header omits.

Each value fills only a missing header keyword: frequency in Hz for RESTFRQ or a frequency axis, restoring-beam full widths and position angle in degrees for BMAJ, BMIN and BPA, and the pixel brightness unit for BUNIT. Supplying a value the header already provides is an input error, so supplied metadata never overrides an image's own description.

SuppliedImageMetadata(reference_frequency_hz=144e6) SuppliedImageMetadata(reference_frequency_hz=144000000.0, beam_major_fwhm_degrees=None, beam_minor_fwhm_degrees=None, beam_position_angle_degrees=None, brightness_unit=None)

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class SuppliedImageMetadata:
    """Physical metadata a caller supplies for keywords a FITS header omits.

    Each value fills only a missing header keyword: frequency in Hz for
    ``RESTFRQ`` or a frequency axis, restoring-beam full widths and position
    angle in degrees for ``BMAJ``, ``BMIN`` and ``BPA``, and the pixel
    brightness unit for ``BUNIT``. Supplying a value the header already
    provides is an input error, so supplied metadata never overrides an
    image's own description.

    >>> SuppliedImageMetadata(reference_frequency_hz=144e6)
    SuppliedImageMetadata(reference_frequency_hz=144000000.0, \
beam_major_fwhm_degrees=None, beam_minor_fwhm_degrees=None, \
beam_position_angle_degrees=None, brightness_unit=None)
    """

    reference_frequency_hz: float | None = None
    beam_major_fwhm_degrees: float | None = None
    beam_minor_fwhm_degrees: float | None = None
    beam_position_angle_degrees: float | None = None
    brightness_unit: str | None = None

    def __post_init__(self) -> None:
        """Require at least one finite value with the header's own limits."""
        frequency = self.reference_frequency_hz
        axes = (self.beam_major_fwhm_degrees, self.beam_minor_fwhm_degrees)
        angle = self.beam_position_angle_degrees
        unit = self.brightness_unit
        if all(value is None for value in (frequency, *axes, angle, unit)):
            raise ValueError("supply at least one image metadata value")
        # Annotations do not bind callers; a number would fail on strip().
        if unit is not None and not isinstance(cast(object, unit), str):
            raise ValueError("supplied brightness unit must be text")
        if unit is not None and not unit.strip():
            raise ValueError("supplied brightness unit must not be blank")
        if frequency is not None and not (
            isfinite(frequency) and frequency > 0
        ):
            raise ValueError(
                "supplied reference frequency must be finite and positive"
            )
        if any(
            axis is not None and not (isfinite(axis) and axis > 0)
            for axis in axes
        ):
            raise ValueError("supplied beam axes must be finite and positive")
        if angle is not None and not isfinite(angle):
            raise ValueError("supplied beam position angle must be finite")
        major, minor = axes
        if major is not None and minor is not None and minor > major:
            raise ValueError("supplied beam minor axis cannot exceed major")

__post_init__()

Require at least one finite value with the header's own limits.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Require at least one finite value with the header's own limits."""
    frequency = self.reference_frequency_hz
    axes = (self.beam_major_fwhm_degrees, self.beam_minor_fwhm_degrees)
    angle = self.beam_position_angle_degrees
    unit = self.brightness_unit
    if all(value is None for value in (frequency, *axes, angle, unit)):
        raise ValueError("supply at least one image metadata value")
    # Annotations do not bind callers; a number would fail on strip().
    if unit is not None and not isinstance(cast(object, unit), str):
        raise ValueError("supplied brightness unit must be text")
    if unit is not None and not unit.strip():
        raise ValueError("supplied brightness unit must not be blank")
    if frequency is not None and not (
        isfinite(frequency) and frequency > 0
    ):
        raise ValueError(
            "supplied reference frequency must be finite and positive"
        )
    if any(
        axis is not None and not (isfinite(axis) and axis > 0)
        for axis in axes
    ):
        raise ValueError("supplied beam axes must be finite and positive")
    if angle is not None and not isfinite(angle):
        raise ValueError("supplied beam position angle must be finite")
    major, minor = axes
    if major is not None and minor is not None and minor > major:
        raise ValueError("supplied beam minor axis cannot exceed major")

UnsupportedSourceFinderConfigurationError

Bases: SourceFinderError

The requested science cannot be executed by the public pipeline.

Source code in src/hebog/pipeline.py

class UnsupportedSourceFinderConfigurationError(SourceFinderError):
    """The requested science cannot be executed by the public pipeline."""

analysis_memory_bytes(image_shape_yx, *, expected_sources)

Return the memory one analysis holds in the process that runs it.

find_sources keeps Hebog's modules, the records each pass returns for every source, and the products it publishes in the process that calls it, so what it holds grows with the sources it finds, and with the image through the background's noise grids. Its tasks are not counted: each declares its own memory to the executor, which a DaskExecutor attaches to the task under the caller's memory resource. Reserve this amount for the task that calls find_sources, for example under the same Dask resource. With a SerialExecutor or ThreadExecutor the tasks run in the calling process, so the memory they are admitted comes on top. A request with a flat-noise image is covered.

Parameters:

Name	Type	Description	Default
image_shape_yx	tuple[int, int]	The input image's height and width in pixels.	required
expected_sources	int	An upper estimate of the sources the catalogue will hold, such as :func:expected_source_count makes from the image; the reservation is too small if more are found.	required

Returns:

Type	Description
int	A fixed amount, an amount per expected source and two and a half
int	bytes a pixel, with a quarter of headroom, in bytes.

Raises:

Type	Description
ValueError	If a dimension is not a positive integer or the source count is not a non-negative integer.

Examples:

>>> analysis_memory_bytes(
...     (3_000, 4_000), expected_sources=10
... ) == analysis_memory_bytes((4_000, 3_000), expected_sources=10)
True

Source code in src/hebog/pipeline.py

def analysis_memory_bytes(
    image_shape_yx: tuple[int, int], *, expected_sources: int
) -> int:
    """Return the memory one analysis holds in the process that runs it.

    ``find_sources`` keeps Hebog's modules, the records each pass returns
    for every source, and the products it publishes in the process that
    calls it, so what it holds grows with the sources it finds, and with
    the image through the background's noise grids. Its tasks are
    not counted: each declares its own memory to the executor, which a
    ``DaskExecutor`` attaches to the task under the caller's memory
    resource. Reserve this amount for the task that calls
    ``find_sources``, for example under the same Dask resource. With a
    ``SerialExecutor`` or ``ThreadExecutor`` the tasks run in the calling
    process, so the memory they are admitted comes on top. A request with a
    flat-noise image is covered.

    Args:
        image_shape_yx: The input image's height and width in pixels.
        expected_sources: An upper estimate of the sources the catalogue
            will hold, such as :func:`expected_source_count` makes from the
            image; the reservation is too small if more are found.

    Returns:
        A fixed amount, an amount per expected source and two and a half
        bytes a pixel, with a quarter of headroom, in bytes.

    Raises:
        ValueError: If a dimension is not a positive integer or the source
            count is not a non-negative integer.

    Examples:
        >>> analysis_memory_bytes(
        ...     (3_000, 4_000), expected_sources=10
        ... ) == analysis_memory_bytes((4_000, 3_000), expected_sources=10)
        True
    """
    for side in image_shape_yx:
        if (
            isinstance(side, bool)
            or not isinstance(side, Integral)
            or side < 1
        ):
            raise ValueError("image dimensions must be positive integers")
    if (
        isinstance(expected_sources, bool)
        or not isinstance(expected_sources, Integral)
        or expected_sources < 0
    ):
        raise ValueError("expected_sources must be a non-negative integer")
    height, width = image_shape_yx
    traced = (
        _ANALYSIS_TRACED_BASE_BYTES
        + _ANALYSIS_TRACED_BYTES_PER_SOURCE * int(expected_sources)
        + int(height)
        * int(width)
        * _ANALYSIS_TRACED_PIXEL_BYTES_NUMERATOR
        // _ANALYSIS_TRACED_PIXEL_BYTES_DENOMINATOR
    )
    return -(-traced * _HEADROOM_NUMERATOR // _HEADROOM_DENOMINATOR)

expected_source_count(image_path, *, detection_threshold_sigma, supplied_metadata=None)

Return an upper estimate of the sources find_sources publishes.

Pass it to :func:analysis_memory_bytes. A deconvolution model's component list is no such estimate: it holds many components for each source. A published source peaks at least the detection threshold times the local noise above the background, so the estimate counts the pixels that are the highest of their 3x3 neighbourhood and at least detection_threshold_sigma times the noise above the median of their 512-pixel block. A block's noise is its valid pixels' 3-sigma-clipped standard deviation, or the median of every block's if that is lower, so neither the bright sources in a block nor a primary-beam correction's noisier edges hide faint sources. A noise peak counts as a source would, so the estimate errs high; on 21 development cases it bounded every published source count, at 1.0 to 21 times it: 1.1 to 1.9 times on the LOFAR and simulated SKA fields, and most where noise peaks sit on one wide source. The image is read twice in bounded blocks, so no image-sized array is held. An image find_sources refuses is refused in the same way before any pixel is read.

Parameters:

Name	Type	Description	Default
image_path	Path	The FITS image find_sources will analyse.	required
detection_threshold_sigma	float	The detection threshold the analysis will use, SourceFinderConfig.detection_threshold_sigma.	required
supplied_metadata	SuppliedImageMetadata | None	Values the header lacks, as the request gives them.	None

Returns:

Type	Description
int	The number of such peaks; 0 for an image with no measurable noise.

Raises:

Type	Description
ValueError	If the threshold is not finite and positive.
InvalidSourceFinderInputError	If the image cannot be read as a FITS continuum image.
UnsupportedSourceFinderConfigurationError	If find_sources refuses the image's unit, frame or beam sampling.
SourceFinderImageTooLargeError	If the image exceeds the size envelope find_sources admits.

Source code in src/hebog/pipeline.py

def expected_source_count(
    image_path: Path,
    *,
    detection_threshold_sigma: float,
    supplied_metadata: SuppliedImageMetadata | None = None,
) -> int:
    """Return an upper estimate of the sources ``find_sources`` publishes.

    Pass it to :func:`analysis_memory_bytes`. A deconvolution model's
    component list is no such estimate: it holds many components for each
    source. A published source peaks at least the detection threshold times
    the local noise above the background, so the estimate counts the pixels
    that are the highest of their 3x3 neighbourhood and at least
    ``detection_threshold_sigma`` times the noise above the median of their
    512-pixel block. A block's noise is its valid pixels' 3-sigma-clipped
    standard deviation, or the median of every block's if that is lower, so
    neither the bright sources in a block nor a primary-beam correction's
    noisier edges hide faint sources. A noise peak counts as a source would,
    so the estimate errs high; on 21 development cases it bounded every
    published source count, at 1.0 to 21 times it: 1.1 to 1.9 times on the
    LOFAR and simulated SKA fields, and most where noise peaks sit on one
    wide source. The image is read twice in bounded blocks, so no
    image-sized array is held. An image ``find_sources`` refuses is
    refused in the same way before any pixel is read.

    Args:
        image_path: The FITS image ``find_sources`` will analyse.
        detection_threshold_sigma: The detection threshold the analysis
            will use, ``SourceFinderConfig.detection_threshold_sigma``.
        supplied_metadata: Values the header lacks, as the request gives
            them.

    Returns:
        The number of such peaks; 0 for an image with no measurable noise.

    Raises:
        ValueError: If the threshold is not finite and positive.
        InvalidSourceFinderInputError: If the image cannot be read as a
            FITS continuum image.
        UnsupportedSourceFinderConfigurationError: If ``find_sources``
            refuses the image's unit, frame or beam sampling.
        SourceFinderImageTooLargeError: If the image exceeds the size
            envelope ``find_sources`` admits.
    """
    from hebog.public_api import (  # noqa: PLC0415
        expected_source_count as _expected_source_count,
    )

    return _expected_source_count(
        image_path,
        detection_threshold_sigma=detection_threshold_sigma,
        supplied_metadata=supplied_metadata,
    )

find_sources(request, config, executor)

Analyse one supported FITS image and atomically publish its products.

When the output already holds the products of a finished run of the same request, with the same run ID, inputs, configuration and science, they are returned without analysing the image again. A failed run raises only once no task it submitted is still running, and removes everything it staged. A killed run leaves a hidden staging directory, which the next run to the same output removes once its owner has provably stopped or reports with a SourceFinderStagingWarning.

The implementation is imported only when called so importing the public scheduler-independent API never loads a concrete I/O or scheduler layer.

Source code in src/hebog/pipeline.py

def find_sources(
    request: SourceFinderRequest,
    config: SourceFinderConfig,
    executor: Executor,
) -> SourceFinderResult:
    """Analyse one supported FITS image and atomically publish its products.

    When the output already holds the products of a finished run of the
    same request, with the same run ID, inputs, configuration and science,
    they are returned without analysing the image again. A failed run
    raises only once no task it submitted is still running, and removes
    everything it staged. A killed run leaves a hidden staging directory,
    which the next run to the same output removes once its owner has
    provably stopped or reports with a ``SourceFinderStagingWarning``.

    The implementation is imported only when called so importing the public
    scheduler-independent API never loads a concrete I/O or scheduler layer.
    """
    from hebog.public_api import find_sources as _find_sources  # noqa: PLC0415

    return _find_sources(request, config, executor)

Request and result records

hebog.data_models

Small serializable scheduler-independent domain records.

BrightRegionCounts

Bases: BaseModel

How far one background estimate refined around bright candidates.

candidate_count counts the bright candidates the estimate found, at or above its bright-candidate threshold. The estimate refines the noise around each; candidates close enough to chain merge into one region, which can span the image, and its fine cells are estimated in bounded blocks. largest_read_pixels is the most pixels one of those tasks read, zero when none read any.

Source code in src/hebog/data_models/source_finding.py

class BrightRegionCounts(BaseModel):
    """How far one background estimate refined around bright candidates.

    ``candidate_count`` counts the bright candidates the estimate found, at
    or above its bright-candidate threshold. The estimate refines the noise
    around each; candidates close enough to chain merge into one region,
    which can span the image, and its fine cells are estimated in bounded
    blocks. ``largest_read_pixels`` is the most pixels one of those tasks
    read, zero when none read any.
    """

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    candidate_count: int = 0
    largest_read_pixels: int = 0

    @model_validator(mode="after")
    def _validate_counts(self) -> Self:
        """Reject a negative count."""
        if min(self.candidate_count, self.largest_read_pixels) < 0:
            raise ValueError("bright-region counts cannot be negative")
        return self

CatalogueSourceMembership dataclass

Canonical exact partition of components assigned to one source.

Source code in src/hebog/data_models/source_association.py

@dataclass(frozen=True, slots=True)
class CatalogueSourceMembership:
    """Canonical exact partition of components assigned to one source."""

    source_id: str
    component_ids: tuple[str, ...]

    def __post_init__(self) -> None:
        """Require a non-empty canonical component membership."""
        _require_identifier(self.source_id, field_name="source ID")
        if not self.component_ids or self.component_ids != tuple(
            sorted(set(self.component_ids))
        ):
            raise ValueError(
                "source component IDs must be non-empty unique and canonical"
            )
        for component_id in self.component_ids:
            _require_identifier(component_id, field_name="component ID")

__post_init__()

Require a non-empty canonical component membership.

Source code in src/hebog/data_models/source_association.py

def __post_init__(self) -> None:
    """Require a non-empty canonical component membership."""
    _require_identifier(self.source_id, field_name="source ID")
    if not self.component_ids or self.component_ids != tuple(
        sorted(set(self.component_ids))
    ):
        raise ValueError(
            "source component IDs must be non-empty unique and canonical"
        )
    for component_id in self.component_ids:
        _require_identifier(component_id, field_name="component ID")

CelestialWcs dataclass

Canonical FITS celestial-WCS cards and their coordinate-frame name.

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class CelestialWcs:
    """Canonical FITS celestial-WCS cards and their coordinate-frame name."""

    fits_header: str
    coordinate_frame: str

    def __post_init__(self) -> None:
        """Require enough metadata to reconstruct and interpret the WCS."""
        if not self.fits_header or not self.coordinate_frame:
            raise ValueError(
                "celestial WCS header and coordinate frame must not be empty"
            )

__post_init__()

Require enough metadata to reconstruct and interpret the WCS.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Require enough metadata to reconstruct and interpret the WCS."""
    if not self.fits_header or not self.coordinate_frame:
        raise ValueError(
            "celestial WCS header and coordinate frame must not be empty"
        )

ContinuumSourceFindingDiagnostics

Bases: BaseModel

Version-two diagnostics with per-extended-source provenance.

Source code in src/hebog/data_models/source_finding.py

class ContinuumSourceFindingDiagnostics(BaseModel):
    """Version-two diagnostics with per-extended-source provenance."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    run_id: str
    source_count: int
    gaussian_component_count: int
    island_count: int
    extended_source_count: int
    terminal_disposition_count: int
    rms_scientific_status: Literal["valid", "unavailable"]
    source_provenance: tuple[SourceScaleProvenance, ...]
    schema_version: Literal[2] = 2

    @model_validator(mode="after")
    def _validate_diagnostics(self) -> Self:
        """Require canonical provenance and consistent populations."""
        if not self.run_id:
            raise ValueError("diagnostics run ID must not be empty")
        _require_population_counts(
            source_count=self.source_count,
            gaussian_component_count=self.gaussian_component_count,
            island_count=self.island_count,
        )
        if not 0 <= self.extended_source_count <= self.source_count:
            raise ValueError(
                "extended source count must fit the source population"
            )
        if len(self.source_provenance) != self.extended_source_count:
            raise ValueError(
                "extended source count must match source provenance"
            )
        source_ids = tuple(item.source_id for item in self.source_provenance)
        association_ids = tuple(
            item.association_id for item in self.source_provenance
        )
        if source_ids != tuple(sorted(set(source_ids))):
            raise ValueError("source provenance must use canonical source IDs")
        if len(set(association_ids)) != len(association_ids):
            raise ValueError("source provenance associations must be unique")
        if self.terminal_disposition_count < self.island_count:
            raise ValueError(
                "terminal dispositions cannot be fewer than output islands"
            )
        return self

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one canonical serialized continuum diagnostic."""
        diagnostics = cls.model_validate_json(payload)
        if diagnostics.canonical_json_bytes() != payload:
            raise ValueError(
                "continuum source-finding diagnostics JSON must be canonical"
            )
        return diagnostics

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/source_finding.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

from_json_bytes(payload) classmethod

Validate one canonical serialized continuum diagnostic.

Source code in src/hebog/data_models/source_finding.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one canonical serialized continuum diagnostic."""
    diagnostics = cls.model_validate_json(payload)
    if diagnostics.canonical_json_bytes() != payload:
        raise ValueError(
            "continuum source-finding diagnostics JSON must be canonical"
        )
    return diagnostics

CrossScaleAssociation

Bases: _MultiscaleModel

Deterministic association of persistent scale detections.

Source code in src/hebog/data_models/multiscale.py

class CrossScaleAssociation(_MultiscaleModel):
    """Deterministic association of persistent scale detections."""

    association_id: str
    scale_detection_ids: tuple[str, ...] = Field(min_length=1)
    selected_scale_detection_id: str
    contributing_scale_orders: tuple[int, ...] = Field(min_length=1)
    schema_version: Literal[3] = 3

    @field_validator("scale_detection_ids")
    @classmethod
    def validate_identifiers(
        cls,
        identifiers: tuple[str, ...],
    ) -> tuple[str, ...]:
        """Require stable association inputs."""
        _require_canonical_identifiers(
            identifiers,
            field_name="association IDs",
        )
        return identifiers

    @model_validator(mode="after")
    def validate_association(self) -> Self:
        """Bind the selected representation to retained provenance."""
        _require_identifier(self.association_id, field_name="association ID")
        if self.selected_scale_detection_id not in self.scale_detection_ids:
            raise ValueError(
                "selected scale detection must belong to the association"
            )
        if self.contributing_scale_orders != tuple(
            sorted(set(self.contributing_scale_orders))
        ) or any(order < 1 for order in self.contributing_scale_orders):
            raise ValueError("contributing scale orders must be canonical")
        return self

validate_association()

Bind the selected representation to retained provenance.

Source code in src/hebog/data_models/multiscale.py

@model_validator(mode="after")
def validate_association(self) -> Self:
    """Bind the selected representation to retained provenance."""
    _require_identifier(self.association_id, field_name="association ID")
    if self.selected_scale_detection_id not in self.scale_detection_ids:
        raise ValueError(
            "selected scale detection must belong to the association"
        )
    if self.contributing_scale_orders != tuple(
        sorted(set(self.contributing_scale_orders))
    ) or any(order < 1 for order in self.contributing_scale_orders):
        raise ValueError("contributing scale orders must be canonical")
    return self

validate_identifiers(identifiers) classmethod

Require stable association inputs.

Source code in src/hebog/data_models/multiscale.py

@field_validator("scale_detection_ids")
@classmethod
def validate_identifiers(
    cls,
    identifiers: tuple[str, ...],
) -> tuple[str, ...]:
    """Require stable association inputs."""
    _require_canonical_identifiers(
        identifiers,
        field_name="association IDs",
    )
    return identifiers

DetectionComponentRecord dataclass

Stable identity and pixel geometry for one immutable owner component.

Source code in src/hebog/data_models/source_association.py

@dataclass(frozen=True, slots=True)
class DetectionComponentRecord:
    """Stable identity and pixel geometry for one immutable owner component."""

    component_id: str
    label_value: int
    canonical_pixel_yx: tuple[int, int]
    centroid_yx: tuple[float, float]
    covariance_pixels_squared: (
        tuple[tuple[float, float], tuple[float, float]] | None
    )
    component_labels_are_identity: Literal[False] = False

    def __post_init__(self) -> None:
        """Require valid immutable geometry without promoting local labels."""
        _require_identifier(self.component_id, field_name="component ID")
        if self.label_value <= 0:
            raise ValueError("component label value must be positive")
        if (
            len(self.canonical_pixel_yx) != _COORDINATE_DIMENSIONS
            or min(self.canonical_pixel_yx) < 0
        ):
            raise ValueError("canonical pixel must contain non-negative y-x")
        if len(self.centroid_yx) != _COORDINATE_DIMENSIONS or not all(
            isfinite(value) and value >= 0.0 for value in self.centroid_yx
        ):
            raise ValueError("component centroid must contain finite y-x")
        covariance = self.covariance_pixels_squared
        if covariance is None:
            return
        (yy, yx), (xy, xx) = covariance
        if (
            not all(isfinite(value) for value in (yy, yx, xy, xx))
            or yy <= 0.0
            or xx <= 0.0
            or yx != xy
            or yy * xx - yx * xy <= 0.0
        ):
            raise ValueError(
                "component covariance must be finite symmetric positive "
                "definite"
            )

__post_init__()

Require valid immutable geometry without promoting local labels.

Source code in src/hebog/data_models/source_association.py

def __post_init__(self) -> None:
    """Require valid immutable geometry without promoting local labels."""
    _require_identifier(self.component_id, field_name="component ID")
    if self.label_value <= 0:
        raise ValueError("component label value must be positive")
    if (
        len(self.canonical_pixel_yx) != _COORDINATE_DIMENSIONS
        or min(self.canonical_pixel_yx) < 0
    ):
        raise ValueError("canonical pixel must contain non-negative y-x")
    if len(self.centroid_yx) != _COORDINATE_DIMENSIONS or not all(
        isfinite(value) and value >= 0.0 for value in self.centroid_yx
    ):
        raise ValueError("component centroid must contain finite y-x")
    covariance = self.covariance_pixels_squared
    if covariance is None:
        return
    (yy, yx), (xy, xx) = covariance
    if (
        not all(isfinite(value) for value in (yy, yx, xy, xx))
        or yy <= 0.0
        or xx <= 0.0
        or yx != xy
        or yy * xx - yx * xy <= 0.0
    ):
        raise ValueError(
            "component covariance must be finite symmetric positive "
            "definite"
        )

FluxMeasurement

Bases: _CatalogueModel

Peak and integrated source flux plus local noise in canonical units.

Source code in src/hebog/data_models/catalogues.py

class FluxMeasurement(_CatalogueModel):
    """Peak and integrated source flux plus local noise in canonical units."""

    peak_flux_jy_per_beam: float
    peak_flux_error_jy_per_beam: float | None
    integrated_flux_jy: float
    integrated_flux_error_jy: float | None
    local_rms_jy_per_beam: float

    @model_validator(mode="after")
    def _validate_flux(self) -> Self:
        """Require positive finite measurements and valid uncertainties."""
        if (
            not isfinite(self.peak_flux_jy_per_beam)
            or not isfinite(self.integrated_flux_jy)
            or self.peak_flux_jy_per_beam <= 0
            or self.integrated_flux_jy <= 0
        ):
            raise ValueError("peak and integrated fluxes must be positive")
        _require_optional_errors(
            (
                self.peak_flux_error_jy_per_beam,
                self.integrated_flux_error_jy,
            ),
            field_name="flux errors",
        )
        if (
            not isfinite(self.local_rms_jy_per_beam)
            or self.local_rms_jy_per_beam <= 0
        ):
            raise ValueError("local RMS must be finite and positive")
        return self

GaussianComponent

Bases: _MeasuredCatalogueObject

One fitted Gaussian with explicit source and support-island links.

Source code in src/hebog/data_models/catalogues.py

class GaussianComponent(_MeasuredCatalogueObject):
    """One fitted Gaussian with explicit source and support-island links."""

    gaussian_component_id: str
    source_id: str
    fitted_shape: GaussianShape

    @model_validator(mode="after")
    def _validate_component_ids(self) -> Self:
        """Require stable component and source identities."""
        _require_domain_identifier(
            self.gaussian_component_id,
            field_name="Gaussian component ID",
        )
        _require_domain_identifier(self.source_id, field_name="source ID")
        return self

GaussianShape

Bases: _CatalogueModel

One fitted FWHM ellipse and optional one-sigma errors in degrees.

Source code in src/hebog/data_models/catalogues.py

class GaussianShape(_CatalogueModel):
    """One fitted FWHM ellipse and optional one-sigma errors in degrees."""

    major_fwhm_degrees: float
    minor_fwhm_degrees: float
    position_angle_degrees: float
    major_fwhm_error_degrees: float | None
    minor_fwhm_error_degrees: float | None
    position_angle_error_degrees: float | None

    @model_validator(mode="after")
    def _validate_shape(self) -> Self:
        """Require a positive ordered ellipse in canonical orientation."""
        if (
            not isfinite(self.major_fwhm_degrees)
            or not isfinite(self.minor_fwhm_degrees)
            or self.major_fwhm_degrees <= 0
            or self.minor_fwhm_degrees <= 0
        ):
            raise ValueError("Gaussian shape axes must be finite and positive")
        if self.minor_fwhm_degrees > self.major_fwhm_degrees:
            raise ValueError("Gaussian shape minor axis cannot exceed major")
        if not isfinite(self.position_angle_degrees) or not (
            0 <= self.position_angle_degrees < _HALF_CIRCLE_DEGREES
        ):
            raise ValueError(
                "Gaussian shape position angle must be within [0, 180)"
            )
        _require_optional_errors(
            (
                self.major_fwhm_error_degrees,
                self.minor_fwhm_error_degrees,
                self.position_angle_error_degrees,
            ),
            field_name="shape errors",
        )
        return self

ImageBounds dataclass

Half-open global pixel bounds in NumPy (y, x) axis order.

Source code in src/hebog/data_models/partitioning.py

@dataclass(frozen=True, slots=True)
class ImageBounds:
    """Half-open global pixel bounds in NumPy ``(y, x)`` axis order."""

    y_start: int
    y_stop: int
    x_start: int
    x_stop: int

    def __post_init__(self) -> None:
        """Require non-negative, non-empty half-open bounds."""
        if min(self.y_start, self.y_stop, self.x_start, self.x_stop) < 0:
            raise ValueError("image bounds must be non-negative")
        if self.y_stop <= self.y_start or self.x_stop <= self.x_start:
            raise ValueError("image bounds must be non-empty")

    @property
    def shape_yx(self) -> tuple[int, int]:
        """Return the bounded array shape in NumPy axis order."""
        return (
            self.y_stop - self.y_start,
            self.x_stop - self.x_start,
        )

    @property
    def center_xy(self) -> tuple[float, float]:
        """Return the centre of the bounded pixels, in global ``(x, y)``.

        Local geometry is derived here rather than at a corner, and every
        caller that derives it must agree, so the definition lives with the
        bounds instead of beside each use.
        """
        return (
            (self.x_start + self.x_stop - 1) / 2,
            (self.y_start + self.y_stop - 1) / 2,
        )

    def require_inside(self, shape_yx: tuple[int, int]) -> None:
        """Reject bounds extending beyond one logical image plane."""
        if self.y_stop > shape_yx[0] or self.x_stop > shape_yx[1]:
            raise ValueError(
                f"image bounds must stay inside image shape {shape_yx}"
            )

    def contains(self, other: ImageBounds) -> bool:
        """Return whether another region lies completely within the bounds."""
        return (
            self.y_start <= other.y_start
            and other.y_stop <= self.y_stop
            and self.x_start <= other.x_start
            and other.x_stop <= self.x_stop
        )

    def expanded(
        self,
        margin_pixels: int,
        image_shape_yx: tuple[int, int],
    ) -> ImageBounds:
        """Return a uniformly expanded region clipped to the image plane."""
        if (
            isinstance(margin_pixels, bool)
            or not isinstance(  # pyright: ignore[reportUnnecessaryIsInstance]
                margin_pixels,
                int,
            )
            or margin_pixels < 0
        ):
            raise ValueError("bounds margin must be a non-negative integer")
        self.require_inside(image_shape_yx)
        return ImageBounds(
            y_start=max(0, self.y_start - margin_pixels),
            y_stop=min(image_shape_yx[0], self.y_stop + margin_pixels),
            x_start=max(0, self.x_start - margin_pixels),
            x_stop=min(image_shape_yx[1], self.x_stop + margin_pixels),
        )

center_xy property

Return the centre of the bounded pixels, in global (x, y).

Local geometry is derived here rather than at a corner, and every caller that derives it must agree, so the definition lives with the bounds instead of beside each use.

shape_yx property

Return the bounded array shape in NumPy axis order.

__post_init__()

Require non-negative, non-empty half-open bounds.

Source code in src/hebog/data_models/partitioning.py

def __post_init__(self) -> None:
    """Require non-negative, non-empty half-open bounds."""
    if min(self.y_start, self.y_stop, self.x_start, self.x_stop) < 0:
        raise ValueError("image bounds must be non-negative")
    if self.y_stop <= self.y_start or self.x_stop <= self.x_start:
        raise ValueError("image bounds must be non-empty")

contains(other)

Return whether another region lies completely within the bounds.

Source code in src/hebog/data_models/partitioning.py

def contains(self, other: ImageBounds) -> bool:
    """Return whether another region lies completely within the bounds."""
    return (
        self.y_start <= other.y_start
        and other.y_stop <= self.y_stop
        and self.x_start <= other.x_start
        and other.x_stop <= self.x_stop
    )

expanded(margin_pixels, image_shape_yx)

Return a uniformly expanded region clipped to the image plane.

Source code in src/hebog/data_models/partitioning.py

def expanded(
    self,
    margin_pixels: int,
    image_shape_yx: tuple[int, int],
) -> ImageBounds:
    """Return a uniformly expanded region clipped to the image plane."""
    if (
        isinstance(margin_pixels, bool)
        or not isinstance(  # pyright: ignore[reportUnnecessaryIsInstance]
            margin_pixels,
            int,
        )
        or margin_pixels < 0
    ):
        raise ValueError("bounds margin must be a non-negative integer")
    self.require_inside(image_shape_yx)
    return ImageBounds(
        y_start=max(0, self.y_start - margin_pixels),
        y_stop=min(image_shape_yx[0], self.y_stop + margin_pixels),
        x_start=max(0, self.x_start - margin_pixels),
        x_stop=min(image_shape_yx[1], self.x_stop + margin_pixels),
    )

require_inside(shape_yx)

Reject bounds extending beyond one logical image plane.

Source code in src/hebog/data_models/partitioning.py

def require_inside(self, shape_yx: tuple[int, int]) -> None:
    """Reject bounds extending beyond one logical image plane."""
    if self.y_stop > shape_yx[0] or self.x_stop > shape_yx[1]:
        raise ValueError(
            f"image bounds must stay inside image shape {shape_yx}"
        )

ImageMetadata dataclass

Scientific metadata required to interpret one logical image plane.

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class ImageMetadata:
    """Scientific metadata required to interpret one logical image plane."""

    shape_yx: tuple[int, int]
    unit: str
    beam: RestoringBeam
    celestial_wcs: CelestialWcs
    reference_frequency_hz: float
    schema_version: Literal[1] = 1

    def __post_init__(self) -> None:
        """Reject unsupported versions and incomplete physical metadata."""
        if self.schema_version != 1:
            raise ValueError("unsupported image metadata schema version")
        if min(self.shape_yx) < 1:
            raise ValueError(
                "image metadata shape dimensions must be positive"
            )
        if not self.unit:
            raise ValueError("image metadata unit must not be empty")
        if (
            not isfinite(self.reference_frequency_hz)
            or self.reference_frequency_hz <= 0
        ):
            raise ValueError(
                "image metadata reference frequency must be finite and "
                "positive"
            )

__post_init__()

Reject unsupported versions and incomplete physical metadata.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Reject unsupported versions and incomplete physical metadata."""
    if self.schema_version != 1:
        raise ValueError("unsupported image metadata schema version")
    if min(self.shape_yx) < 1:
        raise ValueError(
            "image metadata shape dimensions must be positive"
        )
    if not self.unit:
        raise ValueError("image metadata unit must not be empty")
    if (
        not isfinite(self.reference_frequency_hz)
        or self.reference_frequency_hz <= 0
    ):
        raise ValueError(
            "image metadata reference frequency must be finite and "
            "positive"
        )

Island

Bases: _CatalogueModel

One connected detection region with signed original-pixel flux.

Source code in src/hebog/data_models/catalogues.py

class Island(_CatalogueModel):
    """One connected detection region with signed original-pixel flux."""

    island_id: str
    pixel_count: int
    integrated_flux_jy: float
    integrated_flux_error_jy: float | None
    local_rms_jy_per_beam: float
    mean_brightness_jy_per_beam: float

    @model_validator(mode="after")
    def _validate_island(self) -> Self:
        """Validate stable identity and finite island summaries."""
        _require_domain_identifier(self.island_id, field_name="island ID")
        if self.pixel_count <= 0:
            raise ValueError("island pixel count must be positive")
        if not isfinite(self.integrated_flux_jy):
            raise ValueError("island integrated flux must be finite")
        _require_optional_errors(
            (self.integrated_flux_error_jy,),
            field_name="island flux error",
        )
        if (
            not isfinite(self.local_rms_jy_per_beam)
            or self.local_rms_jy_per_beam <= 0
        ):
            raise ValueError("island local RMS must be finite and positive")
        if not isfinite(self.mean_brightness_jy_per_beam):
            raise ValueError("island mean brightness must be finite")
        return self

MaterializedProduct

Bases: BaseModel

Closed restartable output file and its content identity.

Source code in src/hebog/data_models/source_finding.py

class MaterializedProduct(BaseModel):
    """Closed restartable output file and its content identity."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    product_role: ProductRole
    path: Path
    media_type: Literal["application/fits", "image/fits", "application/json"]
    byte_count: int
    content_sha256: str
    scientific_status: Literal["valid", "unavailable"]
    content_schema_version: int
    schema_version: Literal[1] = 1

    @model_validator(mode="after")
    def _validate_product(self) -> Self:
        """Validate a portable file identity without opening the path."""
        if not self.path.name:
            raise ValueError("materialized product file path is required")
        if self.byte_count <= 0:
            raise ValueError(
                "materialized product byte count must be positive"
            )
        if _SHA256.fullmatch(self.content_sha256) is None:
            raise ValueError(
                "materialized product SHA-256 must be lowercase hex"
            )
        if self.content_schema_version < 1:
            raise ValueError(
                "materialized product content schema version must be positive"
            )
        expected_media_type = _PRODUCT_MEDIA_TYPES[self.product_role]
        if self.media_type != expected_media_type:
            raise ValueError(
                f"{self.product_role} media type must be {expected_media_type}"
            )
        return self

PartitionManifest dataclass

Canonical tiling record with one owner per output pixel.

Source code in src/hebog/data_models/partitioning.py

@dataclass(frozen=True, slots=True)
class PartitionManifest:
    """Canonical tiling record with one owner per output pixel."""

    image_shape_yx: tuple[int, int]
    tile_core_shape_yx: tuple[int, int]
    halo_yx: tuple[int, int]
    partition_origin_yx: tuple[int, int]
    tiles: tuple[TilePartition, ...]
    schema_version: Literal[1] = 1

    def __post_init__(self) -> None:
        """Reject unsupported versions and non-canonical ownership records."""
        if self.schema_version != 1:
            raise ValueError("unsupported partition manifest schema version")
        _validate_geometry(
            self.image_shape_yx,
            self.tile_core_shape_yx,
            self.halo_yx,
            self.partition_origin_yx,
        )
        expected = _canonical_tiles(
            self.image_shape_yx,
            self.tile_core_shape_yx,
            self.halo_yx,
            self.partition_origin_yx,
        )
        if self.tiles != expected:
            raise ValueError(
                "partition manifest tiles must match canonical row-major "
                "ownership"
            )

    def owner_for_position_yx(
        self,
        position_yx: tuple[float, float],
    ) -> TilePartition:
        """Return the unique tile owning a source reference position.

        Positions use zero-based continuous ``(y, x)`` pixel coordinates.
        A position exactly on an internal boundary belongs to the core that
        starts at that boundary. A source may overlap other cores and halos;
        only its deterministic reference position selects catalogue ownership.
        """
        y_position, x_position = position_yx
        height, width = self.image_shape_yx
        if (
            not isfinite(y_position)
            or not isfinite(x_position)
            or not 0 <= y_position < height
            or not 0 <= x_position < width
        ):
            raise ValueError(
                "source reference position must be finite and inside the image"
            )
        y_index = _axis_owner_index(
            y_position,
            length=height,
            core=self.tile_core_shape_yx[0],
            origin=self.partition_origin_yx[0],
        )
        x_index = _axis_owner_index(
            x_position,
            length=width,
            core=self.tile_core_shape_yx[1],
            origin=self.partition_origin_yx[1],
        )
        tiles_per_row = _axis_partition_count(
            length=width,
            core=self.tile_core_shape_yx[1],
            origin=self.partition_origin_yx[1],
        )
        return self.tiles[y_index * tiles_per_row + x_index]

    def tiles_meeting(self, bounds: ImageBounds) -> tuple[TilePartition, ...]:
        """Return every tile whose core shares a pixel with ``bounds``.

        Cores form a grid, so the tiles are the index rectangle between the
        owners of the region's first and last pixels; no tile is scanned.

        Examples:
            >>> manifest = PartitionManifest.create(
            ...     image_shape_yx=(10, 10),
            ...     tile_core_shape_yx=(4, 4),
            ...     halo_yx=(0, 0),
            ... )
            >>> region = ImageBounds(3, 5, 7, 10)
            >>> [
            ...     (tile.tile_y_index, tile.tile_x_index)
            ...     for tile in manifest.tiles_meeting(region)
            ... ]
            [(0, 1), (0, 2), (1, 1), (1, 2)]
        """
        bounds.require_inside(self.image_shape_yx)
        first = self.owner_for_position_yx((bounds.y_start, bounds.x_start))
        last = self.owner_for_position_yx(
            (bounds.y_stop - 1, bounds.x_stop - 1)
        )
        tiles_per_row = self.tiles[-1].tile_x_index + 1
        return tuple(
            self.tiles[row * tiles_per_row + column]
            for row in range(first.tile_y_index, last.tile_y_index + 1)
            for column in range(first.tile_x_index, last.tile_x_index + 1)
        )

    def __reduce__(
        self,
    ) -> tuple[
        object,
        tuple[
            tuple[int, int],
            tuple[int, int],
            tuple[int, int],
            tuple[int, int],
        ],
    ]:
        """Pickle as the geometry the tiles are planned from.

        The tiles are the one canonical tiling of that geometry, which
        construction already requires, so they are rebuilt where the
        manifest is unpickled rather than shipped with it. Executor tasks
        carry manifests inside sinks and requests, and Dask serializes a
        task's payload with every task, so a payload that grew with the
        tile count made each task cost the driver time proportional to the
        image: about 280 ms for 14,641 128-pixel cells. Schema version 1 is
        the only one a manifest can have, so it is not carried; a second
        version must be added here.
        """
        return (
            _manifest_for_geometry,
            (
                self.image_shape_yx,
                self.tile_core_shape_yx,
                self.halo_yx,
                self.partition_origin_yx,
            ),
        )

    @classmethod
    def create(
        cls,
        *,
        image_shape_yx: tuple[int, int],
        tile_core_shape_yx: tuple[int, int],
        halo_yx: tuple[int, int],
        partition_origin_yx: tuple[int, int] = (0, 0),
    ) -> Self:
        """Validate geometry and construct its unique canonical manifest."""
        _validate_geometry(
            image_shape_yx,
            tile_core_shape_yx,
            halo_yx,
            partition_origin_yx,
        )
        return cls(
            image_shape_yx=image_shape_yx,
            tile_core_shape_yx=tile_core_shape_yx,
            halo_yx=halo_yx,
            partition_origin_yx=partition_origin_yx,
            tiles=_canonical_tiles(
                image_shape_yx,
                tile_core_shape_yx,
                halo_yx,
                partition_origin_yx,
            ),
        )

__post_init__()

Reject unsupported versions and non-canonical ownership records.

Source code in src/hebog/data_models/partitioning.py

def __post_init__(self) -> None:
    """Reject unsupported versions and non-canonical ownership records."""
    if self.schema_version != 1:
        raise ValueError("unsupported partition manifest schema version")
    _validate_geometry(
        self.image_shape_yx,
        self.tile_core_shape_yx,
        self.halo_yx,
        self.partition_origin_yx,
    )
    expected = _canonical_tiles(
        self.image_shape_yx,
        self.tile_core_shape_yx,
        self.halo_yx,
        self.partition_origin_yx,
    )
    if self.tiles != expected:
        raise ValueError(
            "partition manifest tiles must match canonical row-major "
            "ownership"
        )

__reduce__()

Pickle as the geometry the tiles are planned from.

The tiles are the one canonical tiling of that geometry, which construction already requires, so they are rebuilt where the manifest is unpickled rather than shipped with it. Executor tasks carry manifests inside sinks and requests, and Dask serializes a task's payload with every task, so a payload that grew with the tile count made each task cost the driver time proportional to the image: about 280 ms for 14,641 128-pixel cells. Schema version 1 is the only one a manifest can have, so it is not carried; a second version must be added here.

Source code in src/hebog/data_models/partitioning.py

def __reduce__(
    self,
) -> tuple[
    object,
    tuple[
        tuple[int, int],
        tuple[int, int],
        tuple[int, int],
        tuple[int, int],
    ],
]:
    """Pickle as the geometry the tiles are planned from.

    The tiles are the one canonical tiling of that geometry, which
    construction already requires, so they are rebuilt where the
    manifest is unpickled rather than shipped with it. Executor tasks
    carry manifests inside sinks and requests, and Dask serializes a
    task's payload with every task, so a payload that grew with the
    tile count made each task cost the driver time proportional to the
    image: about 280 ms for 14,641 128-pixel cells. Schema version 1 is
    the only one a manifest can have, so it is not carried; a second
    version must be added here.
    """
    return (
        _manifest_for_geometry,
        (
            self.image_shape_yx,
            self.tile_core_shape_yx,
            self.halo_yx,
            self.partition_origin_yx,
        ),
    )

create(*, image_shape_yx, tile_core_shape_yx, halo_yx, partition_origin_yx=(0, 0)) classmethod

Validate geometry and construct its unique canonical manifest.

Source code in src/hebog/data_models/partitioning.py

@classmethod
def create(
    cls,
    *,
    image_shape_yx: tuple[int, int],
    tile_core_shape_yx: tuple[int, int],
    halo_yx: tuple[int, int],
    partition_origin_yx: tuple[int, int] = (0, 0),
) -> Self:
    """Validate geometry and construct its unique canonical manifest."""
    _validate_geometry(
        image_shape_yx,
        tile_core_shape_yx,
        halo_yx,
        partition_origin_yx,
    )
    return cls(
        image_shape_yx=image_shape_yx,
        tile_core_shape_yx=tile_core_shape_yx,
        halo_yx=halo_yx,
        partition_origin_yx=partition_origin_yx,
        tiles=_canonical_tiles(
            image_shape_yx,
            tile_core_shape_yx,
            halo_yx,
            partition_origin_yx,
        ),
    )

owner_for_position_yx(position_yx)

Return the unique tile owning a source reference position.

Positions use zero-based continuous (y, x) pixel coordinates. A position exactly on an internal boundary belongs to the core that starts at that boundary. A source may overlap other cores and halos; only its deterministic reference position selects catalogue ownership.

Source code in src/hebog/data_models/partitioning.py

def owner_for_position_yx(
    self,
    position_yx: tuple[float, float],
) -> TilePartition:
    """Return the unique tile owning a source reference position.

    Positions use zero-based continuous ``(y, x)`` pixel coordinates.
    A position exactly on an internal boundary belongs to the core that
    starts at that boundary. A source may overlap other cores and halos;
    only its deterministic reference position selects catalogue ownership.
    """
    y_position, x_position = position_yx
    height, width = self.image_shape_yx
    if (
        not isfinite(y_position)
        or not isfinite(x_position)
        or not 0 <= y_position < height
        or not 0 <= x_position < width
    ):
        raise ValueError(
            "source reference position must be finite and inside the image"
        )
    y_index = _axis_owner_index(
        y_position,
        length=height,
        core=self.tile_core_shape_yx[0],
        origin=self.partition_origin_yx[0],
    )
    x_index = _axis_owner_index(
        x_position,
        length=width,
        core=self.tile_core_shape_yx[1],
        origin=self.partition_origin_yx[1],
    )
    tiles_per_row = _axis_partition_count(
        length=width,
        core=self.tile_core_shape_yx[1],
        origin=self.partition_origin_yx[1],
    )
    return self.tiles[y_index * tiles_per_row + x_index]

tiles_meeting(bounds)

Return every tile whose core shares a pixel with bounds.

Cores form a grid, so the tiles are the index rectangle between the owners of the region's first and last pixels; no tile is scanned.

Examples:

>>> manifest = PartitionManifest.create(
...     image_shape_yx=(10, 10),
...     tile_core_shape_yx=(4, 4),
...     halo_yx=(0, 0),
... )
>>> region = ImageBounds(3, 5, 7, 10)
>>> [
...     (tile.tile_y_index, tile.tile_x_index)
...     for tile in manifest.tiles_meeting(region)
... ]
[(0, 1), (0, 2), (1, 1), (1, 2)]

Source code in src/hebog/data_models/partitioning.py

def tiles_meeting(self, bounds: ImageBounds) -> tuple[TilePartition, ...]:
    """Return every tile whose core shares a pixel with ``bounds``.

    Cores form a grid, so the tiles are the index rectangle between the
    owners of the region's first and last pixels; no tile is scanned.

    Examples:
        >>> manifest = PartitionManifest.create(
        ...     image_shape_yx=(10, 10),
        ...     tile_core_shape_yx=(4, 4),
        ...     halo_yx=(0, 0),
        ... )
        >>> region = ImageBounds(3, 5, 7, 10)
        >>> [
        ...     (tile.tile_y_index, tile.tile_x_index)
        ...     for tile in manifest.tiles_meeting(region)
        ... ]
        [(0, 1), (0, 2), (1, 1), (1, 2)]
    """
    bounds.require_inside(self.image_shape_yx)
    first = self.owner_for_position_yx((bounds.y_start, bounds.x_start))
    last = self.owner_for_position_yx(
        (bounds.y_stop - 1, bounds.x_stop - 1)
    )
    tiles_per_row = self.tiles[-1].tile_x_index + 1
    return tuple(
        self.tiles[row * tiles_per_row + column]
        for row in range(first.tile_y_index, last.tile_y_index + 1)
        for column in range(first.tile_x_index, last.tile_x_index + 1)
    )

ProductChunk dataclass

Logical identity of one tile-owned chunk in a product array.

Source code in src/hebog/data_models/products.py

@dataclass(frozen=True, slots=True)
class ProductChunk:
    """Logical identity of one tile-owned chunk in a product array."""

    generation_id: str
    product_name: str
    tile_id: str
    core_bounds: ImageBounds
    dtype: str
    shape_yx: tuple[int, int]
    content_sha256: str
    schema_version: Literal[2] = 2

    def __post_init__(self) -> None:
        """Reject malformed chunk identities before storage access."""
        if self.schema_version != _PRODUCT_CHUNK_SCHEMA_VERSION:
            raise ValueError("unsupported product chunk schema version")
        if not self.generation_id:
            raise ValueError("product chunk generation ID must not be empty")
        validate_product_name(self.product_name)
        validate_tile_id(self.tile_id)
        if not self.dtype:
            raise ValueError("product chunk dtype must not be empty")
        if self.shape_yx != self.core_bounds.shape_yx:
            raise ValueError("product chunk shape must match its core bounds")
        if _SHA256.fullmatch(self.content_sha256) is None:
            raise ValueError("product chunk SHA-256 must be lowercase hex")

__post_init__()

Reject malformed chunk identities before storage access.

Source code in src/hebog/data_models/products.py

def __post_init__(self) -> None:
    """Reject malformed chunk identities before storage access."""
    if self.schema_version != _PRODUCT_CHUNK_SCHEMA_VERSION:
        raise ValueError("unsupported product chunk schema version")
    if not self.generation_id:
        raise ValueError("product chunk generation ID must not be empty")
    validate_product_name(self.product_name)
    validate_tile_id(self.tile_id)
    if not self.dtype:
        raise ValueError("product chunk dtype must not be empty")
    if self.shape_yx != self.core_bounds.shape_yx:
        raise ValueError("product chunk shape must match its core bounds")
    if _SHA256.fullmatch(self.content_sha256) is None:
        raise ValueError("product chunk SHA-256 must be lowercase hex")

ProductGenerationManifest

Bases: BaseModel

Exact chunk set that makes one intermediate generation consumable.

Source code in src/hebog/data_models/generations.py

class ProductGenerationManifest(BaseModel):
    """Exact chunk set that makes one intermediate generation consumable."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    generation_id: str
    partition_manifest: PartitionManifest
    product_names: tuple[str, ...]
    chunks: tuple[ProductChunk, ...]
    schema_version: Literal[1] = 1

    @model_validator(mode="after")
    def _validate_generation(self) -> Self:
        """Reject incomplete, ambiguous, or non-canonical generations."""
        self._validate_identity_and_products()
        tiles_by_id = {
            tile.tile_id: tile for tile in self.partition_manifest.tiles
        }
        self._validate_chunk_records(tiles_by_id)
        self._validate_exact_chunk_set()
        return self

    def _validate_identity_and_products(self) -> None:
        """Validate the generation key and canonical product set."""
        if not self.generation_id:
            raise ValueError("generation ID must not be empty")
        if not self.product_names:
            raise ValueError("generation must contain at least one product")
        for product_name in self.product_names:
            validate_product_name(product_name)
        if self.product_names != tuple(sorted(set(self.product_names))):
            raise ValueError(
                "generation product names must be unique and canonical"
            )

    def _validate_chunk_records(
        self,
        tiles_by_id: dict[str, TilePartition],
    ) -> None:
        """Validate chunk identity, ownership, and product dtype."""
        seen: dict[tuple[str, str], ProductChunk] = {}
        dtypes: dict[str, str] = {}
        for chunk in self.chunks:
            if chunk.generation_id != self.generation_id:
                raise ValueError("generation contains a mixed generation ID")
            key = (chunk.product_name, chunk.tile_id)
            previous = seen.get(key)
            if previous is not None:
                qualifier = "duplicate" if previous == chunk else "conflicting"
                raise ValueError(
                    f"generation contains {qualifier} chunk records"
                )
            seen[key] = chunk
            if chunk.product_name not in self.product_names:
                raise ValueError("generation contains an unexpected product")
            tile = tiles_by_id.get(chunk.tile_id)
            if tile is None:
                raise ValueError("generation contains an unexpected tile")
            if chunk.core_bounds != tile.core_bounds:
                raise ValueError(
                    "generation chunk core bounds disagree with ownership"
                )
            existing_dtype = dtypes.setdefault(
                chunk.product_name,
                chunk.dtype,
            )
            if existing_dtype != chunk.dtype:
                raise ValueError(
                    "generation product chunks must use one dtype"
                )

    def _validate_exact_chunk_set(self) -> None:
        """Require every product-owner pair once in canonical order."""
        expected = tuple(
            (product_name, tile.tile_id)
            for product_name in self.product_names
            for tile in self.partition_manifest.tiles
        )
        actual = tuple(
            (chunk.product_name, chunk.tile_id) for chunk in self.chunks
        )
        missing = set(expected).difference(actual)
        if missing:
            raise ValueError("generation is missing expected chunk records")
        if actual != expected:
            raise ValueError("generation chunks must use canonical order")

    @classmethod
    def create(
        cls,
        *,
        generation_id: str,
        partition_manifest: PartitionManifest,
        product_names: Iterable[str],
        chunks: Iterable[ProductChunk],
    ) -> Self:
        """Canonicalize caller results before validating exact completion."""
        return cls(
            generation_id=generation_id,
            partition_manifest=partition_manifest,
            product_names=tuple(sorted(product_names)),
            chunks=tuple(
                sorted(
                    chunks,
                    key=lambda chunk: (
                        chunk.product_name,
                        chunk.tile_id,
                    ),
                )
            ),
        )

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one canonical serialized completion record."""
        generation = cls.model_validate_json(payload)
        if generation.canonical_json_bytes() != payload:
            raise ValueError(
                "product generation manifest JSON must be canonical"
            )
        return generation

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/generations.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

create(*, generation_id, partition_manifest, product_names, chunks) classmethod

Canonicalize caller results before validating exact completion.

Source code in src/hebog/data_models/generations.py

@classmethod
def create(
    cls,
    *,
    generation_id: str,
    partition_manifest: PartitionManifest,
    product_names: Iterable[str],
    chunks: Iterable[ProductChunk],
) -> Self:
    """Canonicalize caller results before validating exact completion."""
    return cls(
        generation_id=generation_id,
        partition_manifest=partition_manifest,
        product_names=tuple(sorted(product_names)),
        chunks=tuple(
            sorted(
                chunks,
                key=lambda chunk: (
                    chunk.product_name,
                    chunk.tile_id,
                ),
            )
        ),
    )

from_json_bytes(payload) classmethod

Validate one canonical serialized completion record.

Source code in src/hebog/data_models/generations.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one canonical serialized completion record."""
    generation = cls.model_validate_json(payload)
    if generation.canonical_json_bytes() != payload:
        raise ValueError(
            "product generation manifest JSON must be canonical"
        )
    return generation

PublicSourceFindingDiagnostics

Bases: BaseModel

Public-run diagnostics retain explicit measurement dispositions.

bright_regions describes the true-sky background estimate and flat_noise_bright_regions the flat-noise image's, present exactly when the run had one.

Source code in src/hebog/data_models/source_finding.py

class PublicSourceFindingDiagnostics(BaseModel):
    """Public-run diagnostics retain explicit measurement dispositions.

    ``bright_regions`` describes the true-sky background estimate and
    ``flat_noise_bright_regions`` the flat-noise image's, present exactly
    when the run had one.
    """

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    run_id: str
    profile: Literal["continuum", "compact"]
    profile_limitations: tuple[Literal["extended-emission-incomplete"], ...]
    configuration_qualification: Literal[
        "development-unqualified", "custom-unqualified"
    ]
    source_count: int
    gaussian_component_count: int
    island_count: int
    deblended_parent_count: int = 0
    deferred_deblend_parent_count: int = 0
    wide_object_counts: WideObjectCounts = Field(
        default_factory=WideObjectCounts
    )
    bright_regions: BrightRegionCounts = Field(
        default_factory=BrightRegionCounts
    )
    flat_noise_bright_regions: BrightRegionCounts | None = None
    measurement_dispositions: tuple[MeasurementDisposition, ...] = ()
    rms_scientific_status: Literal["valid", "unavailable"]
    provenance: PublicSourceFindingProvenance
    schema_version: Literal[13] = 13

    @model_validator(mode="after")
    def _validate_diagnostics(self) -> Self:
        """Require consistent populations and honest profile limitations."""
        if not self.run_id:
            raise ValueError("diagnostics run ID must not be empty")
        _require_population_counts(
            source_count=self.source_count,
            gaussian_component_count=self.gaussian_component_count,
            island_count=self.island_count,
        )
        if (
            self.deblended_parent_count < 0
            or self.deferred_deblend_parent_count < 0
        ):
            raise ValueError("deblend disposition counts cannot be negative")
        expected_limitations = (
            ("extended-emission-incomplete",)
            if self.profile == "compact"
            else ()
        )
        if self.profile_limitations != expected_limitations:
            raise ValueError(
                "public diagnostics limitations must match the profile"
            )
        if (self.flat_noise_bright_regions is None) != (
            self.provenance.flat_noise_input_sha256 is None
        ):
            raise ValueError(
                "flat-noise bright-region counts must match the flat-noise "
                "input"
            )
        entries = self.measurement_dispositions
        identities = tuple(
            (item.object_kind, item.object_id) for item in entries
        )
        components = {
            item.object_id
            for item in entries
            if item.object_kind == "component"
        }
        sources = tuple(
            item for item in entries if item.object_kind == "source"
        )
        members = tuple(
            member for item in sources for member in item.member_component_ids
        )
        if (
            identities != tuple(sorted(set(identities)))
            or len(members) != len(set(members))
            or set(members) != components
            or sum(item.catalogue_row_published for item in sources)
            != self.source_count
            or sum(
                item.catalogue_row_published
                for item in entries
                if item.object_kind == "component"
            )
            != self.gaussian_component_count
        ):
            raise ValueError("public measurement census is inconsistent")
        return self

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one canonical serialized public diagnostic."""
        diagnostics = cls.model_validate_json(payload)
        if diagnostics.canonical_json_bytes() != payload:
            raise ValueError(
                "public source-finding diagnostics JSON must be canonical"
            )
        return diagnostics

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/source_finding.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

from_json_bytes(payload) classmethod

Validate one canonical serialized public diagnostic.

Source code in src/hebog/data_models/source_finding.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one canonical serialized public diagnostic."""
    diagnostics = cls.model_validate_json(payload)
    if diagnostics.canonical_json_bytes() != payload:
        raise ValueError(
            "public source-finding diagnostics JSON must be canonical"
        )
    return diagnostics

PublicSourceFindingProvenance

Bases: BaseModel

Exact input, configuration, and scientific-composition identity.

Source code in src/hebog/data_models/source_finding.py

class PublicSourceFindingProvenance(BaseModel):
    """Exact input, configuration, and scientific-composition identity."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    input_sha256: str
    configuration_sha256: str
    scientific_profile_sha256: str
    scientific_composition_sha256: str
    scientific_composition: Literal[
        "phase-5-evidence-bound-public-catalogue-v23"
    ]
    supplied_image_metadata: SuppliedImageMetadata | None = None
    flat_noise_input_sha256: str | None = None
    schema_version: Literal[4] = 4

    @model_validator(mode="after")
    def _validate_provenance(self) -> Self:
        """Require every bound identity to be exact lowercase SHA-256."""
        identities = (
            self.input_sha256,
            self.configuration_sha256,
            self.scientific_profile_sha256,
            self.scientific_composition_sha256,
            *(
                ()
                if self.flat_noise_input_sha256 is None
                else (self.flat_noise_input_sha256,)
            ),
        )
        if any(_SHA256.fullmatch(identity) is None for identity in identities):
            raise ValueError("public provenance identities must be SHA-256")
        return self

RestoringBeam dataclass

Elliptical restoring-beam full widths and position angle in degrees.

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class RestoringBeam:
    """Elliptical restoring-beam full widths and position angle in degrees."""

    major_fwhm_degrees: float
    minor_fwhm_degrees: float
    position_angle_degrees: float

    def __post_init__(self) -> None:
        """Require finite positive ordered beam axes."""
        if not all(
            isfinite(value)
            for value in (
                self.major_fwhm_degrees,
                self.minor_fwhm_degrees,
                self.position_angle_degrees,
            )
        ):
            raise ValueError("restoring beam values must be finite")
        if self.major_fwhm_degrees <= 0 or self.minor_fwhm_degrees <= 0:
            raise ValueError("restoring beam axes must be positive")
        if self.minor_fwhm_degrees > self.major_fwhm_degrees:
            raise ValueError("restoring beam minor axis cannot exceed major")

__post_init__()

Require finite positive ordered beam axes.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Require finite positive ordered beam axes."""
    if not all(
        isfinite(value)
        for value in (
            self.major_fwhm_degrees,
            self.minor_fwhm_degrees,
            self.position_angle_degrees,
        )
    ):
        raise ValueError("restoring beam values must be finite")
    if self.major_fwhm_degrees <= 0 or self.minor_fwhm_degrees <= 0:
        raise ValueError("restoring beam axes must be positive")
    if self.minor_fwhm_degrees > self.major_fwhm_degrees:
        raise ValueError("restoring beam minor axis cannot exceed major")

ScaleDetection

Bases: _MultiscaleModel

One bounded significant support region at one configured scale.

Source code in src/hebog/data_models/multiscale.py

class ScaleDetection(_MultiscaleModel):
    """One bounded significant support region at one configured scale."""

    detection_id: str
    parent_island_id: str | None
    scale_order: int = Field(ge=1)
    nominal_scale_beam_fwhm: float = Field(gt=0)
    support_pixel_count: int = Field(ge=1)
    valid_support_fraction: float = Field(gt=0, le=1)
    bounds_yx: tuple[int, int, int, int]
    canonical_pixel_yx: tuple[int, int]
    peak_response_jy_per_beam: float = Field(gt=0)
    peak_signal_to_noise: float
    touches_image_edge: bool
    schema_version: Literal[1] = 1

    @model_validator(mode="after")
    def validate_detection(self) -> Self:
        """Validate global identity, bounds, and finite scale response."""
        _require_identifier(self.detection_id, field_name="detection ID")
        if self.parent_island_id is not None:
            _require_identifier(
                self.parent_island_id,
                field_name="parent island ID",
            )
        y_start, y_stop, x_start, x_stop = self.bounds_yx
        if min(self.bounds_yx) < 0 or y_start >= y_stop or x_start >= x_stop:
            raise ValueError("scale detection bounds must be increasing")
        y_pixel, x_pixel = self.canonical_pixel_yx
        if not (y_start <= y_pixel < y_stop and x_start <= x_pixel < x_stop):
            raise ValueError("canonical scale pixel must be inside bounds")
        response_values = (
            self.nominal_scale_beam_fwhm,
            self.valid_support_fraction,
            self.peak_response_jy_per_beam,
            self.peak_signal_to_noise,
        )
        if not all(isfinite(value) for value in response_values):
            raise ValueError("scale response values must be finite")
        return self

validate_detection()

Validate global identity, bounds, and finite scale response.

Source code in src/hebog/data_models/multiscale.py

@model_validator(mode="after")
def validate_detection(self) -> Self:
    """Validate global identity, bounds, and finite scale response."""
    _require_identifier(self.detection_id, field_name="detection ID")
    if self.parent_island_id is not None:
        _require_identifier(
            self.parent_island_id,
            field_name="parent island ID",
        )
    y_start, y_stop, x_start, x_stop = self.bounds_yx
    if min(self.bounds_yx) < 0 or y_start >= y_stop or x_start >= x_stop:
        raise ValueError("scale detection bounds must be increasing")
    y_pixel, x_pixel = self.canonical_pixel_yx
    if not (y_start <= y_pixel < y_stop and x_start <= x_pixel < x_stop):
        raise ValueError("canonical scale pixel must be inside bounds")
    response_values = (
        self.nominal_scale_beam_fwhm,
        self.valid_support_fraction,
        self.peak_response_jy_per_beam,
        self.peak_signal_to_noise,
    )
    if not all(isfinite(value) for value in response_values):
        raise ValueError("scale response values must be finite")
    return self

SkyPosition

Bases: _CatalogueModel

One ICRS sky position and optional one-sigma errors in degrees.

Source code in src/hebog/data_models/catalogues.py

class SkyPosition(_CatalogueModel):
    """One ICRS sky position and optional one-sigma errors in degrees."""

    right_ascension_degrees: float
    declination_degrees: float
    right_ascension_error_degrees: float | None
    declination_error_degrees: float | None

    @model_validator(mode="after")
    def _validate_position(self) -> Self:
        """Require canonical finite sky coordinates and uncertainties."""
        if not isfinite(self.right_ascension_degrees) or not (
            0 <= self.right_ascension_degrees < _FULL_CIRCLE_DEGREES
        ):
            raise ValueError(
                "right ascension must be finite and within [0, 360) degrees"
            )
        if not isfinite(self.declination_degrees) or not (
            _MINIMUM_DECLINATION_DEGREES
            <= self.declination_degrees
            <= _MAXIMUM_DECLINATION_DEGREES
        ):
            raise ValueError(
                "declination must be finite and within [-90, 90] degrees"
            )
        _require_optional_errors(
            (
                self.right_ascension_error_degrees,
                self.declination_error_degrees,
            ),
            field_name="position errors",
        )
        return self

SourceAssociationEdge dataclass

One accepted pair supported by continuity and directional size.

Source code in src/hebog/data_models/source_association.py

@dataclass(frozen=True, slots=True)
class SourceAssociationEdge:
    """One accepted pair supported by continuity and directional size."""

    first_component_id: str
    second_component_id: str
    saddle_margin_sigma: float
    normalized_separation: float

    def __post_init__(self) -> None:
        """Require one canonical undirected accepted edge."""
        _require_identifier(
            self.first_component_id,
            field_name="first component ID",
        )
        _require_identifier(
            self.second_component_id,
            field_name="second component ID",
        )
        if self.first_component_id >= self.second_component_id:
            raise ValueError("association edge component IDs must be ordered")
        if (
            not isfinite(self.saddle_margin_sigma)
            or self.saddle_margin_sigma < 0.0
        ):
            raise ValueError("association saddle margin must be non-negative")
        if (
            not isfinite(self.normalized_separation)
            or not 0.0 <= self.normalized_separation <= 1.0
        ):
            raise ValueError(
                "association normalized separation must be within [0, 1]"
            )

__post_init__()

Require one canonical undirected accepted edge.

Source code in src/hebog/data_models/source_association.py

def __post_init__(self) -> None:
    """Require one canonical undirected accepted edge."""
    _require_identifier(
        self.first_component_id,
        field_name="first component ID",
    )
    _require_identifier(
        self.second_component_id,
        field_name="second component ID",
    )
    if self.first_component_id >= self.second_component_id:
        raise ValueError("association edge component IDs must be ordered")
    if (
        not isfinite(self.saddle_margin_sigma)
        or self.saddle_margin_sigma < 0.0
    ):
        raise ValueError("association saddle margin must be non-negative")
    if (
        not isfinite(self.normalized_separation)
        or not 0.0 <= self.normalized_separation <= 1.0
    ):
        raise ValueError(
            "association normalized separation must be within [0, 1]"
        )

SourceAssociationResult dataclass

Complete array-free association graph and source partition.

Source code in src/hebog/data_models/source_association.py

@dataclass(frozen=True, slots=True)
class SourceAssociationResult:
    """Complete array-free association graph and source partition."""

    components: tuple[DetectionComponentRecord, ...]
    edges: tuple[SourceAssociationEdge, ...]
    memberships: tuple[CatalogueSourceMembership, ...]
    ambiguous_component_ids: tuple[str, ...] = ()
    hierarchy_diagnostics: SourceHierarchyDiagnostics | None = None

    def __post_init__(self) -> None:
        """Require unique evidence and an exact component partition."""
        component_ids = tuple(item.component_id for item in self.components)
        label_values = tuple(item.label_value for item in self.components)
        if component_ids != tuple(sorted(set(component_ids))) or len(
            set(label_values)
        ) != len(label_values):
            raise ValueError("association component records must be unique")
        edge_keys = tuple(
            (item.first_component_id, item.second_component_id)
            for item in self.edges
        )
        if edge_keys != tuple(sorted(set(edge_keys))):
            raise ValueError("association edges must be unique and canonical")
        if any(
            component_id not in set(component_ids)
            for edge in self.edges
            for component_id in (
                edge.first_component_id,
                edge.second_component_id,
            )
        ):
            raise ValueError("association edge names an unknown component")
        source_ids = tuple(item.source_id for item in self.memberships)
        if source_ids != tuple(sorted(set(source_ids))):
            raise ValueError(
                "association source memberships must be canonical"
            )
        claimed = tuple(
            component_id
            for membership in self.memberships
            for component_id in membership.component_ids
        )
        if len(claimed) != len(set(claimed)) or set(claimed) != set(
            component_ids
        ):
            raise ValueError(
                "source memberships must partition every component "
                "exactly once"
            )
        if self.ambiguous_component_ids != tuple(
            sorted(set(self.ambiguous_component_ids))
        ) or not set(self.ambiguous_component_ids).issubset(component_ids):
            raise ValueError(
                "ambiguous component IDs must be a canonical component subset"
            )
        diagnostics = self.hierarchy_diagnostics
        if diagnostics is not None and (
            diagnostics.direct_component_count != len(self.components)
            or diagnostics.catalogue_source_count != len(self.memberships)
        ):
            raise ValueError(
                "source hierarchy diagnostics must match association counts"
            )

__post_init__()

Require unique evidence and an exact component partition.

Source code in src/hebog/data_models/source_association.py

def __post_init__(self) -> None:
    """Require unique evidence and an exact component partition."""
    component_ids = tuple(item.component_id for item in self.components)
    label_values = tuple(item.label_value for item in self.components)
    if component_ids != tuple(sorted(set(component_ids))) or len(
        set(label_values)
    ) != len(label_values):
        raise ValueError("association component records must be unique")
    edge_keys = tuple(
        (item.first_component_id, item.second_component_id)
        for item in self.edges
    )
    if edge_keys != tuple(sorted(set(edge_keys))):
        raise ValueError("association edges must be unique and canonical")
    if any(
        component_id not in set(component_ids)
        for edge in self.edges
        for component_id in (
            edge.first_component_id,
            edge.second_component_id,
        )
    ):
        raise ValueError("association edge names an unknown component")
    source_ids = tuple(item.source_id for item in self.memberships)
    if source_ids != tuple(sorted(set(source_ids))):
        raise ValueError(
            "association source memberships must be canonical"
        )
    claimed = tuple(
        component_id
        for membership in self.memberships
        for component_id in membership.component_ids
    )
    if len(claimed) != len(set(claimed)) or set(claimed) != set(
        component_ids
    ):
        raise ValueError(
            "source memberships must partition every component "
            "exactly once"
        )
    if self.ambiguous_component_ids != tuple(
        sorted(set(self.ambiguous_component_ids))
    ) or not set(self.ambiguous_component_ids).issubset(component_ids):
        raise ValueError(
            "ambiguous component IDs must be a canonical component subset"
        )
    diagnostics = self.hierarchy_diagnostics
    if diagnostics is not None and (
        diagnostics.direct_component_count != len(self.components)
        or diagnostics.catalogue_source_count != len(self.memberships)
    ):
        raise ValueError(
            "source hierarchy diagnostics must match association counts"
        )

SourceCandidate

Bases: _MeasuredCatalogueObject

One catalogue-level association of measured astrophysical emission.

Source code in src/hebog/data_models/catalogues.py

class SourceCandidate(_MeasuredCatalogueObject):
    """One catalogue-level association of measured astrophysical emission."""

    source_id: str
    fitted_shape: GaussianShape | None
    association_aperture_integrated_flux_jy: float | None = None

    @model_validator(mode="after")
    def _validate_source_id(self) -> Self:
        """Require stable identity and valid optional aperture photometry."""
        _require_domain_identifier(self.source_id, field_name="source ID")
        aperture_flux = self.association_aperture_integrated_flux_jy
        if aperture_flux is not None and (
            not isfinite(aperture_flux) or aperture_flux <= 0
        ):
            raise ValueError(
                "association aperture flux must be finite and positive"
            )
        return self

SourceCatalogue

Bases: _CatalogueModel

Canonical MFS catalogue with distinct islands, sources, and fits.

Source code in src/hebog/data_models/catalogues.py

class SourceCatalogue(_CatalogueModel):
    """Canonical MFS catalogue with distinct islands, sources, and fits."""

    catalogue_id: str
    coordinate_frame: Literal["icrs"]
    position_epoch: str
    reference_frequency_hz: float
    islands: tuple[Island, ...]
    sources: tuple[SourceCandidate, ...]
    gaussian_components: tuple[GaussianComponent, ...]
    schema_version: Literal[3] = 3

    @model_validator(mode="after")
    def _validate_catalogue(self) -> Self:
        """Validate metadata, canonical identities, and relationships."""
        _require_domain_identifier(
            self.catalogue_id,
            field_name="catalogue ID",
        )
        if not self.position_epoch:
            raise ValueError("catalogue position epoch must not be empty")
        if (
            not isfinite(self.reference_frequency_hz)
            or self.reference_frequency_hz <= 0
        ):
            raise ValueError("catalogue reference frequency must be positive")
        self._validate_identity_sets()
        self._validate_relationships()
        self._validate_reference_frequencies()
        return self

    def _validate_identity_sets(self) -> None:
        """Require a deterministic unique order within each object class."""
        _require_unique_canonical_ids(
            tuple(island.island_id for island in self.islands),
            object_name="island",
        )
        _require_unique_canonical_ids(
            tuple(source.source_id for source in self.sources),
            object_name="source",
        )
        _require_unique_canonical_ids(
            tuple(
                component.gaussian_component_id
                for component in self.gaussian_components
            ),
            object_name="Gaussian component",
        )

    def _validate_relationships(self) -> None:
        """Require every source and component relationship to resolve."""
        islands_by_id = {island.island_id: island for island in self.islands}
        sources_by_id = {source.source_id: source for source in self.sources}
        for source in self.sources:
            if {source.island_id, *source.additional_island_ids} - (
                islands_by_id.keys()
            ):
                raise ValueError("source references an unknown island")
        for component in self.gaussian_components:
            source = sources_by_id.get(component.source_id)
            if source is None:
                raise ValueError(
                    "Gaussian component references an unknown source"
                )
            if {component.island_id, *component.additional_island_ids} - {
                source.island_id,
                *source.additional_island_ids,
            }:
                raise ValueError(
                    "source and Gaussian component must share an island"
                )

    def _validate_reference_frequencies(self) -> None:
        """Keep every initial MFS measurement on one explicit frequency."""
        measured_objects = (*self.sources, *self.gaussian_components)
        if any(
            item.spectral_model.reference_frequency_hz
            != self.reference_frequency_hz
            for item in measured_objects
        ):
            raise ValueError(
                "catalogue measurements must share its reference frequency"
            )

    @classmethod
    def create(  # noqa: PLR0913
        cls,
        *,
        catalogue_id: str,
        coordinate_frame: Literal["icrs"],
        position_epoch: str,
        reference_frequency_hz: float,
        islands: Iterable[Island],
        sources: Iterable[SourceCandidate],
        gaussian_components: Iterable[GaussianComponent],
    ) -> Self:
        """Canonicalize reconciled records before schema validation."""
        return cls(
            catalogue_id=catalogue_id,
            coordinate_frame=coordinate_frame,
            position_epoch=position_epoch,
            reference_frequency_hz=reference_frequency_hz,
            islands=tuple(sorted(islands, key=lambda item: item.island_id)),
            sources=tuple(sorted(sources, key=lambda item: item.source_id)),
            gaussian_components=tuple(
                sorted(
                    gaussian_components,
                    key=lambda item: item.gaussian_component_id,
                )
            ),
        )

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one serialized internal catalogue."""
        catalogue = cls.model_validate_json(payload)
        if catalogue.canonical_json_bytes() != payload:
            raise ValueError("source catalogue JSON must be canonical")
        return catalogue

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/catalogues.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

create(*, catalogue_id, coordinate_frame, position_epoch, reference_frequency_hz, islands, sources, gaussian_components) classmethod

Canonicalize reconciled records before schema validation.

Source code in src/hebog/data_models/catalogues.py

@classmethod
def create(  # noqa: PLR0913
    cls,
    *,
    catalogue_id: str,
    coordinate_frame: Literal["icrs"],
    position_epoch: str,
    reference_frequency_hz: float,
    islands: Iterable[Island],
    sources: Iterable[SourceCandidate],
    gaussian_components: Iterable[GaussianComponent],
) -> Self:
    """Canonicalize reconciled records before schema validation."""
    return cls(
        catalogue_id=catalogue_id,
        coordinate_frame=coordinate_frame,
        position_epoch=position_epoch,
        reference_frequency_hz=reference_frequency_hz,
        islands=tuple(sorted(islands, key=lambda item: item.island_id)),
        sources=tuple(sorted(sources, key=lambda item: item.source_id)),
        gaussian_components=tuple(
            sorted(
                gaussian_components,
                key=lambda item: item.gaussian_component_id,
            )
        ),
    )

from_json_bytes(payload) classmethod

Validate one serialized internal catalogue.

Source code in src/hebog/data_models/catalogues.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one serialized internal catalogue."""
    catalogue = cls.model_validate_json(payload)
    if catalogue.canonical_json_bytes() != payload:
        raise ValueError("source catalogue JSON must be canonical")
    return catalogue

SourceFinderRequest dataclass

Inputs for one independent source-finding analysis.

flat_noise_image_path names the primary-beam-uncorrected image of the same sector, on the same pixel grid; when it is given, the run also estimates that image's background and RMS and publishes its RMS map. supplied_metadata applies to both images.

Source code in src/hebog/data_models/source_finding.py

@dataclass(frozen=True, slots=True)
class SourceFinderRequest:
    """Inputs for one independent source-finding analysis.

    ``flat_noise_image_path`` names the primary-beam-uncorrected image of the
    same sector, on the same pixel grid; when it is given, the run also
    estimates that image's background and RMS and publishes its RMS map.
    ``supplied_metadata`` applies to both images.
    """

    image_path: Path
    output_directory: Path
    run_id: str
    schema_version: Literal[2] = 2
    supplied_metadata: SuppliedImageMetadata | None = None
    flat_noise_image_path: Path | None = None

    def __post_init__(self) -> None:
        """Reject unsupported schema versions and empty run identifiers."""
        if self.schema_version != _REQUEST_SCHEMA_VERSION:
            raise ValueError(
                "unsupported source-finder request schema version"
            )
        if not self.run_id:
            raise ValueError("run_id must not be empty")

__post_init__()

Reject unsupported schema versions and empty run identifiers.

Source code in src/hebog/data_models/source_finding.py

def __post_init__(self) -> None:
    """Reject unsupported schema versions and empty run identifiers."""
    if self.schema_version != _REQUEST_SCHEMA_VERSION:
        raise ValueError(
            "unsupported source-finder request schema version"
        )
    if not self.run_id:
        raise ValueError("run_id must not be empty")

SourceFinderResult

Bases: BaseModel

Versioned materialised products from one scientific image analysis.

Source code in src/hebog/data_models/source_finding.py

class SourceFinderResult(BaseModel):
    """Versioned materialised products from one scientific image analysis."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    run_id: str
    catalogue: MaterializedProduct
    rms: MaterializedProduct
    mask: MaterializedProduct
    diagnostics: MaterializedProduct
    source_count: int
    gaussian_component_count: int
    island_count: int
    wall_seconds: float
    flat_noise_rms: MaterializedProduct | None = None
    schema_version: Literal[3] = 3

    @model_validator(mode="after")
    def _validate_result(self) -> Self:
        """Require one exact, scientifically labelled product set."""
        self._validate_identity_and_counts()
        self._validate_product_roles()
        return self

    def _validate_identity_and_counts(self) -> None:
        """Validate the run, population counts, and elapsed time."""
        if not self.run_id:
            raise ValueError("source-finder result run ID must not be empty")
        _require_population_counts(
            source_count=self.source_count,
            gaussian_component_count=self.gaussian_component_count,
            island_count=self.island_count,
        )
        if not isfinite(self.wall_seconds) or self.wall_seconds < 0:
            raise ValueError("wall_seconds must be finite and non-negative")

    def _validate_product_roles(self) -> None:
        """Validate roles, distinct paths, and scientific availability."""
        expected_roles = (
            ("catalogue", self.catalogue, "source-catalogue"),
            ("rms", self.rms, "rms"),
            ("mask", self.mask, "source-filtering-mask"),
            ("diagnostics", self.diagnostics, "diagnostics"),
            *(
                ()
                if self.flat_noise_rms is None
                else (("flat_noise_rms", self.flat_noise_rms, "rms"),)
            ),
        )
        for name, product, expected_role in expected_roles:
            if product.product_role != expected_role:
                raise ValueError(
                    f"{name} product role must be {expected_role}"
                )
        paths = tuple(product.path for _, product, _ in expected_roles)
        if len(set(paths)) != len(paths):
            raise ValueError("materialized product paths must be distinct")
        required_products = (
            self.catalogue,
            self.mask,
            self.diagnostics,
        )
        if any(
            product.scientific_status != "valid"
            for product in required_products
        ):
            raise ValueError(
                "only RMS may be scientifically unavailable in a success"
            )

    @property
    def catalogue_path(self) -> Path:
        """Return the materialised catalogue path for workflow consumers."""
        return self.catalogue.path

    @property
    def rms_path(self) -> Path:
        """Return the materialised RMS image path."""
        return self.rms.path

    @property
    def mask_path(self) -> Path:
        """Return the materialised source-filtering mask path."""
        return self.mask.path

    @property
    def diagnostics_path(self) -> Path:
        """Return the materialised diagnostics path."""
        return self.diagnostics.path

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one serialized materialised result."""
        result = cls.model_validate_json(payload)
        if result.canonical_json_bytes() != payload:
            raise ValueError("source-finder result JSON must be canonical")
        return result

catalogue_path property

Return the materialised catalogue path for workflow consumers.

diagnostics_path property

Return the materialised diagnostics path.

mask_path property

Return the materialised source-filtering mask path.

rms_path property

Return the materialised RMS image path.

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/source_finding.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

from_json_bytes(payload) classmethod

Validate one serialized materialised result.

Source code in src/hebog/data_models/source_finding.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one serialized materialised result."""
    result = cls.model_validate_json(payload)
    if result.canonical_json_bytes() != payload:
        raise ValueError("source-finder result JSON must be canonical")
    return result

SourceFindingDiagnostics

Bases: BaseModel

Versioned scientific population summary for one completed run.

Source code in src/hebog/data_models/source_finding.py

class SourceFindingDiagnostics(BaseModel):
    """Versioned scientific population summary for one completed run."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    run_id: str
    source_count: int
    gaussian_component_count: int
    island_count: int
    rms_scientific_status: Literal["valid", "unavailable"]
    schema_version: Literal[1] = 1

    @model_validator(mode="after")
    def _validate_diagnostics(self) -> Self:
        """Require a named run and internally consistent populations."""
        if not self.run_id:
            raise ValueError("diagnostics run ID must not be empty")
        _require_population_counts(
            source_count=self.source_count,
            gaussian_component_count=self.gaussian_component_count,
            island_count=self.island_count,
        )
        return self

    def canonical_json_bytes(self) -> bytes:
        """Return deterministic UTF-8 JSON with one final newline."""
        document = json.dumps(
            self.model_dump(mode="json"),
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
        return f"{document}\n".encode()

    @classmethod
    def from_json_bytes(cls, payload: bytes) -> Self:
        """Validate one canonical serialized diagnostics document."""
        diagnostics = cls.model_validate_json(payload)
        if diagnostics.canonical_json_bytes() != payload:
            raise ValueError(
                "source-finding diagnostics JSON must be canonical"
            )
        return diagnostics

canonical_json_bytes()

Return deterministic UTF-8 JSON with one final newline.

Source code in src/hebog/data_models/source_finding.py

def canonical_json_bytes(self) -> bytes:
    """Return deterministic UTF-8 JSON with one final newline."""
    document = json.dumps(
        self.model_dump(mode="json"),
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    return f"{document}\n".encode()

from_json_bytes(payload) classmethod

Validate one canonical serialized diagnostics document.

Source code in src/hebog/data_models/source_finding.py

@classmethod
def from_json_bytes(cls, payload: bytes) -> Self:
    """Validate one canonical serialized diagnostics document."""
    diagnostics = cls.model_validate_json(payload)
    if diagnostics.canonical_json_bytes() != payload:
        raise ValueError(
            "source-finding diagnostics JSON must be canonical"
        )
    return diagnostics

SourceScaleProvenance

Bases: BaseModel

Auditable scale and support provenance for one extended source.

Source code in src/hebog/data_models/source_finding.py

class SourceScaleProvenance(BaseModel):
    """Auditable scale and support provenance for one extended source."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    source_id: str
    island_id: str
    association_id: str
    scale_detection_ids: tuple[str, ...]
    selected_scale_detection_id: str
    contributing_scale_orders: tuple[int, ...]
    relationship: Literal[
        "extended-only",
        "contains-compact-support",
        "overlaps-compact-support",
    ]
    support_pixel_count: int
    visible_model_fraction: float
    schema_version: Literal[1] = 1

    @model_validator(mode="after")
    def _validate_provenance(self) -> Self:
        """Require canonical identities and complete selected provenance."""
        identifiers = (
            (self.source_id, "source"),
            (self.island_id, "island"),
            (self.association_id, "association"),
        )
        for identifier, field_name in identifiers:
            if _DOMAIN_IDENTIFIER.fullmatch(identifier) is None:
                raise ValueError(
                    f"{field_name} ID must be a domain identifier"
                )
        if self.scale_detection_ids != tuple(
            sorted(set(self.scale_detection_ids))
        ) or any(
            _DOMAIN_IDENTIFIER.fullmatch(identifier) is None
            for identifier in self.scale_detection_ids
        ):
            raise ValueError(
                "scale detection IDs must be canonical domain identifiers"
            )
        if self.selected_scale_detection_id not in self.scale_detection_ids:
            raise ValueError(
                "selected scale detection must belong to the provenance"
            )
        if (
            not self.contributing_scale_orders
            or self.contributing_scale_orders
            != tuple(sorted(set(self.contributing_scale_orders)))
            or any(order < 1 for order in self.contributing_scale_orders)
        ):
            raise ValueError(
                "contributing scale orders must be positive and canonical"
            )
        if self.support_pixel_count < 1:
            raise ValueError("provenance support pixel count must be positive")
        if not isfinite(self.visible_model_fraction) or not (
            0 < self.visible_model_fraction <= 1
        ):
            raise ValueError(
                "visible model fraction must be finite and in (0, 1]"
            )
        return self

SpectralModel

Bases: _CatalogueModel

Frequency convention for one reported source or component flux.

For log-polynomial, coefficient k multiplies log(frequency / reference_frequency) ** (k + 1) in natural-log flux space. An empty tuple explicitly means that no spectral fit was made.

Source code in src/hebog/data_models/catalogues.py

class SpectralModel(_CatalogueModel):
    """Frequency convention for one reported source or component flux.

    For ``log-polynomial``, coefficient ``k`` multiplies
    ``log(frequency / reference_frequency) ** (k + 1)`` in natural-log flux
    space. An empty tuple explicitly means that no spectral fit was made.
    """

    kind: Literal["reference-frequency-only", "log-polynomial"]
    reference_frequency_hz: float
    coefficients: tuple[float, ...]

    @model_validator(mode="after")
    def _validate_spectrum(self) -> Self:
        """Require one finite, internally consistent spectral convention."""
        if (
            not isfinite(self.reference_frequency_hz)
            or self.reference_frequency_hz <= 0
        ):
            raise ValueError("reference frequency must be finite and positive")
        if not all(isfinite(coefficient) for coefficient in self.coefficients):
            raise ValueError("spectral coefficients must be finite")
        if self.kind == "reference-frequency-only" and self.coefficients:
            raise ValueError(
                "reference-frequency-only model must not contain coefficients"
            )
        if self.kind == "log-polynomial" and not self.coefficients:
            raise ValueError("log-polynomial model requires coefficients")
        return self

SuppliedImageMetadata dataclass

Physical metadata a caller supplies for keywords a FITS header omits.

Each value fills only a missing header keyword: frequency in Hz for RESTFRQ or a frequency axis, restoring-beam full widths and position angle in degrees for BMAJ, BMIN and BPA, and the pixel brightness unit for BUNIT. Supplying a value the header already provides is an input error, so supplied metadata never overrides an image's own description.

SuppliedImageMetadata(reference_frequency_hz=144e6) SuppliedImageMetadata(reference_frequency_hz=144000000.0, beam_major_fwhm_degrees=None, beam_minor_fwhm_degrees=None, beam_position_angle_degrees=None, brightness_unit=None)

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class SuppliedImageMetadata:
    """Physical metadata a caller supplies for keywords a FITS header omits.

    Each value fills only a missing header keyword: frequency in Hz for
    ``RESTFRQ`` or a frequency axis, restoring-beam full widths and position
    angle in degrees for ``BMAJ``, ``BMIN`` and ``BPA``, and the pixel
    brightness unit for ``BUNIT``. Supplying a value the header already
    provides is an input error, so supplied metadata never overrides an
    image's own description.

    >>> SuppliedImageMetadata(reference_frequency_hz=144e6)
    SuppliedImageMetadata(reference_frequency_hz=144000000.0, \
beam_major_fwhm_degrees=None, beam_minor_fwhm_degrees=None, \
beam_position_angle_degrees=None, brightness_unit=None)
    """

    reference_frequency_hz: float | None = None
    beam_major_fwhm_degrees: float | None = None
    beam_minor_fwhm_degrees: float | None = None
    beam_position_angle_degrees: float | None = None
    brightness_unit: str | None = None

    def __post_init__(self) -> None:
        """Require at least one finite value with the header's own limits."""
        frequency = self.reference_frequency_hz
        axes = (self.beam_major_fwhm_degrees, self.beam_minor_fwhm_degrees)
        angle = self.beam_position_angle_degrees
        unit = self.brightness_unit
        if all(value is None for value in (frequency, *axes, angle, unit)):
            raise ValueError("supply at least one image metadata value")
        # Annotations do not bind callers; a number would fail on strip().
        if unit is not None and not isinstance(cast(object, unit), str):
            raise ValueError("supplied brightness unit must be text")
        if unit is not None and not unit.strip():
            raise ValueError("supplied brightness unit must not be blank")
        if frequency is not None and not (
            isfinite(frequency) and frequency > 0
        ):
            raise ValueError(
                "supplied reference frequency must be finite and positive"
            )
        if any(
            axis is not None and not (isfinite(axis) and axis > 0)
            for axis in axes
        ):
            raise ValueError("supplied beam axes must be finite and positive")
        if angle is not None and not isfinite(angle):
            raise ValueError("supplied beam position angle must be finite")
        major, minor = axes
        if major is not None and minor is not None and minor > major:
            raise ValueError("supplied beam minor axis cannot exceed major")

__post_init__()

Require at least one finite value with the header's own limits.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Require at least one finite value with the header's own limits."""
    frequency = self.reference_frequency_hz
    axes = (self.beam_major_fwhm_degrees, self.beam_minor_fwhm_degrees)
    angle = self.beam_position_angle_degrees
    unit = self.brightness_unit
    if all(value is None for value in (frequency, *axes, angle, unit)):
        raise ValueError("supply at least one image metadata value")
    # Annotations do not bind callers; a number would fail on strip().
    if unit is not None and not isinstance(cast(object, unit), str):
        raise ValueError("supplied brightness unit must be text")
    if unit is not None and not unit.strip():
        raise ValueError("supplied brightness unit must not be blank")
    if frequency is not None and not (
        isfinite(frequency) and frequency > 0
    ):
        raise ValueError(
            "supplied reference frequency must be finite and positive"
        )
    if any(
        axis is not None and not (isfinite(axis) and axis > 0)
        for axis in axes
    ):
        raise ValueError("supplied beam axes must be finite and positive")
    if angle is not None and not isfinite(angle):
        raise ValueError("supplied beam position angle must be finite")
    major, minor = axes
    if major is not None and minor is not None and minor > major:
        raise ValueError("supplied beam minor axis cannot exceed major")

TilePartition dataclass

One non-overlapping output core and its clipped read-only halo.

Source code in src/hebog/data_models/partitioning.py

@dataclass(frozen=True, slots=True)
class TilePartition:
    """One non-overlapping output core and its clipped read-only halo."""

    tile_y_index: int
    tile_x_index: int
    core_bounds: ImageBounds
    read_bounds: ImageBounds

    def __post_init__(self) -> None:
        """Require valid indices and a read region containing the core."""
        if min(self.tile_y_index, self.tile_x_index) < 0:
            raise ValueError("tile indices must be non-negative")
        if not self.read_bounds.contains(self.core_bounds):
            raise ValueError("tile read bounds must contain its core bounds")

    @property
    def tile_id(self) -> str:
        """Return a deterministic row-major identifier."""
        return f"tile-{self.tile_y_index:05d}-{self.tile_x_index:05d}"

    @property
    def core_slices_yx(self) -> tuple[slice, slice]:
        """Locate the owned core inside an array read through its halo."""
        return (
            slice(
                self.core_bounds.y_start - self.read_bounds.y_start,
                self.core_bounds.y_stop - self.read_bounds.y_start,
            ),
            slice(
                self.core_bounds.x_start - self.read_bounds.x_start,
                self.core_bounds.x_stop - self.read_bounds.x_start,
            ),
        )

core_slices_yx property

Locate the owned core inside an array read through its halo.

tile_id property

Return a deterministic row-major identifier.

__post_init__()

Require valid indices and a read region containing the core.

Source code in src/hebog/data_models/partitioning.py

def __post_init__(self) -> None:
    """Require valid indices and a read region containing the core."""
    if min(self.tile_y_index, self.tile_x_index) < 0:
        raise ValueError("tile indices must be non-negative")
    if not self.read_bounds.contains(self.core_bounds):
        raise ValueError("tile read bounds must contain its core bounds")

WideObjectCounts

Bases: BaseModel

How many objects each round decided from its cores, not a window.

publication_owners, support_components, islands and segments count objects whose window exceeded a task's read budget: the cores that hold such an object decide it, and the driver then holds its own pixels or seeds, the one term ADR-008 bounds by the object rather than the tile. deferred_fit_parents answers to a different bound: the fit stage defers a parent whose direct window exceeds the reviewed compact admission bound, whatever the read budget, and gathers its component records from the cores instead of reading it whole. Every count is zero unless an object exceeded its bound, which no image within the public envelope has been observed to hold.

Source code in src/hebog/data_models/source_finding.py

class WideObjectCounts(BaseModel):
    """How many objects each round decided from its cores, not a window.

    ``publication_owners``, ``support_components``, ``islands`` and
    ``segments`` count objects whose window exceeded a task's read budget:
    the cores that hold such an object decide it, and the driver then holds
    its own pixels or seeds, the one term ADR-008 bounds by the object
    rather than the tile. ``deferred_fit_parents`` answers to a different
    bound: the fit stage defers a parent whose direct window exceeds the
    reviewed compact admission bound, whatever the read budget, and gathers
    its component records from the cores instead of reading it whole. Every
    count is zero unless an object exceeded its bound, which no image within
    the public envelope has been observed to hold.
    """

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    publication_owners: int = 0
    support_components: int = 0
    deferred_fit_parents: int = 0
    islands: int = 0
    segments: int = 0

    @model_validator(mode="after")
    def _validate_counts(self) -> Self:
        """Reject a negative count."""
        if (
            min(
                self.publication_owners,
                self.support_components,
                self.deferred_fit_parents,
                self.islands,
                self.segments,
            )
            < 0
        ):
            raise ValueError("wide-object counts cannot be negative")
        return self

Executors

hebog.executors

Execution policies for coarse source-finding batches.

DaskExecutor

Submit coarse batches to an existing Dask client.

Hebog never creates, resizes or closes the client. Submission stays within the admitted in-flight bound and results are consumed in input order.

memory_resource and core_resource name Dask worker resources the caller declared, for example with dask worker --resources "memory=4e9,cores=4". Each named resource is attached to every task: its declared memory in bytes and its threads, one when it declares none. Dask then starts a task only on a worker with that much of each still free, so tasks are not co-located past the memory a worker declares. Named resources are also the admitted budget, because a worker's thread count and memory limit include what other work holds.

Source code in src/hebog/executors/dask.py

class DaskExecutor:
    """Submit coarse batches to an existing Dask client.

    Hebog never creates, resizes or closes the client. Submission stays
    within the admitted in-flight bound and results are consumed in input
    order.

    ``memory_resource`` and ``core_resource`` name Dask worker resources
    the caller declared, for example with ``dask worker --resources
    "memory=4e9,cores=4"``. Each named resource is attached to every task:
    its declared memory in bytes and its threads, one when it declares
    none. Dask then starts a task only on a worker with that much of each
    still free, so tasks are not co-located past the memory a worker
    declares. Named resources are also the admitted budget, because a
    worker's thread count and memory limit include what other work holds.
    """

    def __init__(
        self,
        client: Client,
        *,
        capacity: ExecutorCapacity | None = None,
        retry_limit: int = 0,
        memory_resource: str | None = None,
        core_resource: str | None = None,
    ) -> None:
        """Bind one caller-owned client, budget, retry policy and resources."""
        self._client = client
        self._declared_capacity = capacity
        if retry_limit < 0:
            raise ValueError("retry_limit must not be negative")
        self._retry_limit = retry_limit
        self._memory_resource = memory_resource
        self._core_resource = core_resource

    @property
    def client(self) -> Client:
        """Return the caller-owned client this executor submits to."""
        return self._client

    @property
    def capacity(self) -> ExecutorCapacity:
        """Return the declared budget, or the cluster's current one."""
        if self._declared_capacity is None:
            self._declared_capacity = self._cluster_capacity()
        return self._declared_capacity

    def _workers(self) -> list[dict[str, Any]]:
        """Return what the caller's cluster reports of each current worker."""
        information = cast(
            dict[str, Any],
            self._client.scheduler_info(),  # pyright: ignore[reportUnknownMemberType]
        )
        return list(
            cast(dict[str, Any], information.get("workers", {})).values()
        )

    def _cluster_capacity(self) -> ExecutorCapacity:
        """Read the budget the caller's cluster currently reports.

        Raises:
            ExecutorAdmissionError: If no worker declares every named
                resource, since Dask would hold Hebog's tasks forever.
        """
        named = [
            name
            for name in (self._memory_resource, self._core_resource)
            if name is not None
        ]
        workers = [
            worker
            for worker in self._workers()
            if all(name in worker.get("resources", {}) for name in named)
        ]
        if named and not workers:
            raise ExecutorAdmissionError(
                f"no worker declares the resources {', '.join(named)} "
                "that Hebog's tasks are told to request"
            )
        threads = [
            int(worker["resources"][self._core_resource])
            if self._core_resource is not None
            else int(worker.get("nthreads", 1))
            for worker in workers
        ]
        memory = [
            int(worker["resources"][self._memory_resource])
            if self._memory_resource is not None
            else int(worker["memory_limit"])
            for worker in workers
            if self._memory_resource is not None or worker.get("memory_limit")
        ]
        threads_per_worker = min(threads) if threads else 1
        total_threads = sum(threads) if threads else 1
        return ExecutorCapacity(
            worker_count=max(len(workers), 1),
            threads_per_worker=max(threads_per_worker, 1),
            maximum_tasks_in_flight=max(
                total_threads * _TASKS_IN_FLIGHT_PER_THREAD, 1
            ),
            memory_bytes_per_worker=min(memory) if memory else None,
        )

    def _task_resources(
        self, requirement: TaskRequirement | None
    ) -> dict[str, int] | None:
        """Return what one task asks of the named worker resources."""
        resources: dict[str, int] = {}
        if self._memory_resource is not None and requirement is not None:
            resources[self._memory_resource] = requirement.memory_bytes
        if self._core_resource is not None:
            resources[self._core_resource] = (
                1 if requirement is None else requirement.threads
            )
        return resources or None

    def _require_a_worker_for(self, resources: dict[str, int] | None) -> None:
        """Refuse a task no current worker declares enough resources for.

        Dask would hold such a task until a worker that could run it joins,
        which may never happen, so it fails before anything is submitted.

        Raises:
            ExecutorAdmissionError: If no worker declares at least the
                task's amount of every named resource.
        """
        if resources is None:
            return
        if not any(
            all(
                worker.get("resources", {}).get(name, 0) >= amount
                for name, amount in resources.items()
            )
            for worker in self._workers()
        ):
            requested = ", ".join(
                f"{amount} {name}" for name, amount in resources.items()
            )
            raise ExecutorAdmissionError(
                f"no worker declares the {requested} one task requests"
            )

    def _submit(
        self,
        function: Callable[..., Any],
        batch: Any,
        resources: dict[str, int] | None,
    ) -> Any:
        """Submit one task, letting Dask retry an idempotent failure."""
        return self._client.submit(  # pyright: ignore[reportUnknownMemberType]
            function,
            batch,
            pure=False,
            retries=self._retry_limit,
            resources=resources,
        )

    def map_batches(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        *,
        requirement: TaskRequirement | None = None,
    ) -> list[Output]:
        """Submit batches and gather their results in input order.

        No more tasks are in flight than the admitted bound, and the oldest
        is consumed before another is submitted.
        """
        self.capacity.admit(requirement)
        prepared = require_serializable_payloads(function, batches)
        limit = admitted_tasks_in_flight(self.capacity, requirement)
        resources = self._task_resources(requirement)
        self._require_a_worker_for(resources)
        in_flight: deque[Any] = deque()
        results: list[Output] = []
        submitted = 0
        try:
            while submitted < len(prepared) or in_flight:
                while len(in_flight) < limit and submitted < len(prepared):
                    in_flight.append(
                        self._submit(function, prepared[submitted], resources)
                    )
                    submitted += 1
                # The future stays in flight until it has settled, so an
                # exception that reaches the driver while it waits, as a
                # caller's deadline does, still waits for this task below.
                future = in_flight[0]
                wait([future])
                # Futures are consumed in input order, so the first failure
                # observed is the first failing batch.
                results.append(cast(Output, future.result()))
                in_flight.popleft()
            return results
        finally:
            # Nothing more of the plan is submitted, and the call returns only
            # once no task it submitted can still be running, so none writes
            # into storage its caller is already removing. The in-flight
            # bound limits that wait.
            _wait_for_submitted(in_flight)

capacity property

Return the declared budget, or the cluster's current one.

client property

Return the caller-owned client this executor submits to.

__init__(client, *, capacity=None, retry_limit=0, memory_resource=None, core_resource=None)

Bind one caller-owned client, budget, retry policy and resources.

Source code in src/hebog/executors/dask.py

def __init__(
    self,
    client: Client,
    *,
    capacity: ExecutorCapacity | None = None,
    retry_limit: int = 0,
    memory_resource: str | None = None,
    core_resource: str | None = None,
) -> None:
    """Bind one caller-owned client, budget, retry policy and resources."""
    self._client = client
    self._declared_capacity = capacity
    if retry_limit < 0:
        raise ValueError("retry_limit must not be negative")
    self._retry_limit = retry_limit
    self._memory_resource = memory_resource
    self._core_resource = core_resource

map_batches(function, batches, *, requirement=None)

Submit batches and gather their results in input order.

No more tasks are in flight than the admitted bound, and the oldest is consumed before another is submitted.

Source code in src/hebog/executors/dask.py

def map_batches(
    self,
    function: Callable[[Input], Output],
    batches: Iterable[Input],
    *,
    requirement: TaskRequirement | None = None,
) -> list[Output]:
    """Submit batches and gather their results in input order.

    No more tasks are in flight than the admitted bound, and the oldest
    is consumed before another is submitted.
    """
    self.capacity.admit(requirement)
    prepared = require_serializable_payloads(function, batches)
    limit = admitted_tasks_in_flight(self.capacity, requirement)
    resources = self._task_resources(requirement)
    self._require_a_worker_for(resources)
    in_flight: deque[Any] = deque()
    results: list[Output] = []
    submitted = 0
    try:
        while submitted < len(prepared) or in_flight:
            while len(in_flight) < limit and submitted < len(prepared):
                in_flight.append(
                    self._submit(function, prepared[submitted], resources)
                )
                submitted += 1
            # The future stays in flight until it has settled, so an
            # exception that reaches the driver while it waits, as a
            # caller's deadline does, still waits for this task below.
            future = in_flight[0]
            wait([future])
            # Futures are consumed in input order, so the first failure
            # observed is the first failing batch.
            results.append(cast(Output, future.result()))
            in_flight.popleft()
        return results
    finally:
        # Nothing more of the plan is submitted, and the call returns only
        # once no task it submitted can still be running, so none writes
        # into storage its caller is already removing. The in-flight
        # bound limits that wait.
        _wait_for_submitted(in_flight)

Executor

Bases: Protocol

Execute coarse batches without exposing scheduler-specific objects.

Source code in src/hebog/executors/base.py

class Executor(Protocol):
    """Execute coarse batches without exposing scheduler-specific objects."""

    @property
    def capacity(self) -> ExecutorCapacity:
        """Return the execution budget admitted by the caller."""
        ...

    def map_batches(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        *,
        requirement: TaskRequirement | None = None,
    ) -> list[Output]:
        """Apply ``function`` to each batch and return ordered results."""
        ...

capacity property

Return the execution budget admitted by the caller.

map_batches(function, batches, *, requirement=None)

Apply function to each batch and return ordered results.

Source code in src/hebog/executors/base.py

def map_batches(
    self,
    function: Callable[[Input], Output],
    batches: Iterable[Input],
    *,
    requirement: TaskRequirement | None = None,
) -> list[Output]:
    """Apply ``function`` to each batch and return ordered results."""
    ...

ExecutorAdmissionError

Bases: ValueError

One task declared more resource than the caller admitted.

Source code in src/hebog/executors/base.py

class ExecutorAdmissionError(ValueError):
    """One task declared more resource than the caller admitted."""

ExecutorCapacity dataclass

The execution budget the caller admitted for one source finder.

memory_bytes_per_worker is the working set one task may hold. It is None when the caller has declared no budget, in which case memory admission is not enforced and a stage needing a budget must say so.

Source code in src/hebog/executors/base.py

@dataclass(frozen=True, slots=True)
class ExecutorCapacity:
    """The execution budget the caller admitted for one source finder.

    ``memory_bytes_per_worker`` is the working set one task may hold. It is
    ``None`` when the caller has declared no budget, in which case memory
    admission is not enforced and a stage needing a budget must say so.
    """

    worker_count: int
    threads_per_worker: int
    maximum_tasks_in_flight: int
    memory_bytes_per_worker: int | None = None

    def __post_init__(self) -> None:
        """Require a positive, bounded and explicit execution budget."""
        _positive_count(self.worker_count, "worker_count")
        _positive_count(self.threads_per_worker, "threads_per_worker")
        _positive_count(
            self.maximum_tasks_in_flight, "maximum_tasks_in_flight"
        )
        if self.memory_bytes_per_worker is not None:
            _positive_count(
                self.memory_bytes_per_worker, "memory_bytes_per_worker"
            )

    def admit(self, requirement: TaskRequirement | None) -> None:
        """Raise before submission when one task exceeds the budget."""
        if requirement is None:
            return
        admitted = self.memory_bytes_per_worker
        if admitted is not None and requirement.memory_bytes > admitted:
            raise ExecutorAdmissionError(
                f"task requires {requirement.memory_bytes} bytes of memory "
                f"but {admitted} bytes are admitted per worker"
            )
        if requirement.threads > self.threads_per_worker:
            raise ExecutorAdmissionError(
                f"task requires {requirement.threads} threads but "
                f"{self.threads_per_worker} threads are admitted per worker"
            )

__post_init__()

Require a positive, bounded and explicit execution budget.

Source code in src/hebog/executors/base.py

def __post_init__(self) -> None:
    """Require a positive, bounded and explicit execution budget."""
    _positive_count(self.worker_count, "worker_count")
    _positive_count(self.threads_per_worker, "threads_per_worker")
    _positive_count(
        self.maximum_tasks_in_flight, "maximum_tasks_in_flight"
    )
    if self.memory_bytes_per_worker is not None:
        _positive_count(
            self.memory_bytes_per_worker, "memory_bytes_per_worker"
        )

admit(requirement)

Raise before submission when one task exceeds the budget.

Source code in src/hebog/executors/base.py

def admit(self, requirement: TaskRequirement | None) -> None:
    """Raise before submission when one task exceeds the budget."""
    if requirement is None:
        return
    admitted = self.memory_bytes_per_worker
    if admitted is not None and requirement.memory_bytes > admitted:
        raise ExecutorAdmissionError(
            f"task requires {requirement.memory_bytes} bytes of memory "
            f"but {admitted} bytes are admitted per worker"
        )
    if requirement.threads > self.threads_per_worker:
        raise ExecutorAdmissionError(
            f"task requires {requirement.threads} threads but "
            f"{self.threads_per_worker} threads are admitted per worker"
        )

ExecutorPayloadError

Bases: ValueError

One task payload cannot cross a worker boundary.

Source code in src/hebog/executors/base.py

class ExecutorPayloadError(ValueError):
    """One task payload cannot cross a worker boundary."""

SerialExecutor

Run batches serially for testing and scientific reference results.

This executor defines the behaviour every other policy must reproduce: input ordering, payload validation, retry of idempotent tasks and the first failing batch by index.

Source code in src/hebog/executors/serial.py

class SerialExecutor:
    """Run batches serially for testing and scientific reference results.

    This executor defines the behaviour every other policy must reproduce:
    input ordering, payload validation, retry of idempotent tasks and the
    first failing batch by index.
    """

    def __init__(
        self,
        *,
        capacity: ExecutorCapacity | None = None,
        retry_limit: int = 0,
    ) -> None:
        """Bind the caller's admitted budget and retry policy."""
        self._capacity = capacity or _SERIAL_CAPACITY
        if retry_limit < 0:
            raise ValueError("retry_limit must not be negative")
        self._retry_limit = retry_limit

    @property
    def capacity(self) -> ExecutorCapacity:
        """Return the execution budget admitted by the caller."""
        return self._capacity

    def _results(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        requirement: TaskRequirement | None,
    ) -> Iterator[Output]:
        """Yield admitted, validated results in input order."""
        self._capacity.admit(requirement)
        for batch in require_serializable_payloads(function, batches):
            yield call_with_retries(
                function, batch, retry_limit=self._retry_limit
            )

    def map_batches(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        *,
        requirement: TaskRequirement | None = None,
    ) -> list[Output]:
        """Apply ``function`` to each batch in input order."""
        return list(self._results(function, batches, requirement))

capacity property

Return the execution budget admitted by the caller.

__init__(*, capacity=None, retry_limit=0)

Bind the caller's admitted budget and retry policy.

Source code in src/hebog/executors/serial.py

def __init__(
    self,
    *,
    capacity: ExecutorCapacity | None = None,
    retry_limit: int = 0,
) -> None:
    """Bind the caller's admitted budget and retry policy."""
    self._capacity = capacity or _SERIAL_CAPACITY
    if retry_limit < 0:
        raise ValueError("retry_limit must not be negative")
    self._retry_limit = retry_limit

map_batches(function, batches, *, requirement=None)

Apply function to each batch in input order.

Source code in src/hebog/executors/serial.py

def map_batches(
    self,
    function: Callable[[Input], Output],
    batches: Iterable[Input],
    *,
    requirement: TaskRequirement | None = None,
) -> list[Output]:
    """Apply ``function`` to each batch in input order."""
    return list(self._results(function, batches, requirement))

TaskRequirement dataclass

One task's declared peak working set and thread requirement.

Source code in src/hebog/executors/base.py

@dataclass(frozen=True, slots=True)
class TaskRequirement:
    """One task's declared peak working set and thread requirement."""

    memory_bytes: int
    threads: int = 1

    def __post_init__(self) -> None:
        """Require a positive declared working set and thread count."""
        _positive_count(self.memory_bytes, "memory_bytes")
        _positive_count(self.threads, "threads")

__post_init__()

Require a positive declared working set and thread count.

Source code in src/hebog/executors/base.py

def __post_init__(self) -> None:
    """Require a positive declared working set and thread count."""
    _positive_count(self.memory_bytes, "memory_bytes")
    _positive_count(self.threads, "threads")

ThreadExecutor

Run bounded batches on one persistent caller-owned thread pool.

The pool exists for the lifetime of this executor, so tasks never create a nested pool and no stage starts threads of its own. Batches are submitted in input order within a bounded window and consumed in that same order, which keeps results and the propagated failure identical to the serial reference.

Source code in src/hebog/executors/threads.py

class ThreadExecutor:
    """Run bounded batches on one persistent caller-owned thread pool.

    The pool exists for the lifetime of this executor, so tasks never create
    a nested pool and no stage starts threads of its own. Batches are
    submitted in input order within a bounded window and consumed in that
    same order, which keeps results and the propagated failure identical to
    the serial reference.
    """

    def __init__(
        self,
        thread_count: int,
        *,
        memory_bytes_per_worker: int | None = None,
        retry_limit: int = 0,
    ) -> None:
        """Create one pool the caller owns for the whole run."""
        self._capacity = ExecutorCapacity(
            worker_count=1,
            threads_per_worker=thread_count,
            maximum_tasks_in_flight=thread_count,
            memory_bytes_per_worker=memory_bytes_per_worker,
        )
        if retry_limit < 0:
            raise ValueError("retry_limit must not be negative")
        self._retry_limit = retry_limit
        self._pool = ThreadPoolExecutor(
            max_workers=self._capacity.threads_per_worker,
            thread_name_prefix="hebog-executor",
        )

    def __enter__(self) -> ThreadExecutor:
        """Return this executor for use inside a caller-owned scope."""
        return self

    def __exit__(
        self,
        exception_type: type[BaseException] | None,
        exception: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        """Release the pool the caller opened."""
        del exception_type, exception, traceback
        self.close()

    def close(self) -> None:
        """Release the pool and abandon work that has not started."""
        self._pool.shutdown(wait=True, cancel_futures=True)

    @property
    def capacity(self) -> ExecutorCapacity:
        """Return the execution budget admitted by the caller."""
        return self._capacity

    def _results(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        requirement: TaskRequirement | None,
    ) -> Generator[Output]:
        """Yield results in input order from a bounded submission window."""
        self._capacity.admit(requirement)
        prepared = require_serializable_payloads(function, batches)
        limit = admitted_tasks_in_flight(self._capacity, requirement)
        guard = _CallGuard()
        pending: deque[Future[Output]] = deque()
        submitted = 0
        try:
            while True:
                while len(pending) < limit and submitted < len(prepared):
                    pending.append(
                        self._pool.submit(
                            guard.run,
                            function,
                            prepared[submitted],
                            self._retry_limit,
                        )
                    )
                    submitted += 1
                if not pending:
                    return
                yield pending.popleft().result()
        finally:
            # A failure or the caller's deadline cancels the batches that
            # have not started, rather than running work whose result nobody
            # will read, and waits for those that have: a task still running
            # after the call returns could write into storage its caller is
            # already removing.
            for future in pending:
                future.cancel()
            guard.stop()

    def map_batches(
        self,
        function: Callable[[Input], Output],
        batches: Iterable[Input],
        *,
        requirement: TaskRequirement | None = None,
    ) -> list[Output]:
        """Apply ``function`` to each batch and return ordered results."""
        return list(self._results(function, batches, requirement))

capacity property

Return the execution budget admitted by the caller.

__enter__()

Return this executor for use inside a caller-owned scope.

Source code in src/hebog/executors/threads.py

def __enter__(self) -> ThreadExecutor:
    """Return this executor for use inside a caller-owned scope."""
    return self

__exit__(exception_type, exception, traceback)

Release the pool the caller opened.

Source code in src/hebog/executors/threads.py

def __exit__(
    self,
    exception_type: type[BaseException] | None,
    exception: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    """Release the pool the caller opened."""
    del exception_type, exception, traceback
    self.close()

__init__(thread_count, *, memory_bytes_per_worker=None, retry_limit=0)

Create one pool the caller owns for the whole run.

Source code in src/hebog/executors/threads.py

def __init__(
    self,
    thread_count: int,
    *,
    memory_bytes_per_worker: int | None = None,
    retry_limit: int = 0,
) -> None:
    """Create one pool the caller owns for the whole run."""
    self._capacity = ExecutorCapacity(
        worker_count=1,
        threads_per_worker=thread_count,
        maximum_tasks_in_flight=thread_count,
        memory_bytes_per_worker=memory_bytes_per_worker,
    )
    if retry_limit < 0:
        raise ValueError("retry_limit must not be negative")
    self._retry_limit = retry_limit
    self._pool = ThreadPoolExecutor(
        max_workers=self._capacity.threads_per_worker,
        thread_name_prefix="hebog-executor",
    )

close()

Release the pool and abandon work that has not started.

Source code in src/hebog/executors/threads.py

def close(self) -> None:
    """Release the pool and abandon work that has not started."""
    self._pool.shutdown(wait=True, cancel_futures=True)

map_batches(function, batches, *, requirement=None)

Apply function to each batch and return ordered results.

Source code in src/hebog/executors/threads.py

def map_batches(
    self,
    function: Callable[[Input], Output],
    batches: Iterable[Input],
    *,
    requirement: TaskRequirement | None = None,
) -> list[Output]:
    """Apply ``function`` to each batch and return ordered results."""
    return list(self._results(function, batches, requirement))

__getattr__(name)

Load the optional concrete Dask executor only when requested.

Source code in src/hebog/executors/__init__.py

def __getattr__(name: str) -> object:
    """Load the optional concrete Dask executor only when requested."""
    if name == "DaskExecutor":
        from hebog.executors.dask import (  # noqa: PLC0415
            DaskExecutor,
        )

        return DaskExecutor
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

Reading products and images

hebog.io

Image and product input/output boundaries.

FitsImageSource

Read validated logical image planes through bounded FITS sections.

Optional supplied metadata fills keywords the header omits. It is part of the source, so every executor task that re-reads the header sees it.

Source code in src/hebog/io/fits.py

class FitsImageSource:
    """Read validated logical image planes through bounded FITS sections.

    Optional supplied metadata fills keywords the header omits. It is part of
    the source, so every executor task that re-reads the header sees it.
    """

    def __init__(
        self,
        path: Path,
        supplied_metadata: SuppliedImageMetadata | None = None,
    ) -> None:
        """Retain the path and supplied metadata; open files only on use."""
        self._path = path
        self._supplied_metadata = supplied_metadata
        self._metadata: ImageMetadata | None = None
        self._open_files: dict[int, Any] = {}
        self._open_files_lock = threading.Lock()
        # A caller that simply drops a source, as scripts and workers do,
        # would otherwise leave its files to the garbage collector. Dask
        # also keeps sources in reference cycles, whose objects the
        # collector finalizes in no defined order; it runs weakref callbacks
        # before any finalizer, so the files close before they could be
        # finalized and reported unclosed.
        weakref.finalize(
            self, _close_open_files, self._open_files, self._open_files_lock
        )

    def __getstate__(self) -> dict[str, Any]:
        """Serialize the request to read a file, never what it once held.

        A worker opens the file itself, so neither an open file nor
        validated metadata travels with the source: an open file cannot
        cross a process, and metadata could go stale in transit.
        """
        return {
            "_path": self._path,
            "_supplied_metadata": self._supplied_metadata,
        }

    def __setstate__(self, state: dict[str, Any]) -> None:
        """Restore a source that has not yet read its file."""
        self.__init__(  # pyright: ignore[reportUnknownMemberType]
            state["_path"], state["_supplied_metadata"]
        )

    def _primary_hdu(self) -> Any:
        """Return this file's primary HDU, opening it once per thread.

        Opening a FITS file and parsing its header costs several times a
        bounded window read, and tiled stages read hundreds of windows per
        image. Each thread keeps its own open file, because worker threads
        share a source and a file cursor cannot be shared.

        Raises:
            InvalidFitsImageError: If the file cannot be opened.
        """
        thread = threading.get_ident()
        hdus = self._open_files.get(thread)
        if hdus is None:
            try:
                # Stored values are read as they are and scaled window by
                # window in ``read_windows``. Astropy cannot map pixels it
                # scales itself, scales 16-bit ones in single precision and
                # skips a ``BLANK`` of zero.
                # Astropy leaves a file it opened itself to the collector
                # when it refuses the header, so the source opens the file
                # and closes it on that failure; the HDU list closes it
                # otherwise.
                file = self._path.open("rb")
                try:
                    hdus = fits.open(
                        file,
                        mode="readonly",
                        memmap=True,
                        do_not_scale_image_data=True,
                    )
                except BaseException:
                    file.close()
                    raise
            # Astropy raises whatever a malformed BITPIX or NAXIS card
            # first breaks, not one error for a file it cannot open.
            except (KeyError, OSError, TypeError, ValueError) as error:
                raise InvalidFitsImageError(
                    f"cannot read FITS image {self._path}: {error}"
                ) from error
            with self._open_files_lock:
                self._open_files[thread] = hdus
        return hdus[0]

    def close(self) -> None:
        """Release every file this source holds open.

        Worker threads each open the file, so one close frees them all. The
        source stays usable and opens the file again when it is next read.
        """
        _close_open_files(self._open_files, self._open_files_lock)

    def header(self) -> fits.Header:
        """Return the primary header as the finder reads it.

        Each float holds the number Astropy parsed, written so that wcslib
        reads the same one. A WCS built from this header is therefore the
        transform that :meth:`metadata` validates, which one built from the
        file's own header text need not be. The header is the caller's own
        copy.

        Raises:
            InvalidFitsImageError: If the file cannot be opened, or a
                numeric WCS card is not a number.
        """
        return _wcs_header(self._primary_hdu().header, self._path).copy()

    def metadata(self) -> ImageMetadata:
        """Return shape and unit without materialising the image plane.

        A source validates one file once: tiled stages read hundreds of
        windows, and re-parsing the header and rebuilding both WCS objects
        for each of them costs more than reading the pixels.
        """
        if self._metadata is None:
            self._metadata = _metadata(
                self._primary_hdu(), self._path, self._supplied_metadata
            )
        return self._metadata

    def read_window(self, bounds: ImageBounds) -> ImageWindow:
        """Read one half-open global window into owned read-only arrays."""
        return self.read_windows((bounds,))[0]

    def read_windows(
        self,
        bounds_collection: Iterable[ImageBounds],
    ) -> tuple[ImageWindow, ...]:
        """Read bounded windows through one validated FITS open.

        A pixel is valid under the rule of :mod:`hebog.io.pixel_validity`:
        finite, and in no 3x3 square of one repeated value. The rule looks
        at the pixels up to two away, so each window is read two pixels
        wider on every side inside the image, and only the window is kept.
        """
        requested_bounds = tuple(bounds_collection)
        if not requested_bounds:
            return ()
        windows: list[ImageWindow] = []
        metadata = self.metadata()
        primary_hdu = self._primary_hdu()
        scale, zero, blank = _pixel_scaling(primary_hdu.header, self._path)
        leading_indices = (0,) * (len(primary_hdu.shape) - 2)
        for bounds in requested_bounds:
            bounds.require_inside(metadata.shape_yx)
            read = validity_read_bounds(bounds, metadata.shape_yx)
            stored = primary_hdu.section[
                (
                    *leading_indices,
                    slice(read.y_start, read.y_stop),
                    slice(read.x_start, read.x_stop),
                )
            ]
            read_values = np.array(stored, dtype=np.float64, copy=True)
            if (scale, zero) != (1.0, 0.0):
                read_values = zero + scale * read_values
            if blank is not None:
                read_values[stored == blank] = np.nan
            del stored
            valid_pixels = valid_input_pixels(
                read_values, bounds, metadata.shape_yx
            )
            y_offset = bounds.y_start - read.y_start
            x_offset = bounds.x_start - read.x_start
            height, width = bounds.shape_yx
            # The read is this call's own array, so a window that is the
            # whole read, or whole rows of it, is kept without a copy; any
            # other is copied out so it does not hold the margin's memory.
            values = np.ascontiguousarray(
                read_values[
                    y_offset : y_offset + height,
                    x_offset : x_offset + width,
                ]
            )
            del read_values
            values.setflags(write=False)
            valid_pixels.setflags(write=False)
            windows.append(
                ImageWindow(
                    bounds=bounds,
                    values=values,
                    valid_pixels=valid_pixels,
                )
            )
        return tuple(windows)

__getstate__()

Serialize the request to read a file, never what it once held.

A worker opens the file itself, so neither an open file nor validated metadata travels with the source: an open file cannot cross a process, and metadata could go stale in transit.

Source code in src/hebog/io/fits.py

def __getstate__(self) -> dict[str, Any]:
    """Serialize the request to read a file, never what it once held.

    A worker opens the file itself, so neither an open file nor
    validated metadata travels with the source: an open file cannot
    cross a process, and metadata could go stale in transit.
    """
    return {
        "_path": self._path,
        "_supplied_metadata": self._supplied_metadata,
    }

__init__(path, supplied_metadata=None)

Retain the path and supplied metadata; open files only on use.

Source code in src/hebog/io/fits.py

def __init__(
    self,
    path: Path,
    supplied_metadata: SuppliedImageMetadata | None = None,
) -> None:
    """Retain the path and supplied metadata; open files only on use."""
    self._path = path
    self._supplied_metadata = supplied_metadata
    self._metadata: ImageMetadata | None = None
    self._open_files: dict[int, Any] = {}
    self._open_files_lock = threading.Lock()
    # A caller that simply drops a source, as scripts and workers do,
    # would otherwise leave its files to the garbage collector. Dask
    # also keeps sources in reference cycles, whose objects the
    # collector finalizes in no defined order; it runs weakref callbacks
    # before any finalizer, so the files close before they could be
    # finalized and reported unclosed.
    weakref.finalize(
        self, _close_open_files, self._open_files, self._open_files_lock
    )

__setstate__(state)

Restore a source that has not yet read its file.

Source code in src/hebog/io/fits.py

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore a source that has not yet read its file."""
    self.__init__(  # pyright: ignore[reportUnknownMemberType]
        state["_path"], state["_supplied_metadata"]
    )

close()

Release every file this source holds open.

Worker threads each open the file, so one close frees them all. The source stays usable and opens the file again when it is next read.

Source code in src/hebog/io/fits.py

def close(self) -> None:
    """Release every file this source holds open.

    Worker threads each open the file, so one close frees them all. The
    source stays usable and opens the file again when it is next read.
    """
    _close_open_files(self._open_files, self._open_files_lock)

header()

Return the primary header as the finder reads it.

Each float holds the number Astropy parsed, written so that wcslib reads the same one. A WCS built from this header is therefore the transform that :meth:metadata validates, which one built from the file's own header text need not be. The header is the caller's own copy.

Raises:

Type	Description
InvalidFitsImageError	If the file cannot be opened, or a numeric WCS card is not a number.

Source code in src/hebog/io/fits.py

def header(self) -> fits.Header:
    """Return the primary header as the finder reads it.

    Each float holds the number Astropy parsed, written so that wcslib
    reads the same one. A WCS built from this header is therefore the
    transform that :meth:`metadata` validates, which one built from the
    file's own header text need not be. The header is the caller's own
    copy.

    Raises:
        InvalidFitsImageError: If the file cannot be opened, or a
            numeric WCS card is not a number.
    """
    return _wcs_header(self._primary_hdu().header, self._path).copy()

metadata()

Return shape and unit without materialising the image plane.

A source validates one file once: tiled stages read hundreds of windows, and re-parsing the header and rebuilding both WCS objects for each of them costs more than reading the pixels.

Source code in src/hebog/io/fits.py

def metadata(self) -> ImageMetadata:
    """Return shape and unit without materialising the image plane.

    A source validates one file once: tiled stages read hundreds of
    windows, and re-parsing the header and rebuilding both WCS objects
    for each of them costs more than reading the pixels.
    """
    if self._metadata is None:
        self._metadata = _metadata(
            self._primary_hdu(), self._path, self._supplied_metadata
        )
    return self._metadata

read_window(bounds)

Read one half-open global window into owned read-only arrays.

Source code in src/hebog/io/fits.py

def read_window(self, bounds: ImageBounds) -> ImageWindow:
    """Read one half-open global window into owned read-only arrays."""
    return self.read_windows((bounds,))[0]

read_windows(bounds_collection)

Read bounded windows through one validated FITS open.

A pixel is valid under the rule of :mod:hebog.io.pixel_validity: finite, and in no 3x3 square of one repeated value. The rule looks at the pixels up to two away, so each window is read two pixels wider on every side inside the image, and only the window is kept.

Source code in src/hebog/io/fits.py

def read_windows(
    self,
    bounds_collection: Iterable[ImageBounds],
) -> tuple[ImageWindow, ...]:
    """Read bounded windows through one validated FITS open.

    A pixel is valid under the rule of :mod:`hebog.io.pixel_validity`:
    finite, and in no 3x3 square of one repeated value. The rule looks
    at the pixels up to two away, so each window is read two pixels
    wider on every side inside the image, and only the window is kept.
    """
    requested_bounds = tuple(bounds_collection)
    if not requested_bounds:
        return ()
    windows: list[ImageWindow] = []
    metadata = self.metadata()
    primary_hdu = self._primary_hdu()
    scale, zero, blank = _pixel_scaling(primary_hdu.header, self._path)
    leading_indices = (0,) * (len(primary_hdu.shape) - 2)
    for bounds in requested_bounds:
        bounds.require_inside(metadata.shape_yx)
        read = validity_read_bounds(bounds, metadata.shape_yx)
        stored = primary_hdu.section[
            (
                *leading_indices,
                slice(read.y_start, read.y_stop),
                slice(read.x_start, read.x_stop),
            )
        ]
        read_values = np.array(stored, dtype=np.float64, copy=True)
        if (scale, zero) != (1.0, 0.0):
            read_values = zero + scale * read_values
        if blank is not None:
            read_values[stored == blank] = np.nan
        del stored
        valid_pixels = valid_input_pixels(
            read_values, bounds, metadata.shape_yx
        )
        y_offset = bounds.y_start - read.y_start
        x_offset = bounds.x_start - read.x_start
        height, width = bounds.shape_yx
        # The read is this call's own array, so a window that is the
        # whole read, or whole rows of it, is kept without a copy; any
        # other is copied out so it does not hold the margin's memory.
        values = np.ascontiguousarray(
            read_values[
                y_offset : y_offset + height,
                x_offset : x_offset + width,
            ]
        )
        del read_values
        values.setflags(write=False)
        valid_pixels.setflags(write=False)
        windows.append(
            ImageWindow(
                bounds=bounds,
                values=values,
                valid_pixels=valid_pixels,
            )
        )
    return tuple(windows)

FitsProductImageSource

Read a checksum-verified RMS or mask product through bounded windows.

Source code in src/hebog/io/materialization.py

class FitsProductImageSource:
    """Read a checksum-verified RMS or mask product through bounded windows."""

    def __init__(self, product: MaterializedProduct) -> None:
        """Require a supported image-product role without opening its file."""
        if product.product_role not in _IMAGE_ROLES:
            raise UnsupportedMaterializedProductError(
                f"materialized role is not a supported image product: "
                f"{product.product_role}"
            )
        self._product = product
        self._role: Literal["rms", "source-filtering-mask"] = cast(
            Literal["rms", "source-filtering-mask"],
            product.product_role,
        )
        self._identity_validated = False

    def _validate(self) -> None:
        """Verify byte identity and product-level header fields."""
        if self._identity_validated:
            return
        _resolve_product_path(self._product, expected_role=self._role)
        status = _image_header_status(self._product.path, self._role)
        if status != self._product.scientific_status:
            raise InvalidMaterializedProductError(
                "image scientific status does not match its "
                "materialized record"
            )
        self._identity_validated = True

    def scientific_status(self) -> Literal["valid", "unavailable"]:
        """Return the verified scientific availability of this product."""
        self._validate()
        return self._product.scientific_status

    def metadata(self) -> ImageMetadata:
        """Return verified image metadata without materialising the plane."""
        self._validate()
        try:
            return FitsImageSource(self._product.path).metadata()
        except InvalidFitsImageError as error:
            raise InvalidMaterializedProductError(
                f"cannot read image product metadata: {error}"
            ) from error

    def read_window(self, bounds: ImageBounds) -> ImageWindow:
        """Read and validate one bounded product window.

        A product pixel is valid where it is finite. The input image's
        block rule does not apply: a mask, or a constant noise estimate, is
        a product's value, not an unobserved region.
        """
        self._validate()
        try:
            window = FitsImageSource(self._product.path).read_window(bounds)
        except InvalidFitsImageError as error:
            raise InvalidMaterializedProductError(
                f"cannot read image product window: {error}"
            ) from error
        if self._role == "rms":
            finite = np.isfinite(window.values)
            if np.isinf(window.values).any():
                raise InvalidMaterializedProductError(
                    "RMS image contains infinite values"
                )
            if np.any(window.values[finite] < 0):
                raise InvalidMaterializedProductError(
                    "RMS image contains negative values"
                )
            if (
                self._product.scientific_status == "unavailable"
                and finite.any()
            ):
                raise InvalidMaterializedProductError(
                    "unavailable RMS image must contain all NaN pixels"
                )
        elif not np.all(np.isin(window.values, (0, 1))):
            raise InvalidMaterializedProductError(
                "source-filtering mask image must be binary"
            )
        valid_pixels = np.isfinite(window.values)
        valid_pixels.setflags(write=False)
        return replace(window, valid_pixels=valid_pixels)

__init__(product)

Require a supported image-product role without opening its file.

Source code in src/hebog/io/materialization.py

def __init__(self, product: MaterializedProduct) -> None:
    """Require a supported image-product role without opening its file."""
    if product.product_role not in _IMAGE_ROLES:
        raise UnsupportedMaterializedProductError(
            f"materialized role is not a supported image product: "
            f"{product.product_role}"
        )
    self._product = product
    self._role: Literal["rms", "source-filtering-mask"] = cast(
        Literal["rms", "source-filtering-mask"],
        product.product_role,
    )
    self._identity_validated = False

metadata()

Return verified image metadata without materialising the plane.

Source code in src/hebog/io/materialization.py

def metadata(self) -> ImageMetadata:
    """Return verified image metadata without materialising the plane."""
    self._validate()
    try:
        return FitsImageSource(self._product.path).metadata()
    except InvalidFitsImageError as error:
        raise InvalidMaterializedProductError(
            f"cannot read image product metadata: {error}"
        ) from error

read_window(bounds)

Read and validate one bounded product window.

A product pixel is valid where it is finite. The input image's block rule does not apply: a mask, or a constant noise estimate, is a product's value, not an unobserved region.

Source code in src/hebog/io/materialization.py

def read_window(self, bounds: ImageBounds) -> ImageWindow:
    """Read and validate one bounded product window.

    A product pixel is valid where it is finite. The input image's
    block rule does not apply: a mask, or a constant noise estimate, is
    a product's value, not an unobserved region.
    """
    self._validate()
    try:
        window = FitsImageSource(self._product.path).read_window(bounds)
    except InvalidFitsImageError as error:
        raise InvalidMaterializedProductError(
            f"cannot read image product window: {error}"
        ) from error
    if self._role == "rms":
        finite = np.isfinite(window.values)
        if np.isinf(window.values).any():
            raise InvalidMaterializedProductError(
                "RMS image contains infinite values"
            )
        if np.any(window.values[finite] < 0):
            raise InvalidMaterializedProductError(
                "RMS image contains negative values"
            )
        if (
            self._product.scientific_status == "unavailable"
            and finite.any()
        ):
            raise InvalidMaterializedProductError(
                "unavailable RMS image must contain all NaN pixels"
            )
    elif not np.all(np.isin(window.values, (0, 1))):
        raise InvalidMaterializedProductError(
            "source-filtering mask image must be binary"
        )
    valid_pixels = np.isfinite(window.values)
    valid_pixels.setflags(write=False)
    return replace(window, valid_pixels=valid_pixels)

scientific_status()

Return the verified scientific availability of this product.

Source code in src/hebog/io/materialization.py

def scientific_status(self) -> Literal["valid", "unavailable"]:
    """Return the verified scientific availability of this product."""
    self._validate()
    return self._product.scientific_status

ImageBounds dataclass

Half-open global pixel bounds in NumPy (y, x) axis order.

Source code in src/hebog/data_models/partitioning.py

@dataclass(frozen=True, slots=True)
class ImageBounds:
    """Half-open global pixel bounds in NumPy ``(y, x)`` axis order."""

    y_start: int
    y_stop: int
    x_start: int
    x_stop: int

    def __post_init__(self) -> None:
        """Require non-negative, non-empty half-open bounds."""
        if min(self.y_start, self.y_stop, self.x_start, self.x_stop) < 0:
            raise ValueError("image bounds must be non-negative")
        if self.y_stop <= self.y_start or self.x_stop <= self.x_start:
            raise ValueError("image bounds must be non-empty")

    @property
    def shape_yx(self) -> tuple[int, int]:
        """Return the bounded array shape in NumPy axis order."""
        return (
            self.y_stop - self.y_start,
            self.x_stop - self.x_start,
        )

    @property
    def center_xy(self) -> tuple[float, float]:
        """Return the centre of the bounded pixels, in global ``(x, y)``.

        Local geometry is derived here rather than at a corner, and every
        caller that derives it must agree, so the definition lives with the
        bounds instead of beside each use.
        """
        return (
            (self.x_start + self.x_stop - 1) / 2,
            (self.y_start + self.y_stop - 1) / 2,
        )

    def require_inside(self, shape_yx: tuple[int, int]) -> None:
        """Reject bounds extending beyond one logical image plane."""
        if self.y_stop > shape_yx[0] or self.x_stop > shape_yx[1]:
            raise ValueError(
                f"image bounds must stay inside image shape {shape_yx}"
            )

    def contains(self, other: ImageBounds) -> bool:
        """Return whether another region lies completely within the bounds."""
        return (
            self.y_start <= other.y_start
            and other.y_stop <= self.y_stop
            and self.x_start <= other.x_start
            and other.x_stop <= self.x_stop
        )

    def expanded(
        self,
        margin_pixels: int,
        image_shape_yx: tuple[int, int],
    ) -> ImageBounds:
        """Return a uniformly expanded region clipped to the image plane."""
        if (
            isinstance(margin_pixels, bool)
            or not isinstance(  # pyright: ignore[reportUnnecessaryIsInstance]
                margin_pixels,
                int,
            )
            or margin_pixels < 0
        ):
            raise ValueError("bounds margin must be a non-negative integer")
        self.require_inside(image_shape_yx)
        return ImageBounds(
            y_start=max(0, self.y_start - margin_pixels),
            y_stop=min(image_shape_yx[0], self.y_stop + margin_pixels),
            x_start=max(0, self.x_start - margin_pixels),
            x_stop=min(image_shape_yx[1], self.x_stop + margin_pixels),
        )

center_xy property

Return the centre of the bounded pixels, in global (x, y).

Local geometry is derived here rather than at a corner, and every caller that derives it must agree, so the definition lives with the bounds instead of beside each use.

shape_yx property

Return the bounded array shape in NumPy axis order.

__post_init__()

Require non-negative, non-empty half-open bounds.

Source code in src/hebog/data_models/partitioning.py

def __post_init__(self) -> None:
    """Require non-negative, non-empty half-open bounds."""
    if min(self.y_start, self.y_stop, self.x_start, self.x_stop) < 0:
        raise ValueError("image bounds must be non-negative")
    if self.y_stop <= self.y_start or self.x_stop <= self.x_start:
        raise ValueError("image bounds must be non-empty")

contains(other)

Return whether another region lies completely within the bounds.

Source code in src/hebog/data_models/partitioning.py

def contains(self, other: ImageBounds) -> bool:
    """Return whether another region lies completely within the bounds."""
    return (
        self.y_start <= other.y_start
        and other.y_stop <= self.y_stop
        and self.x_start <= other.x_start
        and other.x_stop <= self.x_stop
    )

expanded(margin_pixels, image_shape_yx)

Return a uniformly expanded region clipped to the image plane.

Source code in src/hebog/data_models/partitioning.py

def expanded(
    self,
    margin_pixels: int,
    image_shape_yx: tuple[int, int],
) -> ImageBounds:
    """Return a uniformly expanded region clipped to the image plane."""
    if (
        isinstance(margin_pixels, bool)
        or not isinstance(  # pyright: ignore[reportUnnecessaryIsInstance]
            margin_pixels,
            int,
        )
        or margin_pixels < 0
    ):
        raise ValueError("bounds margin must be a non-negative integer")
    self.require_inside(image_shape_yx)
    return ImageBounds(
        y_start=max(0, self.y_start - margin_pixels),
        y_stop=min(image_shape_yx[0], self.y_stop + margin_pixels),
        x_start=max(0, self.x_start - margin_pixels),
        x_stop=min(image_shape_yx[1], self.x_stop + margin_pixels),
    )

require_inside(shape_yx)

Reject bounds extending beyond one logical image plane.

Source code in src/hebog/data_models/partitioning.py

def require_inside(self, shape_yx: tuple[int, int]) -> None:
    """Reject bounds extending beyond one logical image plane."""
    if self.y_stop > shape_yx[0] or self.x_stop > shape_yx[1]:
        raise ValueError(
            f"image bounds must stay inside image shape {shape_yx}"
        )

ImageMetadata dataclass

Scientific metadata required to interpret one logical image plane.

Source code in src/hebog/data_models/images.py

@dataclass(frozen=True, slots=True)
class ImageMetadata:
    """Scientific metadata required to interpret one logical image plane."""

    shape_yx: tuple[int, int]
    unit: str
    beam: RestoringBeam
    celestial_wcs: CelestialWcs
    reference_frequency_hz: float
    schema_version: Literal[1] = 1

    def __post_init__(self) -> None:
        """Reject unsupported versions and incomplete physical metadata."""
        if self.schema_version != 1:
            raise ValueError("unsupported image metadata schema version")
        if min(self.shape_yx) < 1:
            raise ValueError(
                "image metadata shape dimensions must be positive"
            )
        if not self.unit:
            raise ValueError("image metadata unit must not be empty")
        if (
            not isfinite(self.reference_frequency_hz)
            or self.reference_frequency_hz <= 0
        ):
            raise ValueError(
                "image metadata reference frequency must be finite and "
                "positive"
            )

__post_init__()

Reject unsupported versions and incomplete physical metadata.

Source code in src/hebog/data_models/images.py

def __post_init__(self) -> None:
    """Reject unsupported versions and incomplete physical metadata."""
    if self.schema_version != 1:
        raise ValueError("unsupported image metadata schema version")
    if min(self.shape_yx) < 1:
        raise ValueError(
            "image metadata shape dimensions must be positive"
        )
    if not self.unit:
        raise ValueError("image metadata unit must not be empty")
    if (
        not isfinite(self.reference_frequency_hz)
        or self.reference_frequency_hz <= 0
    ):
        raise ValueError(
            "image metadata reference frequency must be finite and "
            "positive"
        )

ImageSource

Bases: Protocol

Narrow input seam implemented by window-readable image stores.

Source code in src/hebog/io/base.py

class ImageSource(Protocol):
    """Narrow input seam implemented by window-readable image stores."""

    def metadata(self) -> ImageMetadata:
        """Return logical image metadata without materialising the plane."""
        ...

    def read_window(self, bounds: ImageBounds) -> ImageWindow:
        """Read one bounded global window into worker-owned memory.

        The window's ``valid_pixels`` must not depend on ``bounds``: a source
        whose rule looks at a pixel's neighbours, as the FITS source's block
        rule (:mod:`hebog.io.pixel_validity`) does, reads a margin around the
        window and keeps only the window, so any tiling decides the same
        pixels.
        """
        ...

metadata()

Return logical image metadata without materialising the plane.

Source code in src/hebog/io/base.py

def metadata(self) -> ImageMetadata:
    """Return logical image metadata without materialising the plane."""
    ...

read_window(bounds)

Read one bounded global window into worker-owned memory.

The window's valid_pixels must not depend on bounds: a source whose rule looks at a pixel's neighbours, as the FITS source's block rule (:mod:hebog.io.pixel_validity) does, reads a margin around the window and keeps only the window, so any tiling decides the same pixels.

Source code in src/hebog/io/base.py

def read_window(self, bounds: ImageBounds) -> ImageWindow:
    """Read one bounded global window into worker-owned memory.

    The window's ``valid_pixels`` must not depend on ``bounds``: a source
    whose rule looks at a pixel's neighbours, as the FITS source's block
    rule (:mod:`hebog.io.pixel_validity`) does, reads a margin around the
    window and keeps only the window, so any tiling decides the same
    pixels.
    """
    ...

ImageWindow dataclass

One owned, read-only bounded pixel window and its validity mask.

valid_pixels says which pixels the finder measures. A pixel's validity belongs to the pixel, not to the window that holds it: the same pixel must be valid in every window that contains it, however the image is tiled.

Source code in src/hebog/io/base.py

@dataclass(frozen=True, slots=True)
class ImageWindow:
    """One owned, read-only bounded pixel window and its validity mask.

    ``valid_pixels`` says which pixels the finder measures. A pixel's
    validity belongs to the pixel, not to the window that holds it: the same
    pixel must be valid in every window that contains it, however the image
    is tiled.
    """

    bounds: ImageBounds
    values: npt.NDArray[np.float64]
    valid_pixels: npt.NDArray[np.bool_]

InvalidFitsImageError

Bases: ValueError

A FITS input is missing required structural or physical metadata.

Source code in src/hebog/io/fits.py

class InvalidFitsImageError(ValueError):
    """A FITS input is missing required structural or physical metadata."""

InvalidMaterializedProductError

Bases: ProductMaterializationError

A product is corrupt, incomplete, or scientifically invalid.

Source code in src/hebog/io/materialization.py

class InvalidMaterializedProductError(ProductMaterializationError):
    """A product is corrupt, incomplete, or scientifically invalid."""

InvalidProductChunkError

Bases: ProductChunkError

A Zarr product chunk is missing, corrupt, or inconsistent.

Source code in src/hebog/io/zarr.py

class InvalidProductChunkError(ProductChunkError):
    """A Zarr product chunk is missing, corrupt, or inconsistent."""

InvalidProductGenerationError

Bases: ProductGenerationError

A completion manifest or one of its chunks is invalid.

Source code in src/hebog/io/zarr.py

class InvalidProductGenerationError(ProductGenerationError):
    """A completion manifest or one of its chunks is invalid."""

MaterializedProductConflictError

Bases: ProductMaterializationError

A destination already contains different product bytes.

Source code in src/hebog/io/materialization.py

class MaterializedProductConflictError(ProductMaterializationError):
    """A destination already contains different product bytes."""

ProductChunkConflictError

Bases: ProductChunkError

A completed Zarr chunk contains different values for one tile.

Source code in src/hebog/io/zarr.py

class ProductChunkConflictError(ProductChunkError):
    """A completed Zarr chunk contains different values for one tile."""

ProductChunkError

Bases: ValueError

A Zarr product chunk cannot be safely published or consumed.

Source code in src/hebog/io/zarr.py

class ProductChunkError(ValueError):
    """A Zarr product chunk cannot be safely published or consumed."""

ProductGenerationConflictError

Bases: ProductGenerationError

A different completion manifest already publishes this generation.

Source code in src/hebog/io/zarr.py

class ProductGenerationConflictError(ProductGenerationError):
    """A different completion manifest already publishes this generation."""

ProductGenerationError

Bases: ProductChunkError

A Zarr generation cannot be safely published or consumed.

Source code in src/hebog/io/zarr.py

class ProductGenerationError(ProductChunkError):
    """A Zarr generation cannot be safely published or consumed."""

ProductMaterializationError

Bases: ValueError

A restartable product could not be written or validated.

Source code in src/hebog/io/materialization.py

class ProductMaterializationError(ValueError):
    """A restartable product could not be written or validated."""

UnsupportedFitsImageError

Bases: InvalidFitsImageError

A valid FITS input uses an image layout Hebog does not yet support.

Source code in src/hebog/io/fits.py

class UnsupportedFitsImageError(InvalidFitsImageError):
    """A valid FITS input uses an image layout Hebog does not yet support."""

UnsupportedMaterializedProductError

Bases: InvalidMaterializedProductError

A valid product uses an unsupported role or schema version.

Source code in src/hebog/io/materialization.py

class UnsupportedMaterializedProductError(InvalidMaterializedProductError):
    """A valid product uses an unsupported role or schema version."""

ZarrProductSink

Write complete canonical tile cores into one local Zarr v3 group.

Source code in src/hebog/io/zarr.py

class ZarrProductSink:
    """Write complete canonical tile cores into one local Zarr v3 group."""

    def __init__(
        self,
        root: Path,
        manifest: PartitionManifest,
        *,
        generation_id: str,
    ) -> None:
        """Retain plain metadata without opening or creating a store."""
        if manifest.partition_origin_yx != (0, 0):
            raise ValueError(
                "Zarr chunk writes require a zero partition origin"
            )
        if not generation_id:
            raise ValueError("Zarr generation ID must not be empty")
        self._root = root.resolve()
        self._manifest = manifest
        self._generation_id = generation_id
        self._array_cache: dict[str, Any] = {}
        self._published_generation_cache: ProductGenerationManifest | None = (
            None
        )

    def __getstate__(self) -> dict[str, Any]:
        """Pickle the plain metadata only; open handles stay in this process.

        A sink travels inside executor payloads, so a copy on a worker opens
        its own array handles and reads the completion marker once, and the
        pickled bytes never carry a store object. Both are immutable for a
        generation: an array's geometry and policy are fixed when it is
        created and the marker is published once, so a process validates
        them when it first opens them and trusts them for the sink's
        lifetime, while every chunk read is still checked against its
        record. A change to either on disk is not a supported operation; the
        next process or task to open the product rejects it.
        """
        state = self.__dict__.copy()
        state["_array_cache"] = {}
        state["_published_generation_cache"] = None
        return state

    @property
    def manifest(self) -> PartitionManifest:
        """Return the immutable partition geometry used by this sink."""
        return self._manifest

    @property
    def generation_id(self) -> str:
        """Return the immutable run identity stored with every chunk."""
        return self._generation_id

    def initialize_product(
        self,
        *,
        product_name: str,
        dtype: npt.DTypeLike,
    ) -> None:
        """Create or validate one product array before worker submission."""
        validate_product_name(product_name)
        normalized_dtype = _little_endian_dtype(dtype)
        group = zarr.open_group(
            store=LocalStore(self._root),
            mode="a",
            zarr_format=_ZARR_FORMAT,
        )
        expected_group_attributes = {
            **_GROUP_ATTRIBUTES,
            "image_shape_yx": list(self._manifest.image_shape_yx),
            "tile_core_shape_yx": list(self._manifest.tile_core_shape_yx),
            "partition_origin_yx": list(self._manifest.partition_origin_yx),
            "partition_schema_version": self._manifest.schema_version,
            "generation_id": self._generation_id,
        }
        stale = False
        for name, expected in expected_group_attributes.items():
            existing = group.attrs.get(name)
            if existing is not None and existing != expected:
                raise InvalidProductChunkError(
                    f"Zarr group attribute {name!r} conflicts with manifest"
                )
            stale = stale or existing != expected
        if stale:
            group.update_attributes(expected_group_attributes)

        if product_name in group:
            self._require_array_metadata(
                group[product_name],
                product_name=product_name,
                dtype=normalized_dtype,
            )
            return
        group.create_array(
            product_name,
            shape=self._manifest.image_shape_yx,
            chunks=self._manifest.tile_core_shape_yx,
            dtype=normalized_dtype,
            fill_value=0,
            filters=None,
            serializer=BytesCodec(endian="little"),
            compressors=_compressors(normalized_dtype),
            chunk_key_encoding={
                "name": "default",
                "configuration": {"separator": "/"},
            },
            attributes=cast(
                Any,
                {
                    **_ARRAY_ATTRIBUTES,
                    "hebog_generation_id": self._generation_id,
                    "hebog_compression_policy": _compression_policy(
                        normalized_dtype
                    ),
                },
            ),
            config={
                "read_missing_chunks": False,
                "write_empty_chunks": True,
            },
        )

    def _require_array_metadata(
        self,
        array: object,
        *,
        product_name: str,
        dtype: np.dtype[Any] | None = None,
    ) -> Any:
        """Validate the durable geometry and Hebog policy attributes."""
        zarr_array = cast(Any, array)
        if (
            zarr_array.metadata.zarr_format != _ZARR_FORMAT
            or tuple(zarr_array.shape) != self._manifest.image_shape_yx
            or tuple(zarr_array.chunks) != self._manifest.tile_core_shape_yx
            or (dtype is not None and zarr_array.dtype != dtype)
        ):
            raise InvalidProductChunkError(
                f"Zarr product {product_name!r} metadata conflicts with sink"
            )
        expected_attributes = {
            **_ARRAY_ATTRIBUTES,
            "hebog_generation_id": self._generation_id,
            "hebog_compression_policy": _compression_policy(
                np.dtype(zarr_array.dtype)
            ),
        }
        for name, expected in expected_attributes.items():
            if zarr_array.attrs.get(name) != expected:
                raise InvalidProductChunkError(
                    f"Zarr product {product_name!r} policy is invalid"
                )
        return zarr_array.with_config(
            {
                "read_missing_chunks": False,
                "write_empty_chunks": True,
            }
        )

    def _open_array(self, product_name: str) -> Any:
        """Open a pre-created product without racing metadata creation.

        The handle is validated once per process and then cached for the
        sink's lifetime, because array metadata is immutable for a
        generation; see :meth:`__getstate__`.
        """
        validate_product_name(product_name)
        cached = self._array_cache.get(product_name)
        if cached is not None:
            return cached
        try:
            array = zarr.open_array(
                store=LocalStore(self._root),
                path=product_name,
                mode="r+",
                zarr_format=_ZARR_FORMAT,
            )
        except (ArrayNotFoundError, FileNotFoundError) as error:
            raise InvalidProductChunkError(
                f"Zarr product {product_name!r} is not initialized"
            ) from error
        validated = self._require_array_metadata(
            array,
            product_name=product_name,
        )
        self._array_cache[product_name] = validated
        return validated

    def _require_canonical_tile(self, tile: TilePartition) -> None:
        """Reject invented records and storage-chunk misalignment."""
        if self._canonical_tile(tile.tile_id) != tile:
            raise ValueError("tile must belong to the canonical manifest")

    def _canonical_tile(self, tile_id: str) -> TilePartition:
        """Resolve one zero-origin row-major tile in constant time."""
        try:
            _, tile_y_index_value, tile_x_index_value = tile_id.split("-")
            tile_y_index = int(tile_y_index_value)
            tile_x_index = int(tile_x_index_value)
            tiles_per_row = (
                self._manifest.image_shape_yx[1]
                + self._manifest.tile_core_shape_yx[1]
                - 1
            ) // self._manifest.tile_core_shape_yx[1]
            tile = self._manifest.tiles[
                tile_y_index * tiles_per_row + tile_x_index
            ]
        except (IndexError, ValueError) as error:
            raise InvalidProductChunkError(
                "Zarr product chunk is not in the canonical manifest"
            ) from error
        if tile.tile_id != tile_id:
            raise InvalidProductChunkError(
                "Zarr product chunk is not in the canonical manifest"
            )
        return tile

    def _record(
        self,
        *,
        product_name: str,
        tile: TilePartition,
        values: npt.NDArray[np.generic],
    ) -> ProductChunk:
        """Build the small logical identity returned to the scheduler."""
        return ProductChunk(
            generation_id=self._generation_id,
            product_name=product_name,
            tile_id=tile.tile_id,
            core_bounds=tile.core_bounds,
            dtype=values.dtype.str,
            shape_yx=tuple(values.shape),
            content_sha256=_content_sha256(values),
        )

    @staticmethod
    def _chunk_exists(array: Any, tile: TilePartition) -> bool:
        """Return whether one canonical encoded chunk key already exists."""
        chunk_key = array.metadata.encode_chunk_key(
            (tile.tile_y_index, tile.tile_x_index)
        )
        return bool(sync((array.store_path / chunk_key).exists()))

    def write_chunk(
        self,
        *,
        product_name: str,
        tile: TilePartition,
        values: npt.NDArray[np.generic],
    ) -> ProductChunk:
        """Write one aligned chunk or content-validate an existing retry.

        Fresh LocalStore writes are atomic and use the required CRC32C codec.
        Their content SHA-256 is validated in the mandatory full-generation
        check before a completion marker can be published.
        """
        self._require_canonical_tile(tile)
        array = self._open_array(product_name)
        normalized = np.asarray(values)
        if (
            normalized.ndim != _IMAGE_DIMENSIONS
            or tuple(normalized.shape) != tile.core_bounds.shape_yx
        ):
            raise ValueError(
                "Zarr product values must match the tile core shape"
            )
        if normalized.dtype != array.dtype:
            raise ValueError(
                "Zarr product values must match the initialized dtype"
            )
        record = self._record(
            product_name=product_name,
            tile=tile,
            values=normalized,
        )
        if self._chunk_exists(array, tile):
            try:
                self._read_values(array=array, tile=tile, record=record)
            except InvalidProductChunkError as error:
                if not isinstance(error.__cause__, ChunkNotFoundError):
                    raise ProductChunkConflictError(
                        "published Zarr product chunk contains different "
                        "values"
                    ) from error
            else:
                return record

        array[_selection(tile)] = normalized
        return record

    def read_chunk(
        self,
        record: ProductChunk,
    ) -> npt.NDArray[np.generic]:
        """Validate one expected chunk and return an owned read-only array."""
        if record.generation_id != self._generation_id:
            raise InvalidProductChunkError(
                "Zarr product chunk has a different generation ID"
            )
        tile = self._canonical_tile(record.tile_id)
        if tile.core_bounds != record.core_bounds:
            raise InvalidProductChunkError(
                "Zarr product chunk is not in the canonical manifest"
            )
        array = self._open_array(record.product_name)
        return self._read_values(array=array, tile=tile, record=record)

    @staticmethod
    def _require_values_match_record(
        values: npt.NDArray[np.generic],
        record: ProductChunk,
    ) -> npt.NDArray[np.generic]:
        """Validate one owned array against its immutable chunk record."""
        if (
            values.ndim != _IMAGE_DIMENSIONS
            or tuple(values.shape) != record.shape_yx
            or values.dtype.str != record.dtype
        ):
            raise InvalidProductChunkError(
                "Zarr product chunk array metadata disagrees with its record"
            )
        if _content_sha256(values) != record.content_sha256:
            raise InvalidProductChunkError(
                "Zarr product chunk SHA-256 disagrees with its record"
            )
        values.setflags(write=False)
        return values

    @classmethod
    def _read_values(
        cls,
        *,
        array: Any,
        tile: TilePartition,
        record: ProductChunk,
    ) -> npt.NDArray[np.generic]:
        """Validate one chunk selection from an already-open strict array."""
        try:
            values = np.array(
                array[_selection(tile)],
                copy=True,
            )
        except ChunkNotFoundError as error:
            raise InvalidProductChunkError(
                "Zarr product chunk is missing"
            ) from error
        except (OSError, ValueError) as error:
            raise InvalidProductChunkError(
                "Zarr product chunk is corrupt"
            ) from error
        return cls._require_values_match_record(values, record)

    @classmethod
    def _read_product_block(
        cls,
        *,
        array: Any,
        tiles: tuple[TilePartition, ...],
        records: tuple[ProductChunk, ...],
    ) -> tuple[npt.NDArray[np.generic], ...]:
        """Read and validate one contiguous bounded canonical tile block."""
        first_bounds = tiles[0].core_bounds
        last_bounds = tiles[-1].core_bounds
        selection = (
            slice(first_bounds.y_start, last_bounds.y_stop),
            slice(first_bounds.x_start, last_bounds.x_stop),
        )
        try:
            row = np.array(array[selection], copy=True)
        except ChunkNotFoundError as error:
            raise InvalidProductChunkError(
                "Zarr product chunk is missing"
            ) from error
        except (OSError, ValueError) as error:
            raise InvalidProductChunkError(
                "Zarr product chunk is corrupt"
            ) from error
        values_by_tile: list[npt.NDArray[np.generic]] = []
        for tile, record in zip(tiles, records, strict=True):
            bounds = tile.core_bounds
            values = np.ascontiguousarray(
                row[
                    bounds.y_start - first_bounds.y_start : bounds.y_stop
                    - first_bounds.y_start,
                    bounds.x_start - first_bounds.x_start : bounds.x_stop
                    - first_bounds.x_start,
                ]
            )
            values_by_tile.append(
                cls._require_values_match_record(values, record)
            )
        return tuple(values_by_tile)

    def _require_generation_chunks(
        self,
        generation: ProductGenerationManifest,
    ) -> None:
        """Validate identity, geometry, and every referenced store chunk."""
        self._require_generation_identity(generation)
        tile_count = len(self._manifest.tiles)
        tiles_per_row = (
            self._manifest.image_shape_yx[1]
            + self._manifest.tile_core_shape_yx[1]
            - 1
        ) // self._manifest.tile_core_shape_yx[1]
        core_height, core_width = self._manifest.tile_core_shape_yx
        blocks = _validation_blocks(
            tile_count=tile_count,
            tiles_per_row=tiles_per_row,
            tiles_per_block=max(
                1,
                _GENERATION_VALIDATION_BLOCK_PIXELS
                // (core_height * core_width),
            ),
        )
        try:
            for product_index, product_name in enumerate(
                generation.product_names
            ):
                array = self._open_array(product_name)
                product_start = product_index * tile_count
                for block_start, block_stop in blocks:
                    self._read_product_block(
                        array=array,
                        tiles=self._manifest.tiles[block_start:block_stop],
                        records=generation.chunks[
                            product_start + block_start : product_start
                            + block_stop
                        ],
                    )
        except ProductChunkError as error:
            raise InvalidProductGenerationError(
                "completion manifest references an invalid product chunk"
            ) from error

    def _require_generation_identity(
        self,
        generation: ProductGenerationManifest,
    ) -> None:
        """Require a published record for this exact run and partition."""
        if generation.generation_id != self._generation_id:
            raise InvalidProductGenerationError(
                "completion manifest has a different generation ID"
            )
        if generation.partition_manifest != self._manifest:
            raise InvalidProductGenerationError(
                "completion manifest has a different partition manifest"
            )

    def publish_generation(
        self,
        *,
        product_names: Iterable[str],
        chunks: Iterable[ProductChunk],
    ) -> ProductGenerationManifest:
        """Conditionally publish one exact, fully validated generation."""
        try:
            generation = ProductGenerationManifest.create(
                generation_id=self._generation_id,
                partition_manifest=self._manifest,
                product_names=product_names,
                chunks=chunks,
            )
        except ValueError as error:
            raise InvalidProductGenerationError(
                "product chunks do not form a complete generation"
            ) from error
        self._require_generation_chunks(generation)
        payload = generation.canonical_json_bytes()
        buffer = default_buffer_prototype().buffer.from_bytes(payload)
        with LocalStore(self._root) as store:
            sync(store.set_if_not_exists(_COMPLETION_KEY, buffer))
            published = store.get_sync(_COMPLETION_KEY)
        if published is None or published.to_bytes() != payload:
            raise ProductGenerationConflictError(
                "a different completion manifest is already published"
            )
        return generation

    def _read_published_generation(self) -> ProductGenerationManifest:
        """Read and validate the immutable record without rereading chunks."""
        if self._published_generation_cache is not None:
            return self._published_generation_cache
        with LocalStore(self._root, read_only=True) as store:
            stored = store.get_sync(_COMPLETION_KEY)
        if stored is None:
            raise InvalidProductGenerationError(
                "product generation is not published"
            )
        payload = stored.to_bytes()
        try:
            generation = ProductGenerationManifest.from_json_bytes(payload)
        except ValueError as error:
            raise InvalidProductGenerationError(
                f"published completion manifest is invalid: {error}"
            ) from error
        if generation.canonical_json_bytes() != payload:
            raise InvalidProductGenerationError(
                "published completion manifest is not canonical"
            )
        self._require_generation_identity(generation)
        self._published_generation_cache = generation
        return generation

    def read_generation(self) -> ProductGenerationManifest:
        """Read and validate the published completion manifest.

        The marker is parsed, required to be canonical and bound to this
        sink's run and partition. Its chunks are not re-read: every one was
        read and checksummed when the generation was published, and each is
        validated against its record again whenever it is read, so a
        consumer never pays for a whole generation it does not use.
        """
        return self._read_published_generation()

    def read_completed_window(
        self,
        product_name: str,
        bounds: ImageBounds,
    ) -> npt.NDArray[np.generic]:
        """Read one checksum-validated bounded window from owned chunks."""
        return self.read_completed_windows(product_name, (bounds,))[0]

    def read_completed_windows(
        self,
        product_name: str,
        bounds_collection: Iterable[ImageBounds],
    ) -> tuple[npt.NDArray[np.generic], ...]:
        """Read windows through a bounded cache of validated owned chunks."""
        requested_bounds = tuple(bounds_collection)
        if not requested_bounds:
            return ()
        for bounds in requested_bounds:
            bounds.require_inside(self._manifest.image_shape_yx)
        validate_product_name(product_name)
        generation = self._read_published_generation()
        if product_name not in generation.product_names:
            raise InvalidProductGenerationError(
                "published generation does not contain product "
                f"{product_name!r}"
            )
        product_index = generation.product_names.index(product_name)
        tile_count = len(self._manifest.tiles)
        core_y, core_x = self._manifest.tile_core_shape_yx
        tiles_per_row = (
            self._manifest.image_shape_yx[1] + core_x - 1
        ) // core_x
        array = self._open_array(product_name)
        cache: OrderedDict[int, npt.NDArray[np.generic]] = OrderedDict()
        windows: list[npt.NDArray[np.generic]] = []
        for bounds in requested_bounds:
            tile_y_start = bounds.y_start // core_y
            tile_y_stop = (bounds.y_stop - 1) // core_y
            tile_x_start = bounds.x_start // core_x
            tile_x_stop = (bounds.x_stop - 1) // core_x
            first_tile_index = tile_y_start * tiles_per_row + tile_x_start
            first_record = generation.chunks[
                product_index * tile_count + first_tile_index
            ]
            window = np.empty(
                bounds.shape_yx,
                dtype=np.dtype(first_record.dtype),
            )
            for tile_y_index in range(tile_y_start, tile_y_stop + 1):
                tile_indices = tuple(
                    tile_y_index * tiles_per_row + tile_x_index
                    for tile_x_index in range(tile_x_start, tile_x_stop + 1)
                )
                row_values = {
                    tile_index: cache[tile_index]
                    for tile_index in tile_indices
                    if tile_index in cache
                }
                missing_indices = tuple(
                    tile_index
                    for tile_index in tile_indices
                    if tile_index not in row_values
                )
                if len(missing_indices) > 1 and not row_values:
                    missing_tiles = tuple(
                        self._manifest.tiles[tile_index]
                        for tile_index in missing_indices
                    )
                    missing_records = tuple(
                        generation.chunks[
                            product_index * tile_count + tile_index
                        ]
                        for tile_index in missing_indices
                    )
                    row_values.update(
                        zip(
                            missing_indices,
                            self._read_product_block(
                                array=array,
                                tiles=missing_tiles,
                                records=missing_records,
                            ),
                            strict=True,
                        )
                    )
                else:
                    for tile_index in missing_indices:
                        row_values[tile_index] = self._read_values(
                            array=array,
                            tile=self._manifest.tiles[tile_index],
                            record=generation.chunks[
                                product_index * tile_count + tile_index
                            ],
                        )
                for tile_index in tile_indices:
                    values = row_values[tile_index]
                    cache.pop(tile_index, None)
                    if len(cache) >= _WINDOW_CHUNK_CACHE_SIZE:
                        cache.popitem(last=False)
                    cache[tile_index] = values
                    tile = self._manifest.tiles[tile_index]
                    overlap_y_start = max(
                        bounds.y_start,
                        tile.core_bounds.y_start,
                    )
                    overlap_y_stop = min(
                        bounds.y_stop,
                        tile.core_bounds.y_stop,
                    )
                    overlap_x_start = max(
                        bounds.x_start,
                        tile.core_bounds.x_start,
                    )
                    overlap_x_stop = min(
                        bounds.x_stop,
                        tile.core_bounds.x_stop,
                    )
                    window[
                        overlap_y_start - bounds.y_start : overlap_y_stop
                        - bounds.y_start,
                        overlap_x_start - bounds.x_start : overlap_x_stop
                        - bounds.x_start,
                    ] = values[
                        overlap_y_start
                        - tile.core_bounds.y_start : overlap_y_stop
                        - tile.core_bounds.y_start,
                        overlap_x_start
                        - tile.core_bounds.x_start : overlap_x_stop
                        - tile.core_bounds.x_start,
                    ]
            window.setflags(write=False)
            windows.append(window)
        return tuple(windows)

    def iter_completed_row_blocks(
        self,
        product_name: str,
        *,
        max_block_bytes: int,
    ) -> Iterator[npt.NDArray[np.generic]]:
        """Yield validated full-width tile rows for final materialisation.

        Every storage chunk is read and checksum-validated exactly once. A
        multi-tile row is assembled into one owned C-contiguous block. The
        caller must admit enough memory for one full tile row; this is bounded
        by image width times tile-core height rather than full image height.
        A one-tile image returns its already-owned chunk without an assembly
        copy.
        """
        if max_block_bytes < 1:
            raise ValueError("row-block memory budget must be positive")
        validate_product_name(product_name)
        generation = self._read_published_generation()
        if product_name not in generation.product_names:
            raise InvalidProductGenerationError(
                "published generation does not contain product "
                f"{product_name!r}"
            )
        records_by_tile = {
            chunk.tile_id: chunk
            for chunk in generation.chunks
            if chunk.product_name == product_name
        }
        tile_rows: dict[int, list[TilePartition]] = {}
        for tile in self._manifest.tiles:
            tile_rows.setdefault(tile.tile_y_index, []).append(tile)
        for row_tiles in tile_rows.values():
            first_record = records_by_tile[row_tiles[0].tile_id]
            dtype = np.dtype(first_record.dtype)
            row_height = row_tiles[0].core_bounds.shape_yx[0]
            required_bytes = (
                row_height * self._manifest.image_shape_yx[1] * dtype.itemsize
            )
            if required_bytes > max_block_bytes:
                raise ValueError(
                    "row-block memory budget is below one canonical tile row: "
                    f"requires {required_bytes} bytes"
                )
            if len(row_tiles) == 1:
                yield self.read_chunk(first_record)
                continue
            block = np.empty(
                (row_height, self._manifest.image_shape_yx[1]),
                dtype=dtype,
            )
            for tile in row_tiles:
                values = self.read_chunk(records_by_tile[tile.tile_id])
                bounds = tile.core_bounds
                block[:, bounds.x_start : bounds.x_stop] = values
            block.setflags(write=False)
            yield block

generation_id property

Return the immutable run identity stored with every chunk.

manifest property

Return the immutable partition geometry used by this sink.

__getstate__()

Pickle the plain metadata only; open handles stay in this process.

A sink travels inside executor payloads, so a copy on a worker opens its own array handles and reads the completion marker once, and the pickled bytes never carry a store object. Both are immutable for a generation: an array's geometry and policy are fixed when it is created and the marker is published once, so a process validates them when it first opens them and trusts them for the sink's lifetime, while every chunk read is still checked against its record. A change to either on disk is not a supported operation; the next process or task to open the product rejects it.

Source code in src/hebog/io/zarr.py

def __getstate__(self) -> dict[str, Any]:
    """Pickle the plain metadata only; open handles stay in this process.

    A sink travels inside executor payloads, so a copy on a worker opens
    its own array handles and reads the completion marker once, and the
    pickled bytes never carry a store object. Both are immutable for a
    generation: an array's geometry and policy are fixed when it is
    created and the marker is published once, so a process validates
    them when it first opens them and trusts them for the sink's
    lifetime, while every chunk read is still checked against its
    record. A change to either on disk is not a supported operation; the
    next process or task to open the product rejects it.
    """
    state = self.__dict__.copy()
    state["_array_cache"] = {}
    state["_published_generation_cache"] = None
    return state

__init__(root, manifest, *, generation_id)

Retain plain metadata without opening or creating a store.

Source code in src/hebog/io/zarr.py

def __init__(
    self,
    root: Path,
    manifest: PartitionManifest,
    *,
    generation_id: str,
) -> None:
    """Retain plain metadata without opening or creating a store."""
    if manifest.partition_origin_yx != (0, 0):
        raise ValueError(
            "Zarr chunk writes require a zero partition origin"
        )
    if not generation_id:
        raise ValueError("Zarr generation ID must not be empty")
    self._root = root.resolve()
    self._manifest = manifest
    self._generation_id = generation_id
    self._array_cache: dict[str, Any] = {}
    self._published_generation_cache: ProductGenerationManifest | None = (
        None
    )

initialize_product(*, product_name, dtype)

Create or validate one product array before worker submission.

Source code in src/hebog/io/zarr.py

def initialize_product(
    self,
    *,
    product_name: str,
    dtype: npt.DTypeLike,
) -> None:
    """Create or validate one product array before worker submission."""
    validate_product_name(product_name)
    normalized_dtype = _little_endian_dtype(dtype)
    group = zarr.open_group(
        store=LocalStore(self._root),
        mode="a",
        zarr_format=_ZARR_FORMAT,
    )
    expected_group_attributes = {
        **_GROUP_ATTRIBUTES,
        "image_shape_yx": list(self._manifest.image_shape_yx),
        "tile_core_shape_yx": list(self._manifest.tile_core_shape_yx),
        "partition_origin_yx": list(self._manifest.partition_origin_yx),
        "partition_schema_version": self._manifest.schema_version,
        "generation_id": self._generation_id,
    }
    stale = False
    for name, expected in expected_group_attributes.items():
        existing = group.attrs.get(name)
        if existing is not None and existing != expected:
            raise InvalidProductChunkError(
                f"Zarr group attribute {name!r} conflicts with manifest"
            )
        stale = stale or existing != expected
    if stale:
        group.update_attributes(expected_group_attributes)

    if product_name in group:
        self._require_array_metadata(
            group[product_name],
            product_name=product_name,
            dtype=normalized_dtype,
        )
        return
    group.create_array(
        product_name,
        shape=self._manifest.image_shape_yx,
        chunks=self._manifest.tile_core_shape_yx,
        dtype=normalized_dtype,
        fill_value=0,
        filters=None,
        serializer=BytesCodec(endian="little"),
        compressors=_compressors(normalized_dtype),
        chunk_key_encoding={
            "name": "default",
            "configuration": {"separator": "/"},
        },
        attributes=cast(
            Any,
            {
                **_ARRAY_ATTRIBUTES,
                "hebog_generation_id": self._generation_id,
                "hebog_compression_policy": _compression_policy(
                    normalized_dtype
                ),
            },
        ),
        config={
            "read_missing_chunks": False,
            "write_empty_chunks": True,
        },
    )

iter_completed_row_blocks(product_name, *, max_block_bytes)

Yield validated full-width tile rows for final materialisation.

Every storage chunk is read and checksum-validated exactly once. A multi-tile row is assembled into one owned C-contiguous block. The caller must admit enough memory for one full tile row; this is bounded by image width times tile-core height rather than full image height. A one-tile image returns its already-owned chunk without an assembly copy.

Source code in src/hebog/io/zarr.py

def iter_completed_row_blocks(
    self,
    product_name: str,
    *,
    max_block_bytes: int,
) -> Iterator[npt.NDArray[np.generic]]:
    """Yield validated full-width tile rows for final materialisation.

    Every storage chunk is read and checksum-validated exactly once. A
    multi-tile row is assembled into one owned C-contiguous block. The
    caller must admit enough memory for one full tile row; this is bounded
    by image width times tile-core height rather than full image height.
    A one-tile image returns its already-owned chunk without an assembly
    copy.
    """
    if max_block_bytes < 1:
        raise ValueError("row-block memory budget must be positive")
    validate_product_name(product_name)
    generation = self._read_published_generation()
    if product_name not in generation.product_names:
        raise InvalidProductGenerationError(
            "published generation does not contain product "
            f"{product_name!r}"
        )
    records_by_tile = {
        chunk.tile_id: chunk
        for chunk in generation.chunks
        if chunk.product_name == product_name
    }
    tile_rows: dict[int, list[TilePartition]] = {}
    for tile in self._manifest.tiles:
        tile_rows.setdefault(tile.tile_y_index, []).append(tile)
    for row_tiles in tile_rows.values():
        first_record = records_by_tile[row_tiles[0].tile_id]
        dtype = np.dtype(first_record.dtype)
        row_height = row_tiles[0].core_bounds.shape_yx[0]
        required_bytes = (
            row_height * self._manifest.image_shape_yx[1] * dtype.itemsize
        )
        if required_bytes > max_block_bytes:
            raise ValueError(
                "row-block memory budget is below one canonical tile row: "
                f"requires {required_bytes} bytes"
            )
        if len(row_tiles) == 1:
            yield self.read_chunk(first_record)
            continue
        block = np.empty(
            (row_height, self._manifest.image_shape_yx[1]),
            dtype=dtype,
        )
        for tile in row_tiles:
            values = self.read_chunk(records_by_tile[tile.tile_id])
            bounds = tile.core_bounds
            block[:, bounds.x_start : bounds.x_stop] = values
        block.setflags(write=False)
        yield block

publish_generation(*, product_names, chunks)

Conditionally publish one exact, fully validated generation.

Source code in src/hebog/io/zarr.py

def publish_generation(
    self,
    *,
    product_names: Iterable[str],
    chunks: Iterable[ProductChunk],
) -> ProductGenerationManifest:
    """Conditionally publish one exact, fully validated generation."""
    try:
        generation = ProductGenerationManifest.create(
            generation_id=self._generation_id,
            partition_manifest=self._manifest,
            product_names=product_names,
            chunks=chunks,
        )
    except ValueError as error:
        raise InvalidProductGenerationError(
            "product chunks do not form a complete generation"
        ) from error
    self._require_generation_chunks(generation)
    payload = generation.canonical_json_bytes()
    buffer = default_buffer_prototype().buffer.from_bytes(payload)
    with LocalStore(self._root) as store:
        sync(store.set_if_not_exists(_COMPLETION_KEY, buffer))
        published = store.get_sync(_COMPLETION_KEY)
    if published is None or published.to_bytes() != payload:
        raise ProductGenerationConflictError(
            "a different completion manifest is already published"
        )
    return generation

read_chunk(record)

Validate one expected chunk and return an owned read-only array.

Source code in src/hebog/io/zarr.py

def read_chunk(
    self,
    record: ProductChunk,
) -> npt.NDArray[np.generic]:
    """Validate one expected chunk and return an owned read-only array."""
    if record.generation_id != self._generation_id:
        raise InvalidProductChunkError(
            "Zarr product chunk has a different generation ID"
        )
    tile = self._canonical_tile(record.tile_id)
    if tile.core_bounds != record.core_bounds:
        raise InvalidProductChunkError(
            "Zarr product chunk is not in the canonical manifest"
        )
    array = self._open_array(record.product_name)
    return self._read_values(array=array, tile=tile, record=record)

read_completed_window(product_name, bounds)

Read one checksum-validated bounded window from owned chunks.

Source code in src/hebog/io/zarr.py

def read_completed_window(
    self,
    product_name: str,
    bounds: ImageBounds,
) -> npt.NDArray[np.generic]:
    """Read one checksum-validated bounded window from owned chunks."""
    return self.read_completed_windows(product_name, (bounds,))[0]

read_completed_windows(product_name, bounds_collection)

Read windows through a bounded cache of validated owned chunks.

Source code in src/hebog/io/zarr.py

def read_completed_windows(
    self,
    product_name: str,
    bounds_collection: Iterable[ImageBounds],
) -> tuple[npt.NDArray[np.generic], ...]:
    """Read windows through a bounded cache of validated owned chunks."""
    requested_bounds = tuple(bounds_collection)
    if not requested_bounds:
        return ()
    for bounds in requested_bounds:
        bounds.require_inside(self._manifest.image_shape_yx)
    validate_product_name(product_name)
    generation = self._read_published_generation()
    if product_name not in generation.product_names:
        raise InvalidProductGenerationError(
            "published generation does not contain product "
            f"{product_name!r}"
        )
    product_index = generation.product_names.index(product_name)
    tile_count = len(self._manifest.tiles)
    core_y, core_x = self._manifest.tile_core_shape_yx
    tiles_per_row = (
        self._manifest.image_shape_yx[1] + core_x - 1
    ) // core_x
    array = self._open_array(product_name)
    cache: OrderedDict[int, npt.NDArray[np.generic]] = OrderedDict()
    windows: list[npt.NDArray[np.generic]] = []
    for bounds in requested_bounds:
        tile_y_start = bounds.y_start // core_y
        tile_y_stop = (bounds.y_stop - 1) // core_y
        tile_x_start = bounds.x_start // core_x
        tile_x_stop = (bounds.x_stop - 1) // core_x
        first_tile_index = tile_y_start * tiles_per_row + tile_x_start
        first_record = generation.chunks[
            product_index * tile_count + first_tile_index
        ]
        window = np.empty(
            bounds.shape_yx,
            dtype=np.dtype(first_record.dtype),
        )
        for tile_y_index in range(tile_y_start, tile_y_stop + 1):
            tile_indices = tuple(
                tile_y_index * tiles_per_row + tile_x_index
                for tile_x_index in range(tile_x_start, tile_x_stop + 1)
            )
            row_values = {
                tile_index: cache[tile_index]
                for tile_index in tile_indices
                if tile_index in cache
            }
            missing_indices = tuple(
                tile_index
                for tile_index in tile_indices
                if tile_index not in row_values
            )
            if len(missing_indices) > 1 and not row_values:
                missing_tiles = tuple(
                    self._manifest.tiles[tile_index]
                    for tile_index in missing_indices
                )
                missing_records = tuple(
                    generation.chunks[
                        product_index * tile_count + tile_index
                    ]
                    for tile_index in missing_indices
                )
                row_values.update(
                    zip(
                        missing_indices,
                        self._read_product_block(
                            array=array,
                            tiles=missing_tiles,
                            records=missing_records,
                        ),
                        strict=True,
                    )
                )
            else:
                for tile_index in missing_indices:
                    row_values[tile_index] = self._read_values(
                        array=array,
                        tile=self._manifest.tiles[tile_index],
                        record=generation.chunks[
                            product_index * tile_count + tile_index
                        ],
                    )
            for tile_index in tile_indices:
                values = row_values[tile_index]
                cache.pop(tile_index, None)
                if len(cache) >= _WINDOW_CHUNK_CACHE_SIZE:
                    cache.popitem(last=False)
                cache[tile_index] = values
                tile = self._manifest.tiles[tile_index]
                overlap_y_start = max(
                    bounds.y_start,
                    tile.core_bounds.y_start,
                )
                overlap_y_stop = min(
                    bounds.y_stop,
                    tile.core_bounds.y_stop,
                )
                overlap_x_start = max(
                    bounds.x_start,
                    tile.core_bounds.x_start,
                )
                overlap_x_stop = min(
                    bounds.x_stop,
                    tile.core_bounds.x_stop,
                )
                window[
                    overlap_y_start - bounds.y_start : overlap_y_stop
                    - bounds.y_start,
                    overlap_x_start - bounds.x_start : overlap_x_stop
                    - bounds.x_start,
                ] = values[
                    overlap_y_start
                    - tile.core_bounds.y_start : overlap_y_stop
                    - tile.core_bounds.y_start,
                    overlap_x_start
                    - tile.core_bounds.x_start : overlap_x_stop
                    - tile.core_bounds.x_start,
                ]
        window.setflags(write=False)
        windows.append(window)
    return tuple(windows)

read_generation()

Read and validate the published completion manifest.

The marker is parsed, required to be canonical and bound to this sink's run and partition. Its chunks are not re-read: every one was read and checksummed when the generation was published, and each is validated against its record again whenever it is read, so a consumer never pays for a whole generation it does not use.

Source code in src/hebog/io/zarr.py

def read_generation(self) -> ProductGenerationManifest:
    """Read and validate the published completion manifest.

    The marker is parsed, required to be canonical and bound to this
    sink's run and partition. Its chunks are not re-read: every one was
    read and checksummed when the generation was published, and each is
    validated against its record again whenever it is read, so a
    consumer never pays for a whole generation it does not use.
    """
    return self._read_published_generation()

write_chunk(*, product_name, tile, values)

Write one aligned chunk or content-validate an existing retry.

Fresh LocalStore writes are atomic and use the required CRC32C codec. Their content SHA-256 is validated in the mandatory full-generation check before a completion marker can be published.

Source code in src/hebog/io/zarr.py

def write_chunk(
    self,
    *,
    product_name: str,
    tile: TilePartition,
    values: npt.NDArray[np.generic],
) -> ProductChunk:
    """Write one aligned chunk or content-validate an existing retry.

    Fresh LocalStore writes are atomic and use the required CRC32C codec.
    Their content SHA-256 is validated in the mandatory full-generation
    check before a completion marker can be published.
    """
    self._require_canonical_tile(tile)
    array = self._open_array(product_name)
    normalized = np.asarray(values)
    if (
        normalized.ndim != _IMAGE_DIMENSIONS
        or tuple(normalized.shape) != tile.core_bounds.shape_yx
    ):
        raise ValueError(
            "Zarr product values must match the tile core shape"
        )
    if normalized.dtype != array.dtype:
        raise ValueError(
            "Zarr product values must match the initialized dtype"
        )
    record = self._record(
        product_name=product_name,
        tile=tile,
        values=normalized,
    )
    if self._chunk_exists(array, tile):
        try:
            self._read_values(array=array, tile=tile, record=record)
        except InvalidProductChunkError as error:
            if not isinstance(error.__cause__, ChunkNotFoundError):
                raise ProductChunkConflictError(
                    "published Zarr product chunk contains different "
                    "values"
                ) from error
        else:
            return record

    array[_selection(tile)] = normalized
    return record

celestial_wcs_from_metadata(metadata)

Reconstruct an independent celestial WCS from serialized metadata.

Source code in src/hebog/algorithms/astrometry.py

def celestial_wcs_from_metadata(metadata: ImageMetadata) -> WCS:
    """Reconstruct an independent celestial WCS from serialized metadata."""
    header = fits.Header.fromstring(
        metadata.celestial_wcs.fits_header,
        sep="\n",
    )
    return WCS(header, relax=True).celestial

describe_published_product(path, *, role)

Validate one product a finished run published, and describe it.

Each product is read as its writer validated it before publishing it: a catalogue and diagnostics in full, an image by its header. The record is the one the writer returned, so a caller can resume from it.

Source code in src/hebog/io/materialization.py

def describe_published_product(
    path: Path,
    *,
    role: ProductRole,
) -> MaterializedProduct:
    """Validate one product a finished run published, and describe it.

    Each product is read as its writer validated it before publishing it: a
    catalogue and diagnostics in full, an image by its header. The record is
    the one the writer returned, so a caller can resume from it.
    """
    path = Path(path)
    if role == "source-catalogue":
        read_catalogue_fits_product(path)
        return _product_record(
            path,
            role=role,
            media_type=_CATALOGUE_MEDIA_TYPE,
            scientific_status="valid",
        )
    if role == "diagnostics":
        diagnostics = read_diagnostics_product(path)
        return _product_record(
            path,
            role=role,
            media_type=_DIAGNOSTICS_MEDIA_TYPE,
            scientific_status="valid",
            content_schema_version=diagnostics.schema_version,
        )
    status = _image_header_status(path, role)
    _validate_image_path(path, role=role, scientific_status=status)
    return _product_record(
        path,
        role=role,
        media_type=_IMAGE_MEDIA_TYPE,
        scientific_status=status,
    )

read_catalogue_fits_product(product_or_path)

Read and strictly validate an internal catalogue FITS product.

Source code in src/hebog/io/materialization.py

def read_catalogue_fits_product(
    product_or_path: MaterializedProduct | Path,
) -> SourceCatalogue:
    """Read and strictly validate an internal catalogue FITS product."""
    path = _resolve_product_path(
        product_or_path,
        expected_role="source-catalogue",
    )
    try:
        with fits.open(path, mode="readonly", memmap=False) as hdus:
            header = hdus[0].header
            schema_version = header.get("HBGSCHE")
            if schema_version != _CATALOGUE_SCHEMA_VERSION:
                raise UnsupportedMaterializedProductError(
                    f"unsupported catalogue content schema: {schema_version}"
                )
            if header.get("HBGROLE") != "CATALOGUE":
                raise InvalidMaterializedProductError(
                    "catalogue FITS has an invalid product role"
                )
            _require_catalogue_structure(hdus)
            islands = tuple(
                Island(
                    island_id=_text(row["ISLAND_ID"]),
                    pixel_count=int(row["PIXEL_COUNT"]),
                    integrated_flux_jy=float(row["INTEGRATED_FLUX"]),
                    integrated_flux_error_jy=_optional_float(
                        row["INTEGRATED_FLUX_ERROR"]
                    ),
                    local_rms_jy_per_beam=float(row["LOCAL_RMS"]),
                    mean_brightness_jy_per_beam=float(row["MEAN_BRIGHTNESS"]),
                )
                for row in hdus[1].data
            )
            sources = tuple(
                SourceCandidate(
                    source_id=_text(row["SOURCE_ID"]),
                    fitted_shape=_shape_from_row(row, "FITTED"),
                    association_aperture_integrated_flux_jy=(
                        _optional_float(row["ASSOCIATION_APERTURE_FLUX"])
                    ),
                    **_measured_fields(row),
                )
                for row in hdus[2].data
            )
            components: list[GaussianComponent] = []
            for row in hdus[3].data:
                fitted_shape = _shape_from_row(row, "FITTED")
                if fitted_shape is None:
                    raise ValueError(
                        "Gaussian component requires fitted shape"
                    )
                components.append(
                    GaussianComponent(
                        gaussian_component_id=_text(
                            row["GAUSSIAN_COMPONENT_ID"]
                        ),
                        source_id=_text(row["SOURCE_ID"]),
                        fitted_shape=fitted_shape,
                        **_measured_fields(row),
                    )
                )
            return SourceCatalogue(
                catalogue_id=str(header["CATID"]),
                coordinate_frame=cast(
                    Literal["icrs"],
                    str(header["HBGFRAME"]),
                ),
                position_epoch=str(header["HBGEPCH"]),
                reference_frequency_hz=float(header["RESTFRQ"]),
                islands=islands,
                sources=sources,
                gaussian_components=tuple(components),
            )
    except UnsupportedMaterializedProductError:
        raise
    except InvalidMaterializedProductError:
        raise
    except (OSError, ValueError, TypeError, KeyError, IndexError) as error:
        raise InvalidMaterializedProductError(
            f"cannot read catalogue FITS product {path}: {error}"
        ) from error

read_diagnostics_product(product_or_path)

Read and strictly validate canonical diagnostics JSON.

Source code in src/hebog/io/materialization.py

def read_diagnostics_product(
    product_or_path: MaterializedProduct | Path,
) -> DiagnosticsProduct:
    """Read and strictly validate canonical diagnostics JSON."""
    recorded_schema_version = (
        product_or_path.content_schema_version
        if isinstance(product_or_path, MaterializedProduct)
        else None
    )
    path = _resolve_product_path(
        product_or_path,
        expected_role="diagnostics",
    )
    try:
        payload = path.read_bytes()
        raw_document = json.loads(payload)
        if not isinstance(raw_document, dict):
            raise ValueError("diagnostics root must be an object")
        schema_version = raw_document.get("schema_version")
        if (
            recorded_schema_version is not None
            and schema_version != recorded_schema_version
        ):
            raise InvalidMaterializedProductError(
                "diagnostics content schema differs from product record"
            )
        if schema_version == _CONTENT_SCHEMA_VERSION:
            return SourceFindingDiagnostics.from_json_bytes(payload)
        if schema_version == _CONTINUUM_DIAGNOSTICS_SCHEMA_VERSION:
            return ContinuumSourceFindingDiagnostics.from_json_bytes(payload)
        if schema_version == _PUBLIC_DIAGNOSTICS_SCHEMA_VERSION:
            return PublicSourceFindingDiagnostics.from_json_bytes(payload)
        raise UnsupportedMaterializedProductError(
            f"unsupported diagnostics content schema: {schema_version}"
        )
    except UnsupportedMaterializedProductError:
        raise
    except InvalidMaterializedProductError:
        raise
    except (
        OSError,
        UnicodeError,
        json.JSONDecodeError,
        ValidationError,
        ValueError,
    ) as error:
        raise InvalidMaterializedProductError(
            f"cannot read diagnostics product {path}: {error}"
        ) from error

write_catalogue_fits_product(path, catalogue)

Write one validated, idempotent internal catalogue FITS product.

Source code in src/hebog/io/materialization.py

def write_catalogue_fits_product(
    path: Path,
    catalogue: SourceCatalogue,
) -> MaterializedProduct:
    """Write one validated, idempotent internal catalogue FITS product."""

    def write(temporary: Path) -> None:
        with _catalogue_hdus(catalogue) as hdus:
            for hdu in hdus:
                hdu.add_checksum(when=_CATALOGUE_CHECKSUM_COMMENT)
            hdus.writeto(temporary)

    return _materialize(
        path,
        write_temporary=write,
        validate_temporary=read_catalogue_fits_product,
        role="source-catalogue",
        media_type=_CATALOGUE_MEDIA_TYPE,
        scientific_status="valid",
    )

write_diagnostics_product(path, diagnostics)

Write one canonical, idempotent diagnostics JSON product.

Source code in src/hebog/io/materialization.py

def write_diagnostics_product(
    path: Path,
    diagnostics: DiagnosticsProduct,
) -> MaterializedProduct:
    """Write one canonical, idempotent diagnostics JSON product."""

    def write(temporary: Path) -> None:
        temporary.write_bytes(diagnostics.canonical_json_bytes())

    return _materialize(
        path,
        write_temporary=write,
        validate_temporary=read_diagnostics_product,
        role="diagnostics",
        media_type=_DIAGNOSTICS_MEDIA_TYPE,
        scientific_status="valid",
        content_schema_version=diagnostics.schema_version,
    )

write_mask_fits_product(path, metadata, row_blocks)

Write one row-block-bounded binary source-filtering mask.

Source code in src/hebog/io/materialization.py

def write_mask_fits_product(
    path: Path,
    metadata: ImageMetadata,
    row_blocks: Iterable[npt.NDArray[Any]],
) -> MaterializedProduct:
    """Write one row-block-bounded binary source-filtering mask."""
    return _materialize(
        path,
        write_temporary=lambda temporary: _write_mask_temporary(
            temporary,
            metadata,
            row_blocks,
        ),
        validate_temporary=lambda temporary: _validate_image_path(
            temporary,
            role="source-filtering-mask",
            scientific_status="valid",
        ),
        role="source-filtering-mask",
        media_type=_IMAGE_MEDIA_TYPE,
        scientific_status="valid",
    )

write_rms_fits_product(path, metadata, row_blocks, *, dtype, scientific_status)

Write one row-block-bounded RMS FITS product without implicit casts.

Source code in src/hebog/io/materialization.py

def write_rms_fits_product(
    path: Path,
    metadata: ImageMetadata,
    row_blocks: Iterable[npt.NDArray[Any]],
    *,
    dtype: np.dtype[Any],
    scientific_status: str,
) -> MaterializedProduct:
    """Write one row-block-bounded RMS FITS product without implicit casts."""
    dtype = np.dtype(dtype)
    if dtype not in {np.dtype("float32"), np.dtype("float64")}:
        raise InvalidMaterializedProductError(
            "RMS product dtype must be float32 or float64"
        )
    if scientific_status not in {"valid", "unavailable"}:
        raise InvalidMaterializedProductError(
            "RMS scientific status must be valid or unavailable"
        )
    status = cast(Literal["valid", "unavailable"], scientific_status)
    return _materialize(
        path,
        write_temporary=lambda temporary: _write_rms_temporary(
            temporary,
            metadata,
            row_blocks,
            dtype=dtype,
            scientific_status=status,
        ),
        validate_temporary=lambda temporary: _validate_image_path(
            temporary,
            role="rms",
            scientific_status=status,
        ),
        role="rms",
        media_type=_IMAGE_MEDIA_TYPE,
        scientific_status=status,
    )

Errors and entry point

hebog.pipeline

Scheduler-independent public source-finding API.

InvalidSourceFinderInputError

Bases: SourceFinderError

The requested input cannot be read as a supported FITS image.

Source code in src/hebog/pipeline.py

class InvalidSourceFinderInputError(SourceFinderError):
    """The requested input cannot be read as a supported FITS image."""

SourceFinderError

Bases: Exception

Base class for failures at the public source-finding boundary.

Source code in src/hebog/pipeline.py

class SourceFinderError(Exception):
    """Base class for failures at the public source-finding boundary."""

SourceFinderImageTooLargeError

Bases: SourceFinderError

An input exceeds the public finder's bounded image-size envelope.

Source code in src/hebog/pipeline.py

class SourceFinderImageTooLargeError(SourceFinderError):
    """An input exceeds the public finder's bounded image-size envelope."""

SourceFinderOutputExistsError

Bases: SourceFinderError

The output destination exists without this request's products.

Source code in src/hebog/pipeline.py

class SourceFinderOutputExistsError(SourceFinderError):
    """The output destination exists without this request's products."""

SourceFinderStagingWarning

Bases: UserWarning

An earlier run left a hidden staging directory beside the output.

A run that is killed cannot remove its staging directory. The next run to the same output removes one whose owner has provably stopped and leaves any other in place; it reports each with this warning.

Source code in src/hebog/pipeline.py

class SourceFinderStagingWarning(UserWarning):
    """An earlier run left a hidden staging directory beside the output.

    A run that is killed cannot remove its staging directory. The next run
    to the same output removes one whose owner has provably stopped and
    leaves any other in place; it reports each with this warning.
    """

UnsupportedSourceFinderConfigurationError

Bases: SourceFinderError

The requested science cannot be executed by the public pipeline.

Source code in src/hebog/pipeline.py

class UnsupportedSourceFinderConfigurationError(SourceFinderError):
    """The requested science cannot be executed by the public pipeline."""

analysis_memory_bytes(image_shape_yx, *, expected_sources)

Return the memory one analysis holds in the process that runs it.

find_sources keeps Hebog's modules, the records each pass returns for every source, and the products it publishes in the process that calls it, so what it holds grows with the sources it finds, and with the image through the background's noise grids. Its tasks are not counted: each declares its own memory to the executor, which a DaskExecutor attaches to the task under the caller's memory resource. Reserve this amount for the task that calls find_sources, for example under the same Dask resource. With a SerialExecutor or ThreadExecutor the tasks run in the calling process, so the memory they are admitted comes on top. A request with a flat-noise image is covered.

Parameters:

Name	Type	Description	Default
image_shape_yx	tuple[int, int]	The input image's height and width in pixels.	required
expected_sources	int	An upper estimate of the sources the catalogue will hold, such as :func:expected_source_count makes from the image; the reservation is too small if more are found.	required

Returns:

Type	Description
int	A fixed amount, an amount per expected source and two and a half
int	bytes a pixel, with a quarter of headroom, in bytes.

Raises:

Type	Description
ValueError	If a dimension is not a positive integer or the source count is not a non-negative integer.

Examples:

>>> analysis_memory_bytes(
...     (3_000, 4_000), expected_sources=10
... ) == analysis_memory_bytes((4_000, 3_000), expected_sources=10)
True

Source code in src/hebog/pipeline.py

def analysis_memory_bytes(
    image_shape_yx: tuple[int, int], *, expected_sources: int
) -> int:
    """Return the memory one analysis holds in the process that runs it.

    ``find_sources`` keeps Hebog's modules, the records each pass returns
    for every source, and the products it publishes in the process that
    calls it, so what it holds grows with the sources it finds, and with
    the image through the background's noise grids. Its tasks are
    not counted: each declares its own memory to the executor, which a
    ``DaskExecutor`` attaches to the task under the caller's memory
    resource. Reserve this amount for the task that calls
    ``find_sources``, for example under the same Dask resource. With a
    ``SerialExecutor`` or ``ThreadExecutor`` the tasks run in the calling
    process, so the memory they are admitted comes on top. A request with a
    flat-noise image is covered.

    Args:
        image_shape_yx: The input image's height and width in pixels.
        expected_sources: An upper estimate of the sources the catalogue
            will hold, such as :func:`expected_source_count` makes from the
            image; the reservation is too small if more are found.

    Returns:
        A fixed amount, an amount per expected source and two and a half
        bytes a pixel, with a quarter of headroom, in bytes.

    Raises:
        ValueError: If a dimension is not a positive integer or the source
            count is not a non-negative integer.

    Examples:
        >>> analysis_memory_bytes(
        ...     (3_000, 4_000), expected_sources=10
        ... ) == analysis_memory_bytes((4_000, 3_000), expected_sources=10)
        True
    """
    for side in image_shape_yx:
        if (
            isinstance(side, bool)
            or not isinstance(side, Integral)
            or side < 1
        ):
            raise ValueError("image dimensions must be positive integers")
    if (
        isinstance(expected_sources, bool)
        or not isinstance(expected_sources, Integral)
        or expected_sources < 0
    ):
        raise ValueError("expected_sources must be a non-negative integer")
    height, width = image_shape_yx
    traced = (
        _ANALYSIS_TRACED_BASE_BYTES
        + _ANALYSIS_TRACED_BYTES_PER_SOURCE * int(expected_sources)
        + int(height)
        * int(width)
        * _ANALYSIS_TRACED_PIXEL_BYTES_NUMERATOR
        // _ANALYSIS_TRACED_PIXEL_BYTES_DENOMINATOR
    )
    return -(-traced * _HEADROOM_NUMERATOR // _HEADROOM_DENOMINATOR)

expected_source_count(image_path, *, detection_threshold_sigma, supplied_metadata=None)

Return an upper estimate of the sources find_sources publishes.

Pass it to :func:analysis_memory_bytes. A deconvolution model's component list is no such estimate: it holds many components for each source. A published source peaks at least the detection threshold times the local noise above the background, so the estimate counts the pixels that are the highest of their 3x3 neighbourhood and at least detection_threshold_sigma times the noise above the median of their 512-pixel block. A block's noise is its valid pixels' 3-sigma-clipped standard deviation, or the median of every block's if that is lower, so neither the bright sources in a block nor a primary-beam correction's noisier edges hide faint sources. A noise peak counts as a source would, so the estimate errs high; on 21 development cases it bounded every published source count, at 1.0 to 21 times it: 1.1 to 1.9 times on the LOFAR and simulated SKA fields, and most where noise peaks sit on one wide source. The image is read twice in bounded blocks, so no image-sized array is held. An image find_sources refuses is refused in the same way before any pixel is read.

Parameters:

Name	Type	Description	Default
image_path	Path	The FITS image find_sources will analyse.	required
detection_threshold_sigma	float	The detection threshold the analysis will use, SourceFinderConfig.detection_threshold_sigma.	required
supplied_metadata	SuppliedImageMetadata | None	Values the header lacks, as the request gives them.	None

Returns:

Type	Description
int	The number of such peaks; 0 for an image with no measurable noise.

Raises:

Type	Description
ValueError	If the threshold is not finite and positive.
InvalidSourceFinderInputError	If the image cannot be read as a FITS continuum image.
UnsupportedSourceFinderConfigurationError	If find_sources refuses the image's unit, frame or beam sampling.
SourceFinderImageTooLargeError	If the image exceeds the size envelope find_sources admits.

Source code in src/hebog/pipeline.py

def expected_source_count(
    image_path: Path,
    *,
    detection_threshold_sigma: float,
    supplied_metadata: SuppliedImageMetadata | None = None,
) -> int:
    """Return an upper estimate of the sources ``find_sources`` publishes.

    Pass it to :func:`analysis_memory_bytes`. A deconvolution model's
    component list is no such estimate: it holds many components for each
    source. A published source peaks at least the detection threshold times
    the local noise above the background, so the estimate counts the pixels
    that are the highest of their 3x3 neighbourhood and at least
    ``detection_threshold_sigma`` times the noise above the median of their
    512-pixel block. A block's noise is its valid pixels' 3-sigma-clipped
    standard deviation, or the median of every block's if that is lower, so
    neither the bright sources in a block nor a primary-beam correction's
    noisier edges hide faint sources. A noise peak counts as a source would,
    so the estimate errs high; on 21 development cases it bounded every
    published source count, at 1.0 to 21 times it: 1.1 to 1.9 times on the
    LOFAR and simulated SKA fields, and most where noise peaks sit on one
    wide source. The image is read twice in bounded blocks, so no
    image-sized array is held. An image ``find_sources`` refuses is
    refused in the same way before any pixel is read.

    Args:
        image_path: The FITS image ``find_sources`` will analyse.
        detection_threshold_sigma: The detection threshold the analysis
            will use, ``SourceFinderConfig.detection_threshold_sigma``.
        supplied_metadata: Values the header lacks, as the request gives
            them.

    Returns:
        The number of such peaks; 0 for an image with no measurable noise.

    Raises:
        ValueError: If the threshold is not finite and positive.
        InvalidSourceFinderInputError: If the image cannot be read as a
            FITS continuum image.
        UnsupportedSourceFinderConfigurationError: If ``find_sources``
            refuses the image's unit, frame or beam sampling.
        SourceFinderImageTooLargeError: If the image exceeds the size
            envelope ``find_sources`` admits.
    """
    from hebog.public_api import (  # noqa: PLC0415
        expected_source_count as _expected_source_count,
    )

    return _expected_source_count(
        image_path,
        detection_threshold_sigma=detection_threshold_sigma,
        supplied_metadata=supplied_metadata,
    )

find_sources(request, config, executor)

Analyse one supported FITS image and atomically publish its products.

When the output already holds the products of a finished run of the same request, with the same run ID, inputs, configuration and science, they are returned without analysing the image again. A failed run raises only once no task it submitted is still running, and removes everything it staged. A killed run leaves a hidden staging directory, which the next run to the same output removes once its owner has provably stopped or reports with a SourceFinderStagingWarning.

The implementation is imported only when called so importing the public scheduler-independent API never loads a concrete I/O or scheduler layer.

Source code in src/hebog/pipeline.py

def find_sources(
    request: SourceFinderRequest,
    config: SourceFinderConfig,
    executor: Executor,
) -> SourceFinderResult:
    """Analyse one supported FITS image and atomically publish its products.

    When the output already holds the products of a finished run of the
    same request, with the same run ID, inputs, configuration and science,
    they are returned without analysing the image again. A failed run
    raises only once no task it submitted is still running, and removes
    everything it staged. A killed run leaves a hidden staging directory,
    which the next run to the same output removes once its owner has
    provably stopped or reports with a ``SourceFinderStagingWarning``.

    The implementation is imported only when called so importing the public
    scheduler-independent API never loads a concrete I/O or scheduler layer.
    """
    from hebog.public_api import find_sources as _find_sources  # noqa: PLC0415

    return _find_sources(request, config, executor)