weather_integration.py

# ---------------------------------------------------------------------------
#  Weather data for NYC taxi feature engineering
# ---------------------------------------------------------------------------
# Fetches daily weather for NYC (Central Park, GHCND) via NOAA Climate Data
# Online (CDO) API v2. Requires token: https://www.ncdc.noaa.gov/cdo-web/token
# Set env NOAA_CDO_TOKEN. Falls back to local CSV if token missing or API fails.
# Output DataFrame has columns: date, hour, weather_code (for merge with trips).
# ---------------------------------------------------------------------------

import os
import numpy as np
import pandas as pd
import requests as _req
from dotenv import load_dotenv

load_dotenv()
NOAA_CDO_TOKEN = os.getenv("NOAA_CDO_TOKEN")

_BASE = "https://www.ncei.noaa.gov/cdo-web/api/v2"


def _fetch_noaa_weather_nyc(start_date="2022-05-01", end_date="2022-05-31", token=NOAA_CDO_TOKEN):
    """
    Fetches daily GHCND data (PRCP, TMAX, TMIN, etc.) for Central Park NYC.

    Paginates through the CDO data endpoint. Requires token to be passed in
    (from load_weather_nyc, which reads NOAA_CDO_TOKEN from environment).

    Args:
        start_date: Start date (YYYY-MM-DD).
        end_date: End date (YYYY-MM-DD).
        token: NOAA CDO API token (required; no fallback here).

    Returns:
        pd.DataFrame: Long-format rows with columns date, datatype, value.
    """
    if not token:
        raise ValueError("NOAA API token is required for _fetch_noaa_weather_nyc")
    out, offset, limit = [], 0, 1000
    while True:
        url = f"{_BASE}/data?datasetid=GHCND&stationid=GHCND:USW00094728&startdate={start_date}&enddate={end_date}&limit={limit}&offset={offset}&units=standard"
        # Use the token passed in (from env) so the API call succeeds
        r = _req.get(url, headers={"token": token})
        r.raise_for_status()
        data = r.json()
        if "results" not in data:
            break
        for rec in data["results"]:
            out.append({"date": rec["date"][:10], "datatype": rec["datatype"], "value": rec["value"]})
        if len(data["results"]) < limit:
            break
        offset += limit
    return pd.DataFrame(out)


def _noaa_to_weather_code(df_noaa):
    """
    Converts NOAA long-format data to a (date, hour, weather_code) table for trip merge.

    Derives a simple weather_code from PRCP/TMAX/TMIN (e.g. 1=clear, 2=rain, 3=hot, 4=cold),
    then expands to one row per (date, hour) so downstream merge on date+hour works.

    Args:
        df_noaa: DataFrame from _fetch_noaa_weather_nyc (columns date, datatype, value).

    Returns:
        pd.DataFrame: Columns date (str YYYY-MM-DD), hour (0–23), weather_code (int).
    """
    # API can return duplicate (date, datatype); pivot requires unique index. Aggregate duplicates.
    df_noaa = df_noaa.groupby(["date", "datatype"], as_index=False)["value"].mean()
    p = df_noaa.pivot(index="date", columns="datatype", values="value").reset_index()
    p["date"] = pd.to_datetime(p["date"])
    code = np.ones(len(p), dtype=int)
    if "PRCP" in p.columns:
        code[p["PRCP"].fillna(0) > 0.1] = 2
    if "TMAX" in p.columns:
        code[(p["TMAX"].fillna(0) >= 90) & (code != 2)] = 3
    if "TMIN" in p.columns:
        code[(p["TMIN"].fillna(0) <= 40) & (code != 2)] = 4
    p["weather_code"] = code
    rows = []
    for _, row in p.iterrows():
        for h in range(24):
            rows.append({"date": row["date"].strftime("%Y-%m-%d"), "hour": h, "weather_code": row["weather_code"]})
    return pd.DataFrame(rows)


def load_weather_nyc(
    start_date: str = "2022-05-01",
    end_date: str = "2022-05-31",
    csv_fallback_paths: list = None,
) -> pd.DataFrame:
    """
    Loads weather for NYC (date, hour, weather_code) for merge with trip data.

    Uses NOAA CDO API if NOAA_CDO_TOKEN is set; otherwise loads from the first
    existing path in csv_fallback_paths. Ensures columns date (str YYYY-MM-DD),
    hour (int 0–23), weather_code.

    Args:
        start_date: Start date for API (YYYY-MM-DD).
        end_date: End date for API (YYYY-MM-DD).
        csv_fallback_paths: Optional list of paths to try for CSV fallback.
            Defaults to ['data/weather_may_2022.csv', '../data/weather_may_2022.csv'].

    Returns:
        pd.DataFrame: Weather with columns date, hour, weather_code.
    """
    if csv_fallback_paths is None:
        csv_fallback_paths = [
            "data/weather_may_2022.csv",
            os.path.join(os.path.dirname(os.getcwd()), "data", "weather_may_2022.csv"),
        ]
    # Read token from environment (e.g. from .env via load_dotenv)
    token = os.environ.get("NOAA_CDO_TOKEN", "").strip()
    print(f"[Weather] Token present: {bool(token)} (length={len(token) if token else 0})")

    if token:
        try:
            # Pass token so _fetch_noaa_weather_nyc can use it in the API request
            raw = _fetch_noaa_weather_nyc(start_date=start_date, end_date=end_date, token=token)
            print(f"[Weather] API returned {len(raw)} rows")
            return _noaa_to_weather_code(raw)
        except Exception as e:
            # Log the error so you know why we fall back to CSV
            print(f"[Weather] API failed ({e}), falling back to CSV")
            pass
    # No token or API failed: try CSV fallback
    for path in csv_fallback_paths:
        if path and os.path.exists(path):
            weather = pd.read_csv(path)
            if "status" in weather.columns:
                weather = weather.drop(columns=["status"])
            if "code" in weather.columns and "weather_code" not in weather.columns:
                weather = weather.rename(columns={"code": "weather_code"})
            weather["date"] = pd.to_datetime(weather["date"]).dt.strftime("%Y-%m-%d")
            if "hour" not in weather.columns:
                weather["hour"] = 0
            print(f"[Weather] Loaded from CSV: {path}")
            return weather
    raise FileNotFoundError(
        "No NOAA_CDO_TOKEN and no CSV found at " + str(csv_fallback_paths)
    )