Skip to content

omps_nadir

OMPS Nadir Reader

OMPSNadirReader

Bases: GriddedReader

Reader for OMPS (Ozone Mapping and Profiler Suite) Nadir products. Supports NOAA JPSS (V8TOZ, NP/TC SDR/GEO) and NASA (NMTO3) products. Available on AWS Open Data.

Source code in monetio/readers/omps_nadir.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
190
191
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
219
220
221
222
@register_reader("omps_nadir")
class OMPSNadirReader(GriddedReader):
    """
    Reader for OMPS (Ozone Mapping and Profiler Suite) Nadir products.
    Supports NOAA JPSS (V8TOZ, NP/TC SDR/GEO) and NASA (NMTO3) products.
    Available on AWS Open Data.
    """

    def open_dataset(
        self,
        files: str | list[str] = None,
        dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str = None,
        satellite: str = "snpp",
        product: str = "v8toz",
        group: str | list[str] = None,
        **kwargs,
    ) -> xr.Dataset:
        """
        Reads OMPS Nadir data.

        Parameters
        ----------
        files : Union[str, List[str]], optional
            File path(s) or URL(s).
        dates : Union[pd.DatetimeIndex, List[datetime], datetime, str], optional
            Dates to retrieve. If files is None, this is used to build URLs.
        satellite : str, optional
            Satellite identifier: 'snpp', 'n20' (NOAA-20/J01), or 'n21' (NOAA-21/J02).
            Default is 'snpp'.
        product : str, optional
            OMPS product type:
            - 'v8toz': NOAA Total Ozone EDR (default)
            - 'nmto3_l2': NASA Nadir Mapper Total Ozone L2
            - 'nmto3_l3': NASA Nadir Mapper Total Ozone L3
            - 'tc_sdr': NOAA Total Column SDR
            - 'np_sdr': NOAA Nadir Profiler SDR
        group : Union[str, List[str]], optional
            The NetCDF group(s) to open. If None, appropriate groups for the
            product will be selected (e.g. SDR + GEO for SDR products).
        **kwargs : dict
            Additional arguments passed to XarrayDriver.open.

        Returns
        -------
        xr.Dataset
            The OMPS dataset.

        Examples
        --------
        Open standard V8TOZ product:
        >>> reader = OMPSNadirReader()
        >>> ds = reader.open_dataset(dates='2024-01-01', product='v8toz')
        """
        if files is None:
            if dates is None:
                raise ValueError("Either 'files' or 'dates' must be provided.")
            files = self.build_urls(dates, satellite=satellite, product=product)

        if group is None:
            if product.lower() == "tc_sdr":
                groups = ["All_Data/OMPS-TC-SDR_All", "All_Data/OMPS-TC-GEO_All"]
            elif product.lower() == "np_sdr":
                groups = ["All_Data/OMPS-NP-SDR_All", "All_Data/OMPS-NP-GEO_All"]
            else:
                groups = [None]
        elif isinstance(group, str):
            groups = [group]
        else:
            groups = group

        user_preprocess = kwargs.pop("preprocess", None)

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

        if "concat_dim" not in kwargs:
            kwargs["concat_dim"] = "time"
        if "combine" not in kwargs:
            kwargs["combine"] = "nested"

        dsets = []
        for g in groups:
            g_kwargs = kwargs.copy()
            if g:
                g_kwargs["group"] = g
                # Filter files by group to avoid open_mfdataset failure
                if isinstance(files, list) and len(files) > 1:
                    # Look for SDR or GEO keyword in group name
                    g_sdr = "SDR" in g.upper()
                    g_geo = "GEO" in g.upper()
                    g_files = [
                        f
                        for f in files
                        if (g_sdr and "SDR" in f.upper()) or (g_geo and "GEO" in f.upper())
                    ]
                    # If filter returns nothing, it might be a single granule pair, fallback to all
                    if not g_files:
                        g_files = files
                else:
                    g_files = files
            else:
                g_files = files

            try:
                # Open without the preprocessor at this stage
                ds_g = super().open_dataset(g_files, **g_kwargs)
                dsets.append(ds_g)
            except (OSError, RuntimeError, ValueError):
                # Not all groups may be present in all files
                continue

        if not dsets:
            raise RuntimeError(f"No groups could be opened for product {product}.")

        # Merge groups
        ds = xr.merge(dsets, compat="no_conflicts")

        # Now apply OMPS preprocessing to the merged dataset
        ds = omps_nadir_preprocess(ds, product=product)

        if user_preprocess:
            ds = user_preprocess(ds)

        # Update history
        ds = update_history(ds, f"Read OMPS Nadir {product} data from {satellite}.")

        return ds

    def build_urls(
        self,
        dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str,
        satellite: str = "snpp",
        product: str = "v8toz",
    ) -> list[str]:
        """
        Build S3 URLs for OMPS Nadir data based on dates.

        Parameters
        ----------
        dates : Union[pd.DatetimeIndex, List[datetime], datetime, str]
            Dates to retrieve.
        satellite : str, optional
            Satellite identifier ('snpp', 'n20', 'n21', 'j01', 'j02').
        product : str, optional
            OMPS product.

        Returns
        -------
        List[str]
            List of S3 URLs.
        """
        from ..util import _import_required

        s3fs = _import_required("s3fs")

        if isinstance(dates, str | datetime.datetime | pd.Timestamp):
            dates = pd.DatetimeIndex([pd.to_datetime(dates)])
        else:
            dates = pd.to_datetime(dates)

        sat_map = {
            "snpp": "noaa-nesdis-snpp-pds",
            "n20": "noaa-nesdis-n20-pds",
            "n21": "noaa-nesdis-n21-pds",
            "j01": "noaa-nesdis-n20-pds",
            "j02": "noaa-nesdis-n21-pds",
        }
        bucket = sat_map.get(satellite.lower())
        if not bucket:
            raise ValueError(f"Unknown satellite: {satellite}. Choose from {list(sat_map.keys())}")

        prod_map = {
            "v8toz": "OMPS_V8TOZ",
            "tc_sdr": "OMPS-TC-SDR",
            "tc_geo": "OMPS-TC-GEO",
            "np_sdr": "OMPS-NP-SDR",
            "np_geo": "OMPS-NP-GEO",
        }
        dir_name = prod_map.get(product.lower())
        if not dir_name:
            # Fallback for NASA or others if they ever follow this pattern,
            # but usually they don't.
            dir_name = product

        # Determine if we need to fetch multiple directories (SDR + GEO)
        dirs_to_search = [dir_name]
        if product.lower() == "tc_sdr":
            dirs_to_search.append("OMPS-TC-GEO")
        elif product.lower() == "np_sdr":
            dirs_to_search.append("OMPS-NP-GEO")

        fs = s3fs.S3FileSystem(anon=True)
        urls = []
        for d in dates.floor("D").unique():
            for dn in dirs_to_search:
                prefix = f"{bucket}/{dn}/{d.strftime('%Y/%m/%d')}/"
                try:
                    found = fs.glob(f"{prefix}*.nc")
                    # Also try .h5 for SDRs
                    if not found and "SDR" in dn:
                        found = fs.glob(f"{prefix}*.h5")
                    urls.extend([f"s3://{f}" for f in found])
                except Exception:
                    continue

        return sorted(urls)

build_urls(dates, satellite='snpp', product='v8toz')

Build S3 URLs for OMPS Nadir data based on dates.

Parameters:

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

Dates to retrieve.

 required
satellite str

Satellite identifier ('snpp', 'n20', 'n21', 'j01', 'j02').

 'snpp'
product str

OMPS product.

 'v8toz'

Returns:

Type Description
List[str]

List of S3 URLs.
Source code in monetio/readers/omps_nadir.py 
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
190
191
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
219
220
221
222
def build_urls(
    self,
    dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str,
    satellite: str = "snpp",
    product: str = "v8toz",
) -> list[str]:
    """
    Build S3 URLs for OMPS Nadir data based on dates.

    Parameters
    ----------
    dates : Union[pd.DatetimeIndex, List[datetime], datetime, str]
        Dates to retrieve.
    satellite : str, optional
        Satellite identifier ('snpp', 'n20', 'n21', 'j01', 'j02').
    product : str, optional
        OMPS product.

    Returns
    -------
    List[str]
        List of S3 URLs.
    """
    from ..util import _import_required

    s3fs = _import_required("s3fs")

    if isinstance(dates, str | datetime.datetime | pd.Timestamp):
        dates = pd.DatetimeIndex([pd.to_datetime(dates)])
    else:
        dates = pd.to_datetime(dates)

    sat_map = {
        "snpp": "noaa-nesdis-snpp-pds",
        "n20": "noaa-nesdis-n20-pds",
        "n21": "noaa-nesdis-n21-pds",
        "j01": "noaa-nesdis-n20-pds",
        "j02": "noaa-nesdis-n21-pds",
    }
    bucket = sat_map.get(satellite.lower())
    if not bucket:
        raise ValueError(f"Unknown satellite: {satellite}. Choose from {list(sat_map.keys())}")

    prod_map = {
        "v8toz": "OMPS_V8TOZ",
        "tc_sdr": "OMPS-TC-SDR",
        "tc_geo": "OMPS-TC-GEO",
        "np_sdr": "OMPS-NP-SDR",
        "np_geo": "OMPS-NP-GEO",
    }
    dir_name = prod_map.get(product.lower())
    if not dir_name:
        # Fallback for NASA or others if they ever follow this pattern,
        # but usually they don't.
        dir_name = product

    # Determine if we need to fetch multiple directories (SDR + GEO)
    dirs_to_search = [dir_name]
    if product.lower() == "tc_sdr":
        dirs_to_search.append("OMPS-TC-GEO")
    elif product.lower() == "np_sdr":
        dirs_to_search.append("OMPS-NP-GEO")

    fs = s3fs.S3FileSystem(anon=True)
    urls = []
    for d in dates.floor("D").unique():
        for dn in dirs_to_search:
            prefix = f"{bucket}/{dn}/{d.strftime('%Y/%m/%d')}/"
            try:
                found = fs.glob(f"{prefix}*.nc")
                # Also try .h5 for SDRs
                if not found and "SDR" in dn:
                    found = fs.glob(f"{prefix}*.h5")
                urls.extend([f"s3://{f}" for f in found])
            except Exception:
                continue

    return sorted(urls)

open_dataset(files=None, dates=None, satellite='snpp', product='v8toz', group=None, **kwargs)

Reads OMPS Nadir data.

Parameters:

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

File path(s) or URL(s).

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

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

 None
satellite str

Satellite identifier: 'snpp', 'n20' (NOAA-20/J01), or 'n21' (NOAA-21/J02). Default is 'snpp'.

 'snpp'
product str

OMPS product type: - 'v8toz': NOAA Total Ozone EDR (default) - 'nmto3_l2': NASA Nadir Mapper Total Ozone L2 - 'nmto3_l3': NASA Nadir Mapper Total Ozone L3 - 'tc_sdr': NOAA Total Column SDR - 'np_sdr': NOAA Nadir Profiler SDR

 'v8toz'
group Union[str, List[str]]

The NetCDF group(s) to open. If None, appropriate groups for the product will be selected (e.g. SDR + GEO for SDR products).

 None
**kwargs dict

Additional arguments passed to XarrayDriver.open.

 {}

Returns:

Type Description
Dataset

The OMPS dataset.

Examples:

Open standard V8TOZ product:

>>> reader = OMPSNadirReader()
>>> ds = reader.open_dataset(dates='2024-01-01', product='v8toz')
Source code in monetio/readers/omps_nadir.py 
 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
def open_dataset(
    self,
    files: str | list[str] = None,
    dates: pd.DatetimeIndex | list[datetime.datetime] | datetime.datetime | str = None,
    satellite: str = "snpp",
    product: str = "v8toz",
    group: str | list[str] = None,
    **kwargs,
) -> xr.Dataset:
    """
    Reads OMPS Nadir data.

    Parameters
    ----------
    files : Union[str, List[str]], optional
        File path(s) or URL(s).
    dates : Union[pd.DatetimeIndex, List[datetime], datetime, str], optional
        Dates to retrieve. If files is None, this is used to build URLs.
    satellite : str, optional
        Satellite identifier: 'snpp', 'n20' (NOAA-20/J01), or 'n21' (NOAA-21/J02).
        Default is 'snpp'.
    product : str, optional
        OMPS product type:
        - 'v8toz': NOAA Total Ozone EDR (default)
        - 'nmto3_l2': NASA Nadir Mapper Total Ozone L2
        - 'nmto3_l3': NASA Nadir Mapper Total Ozone L3
        - 'tc_sdr': NOAA Total Column SDR
        - 'np_sdr': NOAA Nadir Profiler SDR
    group : Union[str, List[str]], optional
        The NetCDF group(s) to open. If None, appropriate groups for the
        product will be selected (e.g. SDR + GEO for SDR products).
    **kwargs : dict
        Additional arguments passed to XarrayDriver.open.

    Returns
    -------
    xr.Dataset
        The OMPS dataset.

    Examples
    --------
    Open standard V8TOZ product:
    >>> reader = OMPSNadirReader()
    >>> ds = reader.open_dataset(dates='2024-01-01', product='v8toz')
    """
    if files is None:
        if dates is None:
            raise ValueError("Either 'files' or 'dates' must be provided.")
        files = self.build_urls(dates, satellite=satellite, product=product)

    if group is None:
        if product.lower() == "tc_sdr":
            groups = ["All_Data/OMPS-TC-SDR_All", "All_Data/OMPS-TC-GEO_All"]
        elif product.lower() == "np_sdr":
            groups = ["All_Data/OMPS-NP-SDR_All", "All_Data/OMPS-NP-GEO_All"]
        else:
            groups = [None]
    elif isinstance(group, str):
        groups = [group]
    else:
        groups = group

    user_preprocess = kwargs.pop("preprocess", None)

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

    if "concat_dim" not in kwargs:
        kwargs["concat_dim"] = "time"
    if "combine" not in kwargs:
        kwargs["combine"] = "nested"

    dsets = []
    for g in groups:
        g_kwargs = kwargs.copy()
        if g:
            g_kwargs["group"] = g
            # Filter files by group to avoid open_mfdataset failure
            if isinstance(files, list) and len(files) > 1:
                # Look for SDR or GEO keyword in group name
                g_sdr = "SDR" in g.upper()
                g_geo = "GEO" in g.upper()
                g_files = [
                    f
                    for f in files
                    if (g_sdr and "SDR" in f.upper()) or (g_geo and "GEO" in f.upper())
                ]
                # If filter returns nothing, it might be a single granule pair, fallback to all
                if not g_files:
                    g_files = files
            else:
                g_files = files
        else:
            g_files = files

        try:
            # Open without the preprocessor at this stage
            ds_g = super().open_dataset(g_files, **g_kwargs)
            dsets.append(ds_g)
        except (OSError, RuntimeError, ValueError):
            # Not all groups may be present in all files
            continue

    if not dsets:
        raise RuntimeError(f"No groups could be opened for product {product}.")

    # Merge groups
    ds = xr.merge(dsets, compat="no_conflicts")

    # Now apply OMPS preprocessing to the merged dataset
    ds = omps_nadir_preprocess(ds, product=product)

    if user_preprocess:
        ds = user_preprocess(ds)

    # Update history
    ds = update_history(ds, f"Read OMPS Nadir {product} data from {satellite}.")

    return ds

omps_nadir_preprocess(ds, product='v8toz')

Preprocess OMPS Nadir dataset lazily.

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
product str

Product type, by default "v8toz".

 'v8toz'

Returns:

Type Description
Dataset

Processed dataset.

Examples:

>>> ds = omps_nadir_preprocess(ds, product='v8toz')
Source code in monetio/readers/omps_nadir.py 
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
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def omps_nadir_preprocess(ds: xr.Dataset, product: str = "v8toz") -> xr.Dataset:
    """
    Preprocess OMPS Nadir dataset lazily.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset.
    product : str, optional
        Product type, by default "v8toz".

    Returns
    -------
    xr.Dataset
        Processed dataset.

    Examples
    --------
    >>> ds = omps_nadir_preprocess(ds, product='v8toz')
    """
    if product == "v8toz":
        ds = _preprocess_v8toz(ds)
    elif product in ["nmto3_l2", "nmto3_l3"]:
        # Use existing logic from omps.py
        from .omps import omps_preprocess

        ds = omps_preprocess(ds, product=product)
        return ds
    elif product in ["tc_sdr", "np_sdr"]:
        ds = _preprocess_sdr(ds)

    # Standardize coordinates and dimensions
    ds = standardize_satellite_coords(ds)

    # Ensure a time dimension exists for stitching if not already present
    if "time" not in ds.dims:
        if "time" in ds.coords:
            if ds.coords["time"].ndim == 1:
                # If time is 1D and matches y or x, swap it
                # But jpss_time_to_datetime usually produces something indexed by y
                pass
            else:
                # Expand a new time dimension
                ds = ds.expand_dims("time")
        else:
            # Try to add from attributes
            ds = add_time_coord(ds, time_attr="time_coverage_start")

    return ds