Skip to content

ndbc

NDBC Buoy Reader

NDBCReader

Bases: PointReader

Source code in monetio/readers/ndbc.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@register_reader("ndbc")
class NDBCReader(PointReader):
    def open_dataset(
        self,
        files: str | list[str] = None,
        stations: str | list[str] = None,
        years: int | list[int] = None,
        realtime: bool = True,
        wide_fmt: bool = True,
        n_procs: int = 1,
        as_xarray: bool = True,
        lazy: bool = False,
        **kwargs,
    ) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
        """
        Retrieve and load NOAA National Data Buoy Center (NDBC) data.

        Parameters
        ----------
        files : Union[str, List[str]], optional
            File path, list of paths, or glob pattern.
        stations : Union[str, List[str]], optional
            Station IDs to retrieve if files are not provided.
        years : Union[int, List[int]], optional
            Years to retrieve for historical data.
        realtime : bool, optional
            Whether to retrieve real-time data (last 45 days), by default True.
        wide_fmt : bool, optional
            Whether to return data in wide format, by default True.
        n_procs : int, optional
            Number of processors for dask compute (if not lazy), by default 1.
        as_xarray : bool, optional
            Whether to return an xarray.Dataset, by default True.
        lazy : bool, optional
            Whether to return a dask-backed object, by default False.
        **kwargs : dict
            Additional arguments passed to the driver.

        Returns
        -------
        Union[pd.DataFrame, xr.Dataset, dd.DataFrame]
            The loaded NDBC data.
        """
        if files is None:
            if stations is None:
                raise ValueError("Must provide either 'files' or 'stations'.")
            files = build_urls(stations, years=years, realtime=realtime)

        if not files:
            raise ValueError("No files found or URLs built.")

        # Define per-file preprocessing
        read_func = partial(read_ndbc)

        # Use base class to open
        df = super().open_dataset(
            files,
            read_method=read_func,
            as_xarray=False,
            lazy=lazy,
            **kwargs,
        )

        # Post-processing
        df = self._post_process(df)
        df = self.harmonize(df)

        if not lazy and hasattr(df, "compute") and not isinstance(df, pd.DataFrame):
            df = df.compute(num_workers=n_procs)

        if as_xarray:
            # NDBC is already "wide" (one row per time/station)
            # expand2d=True will use ds_to_2d which works fine.
            ds = self.to_xarray(df, expand2d=wide_fmt, **kwargs)
            ds = update_history(ds, "Read NDBC buoy data.")
            return ds

        return df

    def harmonize(self, df):
        """
        Harmonize NDBC data to standard names and units.
        """
        # Backend-agnostic check for empty
        if hasattr(df, "columns") and len(df.columns) == 0:
            return df

        rename_map = {
            "WDIR": "wind_direction",
            "WSPD": "wind_speed",
            "GST": "wind_gust",
            "WVHT": "wave_height",
            "DPD": "dominant_wave_period",
            "APD": "average_wave_period",
            "MWD": "mean_wave_direction",
            "PRES": "air_pressure",
            "ATMP": "air_temperature",
            "WTMP": "sea_surface_temperature",
            "DEWP": "dew_point_temperature",
            "VIS": "visibility",
            "PTDY": "pressure_tendency",
            "TIDE": "tide_height",
        }
        df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns})

        # Add metadata (lat/lon)
        df = add_station_metadata(df)

        return super().harmonize(df)

    def _post_process(
        self,
        df: Union[pd.DataFrame, "dd.DataFrame"],
    ) -> Union[pd.DataFrame, "dd.DataFrame"]:
        """
        Internal post-processing logic.
        """
        # Backend-agnostic check for empty
        if hasattr(df, "columns") and len(df.columns) == 0:
            return df

        # Determine backend
        try:
            import dask.dataframe as dd

            is_dask = isinstance(df, dd.DataFrame)
        except ImportError:
            is_dask = False

        # Convert time
        # Year column can be YY or YYYY
        year_col = "YYYY" if "YYYY" in df.columns else "YY"

        def _to_datetime_ndbc(df_chunk):
            if len(df_chunk.columns) == 0:
                return pd.Series(dtype="datetime64[ns]")

            years = df_chunk[year_col].astype(int)
            # Handle 2-digit years if present (NDBC uses 90s and 00s/10s)
            years = years.apply(
                lambda x: x + 1900 if x >= 70 and x < 100 else (x + 2000 if x < 70 else x)
            )

            months = df_chunk["MM"].astype(int)
            days = df_chunk["DD"].astype(int)
            hours = df_chunk["hh"].astype(int)
            if "mm" in df_chunk.columns:
                minutes = df_chunk["mm"].astype(int)
            else:
                minutes = 0

            return pd.to_datetime(
                {
                    "year": years,
                    "month": months,
                    "day": days,
                    "hour": hours,
                    "minute": minutes,
                }
            )

        if is_dask:
            df["time"] = df.map_partitions(_to_datetime_ndbc)
        else:
            df["time"] = _to_datetime_ndbc(df)

        # Drop internal time columns
        time_cols = ["YY", "MM", "DD", "hh", "mm", "YYYY"]
        df = df.drop(columns=[c for c in time_cols if c in df.columns], errors="ignore")

        df = df.dropna(subset=["time"])

        return df

harmonize(df)

Harmonize NDBC data to standard names and units.

Source code in monetio/readers/ndbc.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def harmonize(self, df):
    """
    Harmonize NDBC data to standard names and units.
    """
    # Backend-agnostic check for empty
    if hasattr(df, "columns") and len(df.columns) == 0:
        return df

    rename_map = {
        "WDIR": "wind_direction",
        "WSPD": "wind_speed",
        "GST": "wind_gust",
        "WVHT": "wave_height",
        "DPD": "dominant_wave_period",
        "APD": "average_wave_period",
        "MWD": "mean_wave_direction",
        "PRES": "air_pressure",
        "ATMP": "air_temperature",
        "WTMP": "sea_surface_temperature",
        "DEWP": "dew_point_temperature",
        "VIS": "visibility",
        "PTDY": "pressure_tendency",
        "TIDE": "tide_height",
    }
    df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns})

    # Add metadata (lat/lon)
    df = add_station_metadata(df)

    return super().harmonize(df)

open_dataset(files=None, stations=None, years=None, realtime=True, wide_fmt=True, n_procs=1, as_xarray=True, lazy=False, **kwargs)

Retrieve and load NOAA National Data Buoy Center (NDBC) data.

Parameters:

Name Type Description Default
files Union[str, List[str]]

File path, list of paths, or glob pattern.

 None
stations Union[str, List[str]]

Station IDs to retrieve if files are not provided.

 None
years Union[int, List[int]]

Years to retrieve for historical data.

 None
realtime bool

Whether to retrieve real-time data (last 45 days), by default True.

 True
wide_fmt bool

Whether to return data in wide format, by default True.

 True
n_procs int

Number of processors for dask compute (if not lazy), by default 1.

 1
as_xarray bool

Whether to return an xarray.Dataset, by default True.

 True
lazy bool

Whether to return a dask-backed object, by default False.

 False
**kwargs dict

Additional arguments passed to the driver.

 {}

Returns:

Type Description
Union[DataFrame, Dataset, DataFrame]

The loaded NDBC data.
Source code in monetio/readers/ndbc.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def open_dataset(
    self,
    files: str | list[str] = None,
    stations: str | list[str] = None,
    years: int | list[int] = None,
    realtime: bool = True,
    wide_fmt: bool = True,
    n_procs: int = 1,
    as_xarray: bool = True,
    lazy: bool = False,
    **kwargs,
) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
    """
    Retrieve and load NOAA National Data Buoy Center (NDBC) data.

    Parameters
    ----------
    files : Union[str, List[str]], optional
        File path, list of paths, or glob pattern.
    stations : Union[str, List[str]], optional
        Station IDs to retrieve if files are not provided.
    years : Union[int, List[int]], optional
        Years to retrieve for historical data.
    realtime : bool, optional
        Whether to retrieve real-time data (last 45 days), by default True.
    wide_fmt : bool, optional
        Whether to return data in wide format, by default True.
    n_procs : int, optional
        Number of processors for dask compute (if not lazy), by default 1.
    as_xarray : bool, optional
        Whether to return an xarray.Dataset, by default True.
    lazy : bool, optional
        Whether to return a dask-backed object, by default False.
    **kwargs : dict
        Additional arguments passed to the driver.

    Returns
    -------
    Union[pd.DataFrame, xr.Dataset, dd.DataFrame]
        The loaded NDBC data.
    """
    if files is None:
        if stations is None:
            raise ValueError("Must provide either 'files' or 'stations'.")
        files = build_urls(stations, years=years, realtime=realtime)

    if not files:
        raise ValueError("No files found or URLs built.")

    # Define per-file preprocessing
    read_func = partial(read_ndbc)

    # Use base class to open
    df = super().open_dataset(
        files,
        read_method=read_func,
        as_xarray=False,
        lazy=lazy,
        **kwargs,
    )

    # Post-processing
    df = self._post_process(df)
    df = self.harmonize(df)

    if not lazy and hasattr(df, "compute") and not isinstance(df, pd.DataFrame):
        df = df.compute(num_workers=n_procs)

    if as_xarray:
        # NDBC is already "wide" (one row per time/station)
        # expand2d=True will use ds_to_2d which works fine.
        ds = self.to_xarray(df, expand2d=wide_fmt, **kwargs)
        ds = update_history(ds, "Read NDBC buoy data.")
        return ds

    return df

add_station_metadata(df)

Merge station metadata (lat/lon) into the dataframe.

Source code in monetio/readers/ndbc.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def add_station_metadata(
    df: Union[pd.DataFrame, "dd.DataFrame"],
) -> Union[pd.DataFrame, "dd.DataFrame"]:
    """
    Merge station metadata (lat/lon) into the dataframe.
    """
    if hasattr(df, "columns") and len(df.columns) == 0:
        return df

    meta = get_station_table()
    if meta.empty:
        return df

    # Remove lat/lon from df if already present to avoid merge suffix
    cols_to_drop = [c for c in meta.columns if c in df.columns and c != "siteid"]
    if cols_to_drop:
        df = df.drop(columns=cols_to_drop)

    try:
        import dask.dataframe as dd

        is_dask = isinstance(df, dd.DataFrame)
    except ImportError:
        is_dask = False

    if is_dask:
        meta_wrap = dd.from_pandas(meta, npartitions=1)
        df = df.merge(meta_wrap, on="siteid", how="left")
    else:
        df = df.merge(meta, on="siteid", how="left")

    return df

build_urls(stations, years=None, realtime=True)

Construct NDBC URLs.

Source code in monetio/readers/ndbc.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def build_urls(
    stations: str | list[str],
    years: int | list[int] = None,
    realtime: bool = True,
) -> list[str]:
    """
    Construct NDBC URLs.
    """
    if isinstance(stations, str):
        stations = [stations]

    urls = []
    if realtime:
        base_url = "https://www.ndbc.noaa.gov/data/realtime2/"
        for s in stations:
            urls.append(f"{base_url}{s.upper()}.txt")
    else:
        if years is None:
            # If no years provided, we might want to default to some, but better to error
            raise ValueError("Years must be provided for historical data.")
        if isinstance(years, int | str):
            years = [years]
        base_url = "https://www.ndbc.noaa.gov/data/historical/stdmet/"
        for s in stations:
            for y in years:
                urls.append(f"{base_url}{s.lower()}h{y}.txt.gz")
    return urls

get_station_table() cached

Download and parse the NDBC station table for metadata.

Source code in monetio/readers/ndbc.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
@lru_cache(maxsize=1)
def get_station_table() -> pd.DataFrame:
    """
    Download and parse the NDBC station table for metadata.
    """
    url = "https://www.ndbc.noaa.gov/data/stations/station_table.txt"
    try:
        df = pd.read_csv(url, sep="|", skiprows=2, header=None, on_bad_lines="skip")
        df.columns = [
            "siteid",
            "owner",
            "ttype",
            "hull",
            "name",
            "payload",
            "location",
            "timezone",
            "forecast",
            "note",
        ]
        df["siteid"] = df["siteid"].astype(str).str.strip().str.upper()

        def parse_loc(loc_str):
            try:
                # 44.794 N 87.313 W (44&#176;47'39" N 87&#176;18'48" W)
                parts = str(loc_str).split()
                lat = float(parts[0])
                if parts[1].upper() == "S":
                    lat = -lat
                lon = float(parts[2])
                if parts[3].upper() == "W":
                    lon = -lon
                return lat, lon
            except (ValueError, IndexError):
                return np.nan, np.nan

        locs = df["location"].apply(parse_loc)
        df["latitude"] = locs.apply(lambda x: x[0])
        df["longitude"] = locs.apply(lambda x: x[1])

        return df[["siteid", "name", "latitude", "longitude"]]
    except Exception:
        return pd.DataFrame()

read_ndbc(fn, **kwargs)

Read a single NDBC standard meteorological data file.

Source code in monetio/readers/ndbc.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def read_ndbc(fn: str, **kwargs) -> pd.DataFrame:
    """
    Read a single NDBC standard meteorological data file.
    """
    try:
        # Read header rows to get column names
        header_df = pd.read_csv(fn, sep=r"\s+", nrows=1, header=None)
        if len(header_df.columns) == 0:
            return pd.DataFrame()

        cols = header_df.iloc[0].tolist()
        if str(cols[0]).startswith("#"):
            cols[0] = str(cols[0]).lstrip("#")
            skip = 2  # Usually second row is units with # too
        else:
            skip = 1

        df = pd.read_csv(
            fn,
            sep=r"\s+",
            skiprows=skip,
            header=None,
            names=cols,
            na_values=["MM", "99", "999", "99.0", "999.0"],
            on_bad_lines="warn",
        )
    except Exception:
        return pd.DataFrame()

    # Extract siteid from filename
    import os

    basename = os.path.basename(fn)
    # 41002.txt -> 41002
    # 41002h2020.txt.gz -> 41002
    siteid = basename.split(".")[0].split("h")[0].upper()
    df["siteid"] = siteid

    return df