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 src/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,
    ) -> list[WeatherStation]:
        """Filter stations by metadata criteria.

        All specified criteria must match (logical AND).
        """
        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
            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.

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 src/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

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 src/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

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 src/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

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 src/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]

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 src/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]

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

Filter stations by metadata criteria.

All specified criteria must match (logical AND).

Source code in src/idfkit/weather/index.py

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

    All specified criteria must match (logical AND).
    """
    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
        result.append(s)
    return result

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 src/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, []))

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 src/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

city instance-attribute

state instance-attribute

country instance-attribute

wmo instance-attribute

latitude instance-attribute

longitude instance-attribute

elevation instance-attribute

url instance-attribute

display_name property

Human-readable station name with location context.

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

SearchResult

idfkit.weather.station.SearchResult dataclass

A text search result with relevance score.

Source code in src/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"``."""

station instance-attribute

score instance-attribute

Relevance score from 0.0 to 1.0, higher is better.

SpatialResult

idfkit.weather.station.SpatialResult dataclass

A spatial proximity result with great-circle distance.

Source code in src/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."""

station instance-attribute

distance_km instance-attribute

Great-circle distance in kilometres.

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 src/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 src/idfkit/weather/geocode.py

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