Skip to content

tropomi

TROPOMI Reader

TROPOMIReader

Bases: GriddedReader

Reader for TROPOMI L2 (Sentinel-5P) data.

Source code in monetio/readers/tropomi.py 
 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
@register_reader("tropomi")
class TROPOMIReader(GriddedReader):
    """
    Reader for TROPOMI L2 (Sentinel-5P) data.
    """

    def open_dataset(
        self,
        files: str | list[str],
        group: str | list[str] | None = None,
        calculate_pressure: bool = True,
        qa_threshold: float | None = None,
        **kwargs,
    ) -> xr.Dataset:
        """
        Reads TROPOMI data.

        Parameters
        ----------
        files : Union[str, List[str]]
            File path(s) or URL(s).
        group : str or list of str, optional
            The NetCDF group(s) to open. If a list is provided, groups will be merged.
            If None, common TROPOMI groups will be opened:
            - "PRODUCT"
            - "PRODUCT/SUPPORT_DATA/INPUT_DATA"
            - "PRODUCT/SUPPORT_DATA/GEOLOCATIONS"
            - "PRODUCT/SUPPORT_DATA/DETAILED_RESULTS"
        calculate_pressure : bool, optional
            Whether to calculate pressure levels if necessary variables are found,
            by default True.
        qa_threshold : float, optional
            If provided, mask data where 'qa_value' is less than this threshold.
        **kwargs : dict
            Additional arguments passed to XarrayDriver.open.

        Returns
        -------
        xr.Dataset
            The TROPOMI dataset.

        Examples
        --------
        Open standard NO2 product:
        >>> reader = TROPOMIReader()
        >>> ds = reader.open_dataset(files="S5P_OFFL_L2_NO2_*.nc", qa_threshold=0.75)
        """
        if group is None:
            groups = [
                "PRODUCT",
                "PRODUCT/SUPPORT_DATA/INPUT_DATA",
                "PRODUCT/SUPPORT_DATA/GEOLOCATIONS",
                "PRODUCT/SUPPORT_DATA/DETAILED_RESULTS",
            ]
        elif isinstance(group, str):
            groups = [group]
        else:
            groups = group

        # To ensure consistent dimensions and that all variables needed for
        # preprocessing (e.g. pressure calculation) are available, we apply
        # preprocessing AFTER merging all groups.
        user_preprocess = kwargs.pop("preprocess", None)

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

        dsets = []
        for g in groups:
            # We copy kwargs to avoid modifying the original dict in the loop
            g_kwargs = kwargs.copy()
            g_kwargs["group"] = g
            try:
                # We open without the TROPOMI preprocess at this stage
                ds_g = super().open_dataset(files, **g_kwargs)
                dsets.append(ds_g)
            except Exception as e:
                warnings.warn(f"Could not open group {g}: {e}")

        if not dsets:
            raise RuntimeError("No groups could be opened.")

        # Merge groups
        # We use compat='no_conflicts' as coordinates should be identical
        ds = xr.merge(dsets, compat="no_conflicts")

        # Now apply TROPOMI preprocessing to the merged dataset
        ds = tropomi_preprocess(
            ds, calculate_pressure=calculate_pressure, qa_threshold=qa_threshold
        )

        if user_preprocess:
            ds = user_preprocess(ds)

        # Update history
        ds = update_history(ds, "Read TROPOMI data.")

        return ds

open_dataset(files, group=None, calculate_pressure=True, qa_threshold=None, **kwargs)

Reads TROPOMI data.

Parameters:

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

File path(s) or URL(s).

 required
group str or list of str

The NetCDF group(s) to open. If a list is provided, groups will be merged. If None, common TROPOMI groups will be opened: - "PRODUCT" - "PRODUCT/SUPPORT_DATA/INPUT_DATA" - "PRODUCT/SUPPORT_DATA/GEOLOCATIONS" - "PRODUCT/SUPPORT_DATA/DETAILED_RESULTS"

 None
calculate_pressure bool

Whether to calculate pressure levels if necessary variables are found, by default True.

 True
qa_threshold float

If provided, mask data where 'qa_value' is less than this threshold.

 None
**kwargs dict

Additional arguments passed to XarrayDriver.open.

 {}

Returns:

Type Description
Dataset

The TROPOMI dataset.

Examples:

Open standard NO2 product:

>>> reader = TROPOMIReader()
>>> ds = reader.open_dataset(files="S5P_OFFL_L2_NO2_*.nc", qa_threshold=0.75)
Source code in monetio/readers/tropomi.py 
 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
def open_dataset(
    self,
    files: str | list[str],
    group: str | list[str] | None = None,
    calculate_pressure: bool = True,
    qa_threshold: float | None = None,
    **kwargs,
) -> xr.Dataset:
    """
    Reads TROPOMI data.

    Parameters
    ----------
    files : Union[str, List[str]]
        File path(s) or URL(s).
    group : str or list of str, optional
        The NetCDF group(s) to open. If a list is provided, groups will be merged.
        If None, common TROPOMI groups will be opened:
        - "PRODUCT"
        - "PRODUCT/SUPPORT_DATA/INPUT_DATA"
        - "PRODUCT/SUPPORT_DATA/GEOLOCATIONS"
        - "PRODUCT/SUPPORT_DATA/DETAILED_RESULTS"
    calculate_pressure : bool, optional
        Whether to calculate pressure levels if necessary variables are found,
        by default True.
    qa_threshold : float, optional
        If provided, mask data where 'qa_value' is less than this threshold.
    **kwargs : dict
        Additional arguments passed to XarrayDriver.open.

    Returns
    -------
    xr.Dataset
        The TROPOMI dataset.

    Examples
    --------
    Open standard NO2 product:
    >>> reader = TROPOMIReader()
    >>> ds = reader.open_dataset(files="S5P_OFFL_L2_NO2_*.nc", qa_threshold=0.75)
    """
    if group is None:
        groups = [
            "PRODUCT",
            "PRODUCT/SUPPORT_DATA/INPUT_DATA",
            "PRODUCT/SUPPORT_DATA/GEOLOCATIONS",
            "PRODUCT/SUPPORT_DATA/DETAILED_RESULTS",
        ]
    elif isinstance(group, str):
        groups = [group]
    else:
        groups = group

    # To ensure consistent dimensions and that all variables needed for
    # preprocessing (e.g. pressure calculation) are available, we apply
    # preprocessing AFTER merging all groups.
    user_preprocess = kwargs.pop("preprocess", None)

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

    dsets = []
    for g in groups:
        # We copy kwargs to avoid modifying the original dict in the loop
        g_kwargs = kwargs.copy()
        g_kwargs["group"] = g
        try:
            # We open without the TROPOMI preprocess at this stage
            ds_g = super().open_dataset(files, **g_kwargs)
            dsets.append(ds_g)
        except Exception as e:
            warnings.warn(f"Could not open group {g}: {e}")

    if not dsets:
        raise RuntimeError("No groups could be opened.")

    # Merge groups
    # We use compat='no_conflicts' as coordinates should be identical
    ds = xr.merge(dsets, compat="no_conflicts")

    # Now apply TROPOMI preprocessing to the merged dataset
    ds = tropomi_preprocess(
        ds, calculate_pressure=calculate_pressure, qa_threshold=qa_threshold
    )

    if user_preprocess:
        ds = user_preprocess(ds)

    # Update history
    ds = update_history(ds, "Read TROPOMI data.")

    return ds

tropomi_preprocess(ds, calculate_pressure=True, qa_threshold=None)

Preprocess TROPOMI dataset: standardize coordinates, handle time, and optionally calculate pressure and apply quality flags.

Parameters:

Name Type Description Default
ds Dataset

Input dataset from a single file/group (or merged groups).

 required
calculate_pressure bool

Whether to calculate pressure levels, by default True.

 True
qa_threshold float

Quality value threshold for masking. If provided, variables are masked where 'qa_value' is less than this threshold.

 None

Returns:

Type Description
Dataset

Processed dataset.

Examples:

>>> ds = tropomi_preprocess(ds, qa_threshold=0.75)
Source code in monetio/readers/tropomi.py 
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
def tropomi_preprocess(
    ds: xr.Dataset, calculate_pressure: bool = True, qa_threshold: float | None = None
) -> xr.Dataset:
    """
    Preprocess TROPOMI dataset: standardize coordinates, handle time, and
    optionally calculate pressure and apply quality flags.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset from a single file/group (or merged groups).
    calculate_pressure : bool, optional
        Whether to calculate pressure levels, by default True.
    qa_threshold : float, optional
        Quality value threshold for masking. If provided, variables are masked
        where 'qa_value' is less than this threshold.

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

    Examples
    --------
    >>> ds = tropomi_preprocess(ds, qa_threshold=0.75)
    """
    # 1. Standardize dimensions and coordinates
    # TROPOMI uses 'scanline' and 'ground_pixel'
    ds = standardize_satellite_coords(ds, lat_name="latitude", lon_name="longitude")

    # 2. Handle Time
    if "time" in ds.coords and "delta_time" in ds.data_vars:
        ref_time = ds.coords["time"]
        delta_time = ds.data_vars["delta_time"]
        if "y" in delta_time.dims:
            scan_time = ref_time + delta_time.astype("timedelta64[ms]")
            ds = ds.assign_coords(time=scan_time)

    # 3. Calculate Pressure (Lazy) - must happen before dim rename if it depends on 'y'
    if calculate_pressure:
        ds = _add_pressure_levels(ds)

    # 4. Handle Altitude/Height
    for h_var in ["altitude", "aerosol_mid_height", "height"]:
        if h_var in ds.data_vars and "height_m_mid" not in ds.data_vars:
            h_mid = ds[h_var].copy()
            units = h_mid.attrs.get("units", "m")
            if units == "km":
                h_mid = h_mid * 1000.0
                h_mid.attrs["units"] = "m"
            ds["height_m_mid"] = h_mid.assign_attrs({"long_name": "mid-layer height"})
            break

    # 5. Coordinate and Dimension Harmonization
    if "time" in ds.coords and "time" not in ds.dims:
        if ds.coords["time"].dims == ("y",):
            ds = ds.swap_dims({"y": "time"})

    # Ensure all data variables have 'time' dimension if it exists and they have 'y'
    if "time" in ds.dims:
        for var in ds.data_vars:
            if "y" in ds[var].dims:
                ds[var] = ds[var].rename({"y": "time"})

    # 6. Quality Flagging (Lazy & Vectorized)
    if qa_threshold is not None:
        ds = apply_qa_mask(ds, qa_var="qa_value", threshold=qa_threshold)

    # Update history
    ds = update_history(
        ds,
        "Preprocessed TROPOMI data: standardized coordinates, handled time, "
        "and optionally applied quality flagging and pressure calculation.",
    )

    return ds