pygx.algo.scalars

Step-based scalars used as hyper-parameters for PyGX algorithms.

scalars

Step-based scalars used as hyper-parameters for PyGX algorithms.

A scalar is a deferred numerical expression that produces a number when sampled at a given step. They are useful as hyper-parameters in evolutionary algorithms when a value should change over the course of a search — for example, a mutation rate that anneals from 1.0 down to 0.1 over 1000 steps.

This module provides:

• The Scalar base type plus Constant for trivial cases.
• Arithmetic combinators (Addition, Multiplication, Division, Mod, ...) so scalars compose with +/*/etc.
• Elementary functions (Cosine, Exp, Log, ...).
• Stochastic primitives (Gaussian, LogNormal).
• make_scalar for converting a plain Python value into a Scalar when an API accepts either.

STEP module-attribute

STEP = _Step()

Scalar that returns the step itself: STEP(i) == i.

Use it as the seed for schedule expressions, e.g. STEP * 0.1 + 1.0.

Absolute

Absolute(x: ScalarValue, **kwargs)

Bases: UnaryOp

Absolute operator.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Addition

Addition(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Add operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

BinaryOp

BinaryOp(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: Scalar

Base class for binary scalar operators.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

operate abstractmethod

operate(x: int | float, y: int | float) -> int | float

Implementation of the operation on two computed values.

Source code in pygx/algo/scalars/_base.py

@abc.abstractmethod
def operate(self, x: int | float, y: int | float) -> int | float:
    """Implementation of the operation on two computed values."""

Ceiling

Ceiling(x: ScalarValue, **kwargs)

Bases: UnaryOp

Ceil operator.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Constant

Constant(value: Any, **kwargs)

Bases: Scalar

A scalar that returns the same value at every step.

Source code in pygx/algo/scalars/_base.py

def __init__(self, value: Any, **kwargs):
    super().__init__(**dict(value=value), **kwargs)

Division

Division(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Divide operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Floor

Floor(x: ScalarValue, **kwargs)

Bases: UnaryOp

Floor operator.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Lambda

Lambda(fn: Callable[[int], Any], **kwargs)

Bases: Scalar

A scalar backed by a user function that maps a step to a value.

Source code in pygx/algo/scalars/_base.py

def __init__(self, fn: Callable[[int], Any], **kwargs):
    super().__init__(**dict(fn=fn), **kwargs)

Mod

Mod(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Mod operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Multiplication

Multiplication(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Multiply operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Negation

Negation(x: ScalarValue, **kwargs)

Bases: UnaryOp

Negation operator.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Power

Power(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Power operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

Scalar

Bases: Object

Interface for step-based scalars.

A scalar maps an integer step to a value. Scalars compose with the standard arithmetic operators, so schedules can be written as expressions of STEP:

Example:

s = pg.algo.scalars.STEP * 0.1 + 1.0
assert s(0) == 1.0
assert s(10) == 2.0

call abstractmethod

call(step: int) -> Any

Implementation. Subclass should override this method.

Source code in pygx/algo/scalars/_base.py

@abc.abstractmethod
def call(self, step: int) -> Any:
    """Implementation. Subclass should override this method."""

floor

floor()

Returns the floor of current scalar.

Source code in pygx/algo/scalars/_base.py

def floor(self):
    """Returns the floor of current scalar."""
    return Floor(x=self)

ceil

ceil()

Returns the ceiling of current scalar.

Source code in pygx/algo/scalars/_base.py

def ceil(self):
    """Returns the ceiling of current scalar."""
    return Ceiling(x=self)

Subtraction

Subtraction(x: ScalarValue, y: ScalarValue, **kwargs)

Bases: BinaryOp

Subtract operation.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, y: ScalarValue, **kwargs):
    super().__init__(**dict(x=x, y=y), **kwargs)

UnaryOp

UnaryOp(x: ScalarValue, **kwargs)

Bases: Scalar

Base class for unary scalar operators.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

operate abstractmethod

operate(x: int | float) -> int | float

Implementation of the operation on a computed value.

Source code in pygx/algo/scalars/_base.py

@abc.abstractmethod
def operate(self, x: int | float) -> int | float:
    """Implementation of the operation on a computed value."""

Cosine

Cosine(x: ScalarValue, **kwargs)

Bases: UnaryOp

Cosine that works for scalars.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Exp

Exp(x: ScalarValue, **kwargs)

Bases: UnaryOp

The exponential scalar e ** x (computed via math.exp).

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Log

Log(x: ScalarValue, base: ScalarValue = e, **kwargs)

Bases: Scalar

The logarithm scalar log(x, base) (natural log by default).

Source code in pygx/algo/scalars/_maths.py

def __init__(
    self,
    x: base.ScalarValue,  # pyright: ignore[reportInvalidTypeForm]
    base: base.ScalarValue = math.e,  # pylint: disable=redefined-outer-name  # pyright: ignore[reportInvalidTypeForm]
    **kwargs,
):
    super().__init__(**dict(x=x, base=base), **kwargs)

Sine

Sine(x: ScalarValue, **kwargs)

Bases: UnaryOp

Sine that works for scalars.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

SquareRoot

SquareRoot(x: ScalarValue, **kwargs)

Bases: UnaryOp

The square root scalar.

Source code in pygx/algo/scalars/_base.py

def __init__(self, x: ScalarValue, **kwargs):
    super().__init__(**dict(x=x), **kwargs)

Gaussian

Gaussian(mean: float, std: float, **kwargs)

Bases: _MeanStdRandom

A random float from a Gaussian distribution.

Source code in pygx/algo/scalars/_randoms.py

def __init__(self, mean: float, std: float, **kwargs):
    super().__init__(**dict(mean=mean, std=std), **kwargs)

LogNormal

LogNormal(mean: float, std: float, **kwargs)

Bases: _MeanStdRandom

A random float from a log-normal distribution.

mean and std parameterize the underlying normal distribution (the distribution of the value's logarithm), not the samples themselves.

Source code in pygx/algo/scalars/_randoms.py

def __init__(self, mean: float, std: float, **kwargs):
    super().__init__(**dict(mean=mean, std=std), **kwargs)

Normal

Normal(mean: float, std: float, **kwargs)

Bases: _MeanStdRandom

A random float from a normal distribution.

Source code in pygx/algo/scalars/_randoms.py

def __init__(self, mean: float, std: float, **kwargs):
    super().__init__(**dict(mean=mean, std=std), **kwargs)

RandomScalar

Bases: Scalar

Base class for random scalars.

A random scalar draws a fresh value from a distribution on every call; the step is ignored. Subclasses implement next_value.

next_value abstractmethod

next_value() -> int | float

Returns the next value.

Source code in pygx/algo/scalars/_randoms.py

@abc.abstractmethod
def next_value(self) -> int | float:
    """Returns the next value."""

Triangular

Triangular(
    low: int | float = 0.0,
    high: int | float = 1.0,
    mode: int | float | None = None,
    **kwargs
)

Bases: RandomScalar

A random number from a triangular distribution.

The result is truncated to an integer when both bounds are integers.

Source code in pygx/algo/scalars/_randoms.py

def __init__(
    self,
    low: int | float = 0.0,
    high: int | float = 1.0,
    mode: int | float | None = None,
    **kwargs,
):
    super().__init__(**dict(low=low, high=high, mode=mode), **kwargs)

Uniform

Uniform(low: int | float = 0.0, high: int | float = 1.0, **kwargs)

Bases: RandomScalar

A uniformly distributed random number between low and high.

When both bounds are integers, an integer is drawn (both bounds inclusive); otherwise a float is drawn.

Source code in pygx/algo/scalars/_randoms.py

def __init__(
    self,
    low: int | float = 0.0,
    high: int | float = 1.0,
    **kwargs,
):
    super().__init__(**dict(low=low, high=high), **kwargs)

StepWise

StepWise(
    phases: list[tuple[int | float, Any]],
    total_steps: int | None = None,
    **kwargs
)

Bases: Scalar

A step-wise schedule that is specified via multiple phases.

Each phase pairs a length with a scheduled value; the value is queried with a step relative to the phase start. The schedule is stateful: it expects to be called with increasing steps (as done during an evolution run), advancing to the next phase after the current phase's last step has been queried. Past the final phase, the last value is returned.

Example:

s = pg.algo.scalars.StepWise([
    # Anneal from 1.0 to 0.0 for the first 2 steps.
    (2, pg.algo.scalars.linear(2, start=1.0, end=0.0)),
    # Then stay at 0.5 for 2 more steps.
    (2, 0.5),
])
assert [s(i) for i in range(5)] == [1.0, 0.5, 0.5, 0.5, 0.5]

Source code in pygx/algo/scalars/_step_wise.py

def __init__(
    self,
    phases: list[tuple[int | float, Any]],
    total_steps: int | None = None,
    **kwargs,
):
    super().__init__(
        **dict(phases=phases, total_steps=total_steps), **kwargs
    )

make_scalar

make_scalar(value: Any) -> Scalar

Converts a value into a Scalar.

A Scalar is returned as is; another callable is wrapped into a Lambda; any other value is wrapped into a Constant.

Parameters:

Name	Type	Description	Default
value	Any	The value to convert.	required

Returns:

Type	Description
Scalar	A Scalar that produces value (or its per-step result) when called.

Source code in pygx/algo/scalars/_base.py

def make_scalar(value: Any) -> 'Scalar':
    """Converts a value into a `Scalar`.

    A `Scalar` is returned as is; another callable is wrapped into a
    [`Lambda`][pygx.algo.scalars.Lambda]; any other value is wrapped into a
    [`Constant`][pygx.algo.scalars.Constant].

    Args:
      value: The value to convert.

    Returns:
      A `Scalar` that produces `value` (or its per-step result) when called.
    """
    if isinstance(value, Scalar):
        return value
    elif callable(value):
        return Lambda(fn=value)  # pytype: disable=bad-return-type
    else:
        return Constant(value=value)  # pytype: disable=bad-return-type

scalar_spec

scalar_spec(value_spec: ValueSpec) -> ValueSpec

Returns the value spec for a field that accepts a scheduled value.

Use this to declare fields that accept either a plain value or a step-based schedule for it (a Scalar or any callable taking a step).

Parameters:

Name	Type	Description	Default
value_spec	ValueSpec	The value spec for the value produced at each step.	required

Returns:

Type	Description
ValueSpec	A value spec that accepts either the value itself or a callable that produces such a value from a step (an integer).

Source code in pygx/algo/scalars/_base.py

def scalar_spec(value_spec: pg.typing.ValueSpec) -> pg.typing.ValueSpec:
    """Returns the value spec for a field that accepts a scheduled value.

    Use this to declare fields that accept either a plain value or a
    step-based schedule for it (a `Scalar` or any callable taking a step).

    Args:
      value_spec: The value spec for the value produced at each step.

    Returns:
      A value spec that accepts either the value itself or a callable that
        produces such a value from a step (an integer).
    """
    return pg.typing.Union(
        [value_spec, pg.typing.Callable([pg.typing.Int()], returns=value_spec)]
    )

scalar_value

scalar_value(value: Any, step: int) -> Any

Evaluates a scheduled value at a step.

Parameters:

Name	Type	Description	Default
value	Any	A plain value, or a callable (e.g. a Scalar) that maps a step to a value.	required
step	int	The step to evaluate the value at.	required

Returns:

Type	Description
Any	value(step) if value is callable; otherwise value unchanged.

Source code in pygx/algo/scalars/_base.py

def scalar_value(value: Any, step: int) -> Any:
    """Evaluates a scheduled value at a step.

    Args:
      value: A plain value, or a callable (e.g. a `Scalar`) that maps a step
        to a value.
      step: The step to evaluate the value at.

    Returns:
      `value(step)` if `value` is callable; otherwise `value` unchanged.
    """
    if callable(value):
        return value(step)
    return value

cosine_decay

cosine_decay(total_steps: int, start: float = 1.0, end: float = 0.0)

Returns a scalar that decays from start to end on a cosine curve.

The value is start at step 0 and reaches end at step total_steps.

Source code in pygx/algo/scalars/_maths.py

def cosine_decay(total_steps: int, start: float = 1.0, end: float = 0.0):
    """Returns a scalar that decays from `start` to `end` on a cosine curve.

    The value is `start` at step 0 and reaches `end` at step `total_steps`.
    """
    return (
        0.5 * (start - end) * (1 + cos(math.pi * base.STEP / total_steps)) + end  # pyright: ignore[reportCallIssue]
    )

cyclic

cyclic(
    cycle: int,
    initial_radiant: float = 0.0,
    high: float = 1.0,
    low: float = 0.0,
)

Returns a scalar oscillating between high and low on a cosine wave.

The oscillation has a period of cycle steps and starts at high when initial_radiant is 0.

Source code in pygx/algo/scalars/_maths.py

def cyclic(
    cycle: int,
    initial_radiant: float = 0.0,
    high: float = 1.0,
    low: float = 0.0,
):
    """Returns a scalar oscillating between `high` and `low` on a cosine wave.

    The oscillation has a period of `cycle` steps and starts at `high` when
    `initial_radiant` is 0.
    """
    return (
        0.5
        * (high - low)
        * (1 + cos(initial_radiant + math.pi * 2 * base.STEP / cycle))  # pyright: ignore[reportCallIssue]
        + low
    )

exponential_decay

exponential_decay(
    decay_rate: float,
    decay_interval: int,
    start: float = 1.0,
    staircase: bool = True,
)

Returns a scalar that multiplies start by decay_rate per interval.

With staircase (the default), the value stays constant within each decay_interval-step interval and drops at interval boundaries; otherwise it decays continuously at every step.

Source code in pygx/algo/scalars/_maths.py

def exponential_decay(
    decay_rate: float,
    decay_interval: int,
    start: float = 1.0,
    staircase: bool = True,
):
    """Returns a scalar that multiplies `start` by `decay_rate` per interval.

    With `staircase` (the default), the value stays constant within each
    `decay_interval`-step interval and drops at interval boundaries;
    otherwise it decays continuously at every step.
    """
    exponent = base.STEP / float(decay_interval)
    if staircase:
        exponent = exponent.floor()
    return start * (decay_rate**exponent)

linear

linear(total_steps: int, start: float = 1.0, end: float = 0.0)

Returns a scalar that moves linearly from start to end.

The value is start at step 0 and reaches end at step total_steps.

Source code in pygx/algo/scalars/_maths.py

def linear(total_steps: int, start: float = 1.0, end: float = 0.0):
    """Returns a scalar that moves linearly from `start` to `end`.

    The value is `start` at step 0 and reaches `end` at step `total_steps`.
    """
    return start + base.STEP * ((end - start) / total_steps)

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2