capturegraph.data.load.types.weather

Weather - Weather data container for CaptureGraph data.

A dataclass for weather conditions captured from Apple WeatherKit, loading from PWeather JSON files.

Example

import capturegraph.data as cg
from datetime import datetime

# Vectorized access with List
sessions.weather.temperature_celsius  # → List of all temperatures
sessions.weather.condition            # → List of all conditions

# Fetch historical weather using meteostat
weather = cg.fetch_past_weather(location, datetime(2024, 6, 15, 12, 0))

Notes

• All temperatures in Celsius
• Wind speed in meters per second
• Pressure in hectopascals (hPa)
• Visibility in meters
• Humidity and cloud cover as ratios (0-1)

Weather dataclass

Weather conditions captured by WeatherKit.

Stores atmospheric conditions at the time and location of capture.

Attributes:

Name	Type	Description
temperature_celsius	float	Current temperature in Celsius.
apparent_temperature_celsius	float	Feels-like temperature in Celsius.
dew_point_celsius	float	Dew point temperature in Celsius.
humidity_ratio	float	Relative humidity (0-1).
pressure_hpa	float	Sea level air pressure in hectopascals.
pressure_trend	str	Direction of pressure change ("rising", "falling", "steady").
wind_speed_mps	float	Wind speed in meters per second.
wind_gust_mps	float	Wind gust speed in m/s (optional).
wind_direction_degrees	float	Wind direction in degrees (0-360).
condition	str	Weather condition description (e.g., "cloudy").
symbol_name	str	SF Symbol name for the condition.
cloud_cover_ratio	float	Cloud cover percentage (0-1).
precipitation_intensity_mmph	float	Precipitation in mm/hr.
visibility_meters	float	Visibility distance in meters.
uv_index	float	UV index (0-11+).
is_daylight	bool	Whether there is daylight.
date	datetime	Date and time of observation.

Example

weather = Weather.from_json(data)
weather.temperature_celsius  # 22.5
weather.condition  # 'partly cloudy'

Source code in capturegraph-lib/capturegraph/data/load/types/weather.py

@dataclass
class Weather:
    """Weather conditions captured by WeatherKit.

    Stores atmospheric conditions at the time and location of capture.

    Attributes:
        temperature_celsius: Current temperature in Celsius.
        apparent_temperature_celsius: Feels-like temperature in Celsius.
        dew_point_celsius: Dew point temperature in Celsius.
        humidity_ratio: Relative humidity (0-1).
        pressure_hpa: Sea level air pressure in hectopascals.
        pressure_trend: Direction of pressure change ("rising", "falling", "steady").
        wind_speed_mps: Wind speed in meters per second.
        wind_gust_mps: Wind gust speed in m/s (optional).
        wind_direction_degrees: Wind direction in degrees (0-360).
        condition: Weather condition description (e.g., "cloudy").
        symbol_name: SF Symbol name for the condition.
        cloud_cover_ratio: Cloud cover percentage (0-1).
        precipitation_intensity_mmph: Precipitation in mm/hr.
        visibility_meters: Visibility distance in meters.
        uv_index: UV index (0-11+).
        is_daylight: Whether there is daylight.
        date: Date and time of observation.

    Example:
        ```python
        weather = Weather.from_json(data)
        weather.temperature_celsius  # 22.5
        weather.condition  # 'partly cloudy'
        ```
    """

    temperature_celsius: float
    apparent_temperature_celsius: float
    dew_point_celsius: float
    humidity_ratio: float
    pressure_hpa: float
    pressure_trend: str
    wind_speed_mps: float
    wind_gust_mps: float
    wind_direction_degrees: float
    condition: str
    symbol_name: str
    cloud_cover_ratio: float
    precipitation_intensity_mmph: float
    visibility_meters: float
    uv_index: float
    is_daylight: bool
    date: datetime

    @classmethod
    def from_json(cls, data: dict) -> "Weather":
        """Create a Weather from a PWeather JSON dict.

        Args:
            data: Dict with weather fields from iOS PWeather JSON.

        Returns:
            Weather instance with all conditions.

        Example:
            ```python
            data = {"temperature_celsius": 22.5, "condition": "cloudy", ...}
            Weather.from_json(data)  # Weather(temp=22.5°C, condition=cloudy)
            ```
        """
        return cls(
            temperature_celsius=data["temperature_celsius"],
            apparent_temperature_celsius=data["apparent_temperature_celsius"],
            dew_point_celsius=data["dew_point_celsius"],
            humidity_ratio=data["humidity_ratio"],
            pressure_hpa=data["pressure_hpa"],
            pressure_trend=data["pressure_trend"],
            wind_speed_mps=data["wind_speed_mps"],
            wind_gust_mps=data.get("wind_gust_mps"),
            wind_direction_degrees=data["wind_direction_degrees"],
            condition=data["condition"],
            symbol_name=data["symbol_name"],
            cloud_cover_ratio=data["cloud_cover_ratio"],
            precipitation_intensity_mmph=data["precipitation_intensity_mmph"],
            visibility_meters=data["visibility_meters"],
            uv_index=data["uv_index"],
            is_daylight=data["is_daylight"],
            date=datetime.fromtimestamp(data["date_timestamp"]),
        )

    def to_json(self) -> dict:
        """Convert to PWeather JSON dict.

        Returns:
            Dict matching the iOS PWeather JSON format.

        Example:
            ```python
            weather.to_json()  # {'temperature_celsius': 22.5, 'condition': 'cloudy', ...}
            ```
        """
        data = {
            "temperature_celsius": self.temperature_celsius,
            "apparent_temperature_celsius": self.apparent_temperature_celsius,
            "dew_point_celsius": self.dew_point_celsius,
            "humidity_ratio": self.humidity_ratio,
            "pressure_hpa": self.pressure_hpa,
            "pressure_trend": self.pressure_trend,
            "wind_speed_mps": self.wind_speed_mps,
            "wind_direction_degrees": self.wind_direction_degrees,
            "condition": self.condition,
            "symbol_name": self.symbol_name,
            "cloud_cover_ratio": self.cloud_cover_ratio,
            "precipitation_intensity_mmph": self.precipitation_intensity_mmph,
            "visibility_meters": self.visibility_meters,
            "uv_index": self.uv_index,
            "is_daylight": self.is_daylight,
            "date_timestamp": self.date.timestamp(),
        }
        if self.wind_gust_mps is not None:
            data["wind_gust_mps"] = self.wind_gust_mps
        return data

    def __repr__(self) -> str:
        return f"Weather(temp={self.temperature_celsius:.1f}°C, condition={self.condition})"

from_json(data) classmethod

Create a Weather from a PWeather JSON dict.

Parameters:

Name	Type	Description	Default
data	dict	Dict with weather fields from iOS PWeather JSON.	required

Returns:

Type	Description
Weather	Weather instance with all conditions.

Example

data = {"temperature_celsius": 22.5, "condition": "cloudy", ...}
Weather.from_json(data)  # Weather(temp=22.5°C, condition=cloudy)

Source code in capturegraph-lib/capturegraph/data/load/types/weather.py

@classmethod
def from_json(cls, data: dict) -> "Weather":
    """Create a Weather from a PWeather JSON dict.

    Args:
        data: Dict with weather fields from iOS PWeather JSON.

    Returns:
        Weather instance with all conditions.

    Example:
        ```python
        data = {"temperature_celsius": 22.5, "condition": "cloudy", ...}
        Weather.from_json(data)  # Weather(temp=22.5°C, condition=cloudy)
        ```
    """
    return cls(
        temperature_celsius=data["temperature_celsius"],
        apparent_temperature_celsius=data["apparent_temperature_celsius"],
        dew_point_celsius=data["dew_point_celsius"],
        humidity_ratio=data["humidity_ratio"],
        pressure_hpa=data["pressure_hpa"],
        pressure_trend=data["pressure_trend"],
        wind_speed_mps=data["wind_speed_mps"],
        wind_gust_mps=data.get("wind_gust_mps"),
        wind_direction_degrees=data["wind_direction_degrees"],
        condition=data["condition"],
        symbol_name=data["symbol_name"],
        cloud_cover_ratio=data["cloud_cover_ratio"],
        precipitation_intensity_mmph=data["precipitation_intensity_mmph"],
        visibility_meters=data["visibility_meters"],
        uv_index=data["uv_index"],
        is_daylight=data["is_daylight"],
        date=datetime.fromtimestamp(data["date_timestamp"]),
    )

to_json()

Convert to PWeather JSON dict.

Returns:

Type	Description
dict	Dict matching the iOS PWeather JSON format.

Example

weather.to_json()  # {'temperature_celsius': 22.5, 'condition': 'cloudy', ...}

Source code in capturegraph-lib/capturegraph/data/load/types/weather.py

def to_json(self) -> dict:
    """Convert to PWeather JSON dict.

    Returns:
        Dict matching the iOS PWeather JSON format.

    Example:
        ```python
        weather.to_json()  # {'temperature_celsius': 22.5, 'condition': 'cloudy', ...}
        ```
    """
    data = {
        "temperature_celsius": self.temperature_celsius,
        "apparent_temperature_celsius": self.apparent_temperature_celsius,
        "dew_point_celsius": self.dew_point_celsius,
        "humidity_ratio": self.humidity_ratio,
        "pressure_hpa": self.pressure_hpa,
        "pressure_trend": self.pressure_trend,
        "wind_speed_mps": self.wind_speed_mps,
        "wind_direction_degrees": self.wind_direction_degrees,
        "condition": self.condition,
        "symbol_name": self.symbol_name,
        "cloud_cover_ratio": self.cloud_cover_ratio,
        "precipitation_intensity_mmph": self.precipitation_intensity_mmph,
        "visibility_meters": self.visibility_meters,
        "uv_index": self.uv_index,
        "is_daylight": self.is_daylight,
        "date_timestamp": self.date.timestamp(),
    }
    if self.wind_gust_mps is not None:
        data["wind_gust_mps"] = self.wind_gust_mps
    return data

fetch_past_weather(location, dt)

Fetch historical weather data for a location and time using Meteostat.

Uses the Meteostat library to retrieve historical weather observations for a specific location and time. This is useful for backfilling weather data for captures that didn't record weather at capture time.

Parameters:

Name	Type	Description	Default
location	Location	Location with latitude, longitude, and altitude.	required
dt	datetime	Date and time to fetch weather for (timezone-naive assumes local).	required

Returns:

Type	Description
Weather	Weather instance with available data (some fields may be estimated).

Raises:

Type	Description
ImportError	If meteostat is not installed.
ValueError	If no weather data is available for the location/time.

Example

import capturegraph.data as cg
from datetime import datetime

loc = cg.Location(latitude=42.445, longitude=-76.480, altitude=261.5)
weather = cg.fetch_past_weather(loc, datetime(2024, 6, 15, 12, 0))
weather.temperature_celsius  # 22.5

Source code in capturegraph-lib/capturegraph/data/load/types/weather.py

def fetch_past_weather(location: Location, dt: datetime) -> Weather:
    """Fetch historical weather data for a location and time using Meteostat.

    Uses the Meteostat library to retrieve historical weather observations
    for a specific location and time. This is useful for backfilling weather
    data for captures that didn't record weather at capture time.

    Args:
        location: Location with latitude, longitude, and altitude.
        dt: Date and time to fetch weather for (timezone-naive assumes local).

    Returns:
        Weather instance with available data (some fields may be estimated).

    Raises:
        ImportError: If meteostat is not installed.
        ValueError: If no weather data is available for the location/time.

    Example:
        ```python
        import capturegraph.data as cg
        from datetime import datetime

        loc = cg.Location(latitude=42.445, longitude=-76.480, altitude=261.5)
        weather = cg.fetch_past_weather(loc, datetime(2024, 6, 15, 12, 0))
        weather.temperature_celsius  # 22.5
        ```
    """
    try:
        import meteostat as ms
    except ImportError as e:
        raise ImportError(
            "fetch_past_weather requires meteostat. Install with: pip install meteostat"
        ) from e

    # Create a Meteostat Point for the location
    point = ms.Point(location.latitude, location.longitude, location.altitude)

    # Get nearby weather stations
    stations = ms.stations.nearby(point, limit=4)

    # Fetch hourly data for the date and interpolate to exact location
    start = dt.replace(hour=0, minute=0, second=0, microsecond=0)
    end = dt.replace(hour=23, minute=59, second=59, microsecond=0)
    ts = ms.hourly(stations, start, end)
    df = ms.interpolate(ts, point).fetch()

    if df is None or df.empty:
        raise ValueError(
            f"No weather data available for ({location.latitude}, {location.longitude}) "
            f"on {dt.date()}"
        )

    # Find the closest hour
    target_hour = dt.replace(minute=0, second=0, microsecond=0)
    if target_hour in df.index:
        row = df.loc[target_hour]
    else:
        # Find nearest hour
        idx = df.index.get_indexer([target_hour], method="nearest")[0]
        row = df.iloc[idx]

    import pandas as pd

    # Map meteostat data to Weather fields
    # Meteostat provides: temp, rhum, prcp, snwd, wdir, wspd, wpgt, pres, tsun, cldc, coco
    # Use pd.isna() to handle pandas NA values safely
    def safe_get(val, default):
        """Return default if val is None or pd.NA."""
        return default if val is None or pd.isna(val) else val

    temp = safe_get(row.get("temp"), 0.0)
    dwpt = safe_get(row.get("dwpt"), temp)
    rhum = safe_get(row.get("rhum"), 50.0)  # Default 50% if missing
    pres = safe_get(row.get("pres"), 1013.25)  # Default sea level
    wspd = safe_get(row.get("wspd"), float("nan"))  # km/h in meteostat
    wpgt = safe_get(row.get("wpgt"), float("nan"))  # Wind gust km/h (None if missing)
    wdir = safe_get(row.get("wdir"), 0.0)
    prcp = safe_get(row.get("prcp"), 0.0)  # Precipitation mm
    coco = safe_get(row.get("coco"), 1)  # Weather condition code
    cldc = safe_get(row.get("cldc"), float("nan"))  # Cloud cover in oktas (0-8)
    tsun = safe_get(row.get("tsun"), float("nan"))  # Sunshine duration in minutes

    # Convert units: meteostat wspd is km/h, we need m/s
    wind_speed_mps = wspd / 3.6 if wspd else 0.0
    wind_gust_mps = wpgt / 3.6 if wpgt else float("nan")

    # Convert cloud cover from oktas (0-8) to ratio (0-1)
    if cldc is not None and not pd.isna(cldc):
        cloud_cover_ratio = float(cldc) / 8.0
    else:
        # Estimate from condition if cldc not available
        cloud_cover_ratio = 0.5 if coco and int(coco) >= 3 else 0.0

    # Map condition code to description
    # Meteostat codes: 1=clear, 2=fair, 3=cloudy, 4=overcast, 5=fog, 6=rain, etc.
    condition_map = {
        1: "Clear",
        2: "Mostly Clear",
        3: "Partly Cloudy",
        4: "Cloudy",
        5: "Fog",
        6: "Freezing Fog",
        7: "Light Rain",
        8: "Rain",
        9: "Heavy Rain",
        10: "Freezing Rain",
        11: "Heavy Freezing Rain",
        12: "Sleet",
        13: "Heavy Sleet",
        14: "Light Snow",
        15: "Snow",
        16: "Heavy Snow",
        17: "Rain Shower",
        18: "Heavy Rain Shower",
        19: "Sleet Shower",
        20: "Heavy Sleet Shower",
        21: "Snow Shower",
        22: "Heavy Snow Shower",
        23: "Lightning",
        24: "Hail",
        25: "Thunderstorm",
        26: "Heavy Thunderstorm",
        27: "Storm",
    }
    condition = condition_map.get(int(coco) if coco else 1, "Unknown")

    # Map condition to SF Symbol (best effort)
    symbol_map = {
        "clear": "sun.max.fill",
        "mostly clear": "sun.max.fill",
        "partly cloudy": "cloud.sun.fill",
        "cloudy": "cloud.fill",
        "fog": "cloud.fog.fill",
        "freezing fog": "cloud.fog.fill",
        "light rain": "cloud.drizzle.fill",
        "rain": "cloud.rain.fill",
        "heavy rain": "cloud.heavyrain.fill",
        "snow": "cloud.snow.fill",
        "heavy snow": "cloud.snow.fill",
        "thunderstorm": "cloud.bolt.rain.fill",
    }
    symbol_name = symbol_map.get(condition, "cloud.fill")

    # Determine if daylight based on sunshine duration if available, else use hour
    if tsun is not None and not pd.isna(tsun) and tsun > 0:
        is_daylight = True
    else:
        hour = dt.hour
        is_daylight = 6 <= hour <= 20

    return Weather(
        temperature_celsius=float(temp),
        apparent_temperature_celsius=float("nan"),
        dew_point_celsius=float(dwpt),
        humidity_ratio=float(rhum) / 100.0,
        pressure_hpa=float(pres),
        pressure_trend="",
        wind_speed_mps=wind_speed_mps,
        wind_gust_mps=wind_gust_mps,
        wind_direction_degrees=float(wdir),
        condition=condition,
        symbol_name=symbol_name,
        cloud_cover_ratio=cloud_cover_ratio,
        precipitation_intensity_mmph=float(prcp),
        visibility_meters=float("nan"),
        uv_index=float("nan"),
        is_daylight=is_daylight,
        date=dt,
    )