Station API

Weather station index, search, and geocoding.

StationIndex

idfkit.weather.index.StationIndex

Searchable index of weather stations from climate.onebuilding.org.

Use load to load the bundled (or user-refreshed) station index. No network access or third-party dependencies are required for load.

Use check_for_updates to see if upstream data has changed, and refresh to re-download and rebuild the index.

Examples:

index = StationIndex.load()
results = index.search("chicago ohare", limit=3)
for r in results:
    print(r.station.display_name, r.score)

Source code in idfkit/weather/index.py

class StationIndex:
    """Searchable index of weather stations from climate.onebuilding.org.

    Use [load][idfkit.weather.index.StationIndex.load] to load the bundled (or user-refreshed) station index.
    No network access or third-party dependencies are required for [load][idfkit.weather.index.StationIndex.load].

    Use [check_for_updates][idfkit.weather.index.StationIndex.check_for_updates] to see if upstream data has changed, and
    [refresh][idfkit.weather.index.StationIndex.refresh] to re-download and rebuild the index.

    Examples:
        ```python
        index = StationIndex.load()
        results = index.search("chicago ohare", limit=3)
        for r in results:
            print(r.station.display_name, r.score)
        ```
    """

    __slots__ = ("_by_filename", "_by_wmo", "_last_modified", "_stations")

    _stations: list[WeatherStation]
    _by_filename: dict[str, list[WeatherStation]]
    _by_wmo: dict[str, list[WeatherStation]]
    _last_modified: dict[str, str]

    def __init__(self, stations: list[WeatherStation]) -> None:
        self._stations = stations
        self._by_wmo: dict[str, list[WeatherStation]] = {}
        self._by_filename: dict[str, list[WeatherStation]] = {}
        for s in stations:
            self._by_wmo.setdefault(s.wmo, []).append(s)
            self._by_filename.setdefault(s.filename_stem.lower(), []).append(s)
        self._last_modified: dict[str, str] = {}

    # --- Construction -------------------------------------------------------

    @classmethod
    def load(cls, *, cache_dir: Path | None = None) -> StationIndex:
        """Load the station index from a local compressed file.

        Checks for a user-refreshed cache first, then falls back to the
        bundled index shipped with the package.  No network access is
        required.

        If the cached file is from an older idfkit version and fails to
        deserialise (e.g. missing schema fields after a format migration),
        it is quarantined to ``stations.json.gz.stale`` and the bundled
        index is used instead.

        Args:
            cache_dir: Override the default cache directory.
        """
        cache = cache_dir or default_cache_dir()
        cached_path = cache / _CACHED_INDEX

        if cached_path.is_file():
            try:
                stations, last_modified, _ = _load_compressed_index(cached_path)
            except (KeyError, ValueError, TypeError, json.JSONDecodeError, OSError, gzip.BadGzipFile) as exc:
                quarantine = cached_path.with_suffix(cached_path.suffix + ".stale")
                logger.warning(
                    "Cached station index at %s could not be loaded (%s: %s). "
                    "Falling back to the bundled index. The stale cache has been "
                    "moved to %s — run StationIndex.refresh() to rebuild it.",
                    cached_path,
                    type(exc).__name__,
                    exc,
                    quarantine,
                )
                with contextlib.suppress(OSError):
                    cached_path.replace(quarantine)
                if not _BUNDLED_INDEX.is_file():
                    msg = (
                        "Cached station index is unreadable and no bundled "
                        "index ships with this build. Run StationIndex.refresh() "
                        "to download one."
                    )
                    raise FileNotFoundError(msg) from exc
                stations, last_modified, _ = _load_compressed_index(_BUNDLED_INDEX)
                logger.info("Loaded station index with %d stations from %s", len(stations), _BUNDLED_INDEX)
            else:
                logger.info("Loaded station index with %d stations from %s", len(stations), cached_path)
        elif _BUNDLED_INDEX.is_file():
            stations, last_modified, _ = _load_compressed_index(_BUNDLED_INDEX)
            logger.info("Loaded station index with %d stations from %s", len(stations), _BUNDLED_INDEX)
        else:
            msg = (
                "No station index found. The bundled index is missing and no "
                "cached index exists. Run StationIndex.refresh() to download one."
            )
            raise FileNotFoundError(msg)

        instance = cls(stations)
        instance._last_modified = last_modified
        _maybe_check_for_updates(instance, cache)
        return instance

    @classmethod
    def from_stations(cls, stations: list[WeatherStation]) -> StationIndex:
        """Create an index from an explicit list of stations (useful for tests)."""
        return cls(stations)

    @classmethod
    def refresh(cls, *, cache_dir: Path | None = None) -> StationIndex:
        """Re-download the regional KML indexes from climate.onebuilding.org and rebuild the cache.

        Uses the Python standard library only — no third-party dependencies.

        Args:
            cache_dir: Override the default cache directory.
        """
        cache = cache_dir or default_cache_dir()

        all_stations: list[WeatherStation] = []
        last_modified: dict[str, str] = {}
        for fname in _INDEX_FILES:
            local_path, lm = _ensure_index_file(fname, cache)
            if lm is not None:
                last_modified[fname] = lm
            all_stations.extend(_parse_kml(local_path))

        dest = cache / _CACHED_INDEX
        _save_compressed_index(all_stations, last_modified, dest)

        instance = cls(all_stations)
        instance._last_modified = last_modified
        return instance

    # --- Freshness ----------------------------------------------------------

    def check_for_updates(self) -> bool:
        """Check if upstream KML files have changed since this index was built.

        Sends lightweight HEAD requests to climate.onebuilding.org.
        Returns ``True`` if any file has a newer ``Last-Modified`` date.
        Returns ``False`` if all files match or if the check fails (offline,
        timeout, etc.).
        """
        if not self._last_modified:
            return False
        for fname in _INDEX_FILES:
            stored = self._last_modified.get(fname)
            if stored is None:
                continue
            url = f"{_SOURCES_BASE_URL}/{fname}"
            upstream = _head_last_modified(url)
            if upstream is not None and upstream != stored:
                return True
        return False

    # --- Properties ---------------------------------------------------------

    @property
    def stations(self) -> list[WeatherStation]:
        """All stations in the index."""
        return list(self._stations)

    def __len__(self) -> int:
        return len(self._stations)

    # --- Exact lookups ------------------------------------------------------

    def get_by_wmo(self, wmo: str) -> list[WeatherStation]:
        """Look up stations by WMO number.

        Args:
            wmo: WMO station number as a string (e.g. ``"722950"``).

        Returns a list because a single WMO number can correspond to
        multiple stations or dataset variants.
        """
        return list(self._by_wmo.get(wmo, []))

    def get_by_filename(self, filename: str) -> list[WeatherStation]:
        """Look up stations by EPW filename.

        The *filename* can include or omit the extension (``.zip``,
        ``.epw``, ``.ddy``).  Matching is case-insensitive.

        When the exact filename matches an indexed entry, those stations
        are returned.  Otherwise the method extracts the WMO number from
        the filename and falls back to
        [get_by_wmo][idfkit.weather.index.StationIndex.get_by_wmo].

        Args:
            filename: An EPW filename or stem, e.g.
                ``"USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023"``
                or ``"GBR_London.Heathrow.AP.037720_TMYx.epw"``.

        Returns:
            Matching stations (may be empty).
        """
        key = _strip_weather_extension(filename).lower()

        exact = self._by_filename.get(key, [])
        if exact:
            return list(exact)

        wmo = _extract_wmo_from_filename(filename)
        if wmo:
            return self.get_by_wmo(wmo)

        return []

    # --- Fuzzy text search --------------------------------------------------

    def search(
        self,
        query: str,
        *,
        limit: int = 10,
        country: str | None = None,
    ) -> list[SearchResult]:
        """Fuzzy-search stations by name, city, state, WMO number, or EPW filename.

        Matching is case-insensitive and uses substring / token-prefix
        heuristics (no external NLP dependencies).  Canonical EPW
        filenames (e.g.
        ``"USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023"``) are
        detected automatically and resolved via
        [get_by_filename][idfkit.weather.index.StationIndex.get_by_filename].

        Args:
            query: Free-text search query or EPW filename.
            limit: Maximum number of results to return.
            country: If given, restrict to stations in this country code.
        """
        raw = query.strip()
        q = raw.lower()
        if not q:
            return []

        # Fast path: canonical EPW filename
        if _is_epw_filename(raw):
            stations = self.get_by_filename(raw)
            if stations:
                results = [
                    SearchResult(station=s, score=1.0, match_field="filename")
                    for s in stations
                    if not country or s.country.upper() == country.upper()
                ]
                return results[:limit]

        tokens = q.split()

        scored: list[SearchResult] = []
        for station in self._stations:
            if country and station.country.upper() != country.upper():
                continue
            score, match_field = _score_station(station, q, tokens)
            if score > 0:
                scored.append(SearchResult(station=station, score=score, match_field=match_field))

        scored.sort(key=lambda r: r.score, reverse=True)
        return scored[:limit]

    # --- Spatial search -----------------------------------------------------

    def nearest(
        self,
        latitude: float,
        longitude: float,
        *,
        limit: int = 5,
        max_distance_km: float | None = None,
        country: str | None = None,
    ) -> list[SpatialResult]:
        """Find stations nearest to a geographic coordinate.

        Uses the Haversine formula for great-circle distance.  A bounding-box
        pre-filter is applied when *max_distance_km* is specified to avoid
        computing distances for stations that are obviously too far.

        Args:
            latitude: Decimal degrees, north positive.
            longitude: Decimal degrees, east positive.
            limit: Maximum results to return.
            max_distance_km: Exclude stations farther than this.
            country: If given, restrict to this country code.
        """
        # Bounding-box pre-filter (~111 km per degree of latitude)
        if max_distance_km is not None:
            delta_deg = max_distance_km / 111.0 + 1.0  # small margin
            lat_min = latitude - delta_deg
            lat_max = latitude + delta_deg
            # Longitude degrees vary with latitude
            cos_lat = math.cos(math.radians(latitude))
            lon_delta = delta_deg / max(cos_lat, 0.01)
            lon_min = longitude - lon_delta
            lon_max = longitude + lon_delta
        else:
            lat_min = lat_max = lon_min = lon_max = 0.0  # unused

        results: list[SpatialResult] = []
        for station in self._stations:
            if country and station.country.upper() != country.upper():
                continue
            if max_distance_km is not None:
                if station.latitude < lat_min or station.latitude > lat_max:
                    continue
                if station.longitude < lon_min or station.longitude > lon_max:
                    continue
            dist = haversine_km(latitude, longitude, station.latitude, station.longitude)
            if max_distance_km is not None and dist > max_distance_km:
                continue
            results.append(SpatialResult(station=station, distance_km=dist))

        results.sort(key=lambda r: r.distance_km)
        return results[:limit]

    # --- Filtering ----------------------------------------------------------

    def filter(
        self,
        *,
        country: str | None = None,
        state: str | None = None,
        wmo_region: int | None = None,
        climate_zone: str | None = None,
        climate_zone_determined: bool | None = None,
    ) -> list[WeatherStation]:
        """Filter stations by metadata criteria.

        All specified criteria must match (logical AND).

        Args:
            country: ISO 3166 country code. Case-insensitive.
            state: State or province abbreviation. Case-insensitive.
            wmo_region: WMO region number, inferred from the download URL.
            climate_zone: ASHRAE climate zone code, e.g. ``"5A"``. Case-insensitive,
                and matched against the code parsed out of
                :attr:`WeatherStation.ashrae_climate_zone` rather than against that
                label's text. The label is not a code: 2,162 records in the shipped
                index read ``7A - ASHRAE Climate Zone could not be determined`` or
                ``8A - ...``, and neither 7A nor 8A is an ASHRAE zone, since zones 7
                and 8 carry no suffix. Matching the first token would invent two zones
                holding 3.1% of the index, so those records match no zone. Ask for them
                with ``climate_zone_determined=False``. An empty string is no
                constraint, matching ``country`` and ``state``.
            climate_zone_determined: Whether the station's zone was determined
                upstream. ``False`` selects the records whose label reports that it
                could not be, which no ``climate_zone`` value returns. A separate
                parameter rather than a reserved ``climate_zone`` value, because that
                parameter's domain is already strings and a magic one could not be told
                from a real code.
        """
        # ``climate_zone=""`` means no constraint, as ``country=""`` and ``state=""``
        # already do below. A UI binding an empty select to this would otherwise get zero
        # stations from one key and every station from the next two.
        wanted_zone = climate_zone.upper() if climate_zone else None
        result: list[WeatherStation] = []
        for s in self._stations:
            if country and s.country.upper() != country.upper():
                continue
            if state and s.state.upper() != state.upper():
                continue
            if wmo_region is not None:
                # Infer WMO region from the URL path
                url_lower = s.url.lower()
                if f"wmo_region_{wmo_region}" not in url_lower:
                    continue
            if wanted_zone is not None or climate_zone_determined is not None:
                # Parsed once per station: both keys ask about the same label, and the
                # shipped index is 69,638 records.
                zone_code = _zone_code_of(s.ashrae_climate_zone)
                if wanted_zone is not None and zone_code != wanted_zone:
                    continue
                if climate_zone_determined is not None:
                    # Deliberately not ``zone_code is None``: that is also true of an
                    # unrecognised code, which is not the same thing as upstream
                    # reporting it could not determine one.
                    matched = (
                        zone_code is not None
                        if climate_zone_determined
                        else _zone_is_undetermined(s.ashrae_climate_zone)
                    )
                    if not matched:
                        continue
            result.append(s)
        return result

    @property
    def countries(self) -> list[str]:
        """Sorted list of unique country codes in the index."""
        return sorted({s.country for s in self._stations})

countries property

Sorted list of unique country codes in the index.

stations property

All stations in the index.

check_for_updates()

Check if upstream KML files have changed since this index was built.

Sends lightweight HEAD requests to climate.onebuilding.org. Returns True if any file has a newer Last-Modified date. Returns False if all files match or if the check fails (offline, timeout, etc.).

Source code in idfkit/weather/index.py

def check_for_updates(self) -> bool:
    """Check if upstream KML files have changed since this index was built.

    Sends lightweight HEAD requests to climate.onebuilding.org.
    Returns ``True`` if any file has a newer ``Last-Modified`` date.
    Returns ``False`` if all files match or if the check fails (offline,
    timeout, etc.).
    """
    if not self._last_modified:
        return False
    for fname in _INDEX_FILES:
        stored = self._last_modified.get(fname)
        if stored is None:
            continue
        url = f"{_SOURCES_BASE_URL}/{fname}"
        upstream = _head_last_modified(url)
        if upstream is not None and upstream != stored:
            return True
    return False

filter(*, country=None, state=None, wmo_region=None, climate_zone=None, climate_zone_determined=None)

Filter stations by metadata criteria.

All specified criteria must match (logical AND).

Parameters:

Name	Type	Description	Default
country	str | None	ISO 3166 country code. Case-insensitive.	None
state	str | None	State or province abbreviation. Case-insensitive.	None
wmo_region	int | None	WMO region number, inferred from the download URL.	None
climate_zone	str | None	ASHRAE climate zone code, e.g. "5A". Case-insensitive, and matched against the code parsed out of :attr:WeatherStation.ashrae_climate_zone rather than against that label's text. The label is not a code: 2,162 records in the shipped index read 7A - ASHRAE Climate Zone could not be determined or 8A - ..., and neither 7A nor 8A is an ASHRAE zone, since zones 7 and 8 carry no suffix. Matching the first token would invent two zones holding 3.1% of the index, so those records match no zone. Ask for them with climate_zone_determined=False. An empty string is no constraint, matching country and state.	None
climate_zone_determined	bool | None	Whether the station's zone was determined upstream. False selects the records whose label reports that it could not be, which no climate_zone value returns. A separate parameter rather than a reserved climate_zone value, because that parameter's domain is already strings and a magic one could not be told from a real code.	None

Source code in idfkit/weather/index.py

def filter(
    self,
    *,
    country: str | None = None,
    state: str | None = None,
    wmo_region: int | None = None,
    climate_zone: str | None = None,
    climate_zone_determined: bool | None = None,
) -> list[WeatherStation]:
    """Filter stations by metadata criteria.

    All specified criteria must match (logical AND).

    Args:
        country: ISO 3166 country code. Case-insensitive.
        state: State or province abbreviation. Case-insensitive.
        wmo_region: WMO region number, inferred from the download URL.
        climate_zone: ASHRAE climate zone code, e.g. ``"5A"``. Case-insensitive,
            and matched against the code parsed out of
            :attr:`WeatherStation.ashrae_climate_zone` rather than against that
            label's text. The label is not a code: 2,162 records in the shipped
            index read ``7A - ASHRAE Climate Zone could not be determined`` or
            ``8A - ...``, and neither 7A nor 8A is an ASHRAE zone, since zones 7
            and 8 carry no suffix. Matching the first token would invent two zones
            holding 3.1% of the index, so those records match no zone. Ask for them
            with ``climate_zone_determined=False``. An empty string is no
            constraint, matching ``country`` and ``state``.
        climate_zone_determined: Whether the station's zone was determined
            upstream. ``False`` selects the records whose label reports that it
            could not be, which no ``climate_zone`` value returns. A separate
            parameter rather than a reserved ``climate_zone`` value, because that
            parameter's domain is already strings and a magic one could not be told
            from a real code.
    """
    # ``climate_zone=""`` means no constraint, as ``country=""`` and ``state=""``
    # already do below. A UI binding an empty select to this would otherwise get zero
    # stations from one key and every station from the next two.
    wanted_zone = climate_zone.upper() if climate_zone else None
    result: list[WeatherStation] = []
    for s in self._stations:
        if country and s.country.upper() != country.upper():
            continue
        if state and s.state.upper() != state.upper():
            continue
        if wmo_region is not None:
            # Infer WMO region from the URL path
            url_lower = s.url.lower()
            if f"wmo_region_{wmo_region}" not in url_lower:
                continue
        if wanted_zone is not None or climate_zone_determined is not None:
            # Parsed once per station: both keys ask about the same label, and the
            # shipped index is 69,638 records.
            zone_code = _zone_code_of(s.ashrae_climate_zone)
            if wanted_zone is not None and zone_code != wanted_zone:
                continue
            if climate_zone_determined is not None:
                # Deliberately not ``zone_code is None``: that is also true of an
                # unrecognised code, which is not the same thing as upstream
                # reporting it could not determine one.
                matched = (
                    zone_code is not None
                    if climate_zone_determined
                    else _zone_is_undetermined(s.ashrae_climate_zone)
                )
                if not matched:
                    continue
        result.append(s)
    return result

from_stations(stations) classmethod

Create an index from an explicit list of stations (useful for tests).

Source code in idfkit/weather/index.py

@classmethod
def from_stations(cls, stations: list[WeatherStation]) -> StationIndex:
    """Create an index from an explicit list of stations (useful for tests)."""
    return cls(stations)

get_by_filename(filename)

Look up stations by EPW filename.

The filename can include or omit the extension (.zip, .epw, .ddy). Matching is case-insensitive.

When the exact filename matches an indexed entry, those stations are returned. Otherwise the method extracts the WMO number from the filename and falls back to get_by_wmo.

Parameters:

Name	Type	Description	Default
filename	str	An EPW filename or stem, e.g. "USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023" or "GBR_London.Heathrow.AP.037720_TMYx.epw".	required

Returns:

Type	Description
list[WeatherStation]	Matching stations (may be empty).

Source code in idfkit/weather/index.py

def get_by_filename(self, filename: str) -> list[WeatherStation]:
    """Look up stations by EPW filename.

    The *filename* can include or omit the extension (``.zip``,
    ``.epw``, ``.ddy``).  Matching is case-insensitive.

    When the exact filename matches an indexed entry, those stations
    are returned.  Otherwise the method extracts the WMO number from
    the filename and falls back to
    [get_by_wmo][idfkit.weather.index.StationIndex.get_by_wmo].

    Args:
        filename: An EPW filename or stem, e.g.
            ``"USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023"``
            or ``"GBR_London.Heathrow.AP.037720_TMYx.epw"``.

    Returns:
        Matching stations (may be empty).
    """
    key = _strip_weather_extension(filename).lower()

    exact = self._by_filename.get(key, [])
    if exact:
        return list(exact)

    wmo = _extract_wmo_from_filename(filename)
    if wmo:
        return self.get_by_wmo(wmo)

    return []

get_by_wmo(wmo)

Look up stations by WMO number.

Parameters:

Name	Type	Description	Default
wmo	str	WMO station number as a string (e.g. "722950").	required

Returns a list because a single WMO number can correspond to multiple stations or dataset variants.

Source code in idfkit/weather/index.py

def get_by_wmo(self, wmo: str) -> list[WeatherStation]:
    """Look up stations by WMO number.

    Args:
        wmo: WMO station number as a string (e.g. ``"722950"``).

    Returns a list because a single WMO number can correspond to
    multiple stations or dataset variants.
    """
    return list(self._by_wmo.get(wmo, []))

load(*, cache_dir=None) classmethod

Load the station index from a local compressed file.

Checks for a user-refreshed cache first, then falls back to the bundled index shipped with the package. No network access is required.

If the cached file is from an older idfkit version and fails to deserialise (e.g. missing schema fields after a format migration), it is quarantined to stations.json.gz.stale and the bundled index is used instead.

Parameters:

Name	Type	Description	Default
cache_dir	Path | None	Override the default cache directory.	None

Source code in idfkit/weather/index.py

@classmethod
def load(cls, *, cache_dir: Path | None = None) -> StationIndex:
    """Load the station index from a local compressed file.

    Checks for a user-refreshed cache first, then falls back to the
    bundled index shipped with the package.  No network access is
    required.

    If the cached file is from an older idfkit version and fails to
    deserialise (e.g. missing schema fields after a format migration),
    it is quarantined to ``stations.json.gz.stale`` and the bundled
    index is used instead.

    Args:
        cache_dir: Override the default cache directory.
    """
    cache = cache_dir or default_cache_dir()
    cached_path = cache / _CACHED_INDEX

    if cached_path.is_file():
        try:
            stations, last_modified, _ = _load_compressed_index(cached_path)
        except (KeyError, ValueError, TypeError, json.JSONDecodeError, OSError, gzip.BadGzipFile) as exc:
            quarantine = cached_path.with_suffix(cached_path.suffix + ".stale")
            logger.warning(
                "Cached station index at %s could not be loaded (%s: %s). "
                "Falling back to the bundled index. The stale cache has been "
                "moved to %s — run StationIndex.refresh() to rebuild it.",
                cached_path,
                type(exc).__name__,
                exc,
                quarantine,
            )
            with contextlib.suppress(OSError):
                cached_path.replace(quarantine)
            if not _BUNDLED_INDEX.is_file():
                msg = (
                    "Cached station index is unreadable and no bundled "
                    "index ships with this build. Run StationIndex.refresh() "
                    "to download one."
                )
                raise FileNotFoundError(msg) from exc
            stations, last_modified, _ = _load_compressed_index(_BUNDLED_INDEX)
            logger.info("Loaded station index with %d stations from %s", len(stations), _BUNDLED_INDEX)
        else:
            logger.info("Loaded station index with %d stations from %s", len(stations), cached_path)
    elif _BUNDLED_INDEX.is_file():
        stations, last_modified, _ = _load_compressed_index(_BUNDLED_INDEX)
        logger.info("Loaded station index with %d stations from %s", len(stations), _BUNDLED_INDEX)
    else:
        msg = (
            "No station index found. The bundled index is missing and no "
            "cached index exists. Run StationIndex.refresh() to download one."
        )
        raise FileNotFoundError(msg)

    instance = cls(stations)
    instance._last_modified = last_modified
    _maybe_check_for_updates(instance, cache)
    return instance

nearest(latitude, longitude, *, limit=5, max_distance_km=None, country=None)

Find stations nearest to a geographic coordinate.

Uses the Haversine formula for great-circle distance. A bounding-box pre-filter is applied when max_distance_km is specified to avoid computing distances for stations that are obviously too far.

Parameters:

Name	Type	Description	Default
latitude	float	Decimal degrees, north positive.	required
longitude	float	Decimal degrees, east positive.	required
limit	int	Maximum results to return.	5
max_distance_km	float | None	Exclude stations farther than this.	None
country	str | None	If given, restrict to this country code.	None

Source code in idfkit/weather/index.py

def nearest(
    self,
    latitude: float,
    longitude: float,
    *,
    limit: int = 5,
    max_distance_km: float | None = None,
    country: str | None = None,
) -> list[SpatialResult]:
    """Find stations nearest to a geographic coordinate.

    Uses the Haversine formula for great-circle distance.  A bounding-box
    pre-filter is applied when *max_distance_km* is specified to avoid
    computing distances for stations that are obviously too far.

    Args:
        latitude: Decimal degrees, north positive.
        longitude: Decimal degrees, east positive.
        limit: Maximum results to return.
        max_distance_km: Exclude stations farther than this.
        country: If given, restrict to this country code.
    """
    # Bounding-box pre-filter (~111 km per degree of latitude)
    if max_distance_km is not None:
        delta_deg = max_distance_km / 111.0 + 1.0  # small margin
        lat_min = latitude - delta_deg
        lat_max = latitude + delta_deg
        # Longitude degrees vary with latitude
        cos_lat = math.cos(math.radians(latitude))
        lon_delta = delta_deg / max(cos_lat, 0.01)
        lon_min = longitude - lon_delta
        lon_max = longitude + lon_delta
    else:
        lat_min = lat_max = lon_min = lon_max = 0.0  # unused

    results: list[SpatialResult] = []
    for station in self._stations:
        if country and station.country.upper() != country.upper():
            continue
        if max_distance_km is not None:
            if station.latitude < lat_min or station.latitude > lat_max:
                continue
            if station.longitude < lon_min or station.longitude > lon_max:
                continue
        dist = haversine_km(latitude, longitude, station.latitude, station.longitude)
        if max_distance_km is not None and dist > max_distance_km:
            continue
        results.append(SpatialResult(station=station, distance_km=dist))

    results.sort(key=lambda r: r.distance_km)
    return results[:limit]

refresh(*, cache_dir=None) classmethod

Re-download the regional KML indexes from climate.onebuilding.org and rebuild the cache.

Uses the Python standard library only — no third-party dependencies.

Parameters:

Name	Type	Description	Default
cache_dir	Path | None	Override the default cache directory.	None

Source code in idfkit/weather/index.py

@classmethod
def refresh(cls, *, cache_dir: Path | None = None) -> StationIndex:
    """Re-download the regional KML indexes from climate.onebuilding.org and rebuild the cache.

    Uses the Python standard library only — no third-party dependencies.

    Args:
        cache_dir: Override the default cache directory.
    """
    cache = cache_dir or default_cache_dir()

    all_stations: list[WeatherStation] = []
    last_modified: dict[str, str] = {}
    for fname in _INDEX_FILES:
        local_path, lm = _ensure_index_file(fname, cache)
        if lm is not None:
            last_modified[fname] = lm
        all_stations.extend(_parse_kml(local_path))

    dest = cache / _CACHED_INDEX
    _save_compressed_index(all_stations, last_modified, dest)

    instance = cls(all_stations)
    instance._last_modified = last_modified
    return instance

search(query, *, limit=10, country=None)

Fuzzy-search stations by name, city, state, WMO number, or EPW filename.

Matching is case-insensitive and uses substring / token-prefix heuristics (no external NLP dependencies). Canonical EPW filenames (e.g. "USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023") are detected automatically and resolved via get_by_filename.

Parameters:

Name	Type	Description	Default
query	str	Free-text search query or EPW filename.	required
limit	int	Maximum number of results to return.	10
country	str | None	If given, restrict to stations in this country code.	None

Source code in idfkit/weather/index.py

def search(
    self,
    query: str,
    *,
    limit: int = 10,
    country: str | None = None,
) -> list[SearchResult]:
    """Fuzzy-search stations by name, city, state, WMO number, or EPW filename.

    Matching is case-insensitive and uses substring / token-prefix
    heuristics (no external NLP dependencies).  Canonical EPW
    filenames (e.g.
    ``"USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023"``) are
    detected automatically and resolved via
    [get_by_filename][idfkit.weather.index.StationIndex.get_by_filename].

    Args:
        query: Free-text search query or EPW filename.
        limit: Maximum number of results to return.
        country: If given, restrict to stations in this country code.
    """
    raw = query.strip()
    q = raw.lower()
    if not q:
        return []

    # Fast path: canonical EPW filename
    if _is_epw_filename(raw):
        stations = self.get_by_filename(raw)
        if stations:
            results = [
                SearchResult(station=s, score=1.0, match_field="filename")
                for s in stations
                if not country or s.country.upper() == country.upper()
            ]
            return results[:limit]

    tokens = q.split()

    scored: list[SearchResult] = []
    for station in self._stations:
        if country and station.country.upper() != country.upper():
            continue
        score, match_field = _score_station(station, q, tokens)
        if score > 0:
            scored.append(SearchResult(station=station, score=score, match_field=match_field))

    scored.sort(key=lambda r: r.score, reverse=True)
    return scored[:limit]

WeatherStation

idfkit.weather.station.WeatherStation dataclass

Metadata for a single weather file entry from climate.onebuilding.org.

Each instance represents one downloadable weather dataset. The same physical station may appear multiple times with different source or year-range variants (e.g. TMYx.2007-2021 vs TMYx.2009-2023).

Attributes:

Name	Type	Description
country	str	ISO 3166 country code (e.g. "USA").
state	str	State or province abbreviation (e.g. "CA").
city	str	City or station name as it appears in the index (e.g. "Marina.Muni.AP").
wmo	str	WMO station number as a string to preserve leading zeros (e.g. "722950" or "012345").
source	str	Dataset source identifier (e.g. "TMYx.2009-2023").
latitude	float	Decimal degrees, north positive.
longitude	float	Decimal degrees, east positive.
timezone	float	Hours offset from GMT (e.g. -8.0).
elevation	float	Meters above sea level.
url	str	Full download URL for the ZIP archive.
ashrae_climate_zone	str	ASHRAE HOF climate zone label (e.g. "4A - Mixed - Humid").
heating_design_db_c	float	99% heating design dry-bulb temperature in °C.
cooling_design_db_c	float	1% cooling design dry-bulb temperature in °C.
hdd18	int	Heating degree-days base 18 °C.
cdd10	int	Cooling degree-days base 10 °C.
design_conditions_source_wmo	str | None	When the station inherits design conditions from a neighbouring station, this holds that station's WMO number; otherwise None.

Source code in idfkit/weather/station.py

@dataclass(frozen=True)
class WeatherStation:
    """Metadata for a single weather file entry from climate.onebuilding.org.

    Each instance represents one downloadable weather dataset. The same physical
    station may appear multiple times with different ``source`` or year-range
    variants (e.g. ``TMYx.2007-2021`` vs ``TMYx.2009-2023``).

    Attributes:
        country: ISO 3166 country code (e.g. ``"USA"``).
        state: State or province abbreviation (e.g. ``"CA"``).
        city: City or station name as it appears in the index
            (e.g. ``"Marina.Muni.AP"``).
        wmo: WMO station number as a string to preserve leading zeros
            (e.g. ``"722950"`` or ``"012345"``).
        source: Dataset source identifier (e.g. ``"TMYx.2009-2023"``).
        latitude: Decimal degrees, north positive.
        longitude: Decimal degrees, east positive.
        timezone: Hours offset from GMT (e.g. ``-8.0``).
        elevation: Meters above sea level.
        url: Full download URL for the ZIP archive.
        ashrae_climate_zone: ASHRAE HOF climate zone label
            (e.g. ``"4A - Mixed - Humid"``).
        heating_design_db_c: 99% heating design dry-bulb temperature in °C.
        cooling_design_db_c: 1% cooling design dry-bulb temperature in °C.
        hdd18: Heating degree-days base 18 °C.
        cdd10: Cooling degree-days base 10 °C.
        design_conditions_source_wmo: When the station inherits design
            conditions from a neighbouring station, this holds that
            station's WMO number; otherwise ``None``.
    """

    country: str
    state: str
    city: str
    wmo: str
    source: str
    latitude: float
    longitude: float
    timezone: float
    elevation: float
    url: str
    ashrae_climate_zone: str
    heating_design_db_c: float
    cooling_design_db_c: float
    hdd18: int
    cdd10: int
    design_conditions_source_wmo: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a plain dictionary for JSON storage."""
        return {
            "country": self.country,
            "state": self.state,
            "city": self.city,
            "wmo": self.wmo,
            "source": self.source,
            "latitude": self.latitude,
            "longitude": self.longitude,
            "timezone": self.timezone,
            "elevation": self.elevation,
            "url": self.url,
            "ashrae_climate_zone": self.ashrae_climate_zone,
            "heating_design_db_c": self.heating_design_db_c,
            "cooling_design_db_c": self.cooling_design_db_c,
            "hdd18": self.hdd18,
            "cdd10": self.cdd10,
            "design_conditions_source_wmo": self.design_conditions_source_wmo,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> WeatherStation:
        """Deserialize from a plain dictionary.

        The required climate fields use ``data[key]`` so a stale or
        corrupt payload fails loudly. ``design_conditions_source_wmo``
        is genuinely optional.
        """
        source_wmo = data.get("design_conditions_source_wmo")
        return cls(
            country=str(data["country"]),
            state=str(data["state"]),
            city=str(data["city"]),
            wmo=str(data["wmo"]),
            source=str(data["source"]),
            latitude=float(data["latitude"]),
            longitude=float(data["longitude"]),
            timezone=float(data["timezone"]),
            elevation=float(data["elevation"]),
            url=str(data["url"]),
            ashrae_climate_zone=str(data["ashrae_climate_zone"]),
            heating_design_db_c=float(data["heating_design_db_c"]),
            cooling_design_db_c=float(data["cooling_design_db_c"]),
            hdd18=int(data["hdd18"]),
            cdd10=int(data["cdd10"]),
            design_conditions_source_wmo=str(source_wmo) if source_wmo is not None else None,
        )

    @property
    def display_name(self) -> str:
        """Human-readable station name with location context.

        Dots in the city name are replaced with spaces for readability.
        """
        name = self.city.replace(".", " ").replace("-", " ").strip()
        parts: list[str] = []
        if name:
            parts.append(name)
        if self.state:
            parts.append(self.state)
        parts.append(self.country)
        return ", ".join(parts)

    @property
    def filename_stem(self) -> str:
        """The canonical EPW filename stem derived from the download URL.

        Returns the ZIP filename without the ``.zip`` extension, e.g.
        ``"USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023"``.
        """
        filename = self.url.rsplit("/", maxsplit=1)[-1]
        return filename.removesuffix(".zip")

    @property
    def dataset_variant(self) -> str:
        """Extract the TMYx dataset variant from the download URL.

        Returns a string like ``"TMYx"``, ``"TMYx.2007-2021"``, or
        ``"TMYx.2009-2023"``.
        """
        # Dataset variant is everything after the last underscore
        # e.g. "USA_CA_Marina.Muni.AP.690070_TMYx" -> "TMYx"
        # e.g. "USA_CA_Twentynine.Palms.SELF.690150_TMYx.2004-2018" -> "TMYx.2004-2018"
        parts = self.filename_stem.rsplit("_", maxsplit=1)
        if len(parts) == 2:
            return parts[1]
        return self.filename_stem

    @property
    def heating_design_db_f(self) -> float:
        """99% heating design dry-bulb temperature in °F."""
        return self.heating_design_db_c * 9 / 5 + 32

    @property
    def cooling_design_db_f(self) -> float:
        """1% cooling design dry-bulb temperature in °F."""
        return self.cooling_design_db_c * 9 / 5 + 32

ashrae_climate_zone instance-attribute

cdd10 instance-attribute

city instance-attribute

cooling_design_db_c instance-attribute

cooling_design_db_f property

1% cooling design dry-bulb temperature in °F.

country instance-attribute

dataset_variant property

Extract the TMYx dataset variant from the download URL.

Returns a string like "TMYx", "TMYx.2007-2021", or "TMYx.2009-2023".

design_conditions_source_wmo = None class-attribute instance-attribute

display_name property

Human-readable station name with location context.

Dots in the city name are replaced with spaces for readability.

elevation instance-attribute

filename_stem property

The canonical EPW filename stem derived from the download URL.

Returns the ZIP filename without the .zip extension, e.g. "USA_IL_Chicago.Ohare.Intl.AP.725300_TMYx.2009-2023".

hdd18 instance-attribute

heating_design_db_c instance-attribute

heating_design_db_f property

99% heating design dry-bulb temperature in °F.

latitude instance-attribute

longitude instance-attribute

source instance-attribute

state instance-attribute

timezone instance-attribute

url instance-attribute

wmo instance-attribute

from_dict(data) classmethod

Deserialize from a plain dictionary.

The required climate fields use data[key] so a stale or corrupt payload fails loudly. design_conditions_source_wmo is genuinely optional.

Source code in idfkit/weather/station.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> WeatherStation:
    """Deserialize from a plain dictionary.

    The required climate fields use ``data[key]`` so a stale or
    corrupt payload fails loudly. ``design_conditions_source_wmo``
    is genuinely optional.
    """
    source_wmo = data.get("design_conditions_source_wmo")
    return cls(
        country=str(data["country"]),
        state=str(data["state"]),
        city=str(data["city"]),
        wmo=str(data["wmo"]),
        source=str(data["source"]),
        latitude=float(data["latitude"]),
        longitude=float(data["longitude"]),
        timezone=float(data["timezone"]),
        elevation=float(data["elevation"]),
        url=str(data["url"]),
        ashrae_climate_zone=str(data["ashrae_climate_zone"]),
        heating_design_db_c=float(data["heating_design_db_c"]),
        cooling_design_db_c=float(data["cooling_design_db_c"]),
        hdd18=int(data["hdd18"]),
        cdd10=int(data["cdd10"]),
        design_conditions_source_wmo=str(source_wmo) if source_wmo is not None else None,
    )

to_dict()

Serialize to a plain dictionary for JSON storage.

Source code in idfkit/weather/station.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a plain dictionary for JSON storage."""
    return {
        "country": self.country,
        "state": self.state,
        "city": self.city,
        "wmo": self.wmo,
        "source": self.source,
        "latitude": self.latitude,
        "longitude": self.longitude,
        "timezone": self.timezone,
        "elevation": self.elevation,
        "url": self.url,
        "ashrae_climate_zone": self.ashrae_climate_zone,
        "heating_design_db_c": self.heating_design_db_c,
        "cooling_design_db_c": self.cooling_design_db_c,
        "hdd18": self.hdd18,
        "cdd10": self.cdd10,
        "design_conditions_source_wmo": self.design_conditions_source_wmo,
    }

SearchResult

idfkit.weather.station.SearchResult dataclass

A text search result with relevance score.

Source code in idfkit/weather/station.py

@dataclass(frozen=True)
class SearchResult:
    """A text search result with relevance score."""

    station: WeatherStation
    score: float
    """Relevance score from 0.0 to 1.0, higher is better."""
    match_field: str
    """Which field matched: ``"wmo"``, ``"name"``, ``"state"``, ``"country"``, or ``"filename"``."""

match_field instance-attribute

Which field matched: "wmo", "name", "state", "country", or "filename".

score instance-attribute

Relevance score from 0.0 to 1.0, higher is better.

station instance-attribute

SpatialResult

idfkit.weather.station.SpatialResult dataclass

A spatial proximity result with great-circle distance.

Source code in idfkit/weather/station.py

@dataclass(frozen=True)
class SpatialResult:
    """A spatial proximity result with great-circle distance."""

    station: WeatherStation
    distance_km: float
    """Great-circle distance in kilometres."""

distance_km instance-attribute

Great-circle distance in kilometres.

station instance-attribute

geocode

idfkit.weather.geocode.geocode(address)

Convert a street address to (latitude, longitude) via Nominatim.

Uses the free OpenStreetMap Nominatim geocoding service. No API key is required. Requests are rate-limited to one per second in compliance with Nominatim usage policy.

This function is thread-safe. Concurrent calls from multiple threads will be serialized to respect the rate limit.

Composable with spatial search: Use the splat operator to combine with nearest for address-based weather station lookup:

```python
from idfkit.weather import StationIndex, geocode

# Find weather stations near an address (one line!)
results = StationIndex.load().nearest(*geocode("350 Fifth Avenue, New York, NY"))

for r in results[:3]:
    print(f"{r.station.display_name}: {r.distance_km:.0f} km")
```

Parameters:

Name	Type	Description	Default
address	str	A free-form address string (e.g. "Willis Tower, Chicago").	required

Returns:

Type	Description
tuple[float, float]	A (latitude, longitude) tuple in decimal degrees.

Raises:

Type	Description
GeocodingError	If the address cannot be resolved or the service is unreachable.

Example

lat, lon = geocode("Empire State Building, NYC") print(f"{lat:.4f}, {lon:.4f}") 40.7484, -73.9857

Source code in idfkit/weather/geocode.py

def geocode(address: str) -> tuple[float, float]:
    """Convert a street address to ``(latitude, longitude)`` via Nominatim.

    Uses the free OpenStreetMap Nominatim geocoding service.  No API key is
    required.  Requests are rate-limited to one per second in compliance with
    Nominatim usage policy.

    This function is thread-safe. Concurrent calls from multiple threads will
    be serialized to respect the rate limit.

    **Composable with spatial search:** Use the splat operator to combine with
    [nearest][idfkit.weather.index.StationIndex.nearest] for address-based
    weather station lookup:

        ```python
        from idfkit.weather import StationIndex, geocode

        # Find weather stations near an address (one line!)
        results = StationIndex.load().nearest(*geocode("350 Fifth Avenue, New York, NY"))

        for r in results[:3]:
            print(f"{r.station.display_name}: {r.distance_km:.0f} km")
        ```

    Args:
        address: A free-form address string (e.g. ``"Willis Tower, Chicago"``).

    Returns:
        A ``(latitude, longitude)`` tuple in decimal degrees.

    Raises:
        GeocodingError: If the address cannot be resolved or the service is
            unreachable.

    Example:
        >>> lat, lon = geocode("Empire State Building, NYC")
        >>> print(f"{lat:.4f}, {lon:.4f}")
        40.7484, -73.9857
    """
    # Wait for rate limit
    _nominatim_limiter.wait()

    params = urllib.parse.urlencode({"q": address, "format": "json", "limit": "1"})
    url = f"{_NOMINATIM_URL}?{params}"

    req = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT})  # noqa: S310
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
            data = json.loads(resp.read())
            if data:
                return float(data[0]["lat"]), float(data[0]["lon"])
    except (URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
        msg = f"Failed to geocode address: {address}"
        raise GeocodingError(msg) from exc
    msg = f"No results found for address: {address}"
    raise GeocodingError(msg)

GeocodingError

idfkit.weather.geocode.GeocodingError

Bases: Exception

Raised when an address cannot be geocoded.

Source code in idfkit/weather/geocode.py

class GeocodingError(Exception):
    """Raised when an address cannot be geocoded."""