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
configure 
configure(*, api_key: str, location: str) -> None

Validate and persist the API key and required location.

Source code in src/openjarvis/connectors/weather.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def configure(self, *, api_key: str, location: str) -> None:
    """Validate and persist the API key and required location."""
    api_key = api_key.strip()
    location = location.strip()
    if not api_key:
        raise ValueError("An OpenWeather API key is required")
    if not location:
        raise ValueError("A weather location is required")
    _weather_api_get(
        "https://api.openweathermap.org/data/2.5/weather",
        params={"q": location, "appid": api_key, "units": "imperial"},
    )
    from openjarvis.security.file_utils import secure_write_json

    secure_write_json(
        self._token_path,
        {"api_key": api_key, "location": location},
    )
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 
 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
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()