Skip to content

tempo

TEMPO Reader

TEMPOReader

Bases: GriddedReader

Reader for TEMPO (Tropospheric Emissions: Monitoring of Pollution) L2 data.

Source code in monetio/readers/tempo.py 
 9
10
11
12
13
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
@register_reader("tempo")
class TEMPOReader(GriddedReader):
    """
    Reader for TEMPO (Tropospheric Emissions: Monitoring of Pollution) L2 data.
    """

    def open_dataset(
        self,
        files: str | list[str],
        group: str | list[str] | None = None,
        variable_dict: dict | None = None,
        **kwargs,
    ) -> xr.Dataset:
        """
        Reads TEMPO 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 TEMPO groups will be opened:
            - "product"
            - "geolocation"
            - "support_data"
            - "qa_statistics"
        variable_dict : dict, optional
            Dictionary mapping variable names to processing options (scale, minimum, maximum).
        **kwargs : dict
            Additional arguments passed to XarrayDriver.open.

        Returns
        -------
        xr.Dataset
            The TEMPO dataset.

        Examples
        --------
        Open standard NO2 product:
        >>> reader = TEMPOReader()
        >>> ds = reader.open_dataset(files="TEMPO_NO2_L2_*.nc")
        """
        if group is None:
            groups = ["product", "geolocation", "support_data", "qa_statistics"]
        elif isinstance(group, str):
            groups = [group]
        else:
            groups = group

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

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

        dsets = []
        for g in groups:
            g_kwargs = kwargs.copy()
            g_kwargs["group"] = g
            try:
                # Open without the preprocessor at this stage
                ds_g = super().open_dataset(files, **g_kwargs)
                dsets.append(ds_g)
            except Exception:
                # Not all groups may be present in all files
                continue

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

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

        # Now apply TEMPO preprocessing to the merged dataset
        ds = tempo_preprocess(ds, variable_dict=variable_dict)

        if user_preprocess:
            ds = user_preprocess(ds)

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

        return ds

open_dataset(files, group=None, variable_dict=None, **kwargs)

Reads TEMPO 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 TEMPO groups will be opened: - "product" - "geolocation" - "support_data" - "qa_statistics"

 None
variable_dict dict

Dictionary mapping variable names to processing options (scale, minimum, maximum).

 None
**kwargs dict

Additional arguments passed to XarrayDriver.open.

 {}

Returns:

Type Description
Dataset

The TEMPO dataset.

Examples:

Open standard NO2 product:

>>> reader = TEMPOReader()
>>> ds = reader.open_dataset(files="TEMPO_NO2_L2_*.nc")
Source code in monetio/readers/tempo.py 
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
def open_dataset(
    self,
    files: str | list[str],
    group: str | list[str] | None = None,
    variable_dict: dict | None = None,
    **kwargs,
) -> xr.Dataset:
    """
    Reads TEMPO 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 TEMPO groups will be opened:
        - "product"
        - "geolocation"
        - "support_data"
        - "qa_statistics"
    variable_dict : dict, optional
        Dictionary mapping variable names to processing options (scale, minimum, maximum).
    **kwargs : dict
        Additional arguments passed to XarrayDriver.open.

    Returns
    -------
    xr.Dataset
        The TEMPO dataset.

    Examples
    --------
    Open standard NO2 product:
    >>> reader = TEMPOReader()
    >>> ds = reader.open_dataset(files="TEMPO_NO2_L2_*.nc")
    """
    if group is None:
        groups = ["product", "geolocation", "support_data", "qa_statistics"]
    elif isinstance(group, str):
        groups = [group]
    else:
        groups = group

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

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

    dsets = []
    for g in groups:
        g_kwargs = kwargs.copy()
        g_kwargs["group"] = g
        try:
            # Open without the preprocessor at this stage
            ds_g = super().open_dataset(files, **g_kwargs)
            dsets.append(ds_g)
        except Exception:
            # Not all groups may be present in all files
            continue

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

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

    # Now apply TEMPO preprocessing to the merged dataset
    ds = tempo_preprocess(ds, variable_dict=variable_dict)

    if user_preprocess:
        ds = user_preprocess(ds)

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

    return ds

tempo_preprocess(ds, variable_dict=None)

Preprocess TEMPO dataset: standardize coordinates, handle units, calculate pressure, and apply variable-specific transformations.

Parameters:

Name Type Description Default
ds Dataset

Input dataset with merged groups.

 required
variable_dict dict

Dictionary mapping variable names to processing options (scale, minimum, maximum).

 None

Returns:

Type Description
Dataset

Processed dataset.

Examples:

>>> ds = tempo_preprocess(ds, variable_dict={"no2_column": {"scale": 1e-15}})
Source code in monetio/readers/tempo.py 
 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
def tempo_preprocess(ds: xr.Dataset, variable_dict: dict | None = None) -> xr.Dataset:
    """
    Preprocess TEMPO dataset: standardize coordinates, handle units,
    calculate pressure, and apply variable-specific transformations.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset with merged groups.
    variable_dict : dict, optional
        Dictionary mapping variable names to processing options (scale, minimum, maximum).

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

    Examples
    --------
    >>> ds = tempo_preprocess(ds, variable_dict={"no2_column": {"scale": 1e-15}})
    """
    # 1. Map variables from groups to root if they are still nested/prefixed
    # (Though open_dataset already handles merging, some engines might prefix)
    mapping = {
        "geolocation/longitude": "longitude",
        "geolocation/latitude": "latitude",
        "geolocation/time": "time",
        "product/vertical_column_troposphere": "vertical_column_troposphere",
    }
    for old, new in mapping.items():
        if old in ds.variables and new not in ds.variables:
            ds = ds.rename({old: new})

    # 2. Standardize dimensions and coordinates
    # TEMPO uses 'x' and 'y' for across-track and along-track.
    # Vertical dims can be 'swt_level' or 'swt_level_stagg'.
    ds = standardize_satellite_coords(
        ds,
        lat_name="latitude",
        lon_name="longitude",
        z_dim=["swt_level", "swt_level_stagg", "level"],
    )

    # 3. Handle Unit Conversion (Lazy)
    # Convert surface_pressure from hPa to Pa if needed
    if "surface_pressure" in ds.variables:
        ps = ds["surface_pressure"]
        if ps.attrs.get("units") == "hPa":
            # Scale attributes if they exist
            new_attrs = ps.attrs.copy()
            new_attrs["units"] = "Pa"
            for attr in ["valid_min", "valid_max", "Eta_A"]:
                if attr in new_attrs:
                    new_attrs[attr] *= 100.0
            ds["surface_pressure"] = (ps * 100.0).assign_attrs(new_attrs)

    # 4. Calculate Pressure (Lazy)
    if variable_dict and "pressure" in variable_dict:
        ds = _add_pressure(ds)

    # 5. Apply variable-specific processing from variable_dict
    if variable_dict:
        for var, options in variable_dict.items():
            if var in ds.variables:
                # Scale
                if "scale" in options:
                    ds[var] = ds[var] * options["scale"]

                # Clipping
                if "minimum" in options:
                    ds[var] = ds[var].where(ds[var] >= options["minimum"])
                if "maximum" in options:
                    ds[var] = ds[var].where(ds[var] <= options["maximum"])

                # Quality Flagging
                if "quality_flag_max" in options and "main_data_quality_flag" in ds.variables:
                    qf = ds["main_data_quality_flag"]
                    ds[var] = ds[var].where(qf <= options["quality_flag_max"])

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

    return ds