Skip to content

Tools: Weather

Weather station search and download tools.

download_weather_file(wmo=None, query=None, country=None, state=None)

Download EPW/DDY files for a station. Auto-used by run_simulation.

Source code in src/idfkit_mcp/tools/weather.py 
 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
@tool(annotations=_DOWNLOAD)
def download_weather_file(
    wmo: Annotated[str | None, Field(description="WMO station number.")] = None,
    query: Annotated[str | None, Field(description="City or airport name.")] = None,
    country: Annotated[str | None, Field(description='Country code, e.g. "USA".')] = None,
    state: Annotated[str | None, Field(description='State code, e.g. "MA".')] = None,
) -> DownloadWeatherFileResult:
    """Download EPW/DDY files for a station. Auto-used by run_simulation."""
    from idfkit.weather import WeatherDownloader

    index = get_state().get_or_load_station_index()

    if query is not None:
        results = index.search(query, limit=30)
        station = None
        for r in results:
            if not _matches_filters(r.station, country, state):
                continue
            station = r.station
            break
        if station is None:
            raise ToolError(f"No weather stations found for query '{query}'.")
    elif wmo is not None:
        results = index.search(wmo, limit=10)
        station = None
        for r in results:
            if r.station.wmo == wmo:
                station = r.station
                break
        if station is None:
            raise ToolError(f"No weather station found with WMO '{wmo}'.")
    else:
        raise ToolError("Provide either 'wmo' or 'query' to identify the weather station.")

    logger.info("Downloading weather file for station %s (%s)", station.wmo, station.city)
    downloader = WeatherDownloader()
    files = downloader.download(station)

    server_state = get_state()
    server_state.weather_file = files.epw
    server_state.save_session()
    logger.info("Downloaded weather file to %s", files.epw)

    return DownloadWeatherFileResult.model_validate({
        "status": "downloaded",
        "station": serialize_station(station),
        "epw_path": str(files.epw),
        "ddy_path": str(files.ddy),
    })

search_weather_stations(query=None, latitude=None, longitude=None, country=None, state=None, limit=5)

Find weather stations by name or coordinates.

Source code in src/idfkit_mcp/tools/weather.py 
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
69
70
71
72
73
74
75
@tool(annotations=_READ_ONLY_OPEN)
def search_weather_stations(
    query: Annotated[str | None, Field(description="City or airport name.")] = None,
    latitude: Annotated[float | None, Field(description="Latitude.")] = None,
    longitude: Annotated[float | None, Field(description="Longitude.")] = None,
    country: Annotated[str | None, Field(description='Country code, e.g. "USA".')] = None,
    state: Annotated[str | None, Field(description='State code, e.g. "MA".')] = None,
    limit: Annotated[int, Field(description="Max results.")] = 5,
) -> SearchWeatherStationsResult:
    """Find weather stations by name or coordinates."""
    index = get_state().get_or_load_station_index()

    if latitude is not None and longitude is not None:
        spatial_results = index.nearest(latitude, longitude, limit=limit)
        spatial_stations: list[dict[str, Any]] = []
        for r in spatial_results:
            if not _matches_filters(r.station, country, state):
                continue
            spatial_stations.append({
                **serialize_station(r.station),
                "distance_km": round(r.distance_km, 1),
            })
        logger.debug(
            "search_weather_stations: spatial lat=%s lon=%s found=%d", latitude, longitude, len(spatial_stations)
        )
        return SearchWeatherStationsResult(
            search_type="spatial",
            count=len(spatial_stations),
            stations=spatial_stations[:limit],
        )

    if query is not None:
        search_results = index.search(query, limit=limit * 3)
        text_stations: list[dict[str, Any]] = []
        for r in search_results:
            if not _matches_filters(r.station, country, state):
                continue
            text_stations.append({
                **serialize_station(r.station),
                "score": round(r.score, 3),
                "match_field": r.match_field,
            })
            if len(text_stations) >= limit:
                break
        logger.debug("search_weather_stations: query=%r found=%d", query, len(text_stations))
        return SearchWeatherStationsResult(
            search_type="text",
            query=query,
            count=len(text_stations),
            stations=text_stations,
        )

    raise ToolError("Provide either 'query' for text search or 'latitude'/'longitude' for spatial search.")