Skip to content

mopitt

MOPITT Reader

MOPITTReader

Bases: GriddedReader

Reader for MOPITT (Measurements Of Pollution In The Troposphere) L3 data.

Source code in monetio/readers/mopitt.py 
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
@register_reader("mopitt")
class MOPITTReader(GriddedReader):
    """
    Reader for MOPITT (Measurements Of Pollution In The Troposphere) L3 data.
    """

    def open_dataset(
        self,
        files: str | list[str],
        **kwargs,
    ) -> xr.Dataset:
        """
        Reads MOPITT data.

        Parameters
        ----------
        files : Union[str, List[str]]
            File path(s) or URL(s).
        **kwargs : dict
            Additional arguments passed to XarrayDriver.open.

        Returns
        -------
        xr.Dataset
            The MOPITT dataset.
        """
        if "preprocess" not in kwargs:
            kwargs["preprocess"] = mopitt_preprocess

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

        ds = super().open_dataset(files, **kwargs)

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

        return ds

open_dataset(files, **kwargs)

Reads MOPITT data.

Parameters:

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

File path(s) or URL(s).

 required
**kwargs dict

Additional arguments passed to XarrayDriver.open.

 {}

Returns:

Type Description
Dataset

The MOPITT dataset.
Source code in monetio/readers/mopitt.py 
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
def open_dataset(
    self,
    files: str | list[str],
    **kwargs,
) -> xr.Dataset:
    """
    Reads MOPITT data.

    Parameters
    ----------
    files : Union[str, List[str]]
        File path(s) or URL(s).
    **kwargs : dict
        Additional arguments passed to XarrayDriver.open.

    Returns
    -------
    xr.Dataset
        The MOPITT dataset.
    """
    if "preprocess" not in kwargs:
        kwargs["preprocess"] = mopitt_preprocess

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

    ds = super().open_dataset(files, **kwargs)

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

    return ds

mopitt_preprocess(ds)

Preprocess MOPITT dataset: standardize coordinates, handle time, and calculate auxiliary variables (pressure, a priori profiles).

Parameters:

Name Type Description Default
ds Dataset

Input dataset from MOPITT L3 file.

 required

Returns:

Type Description
Dataset

Processed dataset with standard names and derived variables.

Examples:

>>> ds = xr.open_dataset("MOP03-20230101-L3V95.hdf", engine="h5netcdf")
>>> ds = mopitt_preprocess(ds)
Source code in monetio/readers/mopitt.py 
 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
def mopitt_preprocess(ds: xr.Dataset) -> xr.Dataset:
    """
    Preprocess MOPITT dataset: standardize coordinates, handle time, and
    calculate auxiliary variables (pressure, a priori profiles).

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset from MOPITT L3 file.

    Returns
    -------
    xr.Dataset
        Processed dataset with standard names and derived variables.

    Examples
    --------
    >>> ds = xr.open_dataset("MOP03-20230101-L3V95.hdf", engine="h5netcdf")
    >>> ds = mopitt_preprocess(ds)
    """
    # 1. Expand Mapping
    # MOPITT L3 HDF5 structure: /HDFEOS/GRIDS/MOP03/Data Fields/
    mapping = {
        "HDFEOS/GRIDS/MOP03/Data Fields/Latitude": "latitude",
        "HDFEOS/GRIDS/MOP03/Data Fields/Longitude": "longitude",
        "HDFEOS/GRIDS/MOP03/Data Fields/Pressure": "pressure_levels",
        "HDFEOS/GRIDS/MOP03/Data Fields/Pressure2": "pressure_levels_ak",
        "HDFEOS/GRIDS/MOP03/Data Fields/RetrievedCOTotalColumnDay": "co_column",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOTotalColumnDay": "apriori_co_column",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOSurfaceMixingRatioDay": "apriori_co_surf",
        "HDFEOS/GRIDS/MOP03/Data Fields/SurfacePressureDay": "surface_pressure",
        "HDFEOS/GRIDS/MOP03/Data Fields/TotalColumnAveragingKernelDay": "co_ak_column",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOMixingRatioProfileDay": "apriori_co_profile",
        # Night versions
        "HDFEOS/GRIDS/MOP03/Data Fields/RetrievedCOTotalColumnNight": "co_column_night",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOTotalColumnNight": "apriori_co_column_night",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOSurfaceMixingRatioNight": "apriori_co_surf_night",
        "HDFEOS/GRIDS/MOP03/Data Fields/SurfacePressureNight": "surface_pressure_night",
        "HDFEOS/GRIDS/MOP03/Data Fields/TotalColumnAveragingKernelNight": "co_ak_column_night",
        "HDFEOS/GRIDS/MOP03/Data Fields/APrioriCOMixingRatioProfileNight": "apriori_co_profile_night",
    }

    actual_rename = {old: new for old, new in mapping.items() if old in ds.variables}
    if actual_rename:
        ds = ds.rename(actual_rename)
        ds = update_history(ds, f"Renamed variables: {list(actual_rename.values())}")

    # Ensure pressure levels are coordinates
    for p_var in ["pressure_levels", "pressure_levels_ak"]:
        if p_var in ds.variables and p_var not in ds.coords:
            ds = ds.set_coords(p_var)

    # 2. Standardize
    ds = standardize_satellite_coords(ds)

    # 3. Handle time from attributes if missing
    if "time" not in ds.coords:
        # Check for StartTime in attributes
        start_time = ds.attrs.get("StartTime")
        if start_time is not None:
            if isinstance(start_time, list | np.ndarray):
                start_time = start_time[0]
            # MOPITT uses seconds since 1993-01-01
            # Wrap in DataArray to use standardized utility
            time_da = xr.DataArray([float(start_time)], dims=("time",))
            if "time" not in ds.dims:
                ds = ds.expand_dims("time")
            ds = ds.assign_coords(time=tai93_to_datetime(time_da))
            ds = update_history(ds, "Assigned time coordinate from StartTime attribute.")

    # 4. Handle Missing Values (Vectorized)
    with xr.set_options(keep_attrs=True):
        ds = ds.where(ds != MOPITT_MISSING)
    ds = update_history(ds, f"Applied vectorized missing value mask (value: {MOPITT_MISSING}).")

    # 5. Calculate Pressure (Lazy)
    if "co_ak_column" in ds.data_vars and "surface_pressure" in ds.data_vars:
        ds = _add_mopitt_pressure(ds)

    # 6. Combine A Priori (Lazy)
    if (
        "apriori_co_profile" in ds.data_vars
        and "apriori_co_surf" in ds.data_vars
        and "pressure_levels_ak" in ds.coords
    ):
        ds = _combine_mopitt_apriori(ds)

    # Update history
    ds = update_history(ds, "Preprocessed MOPITT L3 data using standardized preprocessing.")

    return ds