Skip to content

capturegraph.scheduling.distance.weather #

Weather Distance - Difference in Weather Conditions#

Measures the difference between two weather forecasts across multiple dimensions (temperature, humidity, cloud cover, etc.). Only enabled dimensions contribute to the distance.

WeatherDistanceFunction #

Bases: BatchedDistanceFunction

Weather distance function with batch computation support.

Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class WeatherDistanceFunction(BatchedDistanceFunction):
    """Weather distance function with batch computation support."""

    def __init__(self, dimensions: list[tuple[str, float]]):
        """Initialize weather distance function."""
        self.dimensions = dimensions

    def __call__(self, a: cg.Weather, b: cg.Weather) -> float:
        """Compute distance between two weather objects."""
        if not self.dimensions:
            return 0.0

        distance_sq = 0.0
        for attr, sigma in self.dimensions:
            val_a = getattr(a, attr)
            val_b = getattr(b, attr)
            # Skip if either value is NaN
            if np.isnan(val_a) or np.isnan(val_b):
                continue
            diff = val_a - val_b
            distance_sq += np.square(diff / sigma)

        return float(np.sqrt(distance_sq))

    def extract(self, items: cg.List[Any]) -> np.ndarray:
        """Extract weather dimensions as feature array."""
        n = len(items)
        d = len(self.dimensions)
        features = np.empty((n, d), dtype=np.float64)

        for col, (attr, sigma) in enumerate(self.dimensions):
            for row, item in enumerate(items):
                val = getattr(item, attr)
                # Store normalized value (divide by sigma here for efficiency)
                features[row, col] = val / sigma

        return features

    def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
        """Compute pairwise distances from pre-extracted features."""
        n, d = features_a.shape

        # Expand for broadcasting: (n, 1, d) - (1, m, d) -> (n, m, d)
        diff = features_a[:, np.newaxis, :] - features_b[np.newaxis, :, :]

        # Mask NaN differences (where either side was NaN)
        valid_mask = ~np.isnan(diff)
        diff_sq = np.where(valid_mask, np.square(diff), 0.0)

        # Sum over dimensions and take sqrt
        return np.sqrt(np.sum(diff_sq, axis=2))

__init__(dimensions) #

Initialize weather distance function.

Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
21
22
23
def __init__(self, dimensions: list[tuple[str, float]]):
    """Initialize weather distance function."""
    self.dimensions = dimensions

__call__(a, b) #

Compute distance between two weather objects.

Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __call__(self, a: cg.Weather, b: cg.Weather) -> float:
    """Compute distance between two weather objects."""
    if not self.dimensions:
        return 0.0

    distance_sq = 0.0
    for attr, sigma in self.dimensions:
        val_a = getattr(a, attr)
        val_b = getattr(b, attr)
        # Skip if either value is NaN
        if np.isnan(val_a) or np.isnan(val_b):
            continue
        diff = val_a - val_b
        distance_sq += np.square(diff / sigma)

    return float(np.sqrt(distance_sq))

extract(items) #

Extract weather dimensions as feature array.

Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
def extract(self, items: cg.List[Any]) -> np.ndarray:
    """Extract weather dimensions as feature array."""
    n = len(items)
    d = len(self.dimensions)
    features = np.empty((n, d), dtype=np.float64)

    for col, (attr, sigma) in enumerate(self.dimensions):
        for row, item in enumerate(items):
            val = getattr(item, attr)
            # Store normalized value (divide by sigma here for efficiency)
            features[row, col] = val / sigma

    return features

pairwise(features_a, features_b) #

Compute pairwise distances from pre-extracted features.

Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
def pairwise(self, features_a: np.ndarray, features_b: np.ndarray) -> np.ndarray:
    """Compute pairwise distances from pre-extracted features."""
    n, d = features_a.shape

    # Expand for broadcasting: (n, 1, d) - (1, m, d) -> (n, m, d)
    diff = features_a[:, np.newaxis, :] - features_b[np.newaxis, :, :]

    # Mask NaN differences (where either side was NaN)
    valid_mask = ~np.isnan(diff)
    diff_sq = np.where(valid_mask, np.square(diff), 0.0)

    # Sum over dimensions and take sqrt
    return np.sqrt(np.sum(diff_sq, axis=2))

weather(sigma_temperature_celsius=None, sigma_apparent_temperature_celsius=None, sigma_dew_point_celsius=None, sigma_humidity_ratio=None, sigma_pressure_hpa=None, sigma_wind_speed_mps=None, sigma_wind_gust_mps=None, sigma_wind_direction_degrees=None, sigma_cloud_cover_ratio=None, sigma_precipitation_intensity_mmph=None, sigma_visibility_meters=None, sigma_uv_index=None) #

Create a distance function based on weather conditions.

Only parameters with non-None sigma values contribute to the distance. The combined distance is the Euclidean norm of the normalized differences.

Parameters:

Name Type Description Default
sigma_temperature_celsius float | None

Temperature normalization (°C).

 None
sigma_apparent_temperature_celsius float | None

Feels-like temperature normalization (°C).

 None
sigma_dew_point_celsius float | None

Dew point normalization (°C).

 None
sigma_humidity_ratio float | None

Humidity normalization (0-1 ratio).

 None
sigma_pressure_hpa float | None

Pressure normalization (hPa).

 None
sigma_wind_speed_mps float | None

Wind speed normalization (m/s).

 None
sigma_wind_gust_mps float | None

Wind gust normalization (m/s).

 None
sigma_wind_direction_degrees float | None

Wind direction normalization (degrees).

 None
sigma_cloud_cover_ratio float | None

Cloud cover normalization (0-1 ratio).

 None
sigma_precipitation_intensity_mmph float | None

Precipitation normalization (mm/h).

 None
sigma_visibility_meters float | None

Visibility normalization (m).

 None
sigma_uv_index float | None

UV index normalization.

 None

Returns:

Type Description
WeatherDistanceFunction

A distance function with batch support: (session_a, session_b) -> float
Example 
# Only care about cloud cover for time-lapse (lighting changes)
dist_fn = weather(sigma_cloud_cover_ratio=0.3)

# Care about temp and humidity for outdoor comfort
dist_fn = weather(
    sigma_temperature_celsius=5.0,
    sigma_humidity_ratio=0.2,
)
Source code in capturegraph-lib/capturegraph/scheduling/distance/weather.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def weather(
    sigma_temperature_celsius: float | None = None,
    sigma_apparent_temperature_celsius: float | None = None,
    sigma_dew_point_celsius: float | None = None,
    sigma_humidity_ratio: float | None = None,
    sigma_pressure_hpa: float | None = None,
    sigma_wind_speed_mps: float | None = None,
    sigma_wind_gust_mps: float | None = None,
    sigma_wind_direction_degrees: float | None = None,
    sigma_cloud_cover_ratio: float | None = None,
    sigma_precipitation_intensity_mmph: float | None = None,
    sigma_visibility_meters: float | None = None,
    sigma_uv_index: float | None = None,
) -> WeatherDistanceFunction:
    """Create a distance function based on weather conditions.

    Only parameters with non-None sigma values contribute to the distance.
    The combined distance is the Euclidean norm of the normalized differences.

    Args:
        sigma_temperature_celsius: Temperature normalization (°C).
        sigma_apparent_temperature_celsius: Feels-like temperature normalization (°C).
        sigma_dew_point_celsius: Dew point normalization (°C).
        sigma_humidity_ratio: Humidity normalization (0-1 ratio).
        sigma_pressure_hpa: Pressure normalization (hPa).
        sigma_wind_speed_mps: Wind speed normalization (m/s).
        sigma_wind_gust_mps: Wind gust normalization (m/s).
        sigma_wind_direction_degrees: Wind direction normalization (degrees).
        sigma_cloud_cover_ratio: Cloud cover normalization (0-1 ratio).
        sigma_precipitation_intensity_mmph: Precipitation normalization (mm/h).
        sigma_visibility_meters: Visibility normalization (m).
        sigma_uv_index: UV index normalization.

    Returns:
        A distance function with batch support: `(session_a, session_b) -> float`

    Example:
        ```python
        # Only care about cloud cover for time-lapse (lighting changes)
        dist_fn = weather(sigma_cloud_cover_ratio=0.3)

        # Care about temp and humidity for outdoor comfort
        dist_fn = weather(
            sigma_temperature_celsius=5.0,
            sigma_humidity_ratio=0.2,
        )
        ```
    """
    # Build list of (attribute, sigma) pairs for enabled dimensions
    dimensions = []
    if sigma_temperature_celsius is not None:
        dimensions.append(("temperature_celsius", sigma_temperature_celsius))
    if sigma_apparent_temperature_celsius is not None:
        dimensions.append(
            ("apparent_temperature_celsius", sigma_apparent_temperature_celsius)
        )
    if sigma_dew_point_celsius is not None:
        dimensions.append(("dew_point_celsius", sigma_dew_point_celsius))
    if sigma_humidity_ratio is not None:
        dimensions.append(("humidity_ratio", sigma_humidity_ratio))
    if sigma_pressure_hpa is not None:
        dimensions.append(("pressure_hpa", sigma_pressure_hpa))
    if sigma_wind_speed_mps is not None:
        dimensions.append(("wind_speed_mps", sigma_wind_speed_mps))
    if sigma_wind_gust_mps is not None:
        dimensions.append(("wind_gust_mps", sigma_wind_gust_mps))
    if sigma_wind_direction_degrees is not None:
        dimensions.append(("wind_direction_degrees", sigma_wind_direction_degrees))
    if sigma_cloud_cover_ratio is not None:
        dimensions.append(("cloud_cover_ratio", sigma_cloud_cover_ratio))
    if sigma_precipitation_intensity_mmph is not None:
        dimensions.append(
            ("precipitation_intensity_mmph", sigma_precipitation_intensity_mmph)
        )
    if sigma_visibility_meters is not None:
        dimensions.append(("visibility_meters", sigma_visibility_meters))
    if sigma_uv_index is not None:
        dimensions.append(("uv_index", sigma_uv_index))

    return WeatherDistanceFunction(dimensions)