Skip to content

combination

combination #

Several per-dimension metrics over one session, fused as a Euclidean norm.

CombinedDistance #

Bases: DistanceFunction[Any]

Each term's distance along its own attribute path, combined in L2.

Source code in capturegraph-lib/capturegraph/scheduling/distance/combination.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class CombinedDistance(DistanceFunction[Any]):
    """Each term's distance along its own attribute path, combined in L2."""

    def __init__(self, terms: list[tuple[str, DistanceFunction]]) -> None:
        """Measure each ``(attribute path, metric)`` term and combine them."""
        self._terms = terms

    def extract(self, values: cg.Array) -> np.ndarray:
        """Each term's features for the value at its own attribute path."""
        features = np.empty(len(self._terms), dtype=object)
        for index, (path, metric) in enumerate(self._terms):
            features[index] = metric.extract(cg.Array([access(value, path) for value in values]))
        return features

    def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
        """The L2 norm over the terms both sides define (``nan`` when none do)."""
        squares = np.stack(
            [
                np.square(metric.pairwise(term_a, term_b))
                for (_, metric), term_a, term_b in zip(
                    self._terms,
                    features_a,
                    features_b,
                    strict=True,
                )
            ]
        )
        absent = np.all(np.isnan(squares), axis=0)
        return np.sqrt(np.where(absent, np.nan, np.nansum(squares, axis=0)))

__init__(terms) #

Measure each (attribute path, metric) term and combine them.

Source code in capturegraph-lib/capturegraph/scheduling/distance/combination.py 
18
19
20
def __init__(self, terms: list[tuple[str, DistanceFunction]]) -> None:
    """Measure each ``(attribute path, metric)`` term and combine them."""
    self._terms = terms

extract(values) #

Each term's features for the value at its own attribute path.

Source code in capturegraph-lib/capturegraph/scheduling/distance/combination.py 
22
23
24
25
26
27
def extract(self, values: cg.Array) -> np.ndarray:
    """Each term's features for the value at its own attribute path."""
    features = np.empty(len(self._terms), dtype=object)
    for index, (path, metric) in enumerate(self._terms):
        features[index] = metric.extract(cg.Array([access(value, path) for value in values]))
    return features

pairwise(features_a, features_b) #

The L2 norm over the terms both sides define (nan when none do).

Source code in capturegraph-lib/capturegraph/scheduling/distance/combination.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
    """The L2 norm over the terms both sides define (``nan`` when none do)."""
    squares = np.stack(
        [
            np.square(metric.pairwise(term_a, term_b))
            for (_, metric), term_a, term_b in zip(
                self._terms,
                features_a,
                features_b,
                strict=True,
            )
        ]
    )
    absent = np.all(np.isnan(squares), axis=0)
    return np.sqrt(np.where(absent, np.nan, np.nansum(squares, axis=0)))

combine(**metrics) #

Combine per-dimension distance functions into a single session metric.

Each keyword names the session attribute its metric reads, and the combined distance is the Euclidean norm over those terms: d = sqrt(d1² + d2² + ... + dn²). A term whose attribute is absent on either side drops out; a pair with no surviving term has no distance, and select_sessions never offers such a candidate.

Parameters:

Name Type Description Default
**metrics Callable[[Any, Any], float]

Named distance functions — solar_angle=fn measures session.solar_angle (or session["solar_angle"]) with fn. A plain (a, b) -> float callable works alongside the cgsh.distance.* metrics.

 {}

Returns:

Type Description
CombinedDistance

A combined metric: combined(a, b) for one pair,
CombinedDistance

combined.matrix(sessions_a, sessions_b) for every pair at once.

Raises:

Type Description
ValueError

If no distance function is given.
Example 
import capturegraph.scheduling as cgsh

distance_fn = cgsh.distance.combine(
    solar_angle=cgsh.distance.solar(sigma_deg=2.0),
    location=cgsh.distance.location(sigma_m=100.0),
)
Source code in capturegraph-lib/capturegraph/scheduling/distance/combination.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def combine(**metrics: Callable[[Any, Any], float]) -> CombinedDistance:
    """Combine per-dimension distance functions into a single session metric.

    Each keyword names the session attribute its metric reads, and the combined
    distance is the Euclidean norm over those terms:
    `d = sqrt(d1² + d2² + ... + dn²)`. A term whose attribute is absent on
    either side drops out; a pair with no surviving term has no distance, and
    `select_sessions` never offers such a candidate.

    Args:
        **metrics: Named distance functions — `solar_angle=fn` measures
            `session.solar_angle` (or `session["solar_angle"]`) with `fn`. A
            plain `(a, b) -> float` callable works alongside the
            `cgsh.distance.*` metrics.

    Returns:
        A combined metric: `combined(a, b)` for one pair,
        `combined.matrix(sessions_a, sessions_b)` for every pair at once.

    Raises:
        ValueError: If no distance function is given.

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

        distance_fn = cgsh.distance.combine(
            solar_angle=cgsh.distance.solar(sigma_deg=2.0),
            location=cgsh.distance.location(sigma_m=100.0),
        )
        ```
    """
    if not metrics:
        raise ValueError("combine needs at least one named distance function")
    return CombinedDistance([(path, as_distance(metric)) for path, metric in metrics.items()])