Skip to content

rrfs

RRFS Reader for AWS Open Data

RRFSReader

Bases: NCEPPDSReader

Reader for RRFS (Rapid Refresh Forecast System) on AWS.

Source code in monetio/readers/rrfs.py 
 14
 15
 16
 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
@register_reader("rrfs")
class RRFSReader(NCEPPDSReader):
    """
    Reader for RRFS (Rapid Refresh Forecast System) on AWS.
    """

    def build_urls(
        self,
        dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str,
        hour: int = 0,
        lead_time: int | list[int] = 0,
        product: str = "prslev.3km",
        domain: str = "conus",
        **kwargs: Any,
    ) -> list[str]:
        """
        Build S3 URLs for RRFS data.

        Parameters
        ----------
        dates : Union[pd.DatetimeIndex, List[datetime.datetime], datetime.datetime, str]
            Dates to retrieve.
        hour : int, optional
            Forecast cycle hour, by default 0.
        lead_time : Union[int, List[int]], optional
            Forecast lead time(s), by default 0.
        product : str, optional
            Product string, by default "prslev.3km".
        domain : str, optional
            Domain string (e.g., 'conus', 'ak', 'na', 'pr', 'hi'), by default "conus".
        **kwargs : Any
            Additional arguments.

        Returns
        -------
        List[str]
            List of S3 URLs.

        Examples
        --------
        >>> reader = RRFSReader()
        >>> urls = reader.build_urls("2026-03-28", hour=0, lead_time=1)
        """
        if isinstance(dates, str | datetime.datetime | pd.Timestamp):
            dates = pd.DatetimeIndex([pd.to_datetime(dates)])
        else:
            dates = pd.to_datetime(dates)

        if isinstance(lead_time, int):
            lead_times = [lead_time]
        else:
            lead_times = lead_time

        bucket = "noaa-rrfs-pds"
        urls = []
        for d in dates:
            d_str = d.strftime("%Y%m%d")
            h_str = f"{hour:02d}"
            for lt in lead_times:
                lt_str = f"{lt:03d}"
                # s3://noaa-rrfs-pds/rrfs_a/rrfs.20260328/00/rrfs.t00z.prslev.3km.f000.conus.grib2
                url = f"s3://{bucket}/rrfs_a/rrfs.{d_str}/{h_str}/rrfs.t{h_str}z.{product}.f{lt_str}.{domain}.grib2"
                urls.append(url)
        return urls

    def open_dataset(
        self,
        files: str | list[str] | None = None,
        dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str | None = None,
        hour: int = 0,
        lead_time: int | list[int] = 0,
        product: str = "prslev.3km",
        domain: str = "conus",
        **kwargs: Any,
    ) -> xr.Dataset:
        """
        Reads RRFS data.

        Parameters
        ----------
        files : Union[str, List[str]], optional
            File path(s) or S3 URL(s).
        dates : Union[pd.DatetimeIndex, List[datetime.datetime], datetime.datetime, str], optional
            Dates to retrieve. If files is None, this is used to build URLs.
        hour : int, optional
            Forecast cycle hour, by default 0.
        lead_time : Union[int, List[int]], optional
            Forecast lead time(s), by default 0.
        product : str, optional
            Product string, by default "prslev.3km".
        domain : str, optional
            Domain string, by default "conus".
        **kwargs : Any
            Additional arguments passed to XarrayDriver.open.

        Returns
        -------
        xr.Dataset
            The RRFS dataset.

        Examples
        --------
        >>> reader = RRFSReader()
        >>> # ds = reader.open_dataset(dates="2026-03-28", hour=0, lead_time=0)  # Requires grib2io
        """
        if files is None:
            if dates is None:
                raise ValueError("Either 'files' or 'dates' must be provided.")
            files = self.build_urls(
                dates, hour=hour, lead_time=lead_time, product=product, domain=domain, **kwargs
            )

        if "preprocess" not in kwargs:
            kwargs["preprocess"] = rrfs_preprocess

        if "engine" not in kwargs:
            kwargs["engine"] = "grib2io"

        # Use XarrayDriver (via GriddedReader/BaseReader)
        ds = self.driver.open(files, **kwargs)

        # Apply RRFS-specific harmonization
        ds = self.harmonize(ds)

        # Update history
        ds = update_history(ds, "Read RRFS data from AWS PDS.")

        return ds

    def harmonize(self, ds: xr.Dataset) -> xr.Dataset:
        """
        Harmonize RRFS metadata to monetio standards.

        Parameters
        ----------
        ds : xr.Dataset
            Input RRFS dataset.

        Returns
        -------
        xr.Dataset
            Harmonized dataset.
        """
        # First use parent harmonization (NCEP standards)
        ds = super().harmonize(ds)

        # Add RRFS-specific renaming or transformations if needed
        # (Parent handles most NCEP products well)

        return ds

build_urls(dates, hour=0, lead_time=0, product='prslev.3km', domain='conus', **kwargs)

Build S3 URLs for RRFS data.

Parameters:

Name Type Description Default
dates Union[DatetimeIndex, List[datetime], datetime, str]

Dates to retrieve.

 required
hour int

Forecast cycle hour, by default 0.

 0
lead_time Union[int, List[int]]

Forecast lead time(s), by default 0.

 0
product str

Product string, by default "prslev.3km".

 'prslev.3km'
domain str

Domain string (e.g., 'conus', 'ak', 'na', 'pr', 'hi'), by default "conus".

 'conus'
**kwargs Any

Additional arguments.

 {}

Returns:

Type Description
List[str]

List of S3 URLs.

Examples:

>>> reader = RRFSReader()
>>> urls = reader.build_urls("2026-03-28", hour=0, lead_time=1)
Source code in monetio/readers/rrfs.py 
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
def build_urls(
    self,
    dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str,
    hour: int = 0,
    lead_time: int | list[int] = 0,
    product: str = "prslev.3km",
    domain: str = "conus",
    **kwargs: Any,
) -> list[str]:
    """
    Build S3 URLs for RRFS data.

    Parameters
    ----------
    dates : Union[pd.DatetimeIndex, List[datetime.datetime], datetime.datetime, str]
        Dates to retrieve.
    hour : int, optional
        Forecast cycle hour, by default 0.
    lead_time : Union[int, List[int]], optional
        Forecast lead time(s), by default 0.
    product : str, optional
        Product string, by default "prslev.3km".
    domain : str, optional
        Domain string (e.g., 'conus', 'ak', 'na', 'pr', 'hi'), by default "conus".
    **kwargs : Any
        Additional arguments.

    Returns
    -------
    List[str]
        List of S3 URLs.

    Examples
    --------
    >>> reader = RRFSReader()
    >>> urls = reader.build_urls("2026-03-28", hour=0, lead_time=1)
    """
    if isinstance(dates, str | datetime.datetime | pd.Timestamp):
        dates = pd.DatetimeIndex([pd.to_datetime(dates)])
    else:
        dates = pd.to_datetime(dates)

    if isinstance(lead_time, int):
        lead_times = [lead_time]
    else:
        lead_times = lead_time

    bucket = "noaa-rrfs-pds"
    urls = []
    for d in dates:
        d_str = d.strftime("%Y%m%d")
        h_str = f"{hour:02d}"
        for lt in lead_times:
            lt_str = f"{lt:03d}"
            # s3://noaa-rrfs-pds/rrfs_a/rrfs.20260328/00/rrfs.t00z.prslev.3km.f000.conus.grib2
            url = f"s3://{bucket}/rrfs_a/rrfs.{d_str}/{h_str}/rrfs.t{h_str}z.{product}.f{lt_str}.{domain}.grib2"
            urls.append(url)
    return urls

harmonize(ds)

Harmonize RRFS metadata to monetio standards.

Parameters:

Name Type Description Default
ds Dataset

Input RRFS dataset.

 required

Returns:

Type Description
Dataset

Harmonized dataset.
Source code in monetio/readers/rrfs.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def harmonize(self, ds: xr.Dataset) -> xr.Dataset:
    """
    Harmonize RRFS metadata to monetio standards.

    Parameters
    ----------
    ds : xr.Dataset
        Input RRFS dataset.

    Returns
    -------
    xr.Dataset
        Harmonized dataset.
    """
    # First use parent harmonization (NCEP standards)
    ds = super().harmonize(ds)

    # Add RRFS-specific renaming or transformations if needed
    # (Parent handles most NCEP products well)

    return ds

open_dataset(files=None, dates=None, hour=0, lead_time=0, product='prslev.3km', domain='conus', **kwargs)

Reads RRFS data.

Parameters:

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

File path(s) or S3 URL(s).

 None
dates Union[DatetimeIndex, List[datetime], datetime, str]

Dates to retrieve. If files is None, this is used to build URLs.

 None
hour int

Forecast cycle hour, by default 0.

 0
lead_time Union[int, List[int]]

Forecast lead time(s), by default 0.

 0
product str

Product string, by default "prslev.3km".

 'prslev.3km'
domain str

Domain string, by default "conus".

 'conus'
**kwargs Any

Additional arguments passed to XarrayDriver.open.

 {}

Returns:

Type Description
Dataset

The RRFS dataset.

Examples:

>>> reader = RRFSReader()
>>> # ds = reader.open_dataset(dates="2026-03-28", hour=0, lead_time=0)  # Requires grib2io
Source code in monetio/readers/rrfs.py 
 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
def open_dataset(
    self,
    files: str | list[str] | None = None,
    dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str | None = None,
    hour: int = 0,
    lead_time: int | list[int] = 0,
    product: str = "prslev.3km",
    domain: str = "conus",
    **kwargs: Any,
) -> xr.Dataset:
    """
    Reads RRFS data.

    Parameters
    ----------
    files : Union[str, List[str]], optional
        File path(s) or S3 URL(s).
    dates : Union[pd.DatetimeIndex, List[datetime.datetime], datetime.datetime, str], optional
        Dates to retrieve. If files is None, this is used to build URLs.
    hour : int, optional
        Forecast cycle hour, by default 0.
    lead_time : Union[int, List[int]], optional
        Forecast lead time(s), by default 0.
    product : str, optional
        Product string, by default "prslev.3km".
    domain : str, optional
        Domain string, by default "conus".
    **kwargs : Any
        Additional arguments passed to XarrayDriver.open.

    Returns
    -------
    xr.Dataset
        The RRFS dataset.

    Examples
    --------
    >>> reader = RRFSReader()
    >>> # ds = reader.open_dataset(dates="2026-03-28", hour=0, lead_time=0)  # Requires grib2io
    """
    if files is None:
        if dates is None:
            raise ValueError("Either 'files' or 'dates' must be provided.")
        files = self.build_urls(
            dates, hour=hour, lead_time=lead_time, product=product, domain=domain, **kwargs
        )

    if "preprocess" not in kwargs:
        kwargs["preprocess"] = rrfs_preprocess

    if "engine" not in kwargs:
        kwargs["engine"] = "grib2io"

    # Use XarrayDriver (via GriddedReader/BaseReader)
    ds = self.driver.open(files, **kwargs)

    # Apply RRFS-specific harmonization
    ds = self.harmonize(ds)

    # Update history
    ds = update_history(ds, "Read RRFS data from AWS PDS.")

    return ds

rrfs_preprocess(ds)

Preprocess function for a single RRFS file.

Parameters:

Name Type Description Default
ds Dataset

Input RRFS dataset.

 required

Returns:

Type Description
Dataset

Processed dataset.

Examples:

>>> # ds = rrfs_preprocess(ds)
Source code in monetio/readers/rrfs.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def rrfs_preprocess(ds: xr.Dataset) -> xr.Dataset:
    """
    Preprocess function for a single RRFS file.

    Parameters
    ----------
    ds : xarray.Dataset
        Input RRFS dataset.

    Returns
    -------
    xarray.Dataset
        Processed dataset.

    Examples
    --------
    >>> # ds = rrfs_preprocess(ds)
    """
    # 1. Format Units
    ds = _format_units(ds)

    # 2. Scientific Hygiene
    ds = _scientific_hygiene(ds)

    # Update history
    ds = update_history(ds, "Preprocessed RRFS data.")

    return ds