Skip to content

weather

weather

Weather connector — current conditions and forecast via OpenWeatherMap API.

Uses an API key stored in the connector config dir. All API calls are in module-level functions for easy mocking in tests.

Classes

WeatherConnector 

WeatherConnector(*, token_path: str = _DEFAULT_TOKEN_PATH)

Bases: BaseConnector

Fetch current weather and short-term forecast from OpenWeatherMap.

Source code in src/openjarvis/connectors/weather.py 
38
39
40
def __init__(self, *, token_path: str = _DEFAULT_TOKEN_PATH) -> None:
    self._token_path = Path(token_path)
    self._status = SyncStatus()
Functions
sync 
sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Yield Documents for current weather and forecast.

Source code in src/openjarvis/connectors/weather.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 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
def sync(
    self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
    """Yield Documents for current weather and forecast."""
    config = self._load_config()
    api_key = config["api_key"]
    location = config.get("location", "San Francisco,CA")

    # Current weather
    current = _weather_api_get(
        "https://api.openweathermap.org/data/2.5/weather",
        params={"q": location, "appid": api_key, "units": "imperial"},
    )
    main = current.get("main", {})
    weather_desc = ", ".join(
        w.get("description", "") for w in current.get("weather", [])
    )
    content = (
        f"Temperature: {main.get('temp')}°F, "
        f"Conditions: {weather_desc}, "
        f"Humidity: {main.get('humidity')}%, "
        f"Wind: {current.get('wind', {}).get('speed')} mph"
    )
    yield Document(
        doc_id=f"weather-current-{location}",
        source="weather",
        doc_type="current",
        content=content,
        title=f"Current Weather — {location}",
        timestamp=datetime.now(),
        metadata={
            "location": location,
            "temp": main.get("temp"),
            "conditions": weather_desc,
            "humidity": main.get("humidity"),
            "wind_speed": current.get("wind", {}).get("speed"),
        },
    )

    # Forecast (next ~12 hours, 4 x 3-hour intervals)
    forecast = _weather_api_get(
        "https://api.openweathermap.org/data/2.5/forecast",
        params={
            "q": location,
            "appid": api_key,
            "units": "imperial",
            "cnt": "4",
        },
    )
    summaries = []
    for entry in forecast.get("list", []):
        dt_txt = entry.get("dt_txt", "")
        temp = entry.get("main", {}).get("temp")
        desc = ", ".join(w.get("description", "") for w in entry.get("weather", []))
        summaries.append(f"{dt_txt}: {temp}°F, {desc}")
    forecast_content = "Forecast:\n" + "\n".join(summaries)

    yield Document(
        doc_id=f"weather-forecast-{location}",
        source="weather",
        doc_type="forecast",
        content=forecast_content,
        title=f"Weather Forecast — {location}",
        timestamp=datetime.now(),
        metadata={"location": location},
    )

    self._status.state = "idle"
    self._status.last_sync = datetime.now()