"""High-level Weather API for get-weather-data."""
import logging
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
from typing import TYPE_CHECKING
from get_weather_data.core.config import Config, set_config
from get_weather_data.core.database import INDEX_VERSION, Database
from get_weather_data.core.logging import setup_logging
from get_weather_data.stations import (
build_closest_index,
import_ghcnd_stations,
import_isd_stations,
import_zipcodes,
)
from get_weather_data.weather.batch import process_csv as _process_csv
from get_weather_data.weather.gridded import GriddedLookup
from get_weather_data.weather.hourly import HourlyLookup
from get_weather_data.weather.location import LocationInput
from get_weather_data.weather.lookup import WeatherLookup
from get_weather_data.weather.online import OnlineLookup
from get_weather_data.weather.results import (
Coverage,
HourlyResult,
WeatherResult,
summarize_coverage,
)
from get_weather_data.weather.units import ELEMENTS, Source, Units
if TYPE_CHECKING:
import pandas as pd
logger = logging.getLogger("get_weather_data")
def _has_data(result: WeatherResult) -> bool:
"""Whether a result carries any weather value."""
return any(getattr(result, spec.field) is not None for spec in ELEMENTS.values())
[docs]
@dataclass
class Weather:
"""High-level API for fetching weather data.
This class provides a simple interface for:
- Setting up the station database
- Looking up weather data for ZIP codes
- Processing CSV files
Example:
weather = Weather()
weather.setup()
result = weather.get("10001", "2024-01-15")
With online=True, get() and get_range() query the NOAA CDO API
directly — no setup() download (just a small cached ZIP-coordinates
file), but NCDC_TOKEN must be set.
"""
database_path: Path | str | None = None
verbose: bool = False
online: bool = False
units: Units = "metric"
include_flags: bool = False
include_weather_types: bool = False
explain: bool = False
interpolate: bool = False
source: Source = "station"
_db: Database | None = field(default=None, repr=False)
_lookup: WeatherLookup | None = field(default=None, repr=False)
_online_lookup: OnlineLookup | None = field(default=None, repr=False)
_grid: GriddedLookup | None = field(default=None, repr=False)
_hourly: HourlyLookup | None = field(default=None, repr=False)
def __post_init__(self) -> None:
"""Configure logging and build the selected lookup backend.
Raises:
ValueError: If online=True and no NCDC token is configured.
""" # noqa: DOC502 - raised by OnlineLookup/NOAAClient construction
setup_logging(verbose=self.verbose)
if self.database_path:
config = Config(_database_path=Path(self.database_path))
set_config(config)
if self.online:
# Fail fast on a missing token, and skip the local database
self._online_lookup = OnlineLookup(
units=self.units,
include_weather_types=self.include_weather_types,
explain=self.explain,
)
else:
self._db = Database(self.database_path)
@property
def db(self) -> Database:
"""Get the database instance."""
if self._db is None:
self._db = Database(self.database_path)
return self._db
@property
def lookup(self) -> WeatherLookup:
"""Get the weather lookup instance."""
if self._lookup is None:
self._lookup = WeatherLookup(
db=self.db,
units=self.units,
include_flags=self.include_flags,
include_weather_types=self.include_weather_types,
explain=self.explain,
interpolate=self.interpolate,
)
return self._lookup
@property
def grid(self) -> GriddedLookup:
"""Get the nClimGrid gridded-lookup instance."""
if self._grid is None:
self._grid = GriddedLookup(units=self.units)
return self._grid
@property
def hourly(self) -> HourlyLookup:
"""Get the ISD-Lite hourly-lookup instance."""
if self._hourly is None:
self._hourly = HourlyLookup(db=self.db, units=self.units)
return self._hourly
[docs]
def setup(
self,
force: bool = False,
ghcn_stations: bool = True,
usaf_stations: bool = True,
zipcodes: bool = True,
closest_index: bool = True,
) -> None:
"""Set up the database with station and ZIP code data.
This downloads station lists and ZIP code data, then builds
an index of closest stations for each ZIP code.
Args:
force: If True, rebuild even if database exists.
ghcn_stations: Import GHCN stations.
usaf_stations: Import USAF/WBAN (ISD) stations.
zipcodes: Import ZIP code data.
closest_index: Build closest stations index.
"""
self.db.init_schema()
if (
self.db.exists()
and not force
and self.db.count_stations() > 0
and self.db.count_zipcodes() > 0
):
logger.info("Database already set up. Use force=True to rebuild.")
return
if ghcn_stations:
logger.info("Importing GHCN stations...")
count = import_ghcnd_stations(self.db, force=force)
logger.info(f"Imported {count} GHCN stations")
if usaf_stations:
logger.info("Importing USAF/WBAN stations...")
count = import_isd_stations(self.db, force=force)
logger.info(f"Imported {count} USAF/WBAN stations")
if zipcodes:
logger.info("Importing ZIP codes...")
count = import_zipcodes(self.db, force=force)
logger.info(f"Imported {count} ZIP codes")
if closest_index:
logger.info("Building closest stations index...")
count = build_closest_index(self.db)
logger.info(f"Indexed {count} ZIP codes")
self.db.set_meta("index_version", str(INDEX_VERSION))
[docs]
def get(
self,
location: LocationInput,
target_date: str | date,
elements: list[str] | None = None,
) -> WeatherResult:
"""Get weather data for a location and date.
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
target_date: Date as string (YYYY-MM-DD) or date object.
elements: List of weather elements to retrieve.
Returns:
WeatherResult with available weather data.
"""
if isinstance(target_date, str):
target_date = date.fromisoformat(target_date)
if self._online_lookup is not None:
return self._online_lookup.get_weather(location, target_date, elements)
if self.source == "grid":
return self.grid.get_weather(location, target_date, elements)
result = self.lookup.get_weather(location, target_date, elements)
if self.source == "auto" and not _has_data(result):
grid_result = self.grid.get_weather(location, target_date, elements)
if _has_data(grid_result):
return grid_result
return result
[docs]
def get_range(
self,
location: LocationInput,
start_date: str | date,
end_date: str | date,
elements: list[str] | None = None,
) -> list[WeatherResult]:
"""Get weather data for a location over a date range.
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
start_date: Start date as string (YYYY-MM-DD) or date object.
end_date: End date as string (YYYY-MM-DD) or date object.
elements: List of weather elements to retrieve.
Returns:
List of WeatherResult objects, one per day.
"""
if isinstance(start_date, str):
start_date = date.fromisoformat(start_date)
if isinstance(end_date, str):
end_date = date.fromisoformat(end_date)
if self._online_lookup is not None:
return self._online_lookup.get_weather_range(
location, start_date, end_date, elements
)
if self.source == "grid":
return self.grid.get_weather_range(location, start_date, end_date, elements)
results = self.lookup.get_weather_range(
location, start_date, end_date, elements
)
if self.source == "auto" and any(not _has_data(r) for r in results):
# Fill the days the station network couldn't cover from the grid
grid_results = self.grid.get_weather_range(
location, start_date, end_date, elements
)
results = [
g if (not _has_data(s) and _has_data(g)) else s
for s, g in zip(results, grid_results, strict=True)
]
return results
[docs]
def get_frame(
self,
location: LocationInput,
start_date: str | date,
end_date: str | date | None = None,
elements: list[str] | None = None,
) -> "pd.DataFrame":
"""Get weather for a location as a pandas DataFrame.
One row per day, metadata columns followed by the weather value
columns (in the configured units). Requires the ``pandas`` extra
(``pip install get-weather-data[pandas]``).
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
start_date: Start date (YYYY-MM-DD) or date object.
end_date: End date; defaults to start_date (single day).
elements: List of weather elements to retrieve.
Returns:
A tidy DataFrame of the results.
Raises:
ImportError: If pandas is not installed.
""" # noqa: DOC502 - raised by results_to_frame
from get_weather_data.weather.frame import results_to_frame
if end_date is None:
end_date = start_date
results = self.get_range(location, start_date, end_date, elements)
return results_to_frame(results)
[docs]
def get_hourly(
self,
location: LocationInput,
start_date: str | date,
end_date: str | date | None = None,
) -> list[HourlyResult]:
"""Get hourly observations for a location (ISD-Lite).
Resolves the nearest USAF-WBAN station and reads hourly ISD-Lite
data. Requires the local database (``setup()``); there is no
online equivalent for hourly data.
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
start_date: First UTC date (YYYY-MM-DD) or date object.
end_date: Last UTC date; defaults to start_date (single day).
Returns:
One HourlyResult per available hour, in time order (UTC).
Raises:
ValueError: If this Weather was created with online=True.
"""
if self.online:
raise ValueError(
"get_hourly requires the local database (run setup()); "
"hourly ISD-Lite is not served by the online CDO API."
)
if isinstance(start_date, str):
start_date = date.fromisoformat(start_date)
if isinstance(end_date, str):
end_date = date.fromisoformat(end_date)
return self.hourly.get_hourly(location, start_date, end_date)
[docs]
def get_hourly_frame(
self,
location: LocationInput,
start_date: str | date,
end_date: str | date | None = None,
) -> "pd.DataFrame":
"""Get hourly observations as a pandas DataFrame.
One row per hour, metadata columns followed by the value columns
(in the configured units). Requires the ``pandas`` extra.
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
start_date: First UTC date (YYYY-MM-DD) or date object.
end_date: Last UTC date; defaults to start_date (single day).
Returns:
A tidy DataFrame of the hourly results.
Raises:
ImportError: If pandas is not installed.
""" # noqa: DOC502 - raised by hourly_results_to_frame
from get_weather_data.weather.frame import hourly_results_to_frame
results = self.get_hourly(location, start_date, end_date)
return hourly_results_to_frame(results)
[docs]
def coverage(
self,
location: LocationInput,
start_date: str | date,
end_date: str | date,
elements: list[str] | None = None,
) -> "Coverage":
"""Report how well a location is covered over a date range.
Runs the same lookups as ``get_range`` and summarizes, per
element, the fraction of days with data, plus the station
credited on the most days and its distance.
Args:
location: 5-digit US ZIP code, "lat,lon" string, or
(lat, lon) tuple.
start_date: Start date (YYYY-MM-DD) or date object.
end_date: End date (YYYY-MM-DD) or date object.
elements: List of weather elements to check.
Returns:
A Coverage report.
"""
results = self.get_range(location, start_date, end_date, elements)
return summarize_coverage(results)
[docs]
def process_csv(
self,
input_path: str | Path,
output_path: str | Path,
zipcode_column: str | int = "zip",
lat_column: str | int | None = None,
lon_column: str | int | None = None,
date_column: str | int | None = None,
year_column: str | int | None = "year",
month_column: str | int | None = "month",
day_column: str | int | None = "day",
output_format: str | None = None,
parallel: bool = True,
max_workers: int | None = None,
) -> int:
"""Process a CSV file and add weather data.
Args:
input_path: Path to input CSV file.
output_path: Path to output file (CSV or Parquet).
zipcode_column: Column name or index for ZIP code.
lat_column: Column for latitude (used with lon_column).
lon_column: Column for longitude.
date_column: Column name or index for date (YYYY-MM-DD).
year_column: Column for year (if no date_column).
month_column: Column for month (if no date_column).
day_column: Column for day (if no date_column).
output_format: "csv" or "parquet"; inferred from the output
path suffix when None (Parquet needs the ``parquet`` extra).
parallel: Use parallel processing for faster execution.
max_workers: Number of worker threads (default: CPU count, max 8).
Returns:
Number of rows processed.
Raises:
ValueError: If this Weather was created with online=True —
batch jobs need the local database (the CDO API allows
only 10,000 requests per day).
"""
if self.online:
raise ValueError(
"process_csv requires the local database (run setup()); "
"online mode is impractical for batch jobs given the CDO "
"API quota of 10,000 requests/day."
)
return _process_csv(
input_path=Path(input_path),
output_path=Path(output_path),
zipcode_column=zipcode_column,
lat_column=lat_column,
lon_column=lon_column,
date_column=date_column,
year_column=year_column,
month_column=month_column,
day_column=day_column,
db=self.db,
units=self.units,
include_weather_types=self.include_weather_types,
explain=self.explain,
output_format=output_format,
parallel=parallel,
max_workers=max_workers,
)
[docs]
def info(self) -> dict[str, int]:
"""Get database statistics.
Returns:
Dict with counts of stations and ZIP codes.
Raises:
RuntimeError: In online mode (no local database), or when
the database has not been set up yet.
"""
if self.online:
raise RuntimeError(
"info() reports on the local database; this Weather was "
"created with online=True."
)
if not self.db.exists() or self.db.count_stations() == 0:
raise RuntimeError(
"No station database found. Run setup() (CLI: get-weather setup) first."
)
return {
"ghcn_stations": self.db.count_stations("GHCND"),
"usaf_stations": self.db.count_stations("USAF-WBAN"),
"total_stations": self.db.count_stations(),
"zipcodes": self.db.count_zipcodes(),
}