Utilities

various

affine_case_errors

affine_case_errors(predicted, target, ranges, *, valid=None, empty=float('inf'))

Return Keijzer-scaled case errors without changing the tree.

Fits \(a + b\,f(x)\) with :func:~deap_er.tools.affine_scale on the same valid= mask :func:~deap_er.tools.case_errors uses, applies the scaled series, then reduces to one MSE per case. This is the Darwinian default next to Lamarckian :func:~deap_er.gp.write_affine_scale.

Parameters:

Name	Type	Description	Default
predicted	ndarray	Predicted series f(x).	required
target	ndarray	Target series, same length as predicted.	required
ranges	Sequence[tuple[int, int]] | ndarray	Case bounds forwarded to :func:~deap_er.tools.case_errors.	required
valid	ndarray | None	Optional per-sample mask forwarded to both helpers.	None
empty	float	Value returned when a case has no scorable samples.	float('inf')

Returns:

Type	Description
tuple[float, ...]	One MSE per case, in range order.

Raises:

Type	Description
ValueError	If the inputs are not aligned one-dimensional arrays or if valid has the wrong shape.

Source code in deap_er/private/various/affine_case_errors.py

def affine_case_errors(
    predicted: numpy.ndarray,
    target: numpy.ndarray,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    empty: float = float("inf"),
) -> tuple[float, ...]:
    r"""Return Keijzer-scaled case errors without changing the tree.

    Fits $a + b\,f(x)$ with :func:`~deap_er.tools.affine_scale` on the
    same ``valid=`` mask :func:`~deap_er.tools.case_errors` uses, applies
    the scaled series, then reduces to one MSE per case. This is the
    Darwinian default next to Lamarckian
    :func:`~deap_er.gp.write_affine_scale`.

    Args:
        predicted: Predicted series ``f(x)``.
        target: Target series, same length as ``predicted``.
        ranges: Case bounds forwarded to :func:`~deap_er.tools.case_errors`.
        valid: Optional per-sample mask forwarded to both helpers.
        empty: Value returned when a case has no scorable samples.

    Returns:
        One MSE per case, in range order.

    Raises:
        ValueError: If the inputs are not aligned one-dimensional
            arrays or if ``valid`` has the wrong shape.
    """
    intercept, slope = affine_scale(predicted, target, valid=valid)
    scaled = intercept + slope * numpy.asarray(predicted, dtype=numpy.float64)
    return case_errors(scaled, target, ranges, valid=valid, empty=empty)

affine_scale

affine_scale(predicted, target, *, valid=None)

Fit Keijzer intercept and slope for a + b * f(x).

Least-squares a and b are computed on the same scorable samples :func:~deap_er.tools.case_errors uses: an optional valid= mask intersected with the finite check on both series. Darwinian callers apply a + b * predicted only when writing fitness or case errors; the tree is unchanged.

A series with no scorable samples returns the identity (0.0, 1.0). A constant prediction returns an intercept-only shift (mean(target) - mean(predicted), 1.0).

Parameters:

Name	Type	Description	Default
predicted	ndarray	Predicted series f(x).	required
target	ndarray	Target series, same length as predicted.	required
valid	ndarray | None	Optional per-sample mask. Same contract as :func:~deap_er.tools.case_errors.	None

Returns:

Type	Description
tuple[float, float]	(a, b) so the scaled series is a + b * predicted.

Raises:

Type	Description
ValueError	If the inputs are not aligned one-dimensional arrays, or if valid has the wrong shape.

Source code in deap_er/private/various/affine_scale.py

def affine_scale(
    predicted: numpy.ndarray,
    target: numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
) -> tuple[float, float]:
    """Fit Keijzer intercept and slope for ``a + b * f(x)``.

    Least-squares ``a`` and ``b`` are computed on the same scorable
    samples :func:`~deap_er.tools.case_errors` uses: an optional
    ``valid=`` mask intersected with the finite check on both series.
    Darwinian callers apply ``a + b * predicted`` only when writing
    fitness or case errors; the tree is unchanged.

    A series with no scorable samples returns the identity
    ``(0.0, 1.0)``. A constant prediction returns an intercept-only
    shift ``(mean(target) - mean(predicted), 1.0)``.

    Args:
        predicted: Predicted series ``f(x)``.
        target: Target series, same length as ``predicted``.
        valid: Optional per-sample mask. Same contract as
            :func:`~deap_er.tools.case_errors`.

    Returns:
        ``(a, b)`` so the scaled series is ``a + b * predicted``.

    Raises:
        ValueError: If the inputs are not aligned one-dimensional
            arrays, or if ``valid`` has the wrong shape.
    """
    predicted, target = _as_series(predicted, target)
    sample_valid = case_valid_mask(predicted, target, valid)
    if not numpy.any(sample_valid):
        return 0.0, 1.0
    forecast = predicted[sample_valid]
    observed = target[sample_valid]
    forecast_mean = float(numpy.mean(forecast))
    observed_mean = float(numpy.mean(observed))
    centered = forecast - forecast_mean
    denom = float(numpy.dot(centered, centered))
    if not numpy.isfinite(denom) or denom <= 0.0:
        return observed_mean - forecast_mean, 1.0
    slope = float(numpy.dot(observed - observed_mean, centered) / denom)
    intercept = observed_mean - slope * forecast_mean
    return intercept, slope

bin2float

bin2float(min_, max_, n_bits)

Return a decorator that decodes a binary individual to floats.

Each float uses n_bits bits and is mapped into [min_, max_]. The decorated function receives the decoded float array.

Parameters:

Name	Type	Description	Default
min_	float	Lower bound of each decoded value.	required
max_	float	Upper bound of each decoded value.	required
n_bits	int	Bits used to encode each float.	required

Returns:

Type	Description
Callable[..., Any]	A decorator for an evaluation function.

Source code in deap_er/private/various/bin2float.py

def bin2float(min_: float, max_: float, n_bits: int) -> Callable[..., Any]:
    """Return a decorator that decodes a binary individual to floats.

    Each float uses ``n_bits`` bits and is mapped into
    ``[min_, max_]``. The decorated function receives the decoded
    float array.

    Args:
        min_: Lower bound of each decoded value.
        max_: Upper bound of each decoded value.
        n_bits: Bits used to encode each float.

    Returns:
        A decorator for an evaluation function.
    """

    def wrapper(function: Callable[..., Any]) -> Callable[..., Any]:
        @wraps(function)
        def wrapped(individual: Any, *args: Any, **kwargs: Any) -> Any:
            nelem = len(individual) // n_bits
            decoded = [0] * nelem
            div = 2**n_bits - 1
            span = max_ - min_
            for i in range(nelem):
                gene = 0
                start = i * n_bits
                for bit in individual[start : start + n_bits]:
                    gene = (gene << 1) | (1 if bit else 0)
                decoded[i] = min_ + ((gene / div) * span)
            return function(decoded, *args, **kwargs)

        return wrapped

    return wrapper

case_bounds

ranges_from_mask(mask, length)

Split a boolean mask into one half-open range per True run.

Parameters:

Name	Type	Description	Default
mask	ndarray	One-dimensional bool mask.	required
length	int	Expected mask length.	required

Returns:

Type	Description
list[tuple[int, int]]	Contiguous [start, stop) runs of True.

Raises:

Type	Description
ValueError	If mask is not a 1-D bool array of length.

Source code in deap_er/private/various/case_bounds.py

def ranges_from_mask(mask: numpy.ndarray, length: int) -> list[tuple[int, int]]:
    """Split a boolean mask into one half-open range per ``True`` run.

    Args:
        mask: One-dimensional ``bool`` mask.
        length: Expected mask length.

    Returns:
        Contiguous ``[start, stop)`` runs of ``True``.

    Raises:
        ValueError: If ``mask`` is not a 1-D ``bool`` array of ``length``.
    """
    if mask.ndim != 1:
        raise ValueError("a boolean mask must be one-dimensional")
    if mask.shape[0] != length:
        raise ValueError("a boolean mask must match the series length")
    if mask.dtype != bool:
        raise ValueError("a boolean mask must have dtype bool")
    indices = numpy.flatnonzero(mask)
    if indices.size == 0:
        return []
    breaks = numpy.flatnonzero(numpy.diff(indices) > 1) + 1
    starts = numpy.split(indices, breaks)
    return [(int(run[0]), int(run[-1]) + 1) for run in starts]

normalize_case_ranges(ranges, length)

Normalize ranges or a mask into validated half-open intervals.

Parameters:

Name	Type	Description	Default
ranges	Sequence[tuple[int, int]] | ndarray	(start, stop) pairs, a (n_cases, 2) integer array, or a one-dimensional bool mask.	required
length	int	Exclusive upper bound for endpoints, or the mask length.	required

Returns:

Type	Description
list[tuple[int, int]]	Validated [start, stop) intervals in caller order.

Raises:

Type	Description
ValueError	If a range or mask is invalid.

Source code in deap_er/private/various/case_bounds.py

def normalize_case_ranges(
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    length: int,
) -> list[tuple[int, int]]:
    """Normalize ranges or a mask into validated half-open intervals.

    Args:
        ranges: ``(start, stop)`` pairs, a ``(n_cases, 2)`` integer array,
            or a one-dimensional ``bool`` mask.
        length: Exclusive upper bound for endpoints, or the mask length.

    Returns:
        Validated ``[start, stop)`` intervals in caller order.

    Raises:
        ValueError: If a range or mask is invalid.
    """
    if isinstance(ranges, numpy.ndarray):
        return _ranges_from_array(ranges, length)
    return [_validate_interval(start, stop, length) for start, stop in ranges]

mask_from_ranges(ranges, length)

Paint validated ranges onto a boolean mask of length.

Parameters:

Name	Type	Description	Default
ranges	Sequence[tuple[int, int]] | ndarray	The same range or mask input as normalize_case_ranges.	required
length	int	Length of the returned mask.	required

Returns:

Type	Description
ndarray	A 1-D bool mask that is True on every painted interval.

Raises:

Type	Description
ValueError	If a range or mask is invalid.

Source code in deap_er/private/various/case_bounds.py

def mask_from_ranges(
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    length: int,
) -> numpy.ndarray:
    """Paint validated ranges onto a boolean mask of ``length``.

    Args:
        ranges: The same range or mask input as ``normalize_case_ranges``.
        length: Length of the returned mask.

    Returns:
        A 1-D ``bool`` mask that is ``True`` on every painted interval.

    Raises:
        ValueError: If a range or mask is invalid.
    """
    mask = numpy.zeros(length, dtype=bool)
    for start, stop in normalize_case_ranges(ranges, length):
        mask[start:stop] = True
    return mask

case_errors

case_intervals(ranges, length)

Normalize case bounds to half-open [start, stop) intervals.

Parameters:

Name	Type	Description	Default
ranges	Sequence[tuple[int, int]] | ndarray	Explicit (start, stop) pairs, a (n_cases, 2) integer array, or a one-dimensional bool mask.	required
length	int	Length of the series the intervals index.	required

Returns:

Type	Description
list[tuple[int, int]]	Half-open intervals in caller order.

Raises:

Type	Description
ValueError	If a bound is invalid or a mask has the wrong shape or dtype.

Source code in deap_er/private/various/case_errors.py

def case_intervals(
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    length: int,
) -> list[tuple[int, int]]:
    """Normalize case bounds to half-open ``[start, stop)`` intervals.

    Args:
        ranges: Explicit ``(start, stop)`` pairs, a ``(n_cases, 2)`` integer
            array, or a one-dimensional ``bool`` mask.
        length: Length of the series the intervals index.

    Returns:
        Half-open intervals in caller order.

    Raises:
        ValueError: If a bound is invalid or a mask has the wrong shape
            or dtype.
    """
    return normalize_case_ranges(ranges, length)

case_valid_mask(predicted, target, valid=None)

Return the one-dimensional sample mask used by :func:case_errors.

An optional caller mask is intersected with the finite check on both series. A True in valid that is non-finite on either series stays out of the mask.

Parameters:

Name	Type	Description	Default
predicted	ndarray	Predicted series.	required
target	ndarray	Target series, same length as predicted.	required
valid	ndarray | None	Optional per-sample mask.	None

Returns:

Type	Description
ndarray	One-dimensional bool mask of scorable samples.

Raises:

Type	Description
ValueError	If valid is not a one-dimensional mask matching the series length.

Source code in deap_er/private/various/case_errors.py

def case_valid_mask(
    predicted: numpy.ndarray,
    target: numpy.ndarray,
    valid: numpy.ndarray | None = None,
) -> numpy.ndarray:
    """Return the one-dimensional sample mask used by :func:`case_errors`.

    An optional caller mask is intersected with the finite check on both
    series. A ``True`` in ``valid`` that is non-finite on either series
    stays out of the mask.

    Args:
        predicted: Predicted series.
        target: Target series, same length as ``predicted``.
        valid: Optional per-sample mask.

    Returns:
        One-dimensional ``bool`` mask of scorable samples.

    Raises:
        ValueError: If ``valid`` is not a one-dimensional mask matching
            the series length.
    """
    finite = _finite_mask(predicted, target)
    if valid is None:
        return finite
    sample_valid = numpy.asarray(valid, dtype=bool)
    if sample_valid.ndim != 1 or sample_valid.shape[0] != predicted.shape[0]:
        raise ValueError("valid must be a one-dimensional mask matching the series length")
    return sample_valid & finite

case_errors(predicted, target, ranges, *, valid=None, empty=float('inf'))

Return one mean-squared error per case segment.

Each case is a half-open interval [start, stop) over aligned predicted and target series. Only samples marked valid inside the interval are scored. By default a sample is valid when both series are finite at that index.

Pass an explicit valid mask when comparisons or vwhere can hide a nan warmup while the prediction stays finite. That mask is intersected with the finite check, so non-finite samples never enter the MSE even when valid marks them True. Segment boundaries stay on the caller; this helper does not split a series chronologically.

Parameters:

Name	Type	Description	Default
predicted	ndarray	Predicted series.	required
target	ndarray	Target series, same length as predicted.	required
ranges	Sequence[tuple[int, int]] | ndarray	A sequence of (start, stop) pairs, a (n_cases, 2) integer array of half-open bounds, or a one-dimensional bool mask. A mask defines one case per contiguous run of True values.	required
valid	ndarray | None	Optional per-sample mask intersected with the finite check. Use it to drop trusted prefixes such as hidden warmup.	None
empty	float	Value returned when a case has no scorable samples.	float('inf')

Returns:

Type	Description
tuple[float, ...]	One MSE per case, in range order.

Raises:

Type	Description
ValueError	If the inputs are not aligned one-dimensional arrays, if range endpoints are invalid, or if a mask has the wrong shape or dtype.

Source code in deap_er/private/various/case_errors.py

def case_errors(
    predicted: numpy.ndarray,
    target: numpy.ndarray,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    empty: float = float("inf"),
) -> tuple[float, ...]:
    """Return one mean-squared error per case segment.

    Each case is a half-open interval ``[start, stop)`` over aligned
    ``predicted`` and ``target`` series. Only samples marked valid inside
    the interval are scored. By default a sample is valid when both series
    are finite at that index.

    Pass an explicit ``valid`` mask when comparisons or ``vwhere`` can
    hide a ``nan`` warmup while the prediction stays finite. That mask is
    intersected with the finite check, so non-finite samples never enter
    the MSE even when ``valid`` marks them ``True``. Segment boundaries
    stay on the caller; this helper does not split a series
    chronologically.

    Args:
        predicted: Predicted series.
        target: Target series, same length as ``predicted``.
        ranges: A sequence of ``(start, stop)`` pairs, a ``(n_cases, 2)``
            integer array of half-open bounds, or a one-dimensional
            ``bool`` mask. A mask defines one case per contiguous run of
            ``True`` values.
        valid: Optional per-sample mask intersected with the finite check.
            Use it to drop trusted prefixes such as hidden warmup.
        empty: Value returned when a case has no scorable samples.

    Returns:
        One MSE per case, in range order.

    Raises:
        ValueError: If the inputs are not aligned one-dimensional arrays,
            if range endpoints are invalid, or if a mask has the wrong
            shape or dtype.
    """
    predicted, target, length = _as_series(predicted, target)
    intervals = case_intervals(ranges, length)
    sample_valid = case_valid_mask(predicted, target, valid)
    return tuple(
        _case_mse(predicted, target, sample_valid, start, stop, empty) for start, stop in intervals
    )

case_generalization

CaseGeneralizationRecipe(pool, train_cases, held_cases, n_cases) dataclass

Wiring for the default case-structured generalization path.

Attributes:

Name	Type	Description
pool	CaseExamPool	Train and held-out exams.
train_cases	tuple[int, ...]	Train catalog indices.
held_cases	tuple[int, ...]	Held-out catalog indices, or [] when unset.
n_cases	int	Catalog length.

make_select(*, downsample=None, downsample_mode='informed')

Return a lexicase selector on :attr:train_cases only.

Source code in deap_er/private/various/case_generalization.py

def make_select(
    self,
    *,
    downsample: int | None = None,
    downsample_mode: DownsampleMode = "informed",
) -> Callable[[list[Individual], int], list[Individual]]:
    """Return a lexicase selector on :attr:`train_cases` only."""
    return make_lexicase_train_select(
        self.pool,
        self.n_cases,
        downsample=downsample,
        downsample_mode=downsample_mode,
    )

held_out_tail(n_cases, fraction=0.2)

Return catalog indices for the last fraction of cases.

Chronological meaning stays on the caller. This helper only picks trailing catalog indices.

Parameters:

Name	Type	Description	Default
n_cases	int	Catalog length.	required
fraction	float	Held-out share in (0, 1).	0.2

Returns:

Type	Description
list[int]	Ascending held-out indices.

Raises:

Type	Description
ValueError	If n_cases or fraction is invalid.

Source code in deap_er/private/various/case_generalization.py

def held_out_tail(n_cases: int, fraction: float = 0.2) -> list[int]:
    """Return catalog indices for the last ``fraction`` of cases.

    Chronological meaning stays on the caller. This helper only picks
    trailing catalog indices.

    Args:
        n_cases: Catalog length.
        fraction: Held-out share in ``(0, 1)``.

    Returns:
        Ascending held-out indices.

    Raises:
        ValueError: If ``n_cases`` or ``fraction`` is invalid.
    """
    _validate_n_cases(n_cases)
    _validate_fraction(fraction)
    held_count = max(1, min(n_cases - 1, round(n_cases * fraction)))
    start = n_cases - held_count
    return list(range(start, n_cases))

train_head(n_cases, fraction=0.2)

Return train catalog indices complementary to :func:held_out_tail.

Parameters:

Name	Type	Description	Default
n_cases	int	Catalog length.	required
fraction	float	Held-out share in (0, 1).	0.2

Returns:

Type	Description
list[int]	Ascending train indices.

Raises:

Type	Description
ValueError	If n_cases or fraction is invalid.

Source code in deap_er/private/various/case_generalization.py

def train_head(n_cases: int, fraction: float = 0.2) -> list[int]:
    """Return train catalog indices complementary to :func:`held_out_tail`.

    Args:
        n_cases: Catalog length.
        fraction: Held-out share in ``(0, 1)``.

    Returns:
        Ascending train indices.

    Raises:
        ValueError: If ``n_cases`` or ``fraction`` is invalid.
    """
    held = set(held_out_tail(n_cases, fraction))
    return [idx for idx in range(n_cases) if idx not in held]

case_generalization_pool(n_cases, *, fraction=0.2, held_cases=None)

Build a pool with one train exam and a caller-marked held-out exam.

Parameters:

Name	Type	Description	Default
n_cases	int	Catalog length.	required
fraction	float	Held-out share when held_cases is omitted.	0.2
held_cases	Sequence[int] | None	Optional explicit held-out catalog indices.	None

Returns:

Type	Description
CaseExamPool	A pool whose train exam excludes held_out.

Raises:

Type	Description
ValueError	If n_cases or fraction is invalid.
IndexError	If a held-out index is out of range.

Source code in deap_er/private/various/case_generalization.py

def case_generalization_pool(
    n_cases: int,
    *,
    fraction: float = 0.2,
    held_cases: Sequence[int] | None = None,
) -> CaseExamPool:
    """Build a pool with one train exam and a caller-marked held-out exam.

    Args:
        n_cases: Catalog length.
        fraction: Held-out share when ``held_cases`` is omitted.
        held_cases: Optional explicit held-out catalog indices.

    Returns:
        A pool whose train exam excludes ``held_out``.

    Raises:
        ValueError: If ``n_cases`` or ``fraction`` is invalid.
        IndexError: If a held-out index is out of range.
    """
    _validate_n_cases(n_cases)
    chosen = list(held_cases) if held_cases is not None else held_out_tail(n_cases, fraction)
    if not chosen:
        raise ValueError("held_cases must select at least one case")
    held = CaseExam.from_cases(chosen, n_cases)
    train_indices = [idx for idx in range(n_cases) if idx not in set(chosen)]
    if not train_indices:
        raise ValueError("held_cases must leave at least one train case")
    train = CaseExam.from_cases(train_indices, n_cases)
    return CaseExamPool([train], held_out=held)

make_lexicase_train_select(pool, n_cases, *, downsample=None, downsample_mode='informed')

Return a toolbox.select callable that runs lexicase on train cases.

Held-out catalog indices from pool.held_out are never passed to lexicase. Chronological meaning stays on the caller.

Parameters:

Name	Type	Description	Default
pool	CaseExamPool	Exam pool with a train exam and optional held_out.	required
n_cases	int	Catalog length for CaseExam.as_cases.	required
downsample	int | None	When set, cap the active case count each generation via :func:~deap_er.operators.next_downsample_cases.	None
downsample_mode	DownsampleMode	Downsample mode when downsample is set.	'informed'

Returns:

Type	Description
Callable[[list[Individual], int], list[Individual]]	A selection callable (individuals, sel_count) -> list.

Raises:

Type	Description
ValueError	If the pool has no train exams.

Source code in deap_er/private/various/case_generalization.py

def make_lexicase_train_select(
    pool: CaseExamPool,
    n_cases: int,
    *,
    downsample: int | None = None,
    downsample_mode: DownsampleMode = "informed",
) -> Callable[[list[Individual], int], list[Individual]]:
    """Return a ``toolbox.select`` callable that runs lexicase on train cases.

    Held-out catalog indices from ``pool.held_out`` are never passed to
    lexicase. Chronological meaning stays on the caller.

    Args:
        pool: Exam pool with a train exam and optional ``held_out``.
        n_cases: Catalog length for ``CaseExam.as_cases``.
        downsample: When set, cap the active case count each generation
            via :func:`~deap_er.operators.next_downsample_cases`.
        downsample_mode: Downsample mode when ``downsample`` is set.

    Returns:
        A selection callable ``(individuals, sel_count) -> list``.

    Raises:
        ValueError: If the pool has no train exams.
    """
    if not pool.exams:
        raise ValueError("pool must contain at least one train exam")
    train_cases = pool.exams[0].as_cases(n_cases)
    train_set = set(train_cases)
    held_set = set(pool.held_out.as_cases(n_cases)) if pool.held_out is not None else set()
    generation = 0

    def select(individuals: list[Individual], sel_count: int) -> list[Individual]:
        nonlocal generation
        matrix = fitness_case_matrix(individuals)
        if downsample is None:
            cases = train_cases
        else:
            sampled = next_downsample_cases(
                individuals,
                downsample,
                generation,
                mode=downsample_mode,
                held_out=pool.held_out,
            )
            cases = [case for case in sampled if case in train_set]
            if not cases:
                cases = train_cases
        if held_set.intersection(cases):
            raise ValueError("held-out cases must not reach lexicase selection")
        chosen = sel_lexicase(individuals, sel_count, cases=cases, matrix=matrix)
        generation += 1
        return chosen

    return select

case_generalization_recipe(n_cases, *, fraction=0.2, held_cases=None)

Build the held-out pool and catalog indices for item 41.

Parameters:

Name	Type	Description	Default
n_cases	int	Catalog length.	required
fraction	float	Held-out share when held_cases is omitted.	0.2
held_cases	Sequence[int] | None	Optional explicit held-out catalog indices.	None

Returns:

Type	Description
CaseGeneralizationRecipe	A recipe with pool, train indices, and held-out indices.

Source code in deap_er/private/various/case_generalization.py

def case_generalization_recipe(
    n_cases: int,
    *,
    fraction: float = 0.2,
    held_cases: Sequence[int] | None = None,
) -> CaseGeneralizationRecipe:
    """Build the held-out pool and catalog indices for item 41.

    Args:
        n_cases: Catalog length.
        fraction: Held-out share when ``held_cases`` is omitted.
        held_cases: Optional explicit held-out catalog indices.

    Returns:
        A recipe with ``pool``, train indices, and held-out indices.
    """
    pool = case_generalization_pool(n_cases, fraction=fraction, held_cases=held_cases)
    train = tuple(pool.exams[0].as_cases(n_cases))
    held = tuple(pool.held_out.as_cases(n_cases)) if pool.held_out is not None else ()
    return CaseGeneralizationRecipe(pool=pool, train_cases=train, held_cases=held, n_cases=n_cases)

case_halving

CaseHalvingResult(survivors, nevals, stages_run) dataclass

Outcome of :func:evaluate_case_halving.

Attributes:

Name	Type	Description
survivors	list[Individual]	Individuals that reached the final rung.
nevals	int	Case-eval units charged across all rungs.
stages_run	int	Number of rungs executed.

case_eval_charge(n_individuals, n_cases)

Return case-eval units for budget accounting.

Parameters:

Name	Type	Description	Default
n_individuals	int	Individuals scored at one rung.	required
n_cases	int	Cases scored per individual.	required

Returns:

Type	Description
int	n_individuals * n_cases.

Raises:

Type	Description
ValueError	If either count is negative.

Source code in deap_er/private/various/case_halving.py

def case_eval_charge(n_individuals: int, n_cases: int) -> int:
    """Return case-eval units for budget accounting.

    Args:
        n_individuals: Individuals scored at one rung.
        n_cases: Cases scored per individual.

    Returns:
        ``n_individuals * n_cases``.

    Raises:
        ValueError: If either count is negative.
    """
    if n_individuals < 0 or n_cases < 0:
        raise ValueError("n_individuals and n_cases must be non-negative")
    return n_individuals * n_cases

case_halving_stages(n_train_cases, *, eta=2, min_cases=1)

Return successive-halving rung sizes up to n_train_cases.

Each rung uses the first stage_n entries of the caller's train_cases list. Sizes grow geometrically by eta from min_cases until the full train count is reached.

Parameters:

Name	Type	Description	Default
n_train_cases	int	Number of train catalog indices.	required
eta	int	Elimination factor and geometric growth base. Must be at least 2.	2
min_cases	int	Smallest rung size. Must be at least 1.	1

Returns:

Type	Description
tuple[int, ...]	Distinct ascending rung sizes ending at n_train_cases.

Raises:

Type	Description
ValueError	If inputs are invalid.

Source code in deap_er/private/various/case_halving.py

def case_halving_stages(
    n_train_cases: int,
    *,
    eta: int = 2,
    min_cases: int = 1,
) -> tuple[int, ...]:
    """Return successive-halving rung sizes up to ``n_train_cases``.

    Each rung uses the first ``stage_n`` entries of the caller's
    ``train_cases`` list. Sizes grow geometrically by ``eta`` from
    ``min_cases`` until the full train count is reached.

    Args:
        n_train_cases: Number of train catalog indices.
        eta: Elimination factor and geometric growth base. Must be
            at least ``2``.
        min_cases: Smallest rung size. Must be at least ``1``.

    Returns:
        Distinct ascending rung sizes ending at ``n_train_cases``.

    Raises:
        ValueError: If inputs are invalid.
    """
    if n_train_cases < 1:
        raise ValueError("n_train_cases must be at least 1")
    if eta < 2:
        raise ValueError("eta must be at least 2")
    if min_cases < 1:
        raise ValueError("min_cases must be at least 1")
    if n_train_cases == 1:
        return (1,)
    floor = min(min_cases, n_train_cases)
    stages: list[int] = []
    current = floor
    while current < n_train_cases:
        stages.append(current)
        next_size = current * eta
        if next_size >= n_train_cases:
            break
        current = next_size
    if not stages or stages[-1] != n_train_cases:
        stages.append(n_train_cases)
    return tuple(sorted(set(stages)))

subset_evaluate_cases(evaluate)

Wrap a full-catalog evaluate for partial case scoring.

The wrapped callable still invokes evaluate on the full catalog. For true cheap subsets, pass a custom evaluate_cases that calls evaluate_columnar(..., cases=...) or an equivalent partial scorer.

Parameters:

Name	Type	Description	Default
evaluate	Callable[[Individual], Sequence[float]]	Full-catalog fitness callable.	required

Returns:

Type	Description
Callable[[Individual, Sequence[int]], Sequence[float]]	(individual, cases) -> scores for cases only.

Source code in deap_er/private/various/case_halving.py

def subset_evaluate_cases(
    evaluate: Callable[[Individual], Sequence[float]],
) -> Callable[[Individual, Sequence[int]], Sequence[float]]:
    """Wrap a full-catalog evaluate for partial case scoring.

    The wrapped callable still invokes ``evaluate`` on the full catalog.
    For true cheap subsets, pass a custom ``evaluate_cases`` that calls
    ``evaluate_columnar(..., cases=...)`` or an equivalent partial scorer.

    Args:
        evaluate: Full-catalog fitness callable.

    Returns:
        ``(individual, cases) ->`` scores for ``cases`` only.
    """

    def evaluate_cases(individual: Individual, cases: Sequence[int]) -> Sequence[float]:
        values = evaluate(individual)
        return tuple(values[int(case)] for case in cases)

    return evaluate_cases

evaluate_case_halving(individuals, evaluate_cases, train_cases, *, n_cases, eta=2, min_cases=1, weights=None, evaluate_full=None)

Run successive halving on train catalog indices.

Intermediate rungs rank individuals on prefixes of train_cases without writing fitness.values. The final rung assigns a full-catalog fitness tuple of length n_cases.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Individual]	Candidates to score and filter.	required
evaluate_cases	Callable[[Individual, Sequence[int]], Sequence[float]]	(individual, case_indices) -> per-case scores.	required
train_cases	Sequence[int]	Train catalog indices in caller order.	required
n_cases	int	Full catalog length for final fitness.values.	required
eta	int	Elimination factor between rungs.	2
min_cases	int	Smallest rung size.	1
weights	Sequence[float] | None	Optional per-catalog weights for ranking means.	None
evaluate_full	Callable[[Individual], Sequence[float]] | None	Optional full-catalog scorer for the final rung. Defaults to evaluate_cases(ind, range(n_cases)).	None

Returns:

Type	Description
CaseHalvingResult	Survivors, total case-eval charge, and rung count.

Raises:

Type	Description
ValueError	If n_cases or halving parameters are invalid.

Source code in deap_er/private/various/case_halving.py

def evaluate_case_halving(
    individuals: Sequence[Individual],
    evaluate_cases: Callable[[Individual, Sequence[int]], Sequence[float]],
    train_cases: Sequence[int],
    *,
    n_cases: int,
    eta: int = 2,
    min_cases: int = 1,
    weights: Sequence[float] | None = None,
    evaluate_full: Callable[[Individual], Sequence[float]] | None = None,
) -> CaseHalvingResult:
    """Run successive halving on train catalog indices.

    Intermediate rungs rank individuals on prefixes of ``train_cases``
    without writing ``fitness.values``. The final rung assigns a
    full-catalog fitness tuple of length ``n_cases``.

    Args:
        individuals: Candidates to score and filter.
        evaluate_cases: ``(individual, case_indices) ->`` per-case scores.
        train_cases: Train catalog indices in caller order.
        n_cases: Full catalog length for final ``fitness.values``.
        eta: Elimination factor between rungs.
        min_cases: Smallest rung size.
        weights: Optional per-catalog weights for ranking means.
        evaluate_full: Optional full-catalog scorer for the final
            rung. Defaults to ``evaluate_cases(ind, range(n_cases))``.

    Returns:
        Survivors, total case-eval charge, and rung count.

    Raises:
        ValueError: If ``n_cases`` or halving parameters are invalid.
    """
    if n_cases < 1:
        raise ValueError("n_cases must be at least 1")
    catalog = list(train_cases)
    if not catalog or not individuals:
        return CaseHalvingResult([], 0, 0)

    stages = case_halving_stages(len(catalog), eta=eta, min_cases=min_cases)
    survivors = list(individuals)
    charged = 0

    def assign_full(individual: Individual) -> Sequence[float]:
        return evaluate_cases(individual, list(range(n_cases)))

    assign = evaluate_full if evaluate_full is not None else assign_full
    final_cases = n_cases

    for stage_n in stages[:-1]:
        subset = catalog[:stage_n]
        ranked = _rank_on_cases(survivors, evaluate_cases, subset, weights)
        charged += case_eval_charge(len(ranked), stage_n)
        survivors = _keep_top(ranked, eta)

    charged += case_eval_charge(len(survivors), final_cases)
    for individual in survivors:
        values = tuple(assign(individual))
        if len(values) != n_cases:
            raise ValueError(
                f"final fitness must have length n_cases ({n_cases}), got {len(values)}"
            )
        individual.fitness.values = values

    return CaseHalvingResult(survivors, charged, len(stages))

clone

clone_individual(individual)

Copy a sequence individual and its fitness without a full deepcopy.

Register this on a Toolbox when a shallow gene copy is enough: toolbox.register("clone", tools.clone_individual). Falls back to copy.deepcopy when the individual is a NumPy array or carries extra state such as strategy, ps_, or history_index.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to copy.	required

Returns:

Type	Description
Individual	An independent copy of individual.

Source code in deap_er/private/various/clone.py

def clone_individual(individual: Individual) -> Individual:
    """Copy a sequence individual and its fitness without a full deepcopy.

    Register this on a Toolbox when a shallow gene copy is enough:
    ``toolbox.register("clone", tools.clone_individual)``. Falls back to
    ``copy.deepcopy`` when the individual is a NumPy array or carries
    extra state such as ``strategy``, ``ps_``, or ``history_index``.

    Args:
        individual: Individual to copy.

    Returns:
        An independent copy of ``individual``.
    """
    extra = getattr(individual, "__dict__", None)
    if extra is not None and extra.keys() - {"fitness"}:
        return deepcopy(individual)
    if hasattr(individual, "strategy") or hasattr(individual, "ps_"):
        return deepcopy(individual)
    if hasattr(individual, "history_index"):
        return deepcopy(individual)
    if isinstance(individual, numpy.ndarray):
        return deepcopy(individual)
    if not isinstance(individual, list | array):
        return deepcopy(individual)

    clone = type(individual)(individual)
    if hasattr(individual, "fitness"):
        clone.fitness = deepcopy(individual.fitness)
    return clone

constraints

DeltaPenalty(feasibility, delta, distance=None)

Decorator that penalizes fitness of invalid individuals.

Valid individuals keep the original fitness. Invalid ones receive delta plus an optional distance penalty that grows as the individual moves away from the valid region.

Parameters:

Name	Type	Description	Default
feasibility	Callable[..., Any]	Function that reports whether an individual is valid.	required
delta	NumOrSeq	Constant or sequence of constants used as the base penalty for an invalid individual.	required
distance	Callable[..., Any] | None	Optional function returning the distance between the individual and a valid point.	None

See the class docstring.

Source code in deap_er/private/various/constraints.py

def __init__(
    self,
    feasibility: Callable[..., Any],
    delta: NumOrSeq,
    distance: Callable[..., Any] | None = None,
) -> None:
    """See the class docstring."""
    self.fea_func = feasibility
    delta_arr = numpy.asarray(delta)
    self.delta: Any = repeat(delta) if delta_arr.ndim == 0 else delta_arr
    self.dist_fct = distance

__call__(func)

Wrap a fitness function with the delta penalty.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Fitness function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that returns the original fitness for valid
Callable[..., Any]	individuals and the penalized fitness otherwise.

Source code in deap_er/private/various/constraints.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap a fitness function with the delta penalty.

    Args:
        func: Fitness function to decorate.

    Returns:
        A callable that returns the original fitness for valid
        individuals and the penalized fitness otherwise.
    """

    @wraps(func)
    def wrapper(individual: Individual, *args: Any, **kwargs: Any) -> tuple[Any, ...]:
        if self.fea_func(individual):
            return func(individual, *args, **kwargs)

        weights = tuple(1 if w >= 0 else -1 for w in individual.fitness.weights)

        dists = [0 for _ in individual.fitness.weights]
        if self.dist_fct is not None:
            measured = self.dist_fct(individual)
            dist_arr = numpy.asarray(measured)
            dists = repeat(measured) if dist_arr.ndim == 0 else dist_arr

        return tuple(
            float(d - w * dist) for d, w, dist in zip(self.delta, weights, dists, strict=False)
        )

    return wrapper

ClosestValidPenalty(validity, feasible, alpha, distance=None)

Decorator that penalizes fitness of invalid individuals.

Valid individuals keep the original fitness. Invalid ones receive the fitness of the closest valid individual plus an optional weighted distance penalty that grows as the individual moves away from the valid region.

Parameters:

Name	Type	Description	Default
validity	Callable[..., Any]	Function that reports whether an individual is valid.	required
feasible	Callable[..., Any]	Function that returns the closest feasible individual for an invalid one.	required
alpha	float	Multiplication factor on the distance between the valid and invalid individuals.	required
distance	Callable[..., Any] | None	Optional function returning the distance between the individual and a valid point.	None

See the class docstring.

Source code in deap_er/private/various/constraints.py

def __init__(
    self,
    validity: Callable[..., Any],
    feasible: Callable[..., Any],
    alpha: float,
    distance: Callable[..., Any] | None = None,
) -> None:
    """See the class docstring."""
    self.fea_func = validity
    self.fbl_fct = feasible
    self.alpha = alpha
    self.dist_fct = distance

__call__(func)

Wrap a fitness function with the closest-valid penalty.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Fitness function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that returns the original fitness for valid
Callable[..., Any]	individuals and the penalized fitness otherwise.

Source code in deap_er/private/various/constraints.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap a fitness function with the closest-valid penalty.

    Args:
        func: Fitness function to decorate.

    Returns:
        A callable that returns the original fitness for valid
        individuals and the penalized fitness otherwise.
    """

    @wraps(func)
    def wrapper(individual: Individual, *args: Any, **kwargs: Any) -> tuple[Any, ...]:
        if self.fea_func(individual):
            return func(individual, *args, **kwargs)

        f_ind = self.fbl_fct(individual)
        f_fbl = func(f_ind, *args, **kwargs)

        weights = tuple(1.0 if w >= 0 else -1.0 for w in individual.fitness.weights)

        if len(weights) != len(f_fbl):
            raise IndexError("Fitness weights and computed fitness are of different size.")
        dists = [0 for _ in individual.fitness.weights]
        if self.dist_fct is not None:
            measured = self.dist_fct(f_ind, individual)
            dist_arr = numpy.asarray(measured)
            dists = repeat(measured) if dist_arr.ndim == 0 else dist_arr

        return tuple(
            float(f - w * self.alpha * d)
            for f, w, d in zip(f_fbl, weights, dists, strict=False)
        )

    return wrapper

decorators

Translation(vector)

Decorator that translates an individual before evaluation.

The decorated function receives a plain list of translated values. After decoration, func.translate updates the translation vector.

Parameters:

Name	Type	Description	Default
vector	list[float]	Translation values. Must have the same length as the individual.	required

See the class docstring.

Source code in deap_er/private/various/decorators.py

def __init__(self, vector: list[float]) -> None:
    """See the class docstring."""
    self.translate(vector)

__call__(func)

Wrap an evaluation function with the translation.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Evaluation function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that translates the individual, then calls
Callable[..., Any]	func. The wrapper has a translate method.

Source code in deap_er/private/various/decorators.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap an evaluation function with the translation.

    Args:
        func: Evaluation function to decorate.

    Returns:
        A callable that translates the individual, then calls
        ``func``. The wrapper has a ``translate`` method.
    """

    @wraps(func)
    def wrapper(individual: Any, *args: Any, **kwargs: Any) -> Any:
        translated = [v - t for v, t in zip(individual, self.vector, strict=False)]
        return func(translated, *args, **kwargs)

    decorated: Any = wrapper
    decorated.translate = self.translate
    return decorated

translate(vector)

Update the translation vector.

After decorating the evaluation function, this method is available on the function object.

Parameters:

Name	Type	Description	Default
vector	list[float]	The translation vector.	required

Source code in deap_er/private/various/decorators.py

def translate(self, vector: list[float]) -> None:
    """Update the translation vector.

    After decorating the evaluation function, this method is
    available on the function object.

    Args:
        vector: The translation vector.
    """
    self.vector = vector

Rotation(matrix)

Decorator that rotates an individual before evaluation.

The decorated function receives a plain ndarray of rotated values. After decoration, func.rotate updates the rotation matrix.

Parameters:

Name	Type	Description	Default
matrix	ndarray	Orthogonal N-by-N rotation matrix, where N is the length of the individual.	required

See the class docstring.

Source code in deap_er/private/various/decorators.py

def __init__(self, matrix: numpy.ndarray) -> None:
    """See the class docstring."""
    self.rotate(matrix)

__call__(func)

Wrap an evaluation function with the rotation.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Evaluation function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that rotates the individual, then calls
Callable[..., Any]	func. The wrapper has a rotate method.

Source code in deap_er/private/various/decorators.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap an evaluation function with the rotation.

    Args:
        func: Evaluation function to decorate.

    Returns:
        A callable that rotates the individual, then calls
        ``func``. The wrapper has a ``rotate`` method.
    """

    @wraps(func)
    def wrapper(individual: Any, *args: Any, **kwargs: Any) -> Any:
        rotated = numpy.dot(self.matrix, individual)
        return func(rotated, *args, **kwargs)

    decorated: Any = wrapper
    decorated.rotate = self.rotate
    return decorated

rotate(matrix)

Update the rotation matrix.

After decorating the evaluation function, this method is available on the function object.

Parameters:

Name	Type	Description	Default
matrix	ndarray	The rotation matrix.	required

Source code in deap_er/private/various/decorators.py

def rotate(self, matrix: numpy.ndarray) -> None:
    """Update the rotation matrix.

    After decorating the evaluation function, this method is
    available on the function object.

    Args:
        matrix: The rotation matrix.
    """
    self.matrix = numpy.linalg.inv(matrix)

Scaling(factor)

Decorator that scales an individual before evaluation.

The decorated function receives a plain list of scaled values. After decoration, func.scale updates the scale factors.

Parameters:

Name	Type	Description	Default
factor	list[float]	Scale factors. Must have the same length as the individual.	required

See the class docstring.

Source code in deap_er/private/various/decorators.py

def __init__(self, factor: list[float]) -> None:
    """See the class docstring."""
    self.scale(factor)

__call__(func)

Wrap an evaluation function with the scaling.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Evaluation function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that scales the individual, then calls
Callable[..., Any]	func. The wrapper has a scale method.

Source code in deap_er/private/various/decorators.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap an evaluation function with the scaling.

    Args:
        func: Evaluation function to decorate.

    Returns:
        A callable that scales the individual, then calls
        ``func``. The wrapper has a ``scale`` method.
    """

    @wraps(func)
    def wrapper(individual: Any, *args: Any, **kwargs: Any) -> Any:
        scaled = [v * f for v, f in zip(individual, self.factor, strict=False)]
        return func(scaled, *args, **kwargs)

    decorated: Any = wrapper
    decorated.scale = self.scale
    return decorated

scale(factor)

Update the scale factors.

After decorating the evaluation function, this method is available on the function object.

Parameters:

Name	Type	Description	Default
factor	list[float]	The scale factor.	required

Source code in deap_er/private/various/decorators.py

def scale(self, factor: list[float]) -> None:
    """Update the scale factors.

    After decorating the evaluation function, this method is
    available on the function object.

    Args:
        factor: The scale factor.
    """
    self.factor = tuple(1.0 / f for f in factor)

Noise(funcs)

Decorator that adds noise to an evaluation result.

Each noise generator is called without arguments. After decoration, func.add_noise updates the generators.

Parameters:

Name	Type	Description	Default
funcs	Callable[..., Any] | list[Callable[..., Any] | None]	Noise generator callables. A single callable is applied to every result value. A list must match the result length. A None entry leaves that value unchanged.	required

See the class docstring.

Source code in deap_er/private/various/decorators.py

def __init__(self, funcs: Callable[..., Any] | list[Callable[..., Any] | None]) -> None:
    """See the class docstring."""
    self.add_noise(funcs)

__call__(func)

Wrap an evaluation function with noise on its result.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Evaluation function to decorate.	required

Returns:

Type	Description
Callable[..., Any]	A callable that calls func and adds noise to the
Callable[..., Any]	result. The wrapper has an add_noise method.

Source code in deap_er/private/various/decorators.py

def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Wrap an evaluation function with noise on its result.

    Args:
        func: Evaluation function to decorate.

    Returns:
        A callable that calls ``func`` and adds noise to the
        result. The wrapper has an ``add_noise`` method.
    """

    @wraps(func)
    def wrapper(individual: Any, *args: Any, **kwargs: Any) -> tuple[Any, ...]:
        result = func(individual, *args, **kwargs)
        if not isinstance(result, Iterable):
            result = (result,)
        noisy = []
        for r, f in zip(result, self.rand_funcs, strict=False):
            if f is None:
                noisy.append(r)
            else:
                noisy.append(r + f())
        return tuple(noisy)

    decorated: Any = wrapper
    decorated.add_noise = self.add_noise
    return decorated

add_noise(funcs)

Update the noise generators.

After decorating the evaluation function, this method is available on the function object.

Parameters:

Name	Type	Description	Default
funcs	Callable[..., Any] | list[Callable[..., Any] | None]	The noise function or functions.	required

Source code in deap_er/private/various/decorators.py

def add_noise(self, funcs: Callable[..., Any] | list[Callable[..., Any] | None]) -> None:
    """Update the noise generators.

    After decorating the evaluation function, this method is
    available on the function object.

    Args:
        funcs: The noise function or functions.
    """
    if callable(funcs) and not isinstance(funcs, list):
        self.rand_funcs = repeat(funcs)
    else:
        self.rand_funcs = funcs

eval_cache

EvalCache(evaluate=None, evaluate_batch=None, *, matrix=None, key_fn=None)

Fitness cache keyed by expression plus matrix identity / rows.

Wrap the caller's evaluate / evaluate_batch. A hit returns the stored fitness tuple and does not call the wrapped callable. The default expression key is tree structure via expression_key, source text, or str(individual). Pass key= (or keys= on a batch) to override.

Live instances register with the process-wide invalidation hook: clear_compile_cache clears every cache; invalidate_compiled drops keys for that expression. promote_subtree and tune_ephemerals already call those helpers, so a language mutation or ephemeral write-back cannot keep a stale fitness.

n_evals and logbook nevals still count every fitness assignment through evaluate_invalid. A cache hit skips the wrapped callable only.

Parameters:

Name	Type	Description	Default
evaluate	Callable[[Any], Any] | None	Optional callable(ind) -> fitness tuple.	None
evaluate_batch	Callable[[list[Any]], Any] | None	Optional callable(inds) -> fitness tuples.	None
matrix	Any	Evaluation matrix whose identity and row count join the key. None stores (None, 0).	None
key_fn	Callable[[Any], Any] | None	Optional callable(ind) -> hashable caller key.	None

See the class docstring.

Source code in deap_er/private/various/eval_cache.py

def __init__(
    self,
    evaluate: Callable[[Any], Any] | None = None,
    evaluate_batch: Callable[[list[Any]], Any] | None = None,
    *,
    matrix: Any = None,
    key_fn: Callable[[Any], Any] | None = None,
) -> None:
    """See the class docstring."""
    self._evaluate = evaluate
    self._evaluate_batch = evaluate_batch
    self.matrix = matrix
    self._key_fn = key_fn
    self._entries: dict[CacheKey, Any] = {}
    _eval_caches.add(self)

cache_key(individual, caller_key=None)

Build the key for individual on the current matrix.

Parameters:

Name	Type	Description	Default
individual	Any	Individual or expression being scored.	required
caller_key	Any	Optional explicit key. Overrides key_fn.	None

Returns:

Type	Description
CacheKey	(expression, id(matrix), row_count).

Source code in deap_er/private/various/eval_cache.py

def cache_key(self, individual: Any, caller_key: Any = None) -> CacheKey:
    """Build the key for ``individual`` on the current matrix.

    Args:
        individual: Individual or expression being scored.
        caller_key: Optional explicit key. Overrides ``key_fn``.

    Returns:
        ``(expression, id(matrix), row_count)``.
    """
    override = caller_key
    if override is None and self._key_fn is not None:
        override = self._key_fn(individual)
    ident, rows = _matrix_part(self.matrix)
    return _expression_fragment(individual, override), ident, rows

evaluate(individual, *, key=None)

Return cached fitness or pay evaluate on a miss.

Parameters:

Name	Type	Description	Default
individual	Any	Individual to score.	required
key	Any	Optional caller key for this call.	None

Returns:

Type	Description
Any	The fitness tuple from the cache or from evaluate.

Raises:

Type	Description
ValueError	If no evaluate callable was given.

Source code in deap_er/private/various/eval_cache.py

def evaluate(self, individual: Any, *, key: Any = None) -> Any:
    """Return cached fitness or pay ``evaluate`` on a miss.

    Args:
        individual: Individual to score.
        key: Optional caller key for this call.

    Returns:
        The fitness tuple from the cache or from ``evaluate``.

    Raises:
        ValueError: If no ``evaluate`` callable was given.
    """
    cache_key = self.cache_key(individual, key)
    if cache_key in self._entries:
        return self._entries[cache_key]
    if self._evaluate is None:
        raise ValueError("EvalCache.evaluate requires an evaluate callable.")
    fitness = self._evaluate(individual)
    self._entries[cache_key] = fitness
    return fitness

evaluate_batch(individuals, *, keys=None)

Score a batch, calling evaluate_batch only for misses.

When evaluate_batch is omitted, each miss uses evaluate.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Any]	Individuals to score, in caller order.	required
keys	Sequence[Any] | None	Optional per-individual caller keys.	None

Returns:

Type	Description
list[Any]	Fitness tuples aligned with individuals.

Raises:

Type	Description
ValueError	If neither wrapped callable is set.
ValueError	If keys is given and its length differs.

Source code in deap_er/private/various/eval_cache.py

def evaluate_batch(
    self,
    individuals: Sequence[Any],
    *,
    keys: Sequence[Any] | None = None,
) -> list[Any]:
    """Score a batch, calling ``evaluate_batch`` only for misses.

    When ``evaluate_batch`` is omitted, each miss uses ``evaluate``.

    Args:
        individuals: Individuals to score, in caller order.
        keys: Optional per-individual caller keys.

    Returns:
        Fitness tuples aligned with ``individuals``.

    Raises:
        ValueError: If neither wrapped callable is set.
        ValueError: If ``keys`` is given and its length differs.
    """
    if keys is not None and len(keys) != len(individuals):
        raise ValueError("keys must have one entry per individual.")
    results: list[Any] = [None] * len(individuals)
    misses: list[Any] = []
    miss_idx: list[int] = []
    for index, individual in enumerate(individuals):
        caller_key = None if keys is None else keys[index]
        cache_key = self.cache_key(individual, caller_key)
        if cache_key in self._entries:
            results[index] = self._entries[cache_key]
            continue
        misses.append(individual)
        miss_idx.append(index)
    if not misses:
        return results
    fresh = list(self._score_misses(misses))
    for index, fitness, individual in zip(miss_idx, fresh, misses, strict=True):
        caller_key = None if keys is None else keys[index]
        self._entries[self.cache_key(individual, caller_key)] = fitness
        results[index] = fitness
    return results

invalidate(expr)

Drop entries whose expression fragment names expr.

Parameters:

Name	Type	Description	Default
expr	Any	expression_key fragment, source text, or tree.	required

Returns:

Type	Description
int	The number of entries removed.

Source code in deap_er/private/various/eval_cache.py

def invalidate(self, expr: Any) -> int:
    """Drop entries whose expression fragment names ``expr``.

    Args:
        expr: ``expression_key`` fragment, source text, or tree.

    Returns:
        The number of entries removed.
    """
    fragment = expr if isinstance(expr, str | tuple) else expression_key(expr)
    drop = [key for key in self._entries if _fragment_matches(key[0], fragment)]
    for key in drop:
        self._entries.pop(key, None)
    return len(drop)

clear()

Remove every cached fitness tuple.

Source code in deap_er/private/various/eval_cache.py

def clear(self) -> None:
    """Remove every cached fitness tuple."""
    self._entries.clear()

__len__()

Return how many fitness tuples the cache currently holds.

Source code in deap_er/private/various/eval_cache.py

def __len__(self) -> int:
    """Return how many fitness tuples the cache currently holds."""
    return len(self._entries)

clear_eval_caches()

Drop every entry from each live :class:EvalCache.

clear_compile_cache calls this so promote_subtree (and any other language mutation) cannot leave stale fitness next to a dropped compile-cache table.

Source code in deap_er/private/various/eval_cache.py

def clear_eval_caches() -> None:
    """Drop every entry from each live :class:`EvalCache`.

    ``clear_compile_cache`` calls this so ``promote_subtree`` (and any
    other language mutation) cannot leave stale fitness next to a
    dropped compile-cache table.
    """
    # Mutate cache entries only; the WeakSet itself is not updated.
    for cache in _eval_caches:
        cache.clear()

invalidate_eval(expr)

Drop matching expression keys from every live :class:EvalCache.

Uses the same fragment as invalidate_compiled: a structural tree key, raw source text, or the historical lambda …: {expr} form. tune_ephemerals reaches this through invalidate_compiled.

Parameters:

Name	Type	Description	Default
expr	Any	Expression whose cached fitness should be evicted.	required

Returns:

Type	Description
int	The number of cache entries removed across all live caches.

Source code in deap_er/private/various/eval_cache.py

def invalidate_eval(expr: Any) -> int:
    """Drop matching expression keys from every live :class:`EvalCache`.

    Uses the same fragment as ``invalidate_compiled``: a structural
    tree key, raw source text, or the historical ``lambda …: {expr}``
    form. ``tune_ephemerals`` reaches this through
    ``invalidate_compiled``.

    Args:
        expr: Expression whose cached fitness should be evicted.

    Returns:
        The number of cache entries removed across all live caches.
    """
    fragment = expr if isinstance(expr, str | tuple) else expression_key(expr)
    return sum(cache.invalidate(fragment) for cache in _eval_caches)

fitness_race

RaceStopResult(survivors, nevals, rounds) dataclass

Outcome of :func:race_stop.

Attributes:

Name	Type	Description
survivors	list[Individual]	Individuals that survived every racing round.
nevals	int	Evaluate or cache calls charged across all rounds.
rounds	int	Number of resample rounds executed.

race_eval_charge(n_individuals, n_draws=1)

Return evaluation units for one racing round.

Parameters:

Name	Type	Description	Default
n_individuals	int	Survivors scored in the round.	required
n_draws	int	Draws per individual (default one).	1

Returns:

Type	Description
int	n_individuals * n_draws.

Raises:

Type	Description
ValueError	If either count is negative.

Source code in deap_er/private/various/fitness_race.py

def race_eval_charge(n_individuals: int, n_draws: int = 1) -> int:
    """Return evaluation units for one racing round.

    Args:
        n_individuals: Survivors scored in the round.
        n_draws: Draws per individual (default one).

    Returns:
        ``n_individuals * n_draws``.

    Raises:
        ValueError: If either count is negative.
    """
    if n_individuals < 0 or n_draws < 0:
        raise ValueError("n_individuals and n_draws must be non-negative")
    return n_individuals * n_draws

race_z_score(alpha)

Map a significance level to an approximate normal critical value.

Source code in deap_er/private/various/fitness_race.py

def race_z_score(alpha: float) -> float:
    """Map a significance level to an approximate normal critical value."""
    if alpha in _ALPHA_TO_Z:
        return _ALPHA_TO_Z[alpha]
    if not 0.0 < alpha < 1.0:
        raise ValueError("alpha must be strictly between 0 and 1")
    return 1.96

score_draw(individual, evaluate, *, cache, key_fn, draw)

Score one draw for individual through cache or evaluate.

Source code in deap_er/private/various/fitness_race.py

def score_draw(
    individual: Individual,
    evaluate: Callable[[Individual], Sequence[float]],
    *,
    cache: EvalCache | None,
    key_fn: Callable[[Individual], Any] | None,
    draw: int,
) -> tuple[float, ...]:
    """Score one draw for ``individual`` through cache or ``evaluate``."""
    if cache is not None:
        base = key_fn(individual) if key_fn is not None else id(individual)
        cached = cache.evaluate(individual, key=noisy_draw_key(base, draw))
        return tuple(float(value) for value in cached)
    return tuple(float(value) for value in evaluate(individual))

objective_score(values, objective, maximize)

Return a scalar for ranking on one objective.

Source code in deap_er/private/various/fitness_race.py

def objective_score(values: Sequence[float], objective: int, maximize: bool) -> float:
    """Return a scalar for ranking on one objective."""
    value = float(values[objective])
    if not math.isfinite(value):
        return float("-inf") if maximize else float("inf")
    return value if maximize else -value

sample_mean_std(samples, objective)

Return the sample mean and standard error on one objective.

Source code in deap_er/private/various/fitness_race.py

def sample_mean_std(samples: Sequence[Sequence[float]], objective: int) -> tuple[float, float]:
    """Return the sample mean and standard error on one objective."""
    values = [float(sample[objective]) for sample in samples if math.isfinite(sample[objective])]
    if not values:
        return float("nan"), float("inf")
    mean = sum(values) / len(values)
    if len(values) == 1:
        return mean, float("inf")
    variance = sum((value - mean) ** 2 for value in values) / (len(values) - 1)
    return mean, math.sqrt(variance / len(values))

maximize_first_objective(individuals)

Return whether the first fitness objective is maximized.

Source code in deap_er/private/various/fitness_race.py

def maximize_first_objective(individuals: Sequence[Individual]) -> bool:
    """Return whether the first fitness objective is maximized."""
    for individual in individuals:
        weights = individual.fitness.weights
        if weights:
            return float(weights[0]) > 0.0
    return False

challenger_is_not_worse(challenger_mean, challenger_se, leader_mean, leader_se, *, z_score, maximize)

Return whether a challenger is not confidently worse than the leader.

Source code in deap_er/private/various/fitness_race.py

def challenger_is_not_worse(
    challenger_mean: float,
    challenger_se: float,
    leader_mean: float,
    leader_se: float,
    *,
    z_score: float,
    maximize: bool,
) -> bool:
    """Return whether a challenger is not confidently worse than the leader."""
    if maximize:
        return challenger_mean + z_score * challenger_se >= leader_mean - z_score * leader_se
    return challenger_mean - z_score * challenger_se <= leader_mean + z_score * leader_se

keep_challenger_after_race(challenger_samples, leader_mean, leader_se, *, z_score, maximize)

Return whether a challenger survives comparison with the leader.

Source code in deap_er/private/various/fitness_race.py

def keep_challenger_after_race(
    challenger_samples: Sequence[Sequence[float]],
    leader_mean: float,
    leader_se: float,
    *,
    z_score: float,
    maximize: bool,
) -> bool:
    """Return whether a challenger survives comparison with the leader."""
    if len(challenger_samples) < 2:
        return True
    challenger_mean, challenger_se = sample_mean_std(challenger_samples, 0)
    if not math.isfinite(challenger_mean):
        return False
    return challenger_is_not_worse(
        challenger_mean,
        challenger_se,
        leader_mean,
        leader_se,
        z_score=z_score,
        maximize=maximize,
    )

restore_min_survivors(kept, ranked, min_survivors)

Add back ranked challengers until min_survivors is met.

Source code in deap_er/private/various/fitness_race.py

def restore_min_survivors(
    kept: list[Individual],
    ranked: Sequence[Individual],
    min_survivors: int,
) -> list[Individual]:
    """Add back ranked challengers until ``min_survivors`` is met."""
    if len(kept) >= min_survivors:
        return kept
    restored = list(kept)
    for challenger in ranked[1:]:
        if challenger not in restored:
            restored.append(challenger)
        if len(restored) >= min_survivors:
            break
    return restored

eliminate_losers(survivors, samples, *, aggregate, alpha, maximize, min_survivors)

Drop challengers whose first objective is significantly worse.

Source code in deap_er/private/various/fitness_race.py

def eliminate_losers(
    survivors: list[Individual],
    samples: dict[int, list[tuple[float, ...]]],
    *,
    aggregate: Callable[[Sequence[Sequence[float]]], Sequence[float]],
    alpha: float,
    maximize: bool,
    min_survivors: int,
) -> list[Individual]:
    """Drop challengers whose first objective is significantly worse."""
    if len(survivors) <= min_survivors:
        return survivors
    ranked = sorted(
        survivors,
        key=lambda individual: objective_score(aggregate(samples[id(individual)]), 0, maximize),
        reverse=True,
    )
    leader = ranked[0]
    leader_samples = samples[id(leader)]
    if len(leader_samples) < 2:
        return survivors
    leader_mean, leader_se = sample_mean_std(leader_samples, 0)
    if not math.isfinite(leader_mean):
        return survivors
    z_score = race_z_score(alpha)
    kept = [leader]
    for challenger in ranked[1:]:
        if keep_challenger_after_race(
            samples[id(challenger)],
            leader_mean,
            leader_se,
            z_score=z_score,
            maximize=maximize,
        ):
            kept.append(challenger)
    return restore_min_survivors(kept, ranked, min_survivors)

race_stop(individuals, evaluate, n_rounds, *, cache=None, key_fn=None, alpha=0.05, min_survivors=1, aggregate=resample_aggregate, write=False)

Run an F-Race-shaped stop on noisy fitness draws.

Each round adds one resample per survivor. After the second round, challengers whose first objective is significantly worse than the current leader are dropped. Samples are tracked by id(individual); clones share one history unless they are distinct objects.

Multi-objective racing uses objective zero only. evaluate stays on the caller; this helper does not build a domain metric.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Individual]	Candidates to race.	required
evaluate	Callable[[Individual], Sequence[float]]	One-draw fitness callable.	required
n_rounds	int	Maximum resample rounds.	required
cache	EvalCache | None	Optional :class:~deap_er.tools.EvalCache wrapper.	None
key_fn	Callable[[Individual], Any] | None	Optional per-individual cache key fragment.	None
alpha	float	Significance level mapped to a normal critical value.	0.05
min_survivors	int	Stop eliminating below this count.	1
aggregate	Callable[[Sequence[Sequence[float]]], Sequence[float]]	Reduces each individual's draw list to one tuple.	resample_aggregate
write	bool	When true, write the final aggregate to fitness.values.	False

Returns:

Type	Description
RaceStopResult	Survivors, total evaluate calls, and rounds executed.

Raises:

Type	Description
ValueError	If n_rounds or min_survivors is invalid.

Source code in deap_er/private/various/fitness_race.py

def race_stop(
    individuals: Sequence[Individual],
    evaluate: Callable[[Individual], Sequence[float]],
    n_rounds: int,
    *,
    cache: EvalCache | None = None,
    key_fn: Callable[[Individual], Any] | None = None,
    alpha: float = 0.05,
    min_survivors: int = 1,
    aggregate: Callable[[Sequence[Sequence[float]]], Sequence[float]] = resample_aggregate,
    write: bool = False,
) -> RaceStopResult:
    """Run an F-Race-shaped stop on noisy fitness draws.

    Each round adds one resample per survivor. After the second round,
    challengers whose first objective is significantly worse than the
    current leader are dropped. Samples are tracked by ``id(individual)``;
    clones share one history unless they are distinct objects.

    Multi-objective racing uses objective zero only. ``evaluate`` stays
    on the caller; this helper does not build a domain metric.

    Args:
        individuals: Candidates to race.
        evaluate: One-draw fitness callable.
        n_rounds: Maximum resample rounds.
        cache: Optional :class:`~deap_er.tools.EvalCache` wrapper.
        key_fn: Optional per-individual cache key fragment.
        alpha: Significance level mapped to a normal critical value.
        min_survivors: Stop eliminating below this count.
        aggregate: Reduces each individual's draw list to one tuple.
        write: When true, write the final aggregate to ``fitness.values``.

    Returns:
        Survivors, total evaluate calls, and rounds executed.

    Raises:
        ValueError: If ``n_rounds`` or ``min_survivors`` is invalid.
    """
    if n_rounds < 0:
        raise ValueError("n_rounds must be non-negative")
    if min_survivors < 1:
        raise ValueError("min_survivors must be at least 1")
    survivors = list(individuals)
    if not survivors or n_rounds == 0:
        return RaceStopResult([], 0, 0)

    samples: dict[int, list[tuple[float, ...]]] = {id(individual): [] for individual in survivors}
    charged = 0
    rounds_run = 0
    maximize = maximize_first_objective(survivors)

    for round_idx in range(n_rounds):
        for individual in survivors:
            draw = score_draw(
                individual,
                evaluate,
                cache=cache,
                key_fn=key_fn,
                draw=round_idx,
            )
            samples[id(individual)].append(draw)
        charged += race_eval_charge(len(survivors), 1)
        rounds_run += 1
        if round_idx >= 1:
            survivors = eliminate_losers(
                survivors,
                samples,
                aggregate=aggregate,
                alpha=alpha,
                maximize=maximize,
                min_survivors=min_survivors,
            )
        if len(survivors) <= min_survivors:
            break

    if write:
        for individual in survivors:
            individual.fitness.values = tuple(aggregate(samples[id(individual)]))

    return RaceStopResult(survivors, charged, rounds_run)

fitness_resample

noisy_draw_key(base, draw)

Return a hashable :class:EvalCache caller key that includes the draw.

A noisy evaluate must vary the cache key per draw. Without a per-draw key, EvalCache returns the first draw for every repeat.

Parameters:

Name	Type	Description	Default
base	Any	Caller-owned expression or matrix identity fragment.	required
draw	int	Zero-based resample index.	required

Returns:

Type	Description
tuple[Any, int]	(base, draw) suitable for EvalCache.evaluate(..., key=).

Source code in deap_er/private/various/fitness_resample.py

def noisy_draw_key(base: Any, draw: int) -> tuple[Any, int]:
    """Return a hashable :class:`EvalCache` caller key that includes the draw.

    A noisy ``evaluate`` must vary the cache key per draw. Without a
    per-draw key, ``EvalCache`` returns the first draw for every repeat.

    Args:
        base: Caller-owned expression or matrix identity fragment.
        draw: Zero-based resample index.

    Returns:
        ``(base, draw)`` suitable for ``EvalCache.evaluate(..., key=)``.
    """
    return (base, int(draw))

resample_aggregate(samples)

Average per-objective samples, skipping non-finite values.

Parameters:

Name	Type	Description	Default
samples	Sequence[Sequence[float]]	Fitness tuples from independent draws.	required

Returns:

Type	Description
tuple[float, ...]	Elementwise means across samples.

Raises:

Type	Description
ValueError	If samples is empty or tuple lengths differ.

Source code in deap_er/private/various/fitness_resample.py

def resample_aggregate(samples: Sequence[Sequence[float]]) -> tuple[float, ...]:
    """Average per-objective samples, skipping non-finite values.

    Args:
        samples: Fitness tuples from independent draws.

    Returns:
        Elementwise means across ``samples``.

    Raises:
        ValueError: If ``samples`` is empty or tuple lengths differ.
    """
    if not samples:
        raise ValueError("samples must not be empty")
    width = len(samples[0])
    for sample in samples:
        if len(sample) != width:
            raise ValueError("all samples must have the same length")
    means: list[float] = []
    for objective in range(width):
        values = [
            float(sample[objective]) for sample in samples if math.isfinite(sample[objective])
        ]
        if not values:
            means.append(float("nan"))
        else:
            means.append(sum(values) / len(values))
    return tuple(means)

resample(ind, evaluate, n, *, cache=None, key=None, aggregate=resample_aggregate, write=True)

Score one individual n times and aggregate the noisy draws.

Each repeat goes through cache when it is set. Pass :func:noisy_draw_key (or an equivalent draw-specific key) so independent draws do not share one cache entry.

Parameters:

Name	Type	Description	Default
ind	Individual	Individual to score.	required
evaluate	Callable[[Individual], Sequence[float]]	Callable that returns one fitness tuple per draw.	required
n	int	Number of independent draws. Must be at least 1.	required
cache	EvalCache | None	Optional :class:~deap_er.tools.EvalCache wrapper.	None
key	Any	Optional caller key fragment paired with each draw index.	None
aggregate	Callable[[Sequence[Sequence[float]]], Sequence[float]]	Reduces draw tuples to one fitness tuple.	resample_aggregate
write	bool	When true, assign the aggregate to ind.fitness.values.	True

Returns:

Type	Description
tuple[float, ...]	The aggregated fitness tuple.

Raises:

Type	Description
ValueError	If n is less than 1.

Source code in deap_er/private/various/fitness_resample.py

def resample(
    ind: Individual,
    evaluate: Callable[[Individual], Sequence[float]],
    n: int,
    *,
    cache: EvalCache | None = None,
    key: Any = None,
    aggregate: Callable[[Sequence[Sequence[float]]], Sequence[float]] = resample_aggregate,
    write: bool = True,
) -> tuple[float, ...]:
    """Score one individual ``n`` times and aggregate the noisy draws.

    Each repeat goes through ``cache`` when it is set. Pass
    :func:`noisy_draw_key` (or an equivalent draw-specific key) so
    independent draws do not share one cache entry.

    Args:
        ind: Individual to score.
        evaluate: Callable that returns one fitness tuple per draw.
        n: Number of independent draws. Must be at least ``1``.
        cache: Optional :class:`~deap_er.tools.EvalCache` wrapper.
        key: Optional caller key fragment paired with each draw index.
        aggregate: Reduces draw tuples to one fitness tuple.
        write: When true, assign the aggregate to ``ind.fitness.values``.

    Returns:
        The aggregated fitness tuple.

    Raises:
        ValueError: If ``n`` is less than ``1``.
    """
    if n < 1:
        raise ValueError("n must be at least 1")
    samples: list[Sequence[float]] = []
    for draw in range(n):
        if cache is not None:
            draw_key = noisy_draw_key(key, draw) if key is not None else draw
            sample = cache.evaluate(ind, key=draw_key)
        else:
            sample = evaluate(ind)
        samples.append(tuple(float(value) for value in sample))
    values = tuple(aggregate(samples))
    if write:
        ind.fitness.values = values
    return values

hypervolume

minimized_points(population)

Return objective rows in minimization space (-wvalues).

Parameters:

Name	Type	Description	Default
population	list[Any]	Individuals with a Fitness attribute.	required

Returns:

Type	Description
ndarray	A 2-D array of points, or shape (0, 0) when population
ndarray	is empty.

Source code in deap_er/private/various/hypervolume.py

def minimized_points(population: list[Any]) -> numpy.ndarray:
    """Return objective rows in minimization space (``-wvalues``).

    Args:
        population: Individuals with a Fitness attribute.

    Returns:
        A 2-D array of points, or shape ``(0, 0)`` when ``population``
        is empty.
    """
    if not population:
        return numpy.empty((0, 0), dtype=float)
    return numpy.array([ind.fitness.wvalues for ind in population], dtype=float) * -1

has_fitness(obj)

Return whether obj looks like an individual with Fitness.

Parameters:

Name	Type	Description	Default
obj	object	Value to inspect.	required

Returns:

Type	Description
bool	True if obj has a fitness attribute.

Source code in deap_er/private/various/hypervolume.py

def has_fitness(obj: object) -> bool:
    """Return whether ``obj`` looks like an individual with Fitness.

    Args:
        obj: Value to inspect.

    Returns:
        True if ``obj`` has a ``fitness`` attribute.
    """
    return hasattr(obj, "fitness")

hypervolume(points, ref_point=None)

Return the hypervolume of a point set or a population.

Minimization is assumed. An individual or a sequence of individuals is converted via -wvalues. A bare point matrix (no fitness) is used as-is. ref_point is in that same space. Delegates to moocore.hypervolume.

Parameters:

Name	Type	Description	Default
points	ndarray | list[Individual] | Individual	Minimized objective rows, one individual, or a population with Fitness.	required
ref_point	ndarray | list[float] | None	Reference point. Optional. If omitted, the worst value of each objective plus one is used.	None

Returns:

Type	Description
float	The hypervolume of the point set.

Source code in deap_er/private/various/hypervolume.py

def hypervolume(
    points: numpy.ndarray | list[Individual] | Individual,
    ref_point: numpy.ndarray | list[float] | None = None,
) -> float:
    """Return the hypervolume of a point set or a population.

    Minimization is assumed. An individual or a sequence of
    individuals is converted via ``-wvalues``. A bare point matrix
    (no ``fitness``) is used as-is. ``ref_point`` is in that same
    space. Delegates to ``moocore.hypervolume``.

    Args:
        points: Minimized objective rows, one individual, or a
            population with Fitness.
        ref_point: Reference point. Optional. If omitted, the worst
            value of each objective plus one is used.

    Returns:
        The hypervolume of the point set.
    """
    if has_fitness(points):
        arr = minimized_points([points])
    elif not isinstance(points, numpy.ndarray) and points and has_fitness(points[0]):
        arr = minimized_points(list(points))
    else:
        arr = numpy.asarray(points, dtype=float)
    if arr.size == 0:
        return 0.0
    ref = numpy.max(arr, axis=0) + 1 if ref_point is None else numpy.asarray(ref_point)
    return float(moocore.hypervolume(arr, ref=ref, maximise=False))

initializers

init_repeat(container, func, size)

Call func size times and store the results in container.

Use with a Toolbox to register a generator of filled containers, such as individuals or a population.

Parameters:

Name	Type	Description	Default
container	Callable[..., Any]	Callable that takes an iterable and returns a collection.	required
func	Callable[..., Any]	Function called once per element.	required
size	int	Number of times to call func.	required

Returns:

Type	Description
Any	A collection filled with size results of func.

Source code in deap_er/private/various/initializers.py

def init_repeat(container: Callable[..., Any], func: Callable[..., Any], size: int) -> Any:
    """Call ``func`` ``size`` times and store the results in ``container``.

    Use with a Toolbox to register a generator of filled containers,
    such as individuals or a population.

    Args:
        container: Callable that takes an iterable and returns a collection.
        func: Function called once per element.
        size: Number of times to call ``func``.

    Returns:
        A collection filled with ``size`` results of ``func``.
    """
    return container(func() for _ in range(size))

init_iterate(container, generator)

Call generator and store its results in container.

generator must return an iterable. Use with a Toolbox to register a generator of filled containers, as individuals or a population.

Parameters:

Name	Type	Description	Default
container	Callable[..., Any]	Callable that takes an iterable and returns a collection.	required
generator	Callable[..., Any]	Function that returns the iterable used to fill the container.	required

Returns:

Type	Description
Any	A collection filled with the results of generator.

Source code in deap_er/private/various/initializers.py

def init_iterate(container: Callable[..., Any], generator: Callable[..., Any]) -> Any:
    """Call ``generator`` and store its results in ``container``.

    ``generator`` must return an iterable. Use with a Toolbox to
    register a generator of filled containers, as individuals or a
    population.

    Args:
        container: Callable that takes an iterable and returns a collection.
        generator: Function that returns the iterable used to fill the
            container.

    Returns:
        A collection filled with the results of ``generator``.
    """
    return container(generator())

init_cycle(container, funcs, size=1)

Call each function in funcs size times and store all results.

Use with a Toolbox to register a generator of filled containers, as individuals or a population.

Parameters:

Name	Type	Description	Default
container	Callable[..., Any]	Callable that takes an iterable and returns a collection.	required
funcs	Iterable[Callable[..., Any]]	Sequence of functions to call.	required
size	int	Number of times to iterate through funcs.	1

Returns:

Type	Description
Any	A collection filled with the results of all function calls.

Source code in deap_er/private/various/initializers.py

def init_cycle(
    container: Callable[..., Any], funcs: Iterable[Callable[..., Any]], size: int = 1
) -> Any:
    """Call each function in ``funcs`` ``size`` times and store all results.

    Use with a Toolbox to register a generator of filled containers,
    as individuals or a population.

    Args:
        container: Callable that takes an iterable and returns a collection.
        funcs: Sequence of functions to call.
        size: Number of times to iterate through ``funcs``.

    Returns:
        A collection filled with the results of all function calls.
    """
    return container(func() for _ in range(size) for func in funcs)

least_contrib

least_contrib(population, ref_point=None)

Return the index of the individual with the least hypervolume contribution.

Minimization is implicitly assumed. ref_point is interpreted in that same space (after wvalues are negated). Delegates to moocore.hv_contributions.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Non-dominated individuals, each with a Fitness attribute.	required
ref_point	list[float] | ndarray | None	Reference point for the hypervolume. Optional. If omitted, the worst value of each objective plus one is used.	None

Returns:

Type	Description
int	The index of the individual with the least hypervolume
int	contribution. The first index wins when contributions tie.

Raises:

Type	Description
ValueError	If population is empty.

Source code in deap_er/private/various/least_contrib.py

def least_contrib(
    population: list[Individual], ref_point: list[float] | numpy.ndarray | None = None
) -> int:
    """Return the index of the individual with the least hypervolume contribution.

    Minimization is implicitly assumed. ``ref_point`` is interpreted in
    that same space (after ``wvalues`` are negated). Delegates to
    ``moocore.hv_contributions``.

    Args:
        population: Non-dominated individuals, each with a Fitness
            attribute.
        ref_point: Reference point for the hypervolume. Optional. If
            omitted, the worst value of each objective plus one is
            used.

    Returns:
        The index of the individual with the least hypervolume
        contribution. The first index wins when contributions tie.

    Raises:
        ValueError: If ``population`` is empty.
    """
    if not population:
        raise ValueError("population must not be empty")
    wvals = minimized_points(population)
    point = numpy.max(wvals, axis=0) + 1 if ref_point is None else numpy.asarray(ref_point)
    contrib = moocore.hv_contributions(wvals, ref=point, maximise=False)
    return int(numpy.argmin(contrib))

metrics

duplicate_count(population, key=None)

Return how many individuals are duplicates of an earlier one.

Hashable keys use a set scan in O(n). Unhashable but sortable keys use a sort in O(n log n). Keys that are neither hashable nor mutually sortable fall back to list membership in O(n^2). A NumPy array key is hashed by shape, dtype, and bytes so ndarray individuals take the set path.

Hashable keys must satisfy Python's hash/equality contract: equal keys must hash equally. The set path counts by hash bucket; equal keys with unequal hashes are treated as distinct (unlike a pure == scan).

Parameters:

Name	Type	Description	Default
population	list[Any]	Individuals to scan.	required
key	Any | None	Extracts the compared value. Defaults to the identity.	None

Returns:

Type	Description
int	len(population) minus the number of distinct keys.

Source code in deap_er/private/various/metrics.py

def duplicate_count(population: list[Any], key: Any | None = None) -> int:
    """Return how many individuals are duplicates of an earlier one.

    Hashable keys use a set scan in ``O(n)``. Unhashable but sortable keys
    use a sort in ``O(n log n)``. Keys that are neither hashable nor
    mutually sortable fall back to list membership in ``O(n^2)``.
    A NumPy array key is hashed by shape, dtype, and bytes so ndarray
    individuals take the set path.

    Hashable keys must satisfy Python's hash/equality contract: equal keys
    must hash equally. The set path counts by hash bucket; equal keys with
    unequal hashes are treated as distinct (unlike a pure ``==`` scan).

    Args:
        population: Individuals to scan.
        key: Extracts the compared value. Defaults to the identity.

    Returns:
        ``len(population)`` minus the number of distinct keys.
    """
    extract = key if key is not None else (lambda obj: obj)
    if not population:
        return 0
    keys = []
    for item in population:
        value = extract(item)
        if isinstance(value, numpy.ndarray):
            value = (value.shape, str(value.dtype), value.tobytes())
        keys.append(value)
    try:
        return len(keys) - len(set(keys))
    except TypeError:
        pass
    try:
        ordered = sorted(keys)
    except TypeError:
        unique: list[Any] = []
        for value in keys:
            if value not in unique:
                unique.append(value)
        return len(keys) - len(unique)
    distinct = 1
    for index in range(1, len(ordered)):
        if ordered[index] != ordered[index - 1]:
            distinct += 1
    return len(keys) - distinct

nsga_diversity(population, first, last)

Return the NSGA-II diversity metric of a Pareto front.

population is the front to score. first and last are the extreme points of the optimal Pareto front, as in Deb's original NSGA-II article. Each extreme may be an individual (objectives from fitness.values) or a raw objective vector. Smaller values indicate better spread. An empty front returns 1.0, the same value as a single point or a collapsed front.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Pareto front to evaluate.	required
first	Individual	First extreme point of the optimal Pareto front.	required
last	Individual	Last extreme point of the optimal Pareto front.	required

Returns:

Type	Description
float	The diversity metric of the front.

Source code in deap_er/private/various/metrics.py

def nsga_diversity(population: list[Individual], first: Individual, last: Individual) -> float:
    """Return the NSGA-II diversity metric of a Pareto front.

    ``population`` is the front to score. ``first`` and ``last`` are
    the extreme points of the optimal Pareto front, as in Deb's
    original NSGA-II article. Each extreme may be an individual
    (objectives from ``fitness.values``) or a raw objective vector.
    Smaller values indicate better spread. An empty front returns
    ``1.0``, the same value as a single point or a collapsed front.

    Args:
        population: Pareto front to evaluate.
        first: First extreme point of the optimal Pareto front.
        last: Last extreme point of the optimal Pareto front.

    Returns:
        The diversity metric of the front.
    """
    if not population:
        return 1.0
    ordered = sorted(population, key=lambda ind: ind.fitness.values[0])
    start = _objective_row(first)
    end = _objective_row(last)
    df = hypot(ordered[0].fitness.values[0] - start[0], ordered[0].fitness.values[1] - start[1])
    dl = hypot(ordered[-1].fitness.values[0] - end[0], ordered[-1].fitness.values[1] - end[1])

    def fn(f_: Individual, s_: Individual) -> float:
        return hypot(
            f_.fitness.values[0] - s_.fitness.values[0], f_.fitness.values[1] - s_.fitness.values[1]
        )

    zipper = zip(ordered[:-1], ordered[1:], strict=False)
    dt = [fn(first, second) for first, second in zipper]

    if len(ordered) == 1:
        return 1.0

    dm = sum(dt) / len(dt)
    di = sum(abs(d_i - dm) for d_i in dt)
    denom = df + dl + len(dt) * dm
    if denom <= 0:
        return 1.0
    return (df + dl + di) / denom

nsga_convergence(population, optimal)

Return the NSGA-II convergence metric of a Pareto front.

population is the front to score and optimal is the true Pareto front, as in Deb's original NSGA-II article. Each reference point may be an individual or a raw objective vector. Smaller values indicate closer solutions.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Pareto front to evaluate.	required
optimal	list[Individual]	Optimal Pareto front.	required

Returns:

Type	Description
float	The convergence metric of the front. An empty population
float	or empty optimal returns 0.0.

Source code in deap_er/private/various/metrics.py

def nsga_convergence(population: list[Individual], optimal: list[Individual]) -> float:
    """Return the NSGA-II convergence metric of a Pareto front.

    ``population`` is the front to score and ``optimal`` is the true
    Pareto front, as in Deb's original NSGA-II article. Each reference
    point may be an individual or a raw objective vector. Smaller
    values indicate closer solutions.

    Args:
        population: Pareto front to evaluate.
        optimal: Optimal Pareto front.

    Returns:
        The convergence metric of the front. An empty population
        or empty ``optimal`` returns ``0.0``.
    """
    if not population or not optimal:
        return 0.0
    front = numpy.asarray([ind.fitness.values for ind in population], dtype=float)
    truth = numpy.asarray([_objective_row(opt) for opt in optimal], dtype=float)
    minima = numpy.min(spatial.distance.cdist(front, truth), axis=1)
    return float(numpy.mean(minima))

inv_gen_dist(ind1, ind2)

Compute the inverted generational distance between two point sets.

IGD measures how well one approximation covers another in multi-objective optimization.

Parameters:

Name	Type	Description	Default
ind1	Individual	First point set.	required
ind2	Individual	Second point set.	required

Returns:

Type	Description
Any	The average distance from each point in ind2 to the
Any	nearest point in ind1. An empty ind1 or ind2
Any	returns 0.0.

Source code in deap_er/private/various/metrics.py

def inv_gen_dist(ind1: Individual, ind2: Individual) -> Any:
    """Compute the inverted generational distance between two point sets.

    IGD measures how well one approximation covers another in
    multi-objective optimization.

    Args:
        ind1: First point set.
        ind2: Second point set.

    Returns:
        The average distance from each point in ``ind2`` to the
        nearest point in ``ind1``. An empty ``ind1`` or ``ind2``
        returns ``0.0``.
    """
    if not ind1 or not ind2:
        return 0.0
    first = numpy.asarray([_objective_row(point) for point in ind1], dtype=float)
    second = numpy.asarray([_objective_row(point) for point in ind2], dtype=float)
    distances = spatial.distance.cdist(first, second)
    minima = numpy.min(distances, axis=0)
    return numpy.average(minima)

policy_observe

policy_solve_bits_from_errors(errors)

Build observation solve bits from :func:~deap_er.tools.case_errors.

Parameters:

Name	Type	Description	Default
errors	tuple[float, ...]	One MSE per case segment.	required

Returns:

Type	Description
tuple[int, ...]	1 when a case error is within 1e-12 of zero.

Source code in deap_er/private/various/policy_observe.py

def policy_solve_bits_from_errors(errors: tuple[float, ...]) -> tuple[int, ...]:
    """Build observation solve bits from :func:`~deap_er.tools.case_errors`.

    Args:
        errors: One MSE per case segment.

    Returns:
        ``1`` when a case error is within ``1e-12`` of zero.
    """
    _reject_raw_arrays(errors, label="errors")
    return tuple(1 if numpy.isclose(error, 0.0, atol=_SOLVE_ATOL) else 0 for error in errors)

policy_solve_bits_from_fitness(values)

Build observation solve bits from per-case fitness values.

Uses the same zero threshold as lexicase and :func:~deap_er.tools.score_case_exams.

Parameters:

Name	Type	Description	Default
values	tuple[float, ...]	One fitness value per catalog case.	required

Returns:

Type	Description
tuple[int, ...]	1 when a case is solved at zero error.

Source code in deap_er/private/various/policy_observe.py

def policy_solve_bits_from_fitness(values: tuple[float, ...]) -> tuple[int, ...]:
    """Build observation solve bits from per-case fitness values.

    Uses the same zero threshold as lexicase and
    :func:`~deap_er.tools.score_case_exams`.

    Args:
        values: One fitness value per catalog case.

    Returns:
        ``1`` when a case is solved at zero error.
    """
    return policy_solve_bits_from_errors(values)

policy_solve_bits_from_semantic_row(row)

Coerce one row of :func:~deap_er.gp.semantic_solve_bits output.

Parameters:

Name	Type	Description	Default
row	Sequence[float]	One individual's solve-bit row as a plain sequence.	required

Returns:

Type	Description
tuple[int, ...]	0/1 tuple suitable for :func:policy_observe.

Source code in deap_er/private/various/policy_observe.py

def policy_solve_bits_from_semantic_row(row: Sequence[float]) -> tuple[int, ...]:
    """Coerce one row of :func:`~deap_er.gp.semantic_solve_bits` output.

    Args:
        row: One individual's solve-bit row as a plain sequence.

    Returns:
        ``0/1`` tuple suitable for :func:`policy_observe`.
    """
    _reject_raw_arrays(row, label="row")
    return _coerce_solve_bits(row)

policy_unsolved_count(solve_bits)

Count cases not solved in a solve-bit tuple.

Parameters:

Name	Type	Description	Default
solve_bits	Sequence[int]	Per-case 0/1 flags.	required

Returns:

Type	Description
int	Number of zeros in solve_bits.

Source code in deap_er/private/various/policy_observe.py

def policy_unsolved_count(solve_bits: Sequence[int]) -> int:
    """Count cases not solved in a solve-bit tuple.

    Args:
        solve_bits: Per-case ``0/1`` flags.

    Returns:
        Number of zeros in ``solve_bits``.
    """
    bits = _coerce_solve_bits(solve_bits)
    return _unsolved_count(bits)

policy_exam_scores(exams, elites, *, held_out=None)

Reduce :func:~deap_er.tools.score_case_exams to train / held-out scalars.

Train exams are every exam in exams when it is a sequence. When exams is a :class:~deap_er.records.CaseExamPool, train scores use pool.exams and pool.held_out unless held_out overrides the pool marker.

Parameters:

Name	Type	Description	Default
exams	Sequence[CaseExam] | CaseExamPool	Train exams or a pool with an optional held-out exam.	required
elites	list[Individual]	Evaluated individuals that supply the case pack.	required
held_out	CaseExam | None	Optional held-out exam that overrides a pool marker.	None

Returns:

Type	Description
float	(train_score, held_out_score) where train_score is the
float | None	sum of train-exam difficulties and held_out_score is the
tuple[float, float | None]	held-out difficulty or None.

Source code in deap_er/private/various/policy_observe.py

def policy_exam_scores(
    exams: Sequence[CaseExam] | CaseExamPool,
    elites: list[Individual],
    *,
    held_out: CaseExam | None = None,
) -> tuple[float, float | None]:
    """Reduce :func:`~deap_er.tools.score_case_exams` to train / held-out scalars.

    Train exams are every exam in ``exams`` when it is a sequence. When
    ``exams`` is a :class:`~deap_er.records.CaseExamPool`, train scores
    use ``pool.exams`` and ``pool.held_out`` unless ``held_out`` overrides
    the pool marker.

    Args:
        exams: Train exams or a pool with an optional held-out exam.
        elites: Evaluated individuals that supply the case pack.
        held_out: Optional held-out exam that overrides a pool marker.

    Returns:
        ``(train_score, held_out_score)`` where ``train_score`` is the
        sum of train-exam difficulties and ``held_out_score`` is the
        held-out difficulty or ``None``.
    """
    pool_held_out = held_out
    train_exams = exams
    if isinstance(exams, CaseExamPool):
        pool_held_out = exams.held_out if held_out is None else held_out
        train_exams = exams.exams
    train_score = float(sum(score_case_exams(train_exams, elites)))
    if pool_held_out is None:
        return train_score, None
    return train_score, float(score_case_exams([pool_held_out], elites)[0])

policy_promoted_library_size(prim_set)

Return the promoted-library size for observation fields.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set that may hold a promoted library.	required

Returns:

Type	Description
int	Number of names returned by :func:~deap_er.gp.promoted_names.

Source code in deap_er/private/various/policy_observe.py

def policy_promoted_library_size(prim_set: PrimitiveSetTyped) -> int:
    """Return the promoted-library size for observation fields.

    Args:
        prim_set: Primitive set that may hold a promoted library.

    Returns:
        Number of names returned by :func:`~deap_er.gp.promoted_names`.
    """
    return len(promoted_names(prim_set))

policy_observe(*, solve_bits, train_score, held_out_score=None, archive=None, nevals=0, rows_seen=0, promoted_library_size=0, fitness_invalid=False, last_action_rejected=False)

Build the fixed Push policy observation from summary inputs only.

Accepts reductions from :func:~deap_er.tools.case_errors, :func:~deap_er.tools.score_case_exams, :class:~deap_er.records.ArchiveStats, promoted-library counters, and eval-budget tallies. Raw NumPy packs, column slices, and matrix[t] are rejected at this boundary.

Parameters:

Name	Type	Description	Default
solve_bits	Sequence[int | float | bool]	Per-case 0/1 flags for the observed individual.	required
train_score	float	Sum of train-exam difficulty scores.	required
held_out_score	float | None	Held-out exam difficulty, or None.	None
archive	ArchiveStats | None	Optional archive summary supplying coverage and qd_score.	None
nevals	int	Evaluations consumed this step.	0
rows_seen	int	Rows seen in the evaluation matrix so far.	0
promoted_library_size	int	Count of promoted primitive names.	0
fitness_invalid	bool	Whether the observed individual lacks valid fitness.	False
last_action_rejected	bool	Whether the last policy action was rejected.	False

Returns:

Name	Type	Description
One	PolicyObservation	class:~deap_er.records.PolicyObservation.

Raises:

Type	Description
TypeError	If solve_bits is or contains a raw array.
ValueError	If solve_bits is not a 0/1 sequence or a counter is negative.

Source code in deap_er/private/various/policy_observe.py

def policy_observe(
    *,
    solve_bits: Sequence[int | float | bool],
    train_score: float,
    held_out_score: float | None = None,
    archive: ArchiveStats | None = None,
    nevals: int = 0,
    rows_seen: int = 0,
    promoted_library_size: int = 0,
    fitness_invalid: bool = False,
    last_action_rejected: bool = False,
) -> PolicyObservation:
    """Build the fixed Push policy observation from summary inputs only.

    Accepts reductions from :func:`~deap_er.tools.case_errors`,
    :func:`~deap_er.tools.score_case_exams`,
    :class:`~deap_er.records.ArchiveStats`, promoted-library counters,
    and eval-budget tallies. Raw NumPy packs, column slices, and
    ``matrix[t]`` are rejected at this boundary.

    Args:
        solve_bits: Per-case ``0/1`` flags for the observed individual.
        train_score: Sum of train-exam difficulty scores.
        held_out_score: Held-out exam difficulty, or ``None``.
        archive: Optional archive summary supplying coverage and
            ``qd_score``.
        nevals: Evaluations consumed this step.
        rows_seen: Rows seen in the evaluation matrix so far.
        promoted_library_size: Count of promoted primitive names.
        fitness_invalid: Whether the observed individual lacks valid
            fitness.
        last_action_rejected: Whether the last policy action was
            rejected.

    Returns:
        One :class:`~deap_er.records.PolicyObservation`.

    Raises:
        TypeError: If ``solve_bits`` is or contains a raw array.
        ValueError: If ``solve_bits`` is not a ``0/1`` sequence or a
            counter is negative.
    """
    bits = _coerce_solve_bits(solve_bits)
    if nevals < 0 or rows_seen < 0 or promoted_library_size < 0:
        raise ValueError("nevals, rows_seen, and promoted_library_size must be non-negative")
    coverage = 0.0
    qd_score = 0.0
    if archive is not None:
        coverage = archive.coverage
        qd_score = archive.qd_score
    return PolicyObservation(
        solve_bits=bits,
        unsolved_count=_unsolved_count(bits),
        train_score=float(train_score),
        held_out_score=None if held_out_score is None else float(held_out_score),
        archive_coverage=coverage,
        qd_score=qd_score,
        nevals=nevals,
        rows_seen=rows_seen,
        promoted_library_size=promoted_library_size,
        fitness_invalid=fitness_invalid,
        last_action_rejected=last_action_rejected,
    )

rng

RNG(seed=None)

NumPy Generator facade with buffered uniform and integer streams.

random / uniform / take_floats pop leftover floats. Scalar randint, randrange, choice, and integers pop leftover uint64s. Other methods use the same Generator. Sequences stay Python-indexed.

Create a generator, optionally seeded.

Parameters:

Name	Type	Description	Default
seed	int | None	Seed for the NumPy Generator. Optional; OS entropy if omitted.	None

Source code in deap_er/private/various/rng.py

def __init__(self, seed: int | None = None) -> None:
    """Create a generator, optionally seeded.

    Args:
        seed: Seed for the NumPy Generator. Optional; OS entropy if omitted.
    """
    self._gen = numpy.random.default_rng(seed)
    self._buffers = RngBuffers()

seed(seed=None)

Reseed the generator and discard unused buffered values.

Parameters:

Name	Type	Description	Default
seed	int | None	Seed for a new NumPy Generator. Optional; OS entropy if omitted.	None

Source code in deap_er/private/various/rng.py

def seed(self, seed: int | None = None) -> None:
    """Reseed the generator and discard unused buffered values.

    Args:
        seed: Seed for a new NumPy Generator. Optional; OS entropy if omitted.
    """
    self._gen = numpy.random.default_rng(seed)
    self._buffers.discard()

get_state()

Return the bit-generator state and unused buffers.

Returns:

Type	Description
dict[str, Any]	bit_generator, leftover buf / ibuf, and their indices.

Source code in deap_er/private/various/rng.py

def get_state(self) -> dict[str, Any]:
    """Return the bit-generator state and unused buffers.

    Returns:
        ``bit_generator``, leftover ``buf`` / ``ibuf``, and their indices.
    """
    state = self._buffers.pack()
    state["bit_generator"] = self._gen.bit_generator.state
    return state

set_state(state)

Restore a state previously returned by get_state.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Mapping from get_state. ibuf and iindex are optional.	required

Source code in deap_er/private/various/rng.py

def set_state(self, state: dict[str, Any]) -> None:
    """Restore a state previously returned by ``get_state``.

    Args:
        state: Mapping from ``get_state``. ``ibuf`` and ``iindex`` are optional.
    """
    self._gen.bit_generator.state = state["bit_generator"]
    self._buffers.unpack(state)

random()

Return the next uniform float in [0.0, 1.0).

Returns:

Type	Description
float	A Python float from the buffered stream.

Source code in deap_er/private/various/rng.py

def random(self) -> float:
    """Return the next uniform float in ``[0.0, 1.0)``.

    Returns:
        A Python float from the buffered stream.
    """
    return self._buffers.next_float(self._gen)

take_floats(count)

Return count leftover uniforms, same stream as random.

Parameters:

Name	Type	Description	Default
count	int	Number of floats. 0 returns an empty list.	required

Returns:

Type	Description
list[float]	Uniform floats in [0.0, 1.0).

Raises:

Type	Description
ValueError	If count is negative.

Source code in deap_er/private/various/rng.py

def take_floats(self, count: int) -> list[float]:
    """Return ``count`` leftover uniforms, same stream as ``random``.

    Args:
        count: Number of floats. ``0`` returns an empty list.

    Returns:
        Uniform floats in ``[0.0, 1.0)``.

    Raises:
        ValueError: If ``count`` is negative.
    """
    return self._buffers.take_floats(self._gen, count)

uniform(a, b)

Return a uniform float in [a, b).

Parameters:

Name	Type	Description	Default
a	float	Lower bound.	required
b	float	Upper bound.	required

Returns:

Type	Description
float	a + (b - a) * random().

Source code in deap_er/private/various/rng.py

def uniform(self, a: float, b: float) -> float:
    """Return a uniform float in ``[a, b)``.

    Args:
        a: Lower bound.
        b: Upper bound.

    Returns:
        ``a + (b - a) * random()``.
    """
    return a + (b - a) * self.random()

randint(a, b)

Return a random integer N such that a <= N <= b.

Parameters:

Name	Type	Description	Default
a	int	Inclusive lower bound.	required
b	int	Inclusive upper bound.	required

Returns:

Type	Description
int	An integer in the closed interval.

Raises:

Type	Description
ValueError	If the interval is empty.

Source code in deap_er/private/various/rng.py

def randint(self, a: int, b: int) -> int:
    """Return a random integer ``N`` such that ``a <= N <= b``.

    Args:
        a: Inclusive lower bound.
        b: Inclusive upper bound.

    Returns:
        An integer in the closed interval.

    Raises:
        ValueError: If the interval is empty.
    """
    return self._buffers.offset(self._gen, a, b - a + 1)

randrange(start, stop=None, step=1)

Return a random element from range(start, stop, step).

Parameters:

Name	Type	Description	Default
start	int	Stop if stop is omitted, otherwise the start.	required
stop	int | None	Exclusive stop. Optional.	None
step	int	Range step. Defaults to 1.	1

Returns:

Type	Description
int	An integer from the equivalent range object.

Raises:

Type	Description
ValueError	If the range is empty.

Source code in deap_er/private/various/rng.py

def randrange(self, start: int, stop: int | None = None, step: int = 1) -> int:
    """Return a random element from ``range(start, stop, step)``.

    Args:
        start: Stop if ``stop`` is omitted, otherwise the start.
        stop: Exclusive stop. Optional.
        step: Range step. Defaults to 1.

    Returns:
        An integer from the equivalent ``range`` object.

    Raises:
        ValueError: If the range is empty.
    """
    if stop is None:
        start, stop = 0, start
    values = range(start, stop, step)
    n = len(values)
    if n == 0:
        raise ValueError("empty range for randrange()")
    return values[self._buffers.index(self._gen, n)]

choice(seq)

Return one element of seq.

Parameters:

Name	Type	Description	Default
seq	Sequence[T]	Non-empty sequence. Indexed as a Python sequence.	required

Returns:

Type	Description
T	One element of seq.

Raises:

Type	Description
IndexError	If seq is empty.

Source code in deap_er/private/various/rng.py

def choice[T](self, seq: Sequence[T]) -> T:
    """Return one element of ``seq``.

    Args:
        seq: Non-empty sequence. Indexed as a Python sequence.

    Returns:
        One element of ``seq``.

    Raises:
        IndexError: If ``seq`` is empty.
    """
    n = len(seq)
    if n == 0:
        raise IndexError("Cannot choose from an empty sequence")
    return seq[self._buffers.index(self._gen, n)]

sample(population, k)

Return k unique elements from population.

Parameters:

Name	Type	Description	Default
population	Sequence[T]	Sequence to sample from.	required
k	int	Number of elements. May be passed by keyword.	required

Returns:

Type	Description
list[T]	A new list of k elements.

Raises:

Type	Description
ValueError	If k is negative or larger than len(population).

Source code in deap_er/private/various/rng.py

def sample[T](self, population: Sequence[T], k: int) -> list[T]:
    """Return ``k`` unique elements from ``population``.

    Args:
        population: Sequence to sample from.
        k: Number of elements. May be passed by keyword.

    Returns:
        A new list of ``k`` elements.

    Raises:
        ValueError: If ``k`` is negative or larger than ``len(population)``.
    """
    n = len(population)
    if k < 0 or k > n:
        raise ValueError("Sample larger than population or is negative")
    if k == 0:
        return []
    indices = self._gen.choice(n, size=k, replace=False)
    return [population[int(i)] for i in indices]

shuffle(x)

Shuffle x in place.

Sequences of individuals are permuted by index. A 1-D ndarray is shuffled by the underlying Generator.

Parameters:

Name	Type	Description	Default
x	MutableSequence[Any] | ndarray	Mutable sequence or ndarray to shuffle.	required

Source code in deap_er/private/various/rng.py

def shuffle(self, x: MutableSequence[Any] | numpy.ndarray) -> None:
    """Shuffle ``x`` in place.

    Sequences of individuals are permuted by index. A 1-D ndarray
    is shuffled by the underlying Generator.

    Args:
        x: Mutable sequence or ndarray to shuffle.
    """
    if isinstance(x, numpy.ndarray):
        self._gen.shuffle(x)
        return
    n = len(x)
    if n < 2:
        return
    order = self._gen.permutation(n)
    shuffled = [x[int(i)] for i in order]
    x[:] = shuffled

gauss(mu, sigma)

Return a sample from a Gaussian distribution.

Parameters:

Name	Type	Description	Default
mu	float	Mean.	required
sigma	float	Standard deviation.	required

Returns:

Type	Description
float	A Python float from the normal distribution.

Source code in deap_er/private/various/rng.py

def gauss(self, mu: float, sigma: float) -> float:
    """Return a sample from a Gaussian distribution.

    Args:
        mu: Mean.
        sigma: Standard deviation.

    Returns:
        A Python float from the normal distribution.
    """
    return float(self._gen.normal(mu, sigma))

standard_normal(size=None)

standard_normal(size: None = None) -> float

standard_normal(
    size: int | tuple[int, ...],
) -> numpy.ndarray

Return samples from the standard normal distribution.

Parameters:

Name	Type	Description	Default
size	int | tuple[int, ...] | None	Output shape. A single float is returned when omitted.	None

Returns:

Type	Description
float | ndarray	A float, or an ndarray when size is given.

Source code in deap_er/private/various/rng.py

def standard_normal(self, size: int | tuple[int, ...] | None = None) -> float | numpy.ndarray:
    """Return samples from the standard normal distribution.

    Args:
        size: Output shape. A single float is returned when omitted.

    Returns:
        A float, or an ndarray when ``size`` is given.
    """
    if size is None:
        return float(self._gen.standard_normal())
    return self._gen.standard_normal(size)

integers(low, high=None, size=None, endpoint=False)

integers(
    low: int,
    high: int | None = None,
    size: None = None,
    endpoint: bool = False,
) -> int

integers(
    low: int,
    high: int | None = None,
    *,
    size: int | tuple[int, ...],
    endpoint: bool = False,
) -> numpy.ndarray

Return random integers from the buffered or Generator path.

Parameters:

Name	Type	Description	Default
low	int	Inclusive lower bound, or exclusive high when high is omitted.	required
high	int | None	Exclusive upper bound unless endpoint is True.	None
size	int | tuple[int, ...] | None	Output shape. A single int is returned when omitted.	None
endpoint	bool	If True, high is inclusive.	False

Returns:

Type	Description
int | ndarray	An int, or an ndarray when size is given.

Raises:

Type	Description
ValueError	If the interval is empty.

Source code in deap_er/private/various/rng.py

def integers(
    self,
    low: int,
    high: int | None = None,
    size: int | tuple[int, ...] | None = None,
    endpoint: bool = False,
) -> int | numpy.ndarray:
    """Return random integers from the buffered or Generator path.

    Args:
        low: Inclusive lower bound, or exclusive high when ``high`` is omitted.
        high: Exclusive upper bound unless ``endpoint`` is True.
        size: Output shape. A single int is returned when omitted.
        endpoint: If True, ``high`` is inclusive.

    Returns:
        An int, or an ndarray when ``size`` is given.

    Raises:
        ValueError: If the interval is empty.
    """
    return draw_integers(self._gen, self._buffers, low, high, size, endpoint)

rng_buf

RngBuffers()

Refillable uniform-float and raw-uint64 streams.

Allocate zeroed buffers that refill on the first pop.

Source code in deap_er/private/various/rng_buf.py

def __init__(self) -> None:
    """Allocate zeroed buffers that refill on the first pop."""
    self._fbuf = numpy.zeros(_BUFSIZE, dtype=numpy.float64)
    self._floats: list[float] = []
    self._fi = _BUFSIZE
    self._reset_ints()

discard()

Drop leftover values and zero unused buffer storage.

Source code in deap_er/private/various/rng_buf.py

def discard(self) -> None:
    """Drop leftover values and zero unused buffer storage."""
    self._fbuf.fill(0)
    self._floats = []
    self._fi = _BUFSIZE
    self._reset_ints()

pack()

Copy leftover buffer arrays and their read indices.

Spent buffers (index at capacity) are exported as zeros so a checkpoint never carries uninitialized or discarded words.

Returns:

Type	Description
dict[str, Any]	buf, index, ibuf, and iindex.

Source code in deap_er/private/various/rng_buf.py

def pack(self) -> dict[str, Any]:
    """Copy leftover buffer arrays and their read indices.

    Spent buffers (index at capacity) are exported as zeros so a
    checkpoint never carries uninitialized or discarded words.

    Returns:
        ``buf``, ``index``, ``ibuf``, and ``iindex``.
    """
    fbuf = self._fbuf.copy()
    ibuf = self._ibuf.copy()
    if self._fi >= _BUFSIZE:
        fbuf.fill(0)
    if self._ii >= _BUFSIZE:
        ibuf.fill(0)
    return {
        "buf": fbuf,
        "index": self._fi,
        "ibuf": ibuf,
        "iindex": self._ii,
    }

unpack(state)

Restore buffers from pack or a legacy uniform-only mapping.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Mapping with buf and index. ibuf and iindex are optional; omit them for an empty integer buffer.	required

Source code in deap_er/private/various/rng_buf.py

def unpack(self, state: dict[str, Any]) -> None:
    """Restore buffers from ``pack`` or a legacy uniform-only mapping.

    Args:
        state: Mapping with ``buf`` and ``index``. ``ibuf`` and
            ``iindex`` are optional; omit them for an empty
            integer buffer.
    """
    self._fbuf = numpy.array(state["buf"], dtype=numpy.float64, copy=True)
    self._floats = cast(list[float], self._fbuf.tolist())
    self._fi = int(state["index"])
    if "ibuf" in state and "iindex" in state:
        self._ibuf = numpy.array(state["ibuf"], dtype=numpy.uint64, copy=True)
        self._u64s = [int(value) for value in self._ibuf.tolist()]
        self._ii = int(state["iindex"])
        return
    self._reset_ints()

next_float(gen)

Pop the next uniform float in [0.0, 1.0).

Parameters:

Name	Type	Description	Default
gen	Generator	Generator used to refill an empty float buffer.	required

Returns:

Type	Description
float	A Python float from the leftover uniforms.

Source code in deap_er/private/various/rng_buf.py

def next_float(self, gen: numpy.random.Generator) -> float:
    """Pop the next uniform float in ``[0.0, 1.0)``.

    Args:
        gen: Generator used to refill an empty float buffer.

    Returns:
        A Python float from the leftover uniforms.
    """
    if self._fi >= _BUFSIZE:
        gen.random(out=self._fbuf)
        self._floats = cast(list[float], self._fbuf.tolist())
        self._fi = 0
    value = self._floats[self._fi]
    self._fi += 1
    return value

take_floats(gen, count)

Pop count leftover uniforms, same stream as next_float.

Parameters:

Name	Type	Description	Default
gen	Generator	Generator used to refill an empty float buffer.	required
count	int	Number of floats. 0 returns an empty list.	required

Returns:

Type	Description
list[float]	Uniform floats in [0.0, 1.0).

Raises:

Type	Description
ValueError	If count is negative.

Source code in deap_er/private/various/rng_buf.py

def take_floats(self, gen: numpy.random.Generator, count: int) -> list[float]:
    """Pop ``count`` leftover uniforms, same stream as ``next_float``.

    Args:
        gen: Generator used to refill an empty float buffer.
        count: Number of floats. ``0`` returns an empty list.

    Returns:
        Uniform floats in ``[0.0, 1.0)``.

    Raises:
        ValueError: If ``count`` is negative.
    """
    if count < 0:
        raise ValueError("count must be non-negative")
    out: list[float] = []
    remaining = count
    while remaining:
        if self._fi >= _BUFSIZE:
            gen.random(out=self._fbuf)
            self._floats = cast(list[float], self._fbuf.tolist())
            self._fi = 0
        take = min(remaining, _BUFSIZE - self._fi)
        out.extend(self._floats[self._fi : self._fi + take])
        self._fi += take
        remaining -= take
    return out

next_u64(gen)

Pop the next raw 64-bit word.

Parameters:

Name	Type	Description	Default
gen	Generator	Generator whose bit generator fills an empty buffer.	required

Returns:

Type	Description
int	A Python int in [0, 2**64).

Source code in deap_er/private/various/rng_buf.py

def next_u64(self, gen: numpy.random.Generator) -> int:
    """Pop the next raw 64-bit word.

    Args:
        gen: Generator whose bit generator fills an empty buffer.

    Returns:
        A Python int in ``[0, 2**64)``.
    """
    if self._ii >= _BUFSIZE:
        self._ibuf = numpy.asarray(gen.bit_generator.random_raw(_BUFSIZE), dtype=numpy.uint64)
        self._u64s = [int(value) for value in self._ibuf.tolist()]
        self._ii = 0
    value = int(self._u64s[self._ii])
    self._ii += 1
    return value

offset(gen, start, span)

Return start plus an unbiased integer in [0, span).

Parameters:

Name	Type	Description	Default
gen	Generator	Generator used to refill the uint64 buffer.	required
start	int	Inclusive origin of the interval.	required
span	int	Number of integers in the interval.	required

Returns:

Type	Description
int	An integer in [start, start + span).

Raises:

Type	Description
ValueError	If span is not positive.
OverflowError	If span exceeds 2**64.

Source code in deap_er/private/various/rng_buf.py

def offset(self, gen: numpy.random.Generator, start: int, span: int) -> int:
    """Return ``start`` plus an unbiased integer in ``[0, span)``.

    Args:
        gen: Generator used to refill the uint64 buffer.
        start: Inclusive origin of the interval.
        span: Number of integers in the interval.

    Returns:
        An integer in ``[start, start + span)``.

    Raises:
        ValueError: If ``span`` is not positive.
        OverflowError: If ``span`` exceeds ``2**64``.
    """
    return start + self.index(gen, span)

index(gen, n)

Return an unbiased integer in [0, n).

Always consumes at least one uint64, including when n is 1.

Parameters:

Name	Type	Description	Default
gen	Generator	Generator used to refill the uint64 buffer.	required
n	int	Exclusive upper bound. Must be in 1 .. 2**64.	required

Returns:

Type	Description
int	An integer in the half-open interval.

Raises:

Type	Description
ValueError	If n is not positive.
OverflowError	If n exceeds 2**64.

Source code in deap_er/private/various/rng_buf.py

def index(self, gen: numpy.random.Generator, n: int) -> int:
    """Return an unbiased integer in ``[0, n)``.

    Always consumes at least one uint64, including when ``n`` is 1.

    Args:
        gen: Generator used to refill the uint64 buffer.
        n: Exclusive upper bound. Must be in ``1 .. 2**64``.

    Returns:
        An integer in the half-open interval.

    Raises:
        ValueError: If ``n`` is not positive.
        OverflowError: If ``n`` exceeds ``2**64``.
    """
    if n <= 0:
        raise ValueError("low >= high")
    if n > _U64_SPACE:
        raise OverflowError("high - low is too large for 64-bit sampling")
    limit = _U64_SPACE - (_U64_SPACE % n)
    while True:
        value = self.next_u64(gen)
        if value < limit:
            return value % n

draw_integers(gen, buffers, low, high, size, endpoint)

Draw a scalar int from the uint64 buffer, or a sized array from gen.

A sized draw discards leftover uint64s so later scalar ints refill from the advanced bit generator. Leftover uniforms are kept.

Parameters:

Name	Type	Description	Default
gen	Generator	Underlying NumPy Generator.	required
buffers	RngBuffers	Float and uint64 leftovers.	required
low	int	Inclusive lower bound, or exclusive high when high is omitted.	required
high	int | None	Exclusive upper bound unless endpoint is True.	required
size	int | tuple[int, ...] | None	Output shape. A single int is returned when omitted.	required
endpoint	bool	If True, high is inclusive.	required

Returns:

Type	Description
int | ndarray	An int, or an ndarray when size is given.

Raises:

Type	Description
ValueError	If the interval is empty.
OverflowError	If the span exceeds 64 bits.

Source code in deap_er/private/various/rng_buf.py

def draw_integers(
    gen: numpy.random.Generator,
    buffers: RngBuffers,
    low: int,
    high: int | None,
    size: int | tuple[int, ...] | None,
    endpoint: bool,
) -> int | numpy.ndarray:
    """Draw a scalar int from the uint64 buffer, or a sized array from ``gen``.

    A sized draw discards leftover uint64s so later scalar ints refill
    from the advanced bit generator. Leftover uniforms are kept.

    Args:
        gen: Underlying NumPy Generator.
        buffers: Float and uint64 leftovers.
        low: Inclusive lower bound, or exclusive high when ``high``
            is omitted.
        high: Exclusive upper bound unless ``endpoint`` is True.
        size: Output shape. A single int is returned when omitted.
        endpoint: If True, ``high`` is inclusive.

    Returns:
        An int, or an ndarray when ``size`` is given.

    Raises:
        ValueError: If the interval is empty.
        OverflowError: If the span exceeds 64 bits.
    """
    if size is not None:
        buffers._reset_ints()
        return gen.integers(low, high, size=size, endpoint=endpoint)
    if high is None:
        low, high = 0, low
    span = high - low + 1 if endpoint else high - low
    return buffers.offset(gen, low, span)

rng_spawn

spawn_rng(seed, worker_id)

Return an independent RNG derived from seed and worker_id.

Uses a NumPy SeedSequence child of the run seed. The process-wide generator is not read or advanced, so a parent seeded with the same value keeps its own stream. The same pair always yields the same stream, regardless of which process draws it or when.

Parameters:

Name	Type	Description	Default
seed	int | Sequence[int]	Run seed. The same value passed to rng.seed.	required
worker_id	int	Stable non-negative task index, not an OS pid.	required

Returns:

Type	Description
RNG	A new RNG that does not share state with the parent.

Raises:

Type	Description
TypeError	If seed or worker_id has a bad type.
ValueError	If seed is an empty sequence or worker_id is negative.

Source code in deap_er/private/various/rng_spawn.py

def spawn_rng(seed: int | Sequence[int], worker_id: int) -> RNG:
    """Return an independent RNG derived from ``seed`` and ``worker_id``.

    Uses a NumPy ``SeedSequence`` child of the run seed. The process-wide
    generator is not read or advanced, so a parent seeded with the same
    value keeps its own stream. The same pair always yields the same
    stream, regardless of which process draws it or when.

    Args:
        seed: Run seed. The same value passed to ``rng.seed``.
        worker_id: Stable non-negative task index, not an OS pid.

    Returns:
        A new ``RNG`` that does not share state with the parent.

    Raises:
        TypeError: If ``seed`` or ``worker_id`` has a bad type.
        ValueError: If ``seed`` is an empty sequence or ``worker_id``
            is negative.
    """
    spawn_key = (_worker_index(worker_id),)
    sequence = numpy.random.SeedSequence(_entropy(seed), spawn_key=spawn_key)
    mixed = 0
    for word in sequence.generate_state(4):
        mixed = (mixed << 32) | int(word)
    return RNG(mixed)

bind_spawned_rng(seed, worker_id)

Install spawn_rng(seed, worker_id) as the process-wide generator.

Parameters:

Name	Type	Description	Default
seed	int | Sequence[int]	Run seed. The same value passed to rng.seed.	required
worker_id	int	Stable non-negative task index.	required

Returns:

Type	Description
RNG	The process-wide rng singleton, now on the spawned stream.

Source code in deap_er/private/various/rng_spawn.py

def bind_spawned_rng(seed: int | Sequence[int], worker_id: int) -> RNG:
    """Install ``spawn_rng(seed, worker_id)`` as the process-wide generator.

    Args:
        seed: Run seed. The same value passed to ``rng.seed``.
        worker_id: Stable non-negative task index.

    Returns:
        The process-wide ``rng`` singleton, now on the spawned stream.
    """
    rng.set_state(spawn_rng(seed, worker_id).get_state())
    return rng

call_spawned(payload)

Bind a spawned stream, call func(item), then restore rng.

The process-wide generator is restored even if func raises, so an in-process map does not leak a worker stream into the parent.

Parameters:

Name	Type	Description	Default
payload	SpawnedPayload[T]	(seed, worker_id, func, item).	required

Returns:

Type	Description
tuple[int, Any]	(worker_id, func(item)) so results can be ordered by id.

Source code in deap_er/private/various/rng_spawn.py

def call_spawned[T](payload: SpawnedPayload[T]) -> tuple[int, Any]:
    """Bind a spawned stream, call ``func(item)``, then restore ``rng``.

    The process-wide generator is restored even if ``func`` raises, so an
    in-process map does not leak a worker stream into the parent.

    Args:
        payload: ``(seed, worker_id, func, item)``.

    Returns:
        ``(worker_id, func(item))`` so results can be ordered by id.
    """
    seed, worker_id, func, item = payload
    previous = rng.get_state()
    try:
        bind_spawned_rng(seed, worker_id)
        return worker_id, func(item)
    finally:
        rng.set_state(previous)

map_spawned(func, iterable, *, seed, map_func=map)

Map func over iterable with one spawned stream per item.

Item i sees spawn_rng(seed, i) as the process-wide rng. Results are returned in input order. map_func may complete tasks in any order; only the worker_id on each result matters.

Parameters:

Name	Type	Description	Default
func	Callable[[T], R]	Callable applied to each item. Library operators that read rng see the spawned stream.	required
iterable	Iterable[T]	Inputs. Materialized once so ids are stable.	required
seed	int | Sequence[int]	Run seed shared with rng.seed on the parent.	required
map_func	Callable[..., Iterable[Any]]	map-like callable. Defaults to builtin map. Pass pool.map or an unordered mapper.	map

Returns:

Type	Description
list[R]	func results in the same order as iterable.

Raises:

Type	Description
ValueError	If map_func does not return one result per item.

Source code in deap_er/private/various/rng_spawn.py

def map_spawned[T, R](
    func: Callable[[T], R],
    iterable: Iterable[T],
    *,
    seed: int | Sequence[int],
    map_func: Callable[..., Iterable[Any]] = map,
) -> list[R]:
    """Map ``func`` over ``iterable`` with one spawned stream per item.

    Item ``i`` sees ``spawn_rng(seed, i)`` as the process-wide ``rng``.
    Results are returned in input order. ``map_func`` may complete
    tasks in any order; only the ``worker_id`` on each result matters.

    Args:
        func: Callable applied to each item. Library operators that
            read ``rng`` see the spawned stream.
        iterable: Inputs. Materialized once so ids are stable.
        seed: Run seed shared with ``rng.seed`` on the parent.
        map_func: ``map``-like callable. Defaults to builtin ``map``.
            Pass ``pool.map`` or an unordered mapper.

    Returns:
        ``func`` results in the same order as ``iterable``.

    Raises:
        ValueError: If ``map_func`` does not return one result per item.
    """
    items = list(iterable)
    if not items:
        return []
    payloads: list[SpawnedPayload[T]] = [
        (seed, index, func, item) for index, item in enumerate(items)
    ]
    ordered = sorted(map_func(call_spawned, payloads), key=lambda pair: pair[0])
    ids = [pair[0] for pair in ordered]
    if ids != list(range(len(items))):
        raise ValueError("map_func must return one (worker_id, value) per item")
    return [pair[1] for pair in ordered]

semantic_descriptors

semantic_moments(matrix, *, valid=None, individuals=None, trust_matrix=False)

Return per-individual mean, population std, min, and max.

Moments use samples where valid (broadcast) and the row are finite. An empty row is all-NaN. A single sample has std = 0.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array of shape (n_individuals, 4).

Source code in deap_er/private/various/semantic_descriptors.py

def semantic_moments(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    valid: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return per-individual mean, population std, min, and max.

    Moments use samples where ``valid`` (broadcast) and the row are
    finite. An empty row is all-NaN. A single sample has ``std = 0``.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array of shape ``(n_individuals, 4)``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    return _row_moments(packed, semantic_valid_mask(packed, valid))

semantic_solve_bits(matrix, target, ranges, *, valid=None, empty=float('inf'), individuals=None, trust_matrix=False)

Return one solved-case bit per individual and case.

A case is solved when its MSE is isclose to 0 with abs_tol=1e-12, matching :func:~deap_er.tools.sample_informed_cases. Segment bounds and valid= follow :func:~deap_er.tools.case_errors.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Predicted pack of shape (n_individuals, n_rows).	required
target	ndarray	Target series of length n_rows.	required
ranges	Sequence[tuple[int, int]] | ndarray	Case bounds accepted by :func:~deap_er.tools.case_intervals.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
empty	float	MSE used when a case has no scorable samples.	float('inf')
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Float 0/1 array of shape (n_individuals, n_cases).

Source code in deap_er/private/various/semantic_descriptors.py

def semantic_solve_bits(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    target: numpy.ndarray,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    empty: float = float("inf"),
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return one solved-case bit per individual and case.

    A case is solved when its MSE is ``isclose`` to ``0`` with
    ``abs_tol=1e-12``, matching :func:`~deap_er.tools.sample_informed_cases`.
    Segment bounds and ``valid=`` follow :func:`~deap_er.tools.case_errors`.

    Args:
        matrix: Predicted pack of shape ``(n_individuals, n_rows)``.
        target: Target series of length ``n_rows``.
        ranges: Case bounds accepted by :func:`~deap_er.tools.case_intervals`.
        valid: Optional per-row warmup mask of length ``n_rows``.
        empty: MSE used when a case has no scorable samples.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Float ``0/1`` array of shape ``(n_individuals, n_cases)``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    series = numpy.asarray(target, dtype=numpy.float64)
    if series.ndim != 1 or series.shape[0] != packed.shape[1]:
        raise ValueError("target must be a one-dimensional series matching n_rows")
    intervals = case_intervals(ranges, packed.shape[1])
    row_valid = semantic_valid_mask(packed, valid) & numpy.isfinite(series)
    bits = numpy.empty((packed.shape[0], len(intervals)), dtype=numpy.float64)
    for index, (start, stop) in enumerate(intervals):
        sample = row_valid[:, start:stop]
        diff = packed[:, start:stop] - series[start:stop]
        sq = numpy.where(sample, diff * diff, 0.0)
        counts = sample.sum(axis=1)
        mse = numpy.where(counts > 0, sq.sum(axis=1) / counts, empty)
        bits[:, index] = numpy.isclose(mse, 0.0, atol=1e-12)
    return bits

semantic_descriptors(matrix, *, kind='moments', valid=None, target=None, ranges=None, basis=None, center=None, empty=float('inf'), individuals=None, trust_matrix=False)

Dispatch a semantic pack to moments, solve bits, or a projection.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
kind	DescriptorKind	moments, solve, or project.	'moments'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Target series. Required for solve.	None
ranges	Sequence[tuple[int, int]] | ndarray | None	Case bounds. Required for solve.	None
basis	ndarray | None	Projection matrix. Required for project.	None
center	ndarray | None	Optional center passed to :func:semantic_project.	None
empty	float	Empty-case MSE for solve.	float('inf')
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array whose width depends on kind.

Raises:

Type	Description
ValueError	If kind is unknown or a required argument is missing.

Source code in deap_er/private/various/semantic_descriptors.py

def semantic_descriptors(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    kind: DescriptorKind = "moments",
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray | None = None,
    basis: numpy.ndarray | None = None,
    center: numpy.ndarray | None = None,
    empty: float = float("inf"),
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Dispatch a semantic pack to moments, solve bits, or a projection.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        kind: ``moments``, ``solve``, or ``project``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Target series. Required for ``solve``.
        ranges: Case bounds. Required for ``solve``.
        basis: Projection matrix. Required for ``project``.
        center: Optional center passed to :func:`semantic_project`.
        empty: Empty-case MSE for ``solve``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array whose width depends on ``kind``.

    Raises:
        ValueError: If ``kind`` is unknown or a required argument is
            missing.
    """
    if kind == "moments":
        return semantic_moments(
            matrix, valid=valid, individuals=individuals, trust_matrix=trust_matrix
        )
    if kind == "solve":
        if target is None or ranges is None:
            raise ValueError("kind='solve' requires target and ranges")
        return semantic_solve_bits(
            matrix,
            target,
            ranges,
            valid=valid,
            empty=empty,
            individuals=individuals,
            trust_matrix=trust_matrix,
        )
    if kind == "project":
        if basis is None:
            raise ValueError("kind='project' requires basis")
        return semantic_project(
            matrix,
            basis,
            valid=valid,
            target=target,
            center=center,
            individuals=individuals,
            trust_matrix=trust_matrix,
        )
    raise ValueError(f"unknown descriptor kind {kind!r}")

semantic_mask

as_semantic_matrix(matrix)

Pack a semantic matrix as float64.

The result is two-dimensional. Layout follows NumPy asarray: a Fortran-order input keeps its memory order.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Caller-supplied (n_individuals, n_rows) pack.	required

Returns:

Type	Description
ndarray	A two-dimensional float64 array.

Raises:

Type	Description
ValueError	If matrix is not two-dimensional.

Source code in deap_er/private/various/semantic_mask.py

def as_semantic_matrix(matrix: numpy.ndarray | Sequence[Sequence[float]]) -> numpy.ndarray:
    """Pack a semantic matrix as ``float64``.

    The result is two-dimensional. Layout follows NumPy ``asarray``:
    a Fortran-order input keeps its memory order.

    Args:
        matrix: Caller-supplied ``(n_individuals, n_rows)`` pack.

    Returns:
        A two-dimensional ``float64`` array.

    Raises:
        ValueError: If ``matrix`` is not two-dimensional.
    """
    packed = numpy.asarray(matrix, dtype=numpy.float64)
    if packed.ndim != 2:
        raise ValueError(
            "semantic matrix must be two-dimensional "
            f"(n_individuals, n_rows), got ndim={packed.ndim}"
        )
    return packed

semantic_valid_mask(matrix, valid=None)

Return the finite sample mask for a semantic matrix.

A one-dimensional valid is broadcast across individuals and intersected with isfinite(matrix). A sample that is True in valid but non-finite still stays out of the mask.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None

Returns:

Type	Description
ndarray	Boolean mask with the same shape as matrix.

Raises:

Type	Description
ValueError	If matrix is not two-dimensional, or valid is not a one-dimensional mask of length n_rows.

Source code in deap_er/private/various/semantic_mask.py

def semantic_valid_mask(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    valid: numpy.ndarray | None = None,
) -> numpy.ndarray:
    """Return the finite sample mask for a semantic matrix.

    A one-dimensional ``valid`` is broadcast across individuals and
    intersected with ``isfinite(matrix)``. A sample that is ``True`` in
    ``valid`` but non-finite still stays out of the mask.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        valid: Optional per-row warmup mask of length ``n_rows``.

    Returns:
        Boolean mask with the same shape as ``matrix``.

    Raises:
        ValueError: If ``matrix`` is not two-dimensional, or ``valid`` is
            not a one-dimensional mask of length ``n_rows``.
    """
    packed = as_semantic_matrix(matrix)
    finite = numpy.isfinite(packed)
    if valid is None:
        return finite
    sample_valid = numpy.asarray(valid, dtype=bool)
    if sample_valid.ndim != 1 or sample_valid.shape[0] != packed.shape[1]:
        raise ValueError("valid must be a one-dimensional mask matching the series length")
    return sample_valid[numpy.newaxis, :] & finite

validate_semantic_matrix(matrix, individuals, *, trust_matrix=False)

Check that a semantic pack is row-aligned with individuals.

Unlike lexicase, the series cannot be compared to fitness.values. When trust_matrix is False, every individual must have a valid fitness. When True, only the leading shape is checked.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
individuals	Sequence[Any] | None	Population the rows describe, or None to skip the alignment check.	required
trust_matrix	bool	When True, accept the pack on shape alone.	False

Returns:

Type	Description
ndarray	The packed float64 matrix.

Raises:

Type	Description
ValueError	If matrix is not two-dimensional, its leading length does not match individuals, or a fitness is missing when trust_matrix is False.

Source code in deap_er/private/various/semantic_mask.py

def validate_semantic_matrix(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    individuals: Sequence[Any] | None,
    *,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Check that a semantic pack is row-aligned with ``individuals``.

    Unlike lexicase, the series cannot be compared to ``fitness.values``.
    When ``trust_matrix`` is ``False``, every individual must have a
    valid fitness. When ``True``, only the leading shape is checked.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        individuals: Population the rows describe, or ``None`` to skip
            the alignment check.
        trust_matrix: When ``True``, accept the pack on shape alone.

    Returns:
        The packed ``float64`` matrix.

    Raises:
        ValueError: If ``matrix`` is not two-dimensional, its leading
            length does not match ``individuals``, or a fitness is
            missing when ``trust_matrix`` is ``False``.
    """
    packed = as_semantic_matrix(matrix)
    if individuals is None:
        return packed
    expected = (len(individuals), packed.shape[1])
    if packed.shape[0] != expected[0]:
        raise ValueError(f"matrix must have shape {expected}, got {packed.shape}")
    if trust_matrix:
        return packed
    for individual in individuals:
        fitness = getattr(individual, "fitness", None)
        if fitness is None or not fitness.is_valid():
            raise ValueError("every individual must have a valid fitness")
    return packed

packed_semantics(matrix, individuals, trust_matrix)

Pack a semantic matrix, optionally checking individuals.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
individuals	Sequence[Any] | None	Population the rows describe, or None.	required
trust_matrix	bool	When True, accept the pack on shape alone.	required

Returns:

Type	Description
ndarray	The packed float64 matrix.

Source code in deap_er/private/various/semantic_mask.py

def packed_semantics(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    individuals: Sequence[Any] | None,
    trust_matrix: bool,
) -> numpy.ndarray:
    """Pack a semantic matrix, optionally checking ``individuals``.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        individuals: Population the rows describe, or ``None``.
        trust_matrix: When ``True``, accept the pack on shape alone.

    Returns:
        The packed ``float64`` matrix.
    """
    if individuals is None:
        return as_semantic_matrix(matrix)
    return validate_semantic_matrix(matrix, individuals, trust_matrix=trust_matrix)

semantic_column_keep(n_rows, valid, target)

Return the shared column mask used by projection helpers.

Parameters:

Name	Type	Description	Default
n_rows	int	Number of semantic coordinates.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	required
target	ndarray | None	Optional target whose non-finite samples are dropped.	required

Returns:

Type	Description
ndarray	One-dimensional bool mask of kept columns.

Raises:

Type	Description
ValueError	If valid or target does not match n_rows.

Source code in deap_er/private/various/semantic_mask.py

def semantic_column_keep(
    n_rows: int,
    valid: numpy.ndarray | None,
    target: numpy.ndarray | None,
) -> numpy.ndarray:
    """Return the shared column mask used by projection helpers.

    Args:
        n_rows: Number of semantic coordinates.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Optional target whose non-finite samples are dropped.

    Returns:
        One-dimensional ``bool`` mask of kept columns.

    Raises:
        ValueError: If ``valid`` or ``target`` does not match ``n_rows``.
    """
    keep = numpy.ones(n_rows, dtype=bool) if valid is None else numpy.asarray(valid, dtype=bool)
    if keep.ndim != 1 or keep.shape[0] != n_rows:
        raise ValueError("valid must be a one-dimensional mask matching the series length")
    if target is None:
        return keep
    series = numpy.asarray(target, dtype=numpy.float64)
    if series.ndim != 1 or series.shape[0] != n_rows:
        raise ValueError("target must be a one-dimensional series matching n_rows")
    return keep & numpy.isfinite(series)

semantic_neighbors

semantic_distance(query, matrix, *, metric='euclidean', valid=None)

Return finite-mask distances from query to each packed row.

Only coordinates that are finite on both sides and marked valid enter the distance. An empty overlap, or a zero cosine norm, is +inf. A one-dimensional matrix is treated as a single row and still returns a length-1 array (not a Python float) so the return type stays ndarray for the type checker.

Parameters:

Name	Type	Description	Default
query	ndarray | Sequence[float]	Semantic row of length n_rows.	required
matrix	ndarray | Sequence[Sequence[float]] | Sequence[float]	Pack of shape (n_individuals, n_rows), or one row of length n_rows.	required
metric	SemanticMetric	euclidean or cosine (1 - cosine similarity).	'euclidean'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None

Returns:

Type	Description
ndarray	Distances of length n_individuals. One-dimensional
ndarray	matrix yields shape (1,).

Raises:

Type	Description
ValueError	If shapes do not match or metric is unknown.

Source code in deap_er/private/various/semantic_neighbors.py

def semantic_distance(
    query: numpy.ndarray | Sequence[float],
    matrix: numpy.ndarray | Sequence[Sequence[float]] | Sequence[float],
    *,
    metric: SemanticMetric = "euclidean",
    valid: numpy.ndarray | None = None,
) -> numpy.ndarray:
    """Return finite-mask distances from ``query`` to each packed row.

    Only coordinates that are finite on both sides and marked ``valid``
    enter the distance. An empty overlap, or a zero cosine norm, is
    ``+inf``. A one-dimensional ``matrix`` is treated as a single row
    and still returns a length-1 array (not a Python float) so the
    return type stays ``ndarray`` for the type checker.

    Args:
        query: Semantic row of length ``n_rows``.
        matrix: Pack of shape ``(n_individuals, n_rows)``, or one row of
            length ``n_rows``.
        metric: ``euclidean`` or ``cosine`` (``1 -`` cosine similarity).
        valid: Optional per-row warmup mask of length ``n_rows``.

    Returns:
        Distances of length ``n_individuals``. One-dimensional
        ``matrix`` yields shape ``(1,)``.

    Raises:
        ValueError: If shapes do not match or ``metric`` is unknown.
    """
    packed = numpy.asarray(matrix, dtype=numpy.float64)
    if packed.ndim == 1:
        packed = packed.reshape(1, -1)
    packed = as_semantic_matrix(packed)
    row = _query_row(query, packed.shape[1])
    return _masked_distance(packed, row, _pair_mask(packed, row, valid), metric)

semantic_nearest(query, matrix, *, k=1, metric='euclidean', valid=None, individuals=None, trust_matrix=False)

Return the lowest-index nearest neighbors of query.

Infinite distances are skipped. Ties keep the lowest pack index. individuals is trust-alignment only: the return value is always integer pack indices, never the individual objects.

Parameters:

Name	Type	Description	Default
query	ndarray | Sequence[float]	Semantic row of length n_rows.	required
matrix	ndarray | Sequence[Sequence[float]]	Pack of shape (n_individuals, n_rows).	required
k	int	Maximum number of neighbors to return.	1
metric	SemanticMetric	euclidean or cosine.	'euclidean'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Neighbor indices in increasing distance order, length at most
ndarray	k. Empty when every distance is infinite.

Raises:

Type	Description
ValueError	If k is less than 1, or the pack does not match individuals.

Source code in deap_er/private/various/semantic_neighbors.py

def semantic_nearest(
    query: numpy.ndarray | Sequence[float],
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    k: int = 1,
    metric: SemanticMetric = "euclidean",
    valid: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return the lowest-index nearest neighbors of ``query``.

    Infinite distances are skipped. Ties keep the lowest pack index.
    ``individuals`` is trust-alignment only: the return value is always
    integer pack indices, never the individual objects.

    Args:
        query: Semantic row of length ``n_rows``.
        matrix: Pack of shape ``(n_individuals, n_rows)``.
        k: Maximum number of neighbors to return.
        metric: ``euclidean`` or ``cosine``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Neighbor indices in increasing distance order, length at most
        ``k``. Empty when every distance is infinite.

    Raises:
        ValueError: If ``k`` is less than 1, or the pack does not match
            ``individuals``.
    """
    if k < 1:
        raise ValueError("k must be at least 1")
    if individuals is None:
        packed = as_semantic_matrix(matrix)
    else:
        packed = validate_semantic_matrix(matrix, individuals, trust_matrix=trust_matrix)
    dist = semantic_distance(query, packed, metric=metric, valid=valid)
    finite = numpy.flatnonzero(numpy.isfinite(dist))
    if finite.size == 0:
        return numpy.empty(0, dtype=int)
    order = finite[numpy.argsort(dist[finite], kind="stable")]
    return order[:k]

semantic_project

semantic_random_basis(n_rows, n_dims)

Return a Gaussian random-projection basis.

Columns are scaled by 1 / sqrt(n_dims). Draws use the process :data:~deap_er.tools.rng.

Parameters:

Name	Type	Description	Default
n_rows	int	Number of semantic coordinates (rows of the pack).	required
n_dims	int	Number of projected dimensions.	required

Returns:

Type	Description
ndarray	Basis of shape (n_rows, n_dims).

Raises:

Type	Description
ValueError	If n_rows or n_dims is not positive.

Source code in deap_er/private/various/semantic_project.py

def semantic_random_basis(n_rows: int, n_dims: int) -> numpy.ndarray:
    """Return a Gaussian random-projection basis.

    Columns are scaled by ``1 / sqrt(n_dims)``. Draws use the process
    :data:`~deap_er.tools.rng`.

    Args:
        n_rows: Number of semantic coordinates (rows of the pack).
        n_dims: Number of projected dimensions.

    Returns:
        Basis of shape ``(n_rows, n_dims)``.

    Raises:
        ValueError: If ``n_rows`` or ``n_dims`` is not positive.
    """
    if n_rows < 1 or n_dims < 1:
        raise ValueError("n_rows and n_dims must be positive")
    draw = rng.standard_normal((n_rows, n_dims))
    return draw / numpy.sqrt(n_dims)

semantic_pca_basis(matrix, n_dims, *, valid=None, target=None)

Return a thin-SVD basis and column center for :func:semantic_project.

Columns outside valid (and non-finite target samples) are dropped before centering. Unused rows of the returned basis and center are zero.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
n_dims	int	Number of components to keep.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Optional target whose non-finite samples are dropped.	None

Returns:

Type	Description
ndarray	(basis, center) with shapes (n_rows, n_dims) and
ndarray	(n_rows,).

Raises:

Type	Description
ValueError	If n_dims is not positive, or no finite row remains on the kept columns.

Source code in deap_er/private/various/semantic_project.py

def semantic_pca_basis(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    n_dims: int,
    *,
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
) -> tuple[numpy.ndarray, numpy.ndarray]:
    """Return a thin-SVD basis and column center for :func:`semantic_project`.

    Columns outside ``valid`` (and non-finite ``target`` samples) are
    dropped before centering. Unused rows of the returned basis and
    center are zero.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        n_dims: Number of components to keep.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Optional target whose non-finite samples are dropped.

    Returns:
        ``(basis, center)`` with shapes ``(n_rows, n_dims)`` and
        ``(n_rows,)``.

    Raises:
        ValueError: If ``n_dims`` is not positive, or no finite row
            remains on the kept columns.
    """
    if n_dims < 1:
        raise ValueError("n_dims must be positive")
    packed = as_semantic_matrix(matrix)
    keep = semantic_column_keep(packed.shape[1], valid, target)
    if not numpy.any(keep):
        raise ValueError("semantic_pca_basis needs at least one finite row on kept columns")
    kept = packed[:, keep]
    finite_rows = numpy.all(numpy.isfinite(kept), axis=1)
    kept = kept[finite_rows]
    if kept.shape[0] == 0:
        raise ValueError("semantic_pca_basis needs at least one finite row on kept columns")
    center_keep = kept.mean(axis=0)
    _, _, vt = numpy.linalg.svd(kept - center_keep, full_matrices=False)
    n_comp = min(n_dims, vt.shape[0])
    basis = numpy.zeros((packed.shape[1], n_dims), dtype=numpy.float64)
    center = numpy.zeros(packed.shape[1], dtype=numpy.float64)
    center[keep] = center_keep
    if n_comp:
        basis[numpy.ix_(numpy.flatnonzero(keep), numpy.arange(n_comp))] = vt[:n_comp].T
    return basis, center

semantic_project(matrix, basis, *, valid=None, target=None, center=None, individuals=None, trust_matrix=False)

Project a semantic pack through a caller-supplied basis.

Columns outside valid (and non-finite target samples) are zeroed so warmup does not enter the product. A row that is still non-finite on a kept column yields a non-finite descriptor.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
basis	ndarray	Projection matrix of shape (n_rows, n_dims).	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Optional target whose non-finite samples are dropped.	None
center	ndarray | None	Optional length-n_rows vector subtracted before the product. Used with :func:semantic_pca_basis.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array of shape (n_individuals, n_dims).

Raises:

Type	Description
ValueError	If basis or center does not match n_rows.

Source code in deap_er/private/various/semantic_project.py

def semantic_project(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    basis: numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
    center: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Project a semantic pack through a caller-supplied basis.

    Columns outside ``valid`` (and non-finite ``target`` samples) are
    zeroed so warmup does not enter the product. A row that is still
    non-finite on a kept column yields a non-finite descriptor.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        basis: Projection matrix of shape ``(n_rows, n_dims)``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Optional target whose non-finite samples are dropped.
        center: Optional length-``n_rows`` vector subtracted before the
            product. Used with :func:`semantic_pca_basis`.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array of shape ``(n_individuals, n_dims)``.

    Raises:
        ValueError: If ``basis`` or ``center`` does not match ``n_rows``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    components = numpy.asarray(basis, dtype=numpy.float64)
    if components.ndim != 2 or components.shape[0] != packed.shape[1]:
        raise ValueError(
            f"basis must have shape ({packed.shape[1]}, n_dims), got {components.shape}"
        )
    keep = semantic_column_keep(packed.shape[1], valid, target)
    shifted = packed.copy()
    if center is not None:
        mean = numpy.asarray(center, dtype=numpy.float64)
        if mean.ndim != 1 or mean.shape[0] != packed.shape[1]:
            raise ValueError("center must be a one-dimensional vector matching n_rows")
        shifted = shifted - mean
    shifted[:, ~keep] = 0.0
    return shifted @ components

sort_non_dominated

sort_non_dominated(individuals, sel_count)

Sort individuals into non-dominated Pareto fronts.

Uses moocore.pareto_rank on fitness.wvalues (higher is better). Fronts are truncated once they hold at least sel_count individuals. Individuals without a comparable fitness (missing, invalid, or non-finite) are ignored.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to sort.	required
sel_count	int	Number of individuals to place into fronts.	required

Returns:

Type	Description
list[list[Individual]]	A list of Pareto fronts. The first element is the true
list[list[Individual]]	Pareto front. An empty list if sel_count is not
list[list[Individual]]	positive. A single empty front if no rankable individual
list[list[Individual]]	remains and sel_count is positive.

Source code in deap_er/private/various/sort_non_dominated.py

def sort_non_dominated(individuals: list[Individual], sel_count: int) -> list[list[Individual]]:
    """Sort individuals into non-dominated Pareto fronts.

    Uses ``moocore.pareto_rank`` on ``fitness.wvalues`` (higher is
    better). Fronts are truncated once they hold at least
    ``sel_count`` individuals. Individuals without a comparable
    fitness (missing, invalid, or non-finite) are ignored.

    Args:
        individuals: Individuals to sort.
        sel_count: Number of individuals to place into fronts.

    Returns:
        A list of Pareto fronts. The first element is the true
        Pareto front. An empty list if ``sel_count`` is not
        positive. A single empty front if no rankable individual
        remains and ``sel_count`` is positive.
    """
    if sel_count <= 0:
        return []
    ranked = [ind for ind in individuals if has_comparable_fitness(ind)]
    if not ranked:
        return [[]]

    points = numpy.array([ind.fitness.wvalues for ind in ranked], dtype=float)
    ranks = moocore.pareto_rank(points, maximise=True)

    by_rank: defaultdict[int, list[Individual]] = defaultdict(list)
    for ind, rank in zip(ranked, ranks, strict=True):
        by_rank[int(rank)].append(ind)

    fronts: list[list[Individual]] = []
    placed = 0
    for rank in range(int(max(ranks)) + 1):
        front = by_rank[rank]
        fronts.append(front)
        placed += len(front)
        if placed >= sel_count:
            break
    return fronts

sorting_network

SortingNetwork(dimension, connectors=None)

A network of wires and comparators that sorts a sequence.

Wires run from left to right and carry one value each. A comparator connects two wires and swaps their values when the upper wire is greater than the lower wire.

Parameters:

Name	Type	Description	Default
dimension	int	Number of wires in the network.	required
connectors	list[tuple[int, int]] | None	Optional list of wire pairs connected by a comparator.	None

See the class docstring.

Source code in deap_er/private/various/sorting_network.py

def __init__(self, dimension: int, connectors: list[tuple[int, int]] | None = None) -> None:
    """See the class docstring."""
    self.dimension = dimension
    self.data: list[list[tuple[int, int]]] = []
    if connectors:
        for wire1, wire2 in connectors:
            self.add_connector(wire1, wire2)
    super().__init__()

depth property

Returns the depth of the network.

length property

Returns the length of the network.

__iter__()

Iterate over comparator levels.

Source code in deap_er/private/various/sorting_network.py

def __iter__(self) -> Iterator[list[tuple[int, int]]]:
    """Iterate over comparator levels."""
    return iter(self.data)

__contains__(item)

Return whether item is a stored level.

Source code in deap_er/private/various/sorting_network.py

def __contains__(self, item: object) -> bool:
    """Return whether ``item`` is a stored level."""
    return item in self.data

__getitem__(key)

Return the comparator level at key.

Source code in deap_er/private/various/sorting_network.py

def __getitem__(self, key: int) -> list[tuple[int, int]]:
    """Return the comparator level at ``key``."""
    return self.data[key]

__setitem__(key, value)

Replace the comparator level at key.

Source code in deap_er/private/various/sorting_network.py

def __setitem__(self, key: int, value: list[tuple[int, int]]) -> None:
    """Replace the comparator level at ``key``."""
    self.data[key] = value

__delitem__(key)

Delete the comparator level at key.

Source code in deap_er/private/various/sorting_network.py

def __delitem__(self, key: int) -> None:
    """Delete the comparator level at ``key``."""
    del self.data[key]

__len__()

Return the number of comparator levels.

Source code in deap_er/private/various/sorting_network.py

def __len__(self) -> int:
    """Return the number of comparator levels."""
    return len(self.data)

check_conflict(level, wire1, wire2) staticmethod

Return whether the wires conflict on the given level.

Parameters:

Name	Type	Description	Default
level	list[tuple[int, int]]	Comparators already present on the level.	required
wire1	int	Index of the first wire.	required
wire2	int	Index of the second wire.	required

Returns:

Type	Description
bool	True if the wires conflict, False otherwise.

Source code in deap_er/private/various/sorting_network.py

@staticmethod
def check_conflict(level: list[tuple[int, int]], wire1: int, wire2: int) -> bool:
    """Return whether the wires conflict on the given level.

    Args:
        level: Comparators already present on the level.
        wire1: Index of the first wire.
        wire2: Index of the second wire.

    Returns:
        True if the wires conflict, False otherwise.
    """
    return any(wires[1] >= wire1 and wires[0] <= wire2 for wires in level)

add_connector(wire1, wire2)

Add a comparator between the two wires.

Same-index wires are ignored.

Parameters:

Name	Type	Description	Default
wire1	int	Index of the first wire.	required
wire2	int	Index of the second wire.	required

Source code in deap_er/private/various/sorting_network.py

def add_connector(self, wire1: int, wire2: int) -> None:
    """Add a comparator between the two wires.

    Same-index wires are ignored.

    Args:
        wire1: Index of the first wire.
        wire2: Index of the second wire.
    """
    if wire1 == wire2:
        return

    if wire1 > wire2:
        wire1, wire2 = wire2, wire1

    index = 0
    for level in reversed(self.data):
        if self.check_conflict(level, wire1, wire2):
            break
        index -= 1

    cnx = (wire1, wire2)
    if index == 0:
        self.data.append([cnx])
    else:
        self.data[index].append(cnx)

sort(values)

Sort values in place using this network.

Parameters:

Name	Type	Description	Default
values	list[Any]	Sequence to sort. Must have at least dimension elements.	required

Source code in deap_er/private/various/sorting_network.py

def sort(self, values: list[Any]) -> None:
    """Sort ``values`` in place using this network.

    Args:
        values: Sequence to sort. Must have at least ``dimension``
            elements.
    """
    for level in self.data:
        for wire1, wire2 in level:
            if values[wire1] > values[wire2]:
                values[wire1], values[wire2] = values[wire2], values[wire1]

evaluate(cases=None)

Count how many cases the network fails to sort.

When cases is omitted, every binary sequence of length dimension is tested.

Parameters:

Name	Type	Description	Default
cases	Iterable[Iterable[Any]] | None	Sequences to sort and check. Optional.	None

Returns:

Type	Description
int	The number of incorrectly sorted cases.

Source code in deap_er/private/various/sorting_network.py

def evaluate(self, cases: Iterable[Iterable[Any]] | None = None) -> int:
    """Count how many ``cases`` the network fails to sort.

    When ``cases`` is omitted, every binary sequence of length
    ``dimension`` is tested.

    Args:
        cases: Sequences to sort and check. Optional.

    Returns:
        The number of incorrectly sorted cases.
    """
    if cases is None:
        cases = product((0, 1), repeat=self.dimension)

    errors = 0
    for sequence in cases:
        original = list(sequence)
        seq = list(original)
        self.sort(seq)
        errors += int(seq != sorted(original))
    return errors

draw()

Return an ASCII diagram of the network.

Returns:

Type	Description
str	A schematic of the wires and comparators.

Source code in deap_er/private/various/sorting_network.py

def draw(self) -> str:
    """Return an ASCII diagram of the network.

    Returns:
        A schematic of the wires and comparators.
    """
    cols = max(7 * max(self.depth, 1), 6 * self.depth + 2)
    str_wires = [["-"] * cols]
    str_wires[0][0] = "0"
    str_wires[0][1] = " o"
    str_spaces = []

    for i in range(1, self.dimension):
        str_wires.append(["-"] * cols)
        str_spaces.append([" "] * cols)
        str_wires[i][0] = str(i)
        str_wires[i][1] = " o"

    for index, level in enumerate(self.data):
        for wire1, wire2 in level:
            str_wires[wire1][(index + 1) * 6] = "x"
            str_wires[wire2][(index + 1) * 6] = "x"
            for i in range(wire1, wire2):
                str_spaces[i][(index + 1) * 6 + 1] = "|"
            for i in range(wire1 + 1, wire2):
                str_wires[i][(index + 1) * 6] = "|"

    network_draw = "".join(str_wires[0])

    for line, space in zip(str_wires[1:], str_spaces, strict=False):
        network_draw += "\n"
        network_draw += "".join(space)
        network_draw += "\n"
        network_draw += "".join(line)

    return network_draw

structural_meta_case

structural_meta_case_weights(columns=None)

Return default lexicase signs for structural meta-case columns.

Bloat metrics default to minimize (-1). non_finite_fraction defaults to maximize (+1) so lexicase prefers programs that are not finite on every bar.

Parameters:

Name	Type	Description	Default
columns	Sequence[str] | None	Subset of :data:STRUCTURAL_META_CASES. All columns are used when omitted.	None

Returns:

Type	Description
tuple[float, ...]	One maximize/minimize sign per column.

Raises:

Type	Description
ValueError	If a column name is unknown.

Source code in deap_er/private/various/structural_meta_case.py

def structural_meta_case_weights(
    columns: Sequence[str] | None = None,
) -> tuple[float, ...]:
    """Return default lexicase signs for structural meta-case columns.

    Bloat metrics default to minimize (-1). ``non_finite_fraction``
    defaults to maximize (+1) so lexicase prefers programs that are
    not finite on every bar.

    Args:
        columns: Subset of :data:`STRUCTURAL_META_CASES`. All
            columns are used when omitted.

    Returns:
        One maximize/minimize sign per column.

    Raises:
        ValueError: If a column name is unknown.
    """
    names = _resolve_columns(columns)
    return tuple(1.0 if name == "non_finite_fraction" else -1.0 for name in names)

structural_meta_case_columns(individuals, *, prim_set=None, predicted=None, columns=None)

Return cheap structural meta-case columns for a population.

Shape (len(individuals), len(columns)). Tree metrics need no predicted. non_finite_fraction needs one row per individual in predicted ((n_ind, n_rows) or a sequence of 1-D series). When predicted is missing, that column is filled with numpy.nan.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Any]	Population whose genomes are PrimitiveTree or list-like node sequences.	required
prim_set	PrimitiveSetTyped | None	Primitive set for unique_opcodes and promote_hits. Required when either column is requested.	None
predicted	ndarray | Sequence[Sequence[float]] | None	Optional per-individual output series.	None
columns	Sequence[str] | None	Subset of :data:STRUCTURAL_META_CASES. All columns are used when omitted.	None

Returns:

Type	Description
ndarray	Structural scalars with one row per individual.

Raises:

Type	Description
ValueError	If individuals is empty, a column name is unknown, prim_set is missing for opcode or promote columns, or predicted has the wrong shape.

Source code in deap_er/private/various/structural_meta_case.py

def structural_meta_case_columns(
    individuals: Sequence[Any],
    *,
    prim_set: PrimitiveSetTyped | None = None,
    predicted: numpy.ndarray | Sequence[Sequence[float]] | None = None,
    columns: Sequence[str] | None = None,
) -> numpy.ndarray:
    """Return cheap structural meta-case columns for a population.

    Shape ``(len(individuals), len(columns))``. Tree metrics need no
    ``predicted``. ``non_finite_fraction`` needs one row per
    individual in ``predicted`` (``(n_ind, n_rows)`` or a sequence of
    1-D series). When ``predicted`` is missing, that column is filled
    with ``numpy.nan``.

    Args:
        individuals: Population whose genomes are ``PrimitiveTree`` or
            list-like node sequences.
        prim_set: Primitive set for ``unique_opcodes`` and
            ``promote_hits``. Required when either column is requested.
        predicted: Optional per-individual output series.
        columns: Subset of :data:`STRUCTURAL_META_CASES`. All columns
            are used when omitted.

    Returns:
        Structural scalars with one row per individual.

    Raises:
        ValueError: If ``individuals`` is empty, a column name is
            unknown, ``prim_set`` is missing for opcode or promote
            columns, or ``predicted`` has the wrong shape.
    """
    if not individuals:
        raise ValueError("individuals must be non-empty")
    names = _resolve_columns(columns)
    if _NEED_PRIM_SET.intersection(names) and prim_set is None:
        raise ValueError("prim_set is required for unique_opcodes and promote_hits")

    predicted_rows = (
        _predicted_rows(individuals, predicted) if "non_finite_fraction" in names else None
    )
    promoted = frozenset(promoted_names(prim_set)) if prim_set is not None else frozenset()
    tape_cache: dict[str, Any] = {}
    matrix = numpy.empty((len(individuals), len(names)), dtype=numpy.float64)

    for row, individual in enumerate(individuals):
        tree = _as_tree(individual)
        for col, name in enumerate(names):
            matrix[row, col] = _column_scalar(
                name,
                tree=tree,
                individual=individual,
                typed_set=prim_set,
                promoted=promoted,
                tape_cache=tape_cache,
                predicted_rows=predicted_rows,
                row=row,
            )
    return matrix