Skip to content

scalar

scalar #

Simple numeric distance between two scalar values.

Measures the absolute difference between two scalar values.

ScalarDistanceFunction #

Bases: BatchedDistanceFunction[float]

Absolute numeric difference, normalized by sigma.

Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
class ScalarDistanceFunction(BatchedDistanceFunction[float]):
    """Absolute numeric difference, normalized by ``sigma``."""

    def __init__(self, sigma: float = 1.0) -> None:
        """Normalize the absolute difference by ``sigma``."""
        self.sigma = sigma

    def __call__(self, a: float, b: float) -> float:
        """Absolute difference between ``a`` and ``b``, normalized by ``sigma``."""
        return np.abs(a - b) / self.sigma

    def extract(self, items: cg.Array[Any]) -> np.ndarray:
        """Extract raw scalar values as a single-column feature array."""
        return np.array(list(items), dtype=np.float64).reshape(-1, 1)

    def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
        """Compute pairwise absolute differences from extracted scalar features."""
        return np.abs(features_a[:, 0][:, None] - features_b[:, 0][None, :]) / self.sigma

__call__(a, b) #

Absolute difference between a and b, normalized by sigma.

Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
def __call__(self, a: float, b: float) -> float:
    """Absolute difference between ``a`` and ``b``, normalized by ``sigma``."""
    return np.abs(a - b) / self.sigma

__init__(sigma=1.0) #

Normalize the absolute difference by sigma.

Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
def __init__(self, sigma: float = 1.0) -> None:
    """Normalize the absolute difference by ``sigma``."""
    self.sigma = sigma

extract(items) #

Extract raw scalar values as a single-column feature array.

Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
def extract(self, items: cg.Array[Any]) -> np.ndarray:
    """Extract raw scalar values as a single-column feature array."""
    return np.array(list(items), dtype=np.float64).reshape(-1, 1)

pairwise(features_a, features_b) #

Compute pairwise absolute differences from extracted scalar features.

Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
    """Compute pairwise absolute differences from extracted scalar features."""
    return np.abs(features_a[:, 0][:, None] - features_b[:, 0][None, :]) / self.sigma

scalar(sigma=1.0) #

Create a simple scalar distance function.

Computes the absolute difference between two values, normalized by sigma.

Parameters:

Name Type Description Default
sigma float

Normalization factor. Values within sigma of each other have distance < 1.0. Default is 1.0.

1.0

Returns:

Type Description
ScalarDistanceFunction

A distance function (value_a, value_b) -> float with batch support.

Example
import capturegraph.scheduling as cgsh

# Temperature distance (5°C difference is "significant")
dist_fn = cgsh.distance.scalar(sigma=5.0)

# Use with combine for custom attributes
combined = cgsh.distance.combine(
    temperature=cgsh.distance.scalar(sigma=5.0),
    humidity=cgsh.distance.scalar(sigma=0.2),
)
Source code in capturegraph-lib/capturegraph/scheduling/distance/scalar.py
def scalar(
    sigma: float = 1.0,
) -> ScalarDistanceFunction:
    """Create a simple scalar distance function.

    Computes the absolute difference between two values, normalized by sigma.

    Args:
        sigma: Normalization factor. Values within `sigma` of each other
            have distance < 1.0. Default is 1.0.

    Returns:
        A distance function `(value_a, value_b) -> float` with batch support.

    Example:
        ```python
        import capturegraph.scheduling as cgsh

        # Temperature distance (5°C difference is "significant")
        dist_fn = cgsh.distance.scalar(sigma=5.0)

        # Use with combine for custom attributes
        combined = cgsh.distance.combine(
            temperature=cgsh.distance.scalar(sigma=5.0),
            humidity=cgsh.distance.scalar(sigma=0.2),
        )
        ```
    """
    return ScalarDistanceFunction(sigma)