Skip to content

sat_utils

Satellite Reader Utilities

add_time_coord(ds, time_val=None, time_attr=None)

Add a time dimension and coordinate to the dataset.

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
time_val datetime

Explicit time value to use.

 None
time_attr str

Attribute name to extract time from if time_val is None.

 None

Returns:

Type Description
Dataset

Dataset with 'time' dimension and coordinate.
Source code in monetio/readers/sat_utils.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
def add_time_coord(
    ds: xr.Dataset,
    time_val: datetime.datetime | None = None,
    time_attr: str | None = None,
) -> xr.Dataset:
    """
    Add a time dimension and coordinate to the dataset.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset.
    time_val : datetime.datetime, optional
        Explicit time value to use.
    time_attr : str, optional
        Attribute name to extract time from if time_val is None.

    Returns
    -------
    xr.Dataset
        Dataset with 'time' dimension and coordinate.
    """
    if time_val is None and time_attr is not None:
        if time_attr in ds.attrs:
            try:
                time_val = pd.to_datetime(ds.attrs[time_attr])
            except (ValueError, TypeError):
                pass

    if time_val is not None:
        if isinstance(time_val, str):
            time_val = pd.to_datetime(time_val)
        ds = ds.expand_dims("time")
        ds["time"] = [time_val]

    return ds

apply_lazy_conversion(data, func, output_dtype)

Apply a conversion function lazily to a DataArray backend-agnostic.

Parameters:

Name Type Description Default
data DataArray

Input DataArray.

 required
func Callable

Function to apply. Should work on NumPy arrays.

 required
output_dtype Union[str, dtype, type]

Expected output dtype.

 required

Returns:

Type Description
DataArray

Converted DataArray.
Source code in monetio/readers/sat_utils.py 
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
def apply_lazy_conversion(
    data: xr.DataArray, func: Callable, output_dtype: str | np.dtype | type
) -> xr.DataArray:
    """
    Apply a conversion function lazily to a DataArray backend-agnostic.

    Parameters
    ----------
    data : xr.DataArray
        Input DataArray.
    func : Callable
        Function to apply. Should work on NumPy arrays.
    output_dtype : Union[str, np.dtype, type]
        Expected output dtype.

    Returns
    -------
    xr.DataArray
        Converted DataArray.
    """

    def _wrapped_func(x):
        res = func(x)
        # Ensure result is a NumPy array to avoid issues with Dask/Xarray
        # (e.g. DatetimeIndex causing transpose errors)
        if hasattr(res, "to_numpy"):
            return res.to_numpy()
        return np.asarray(res)

    return xr.apply_ufunc(
        _wrapped_func,
        data,
        dask="parallelized",
        output_dtypes=[output_dtype],
    )

apply_qa_mask(ds, qa_var='qa_value', threshold=0.5)

Apply quality flag masking to a dataset.

Masks all data variables where the QA variable is below the threshold, while preserving the QA variable itself and all coordinates.

Parameters:

Name Type Description Default
ds Dataset

Input dataset containing a QA variable.

 required
qa_var str

Name of the quality flag variable.

 'qa_value'
threshold float

Minimum acceptable quality value.

 0.5

Returns:

Type Description
Dataset

Dataset with low-quality values masked as NaN.
Source code in monetio/readers/sat_utils.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
def apply_qa_mask(ds: xr.Dataset, qa_var: str = "qa_value", threshold: float = 0.5) -> xr.Dataset:
    """Apply quality flag masking to a dataset.

    Masks all data variables where the QA variable is below the threshold,
    while preserving the QA variable itself and all coordinates.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset containing a QA variable.
    qa_var : str
        Name of the quality flag variable.
    threshold : float
        Minimum acceptable quality value.

    Returns
    -------
    xr.Dataset
        Dataset with low-quality values masked as NaN.
    """
    if qa_var not in ds.data_vars:
        return ds
    qa = ds[qa_var]
    mask = qa >= threshold
    ds = ds.where(mask)
    ds[qa_var] = qa  # Restore unmasked QA values
    return ds

convert_ppmv_to_ppbv(ds, variables=None)

Convert gas-phase variables from ppmV to ppbV (multiply by 1000).

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
variables list of str

Variables to convert. If None, converts all variables with units containing 'ppm'.

 None

Returns:

Type Description
Dataset

Dataset with converted units.
Source code in monetio/readers/sat_utils.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def convert_ppmv_to_ppbv(ds: xr.Dataset, variables: list[str] | None = None) -> xr.Dataset:
    """Convert gas-phase variables from ppmV to ppbV (multiply by 1000).

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset.
    variables : list of str, optional
        Variables to convert. If None, converts all variables with
        units containing 'ppm'.

    Returns
    -------
    xr.Dataset
        Dataset with converted units.
    """
    if variables is None:
        variables = [v for v in ds.data_vars if "ppm" in ds[v].attrs.get("units", "").lower()]
    for var in variables:
        if var in ds.data_vars:
            ds[var] = ds[var] * 1000.0
            if "units" in ds[var].attrs:
                ds[var].attrs["units"] = ds[var].attrs["units"].replace("ppm", "ppb")
    return ds

jpss_time_to_datetime(time_array, origin='1958-01-01', unit='us')

Convert JPSS time (usually microseconds since 1958) to datetime64[ns].

Parameters:

Name Type Description Default
time_array DataArray

Input time array.

 required
origin str

Origin date, by default "1958-01-01".

 '1958-01-01'
unit str

Time unit, by default "us" (microseconds).

 'us'

Returns:

Type Description
DataArray

Time array in datetime64[ns].
Source code in monetio/readers/sat_utils.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def jpss_time_to_datetime(
    time_array: xr.DataArray, origin: str = "1958-01-01", unit: str = "us"
) -> xr.DataArray:
    """
    Convert JPSS time (usually microseconds since 1958) to datetime64[ns].

    Parameters
    ----------
    time_array : xr.DataArray
        Input time array.
    origin : str, optional
        Origin date, by default "1958-01-01".
    unit : str, optional
        Time unit, by default "us" (microseconds).

    Returns
    -------
    xr.DataArray
        Time array in datetime64[ns].
    """

    def _convert(t):
        return pd.to_datetime(t, unit=unit, origin=origin)

    return apply_lazy_conversion(time_array, _convert, "datetime64[ns]")

lazy_index_along_axis(data, index, dim)

Index a dimension using another DataArray lazily, handling both Eager and Dask. Fixes the 'vindex does not support indexing with dask objects' limitation.

Parameters:

Name Type Description Default
data DataArray

DataArray to index. Must have dimension dim.

 required
index DataArray

DataArray of indices.

 required
dim str

Dimension name to index along.

 required

Returns:

Type Description
DataArray

The indexed DataArray.
Source code in monetio/readers/sat_utils.py 
 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
def lazy_index_along_axis(data: xr.DataArray, index: xr.DataArray, dim: str) -> xr.DataArray:
    """
    Index a dimension using another DataArray lazily, handling both Eager and Dask.
    Fixes the 'vindex does not support indexing with dask objects' limitation.

    Parameters
    ----------
    data : xr.DataArray
        DataArray to index. Must have dimension `dim`.
    index : xr.DataArray
        DataArray of indices.
    dim : str
        Dimension name to index along.

    Returns
    -------
    xr.DataArray
        The indexed DataArray.
    """

    def _index_func(arr, idx):
        # In apply_ufunc with input_core_dims=[[dim], []], the core dimension
        # is moved to the last axis of arr.
        # arr shape: (..., dim_size)
        # idx shape: (...)
        idx_expanded = idx[..., np.newaxis]
        return np.take_along_axis(arr, idx_expanded, axis=-1).squeeze(axis=-1)

    return xr.apply_ufunc(
        _index_func,
        data,
        index,
        input_core_dims=[[dim], []],
        output_core_dims=[[]],
        dask="parallelized",
        output_dtypes=[data.dtype],
        dask_gufunc_kwargs={"allow_rechunk": True},
    )

standardize_satellite_coords(ds, lat_name='Latitude', lon_name='Longitude', y_dim=['Rows', 'scanline', 'nlat', 'lat', 'nscan', 'nTimes'], x_dim=['Columns', 'ground_pixel', 'nlon', 'lon', 'nstep', 'nIFOV'], z_dim=['Levels', 'layer', 'level', 'nLayer'], time_name='Time')

Standardize satellite swath/gridded coordinates and dimensions.

Parameters:

Name Type Description Default
ds Dataset

Input dataset.

 required
lat_name str

Name of the latitude coordinate in the file, by default "Latitude".

 'Latitude'
lon_name str

Name of the longitude coordinate in the file, by default "Longitude".

 'Longitude'
y_dim str or list of str

Name(s) of the y/row dimension in the file, by default ["Rows", "scanline"].

 ['Rows', 'scanline', 'nlat', 'lat', 'nscan', 'nTimes']
x_dim str or list of str

Name(s) of the x/column dimension in the file, by default ["Columns", "ground_pixel"].

 ['Columns', 'ground_pixel', 'nlon', 'lon', 'nstep', 'nIFOV']

Returns:

Type Description
Dataset

Dataset with standardized dimensions (y, x) and coordinates (latitude, longitude).
Source code in monetio/readers/sat_utils.py 
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
223
224
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
def standardize_satellite_coords(
    ds: xr.Dataset,
    lat_name: str = "Latitude",
    lon_name: str = "Longitude",
    y_dim: str | list[str] = ["Rows", "scanline", "nlat", "lat", "nscan", "nTimes"],
    x_dim: str | list[str] = ["Columns", "ground_pixel", "nlon", "lon", "nstep", "nIFOV"],
    z_dim: str | list[str] = ["Levels", "layer", "level", "nLayer"],
    time_name: str = "Time",
) -> xr.Dataset:
    """
    Standardize satellite swath/gridded coordinates and dimensions.

    Parameters
    ----------
    ds : xr.Dataset
        Input dataset.
    lat_name : str, optional
        Name of the latitude coordinate in the file, by default "Latitude".
    lon_name : str, optional
        Name of the longitude coordinate in the file, by default "Longitude".
    y_dim : str or list of str, optional
        Name(s) of the y/row dimension in the file, by default ["Rows", "scanline"].
    x_dim : str or list of str, optional
        Name(s) of the x/column dimension in the file, by default ["Columns", "ground_pixel"].

    Returns
    -------
    xr.Dataset
        Dataset with standardized dimensions (y, x) and coordinates (latitude, longitude).
    """
    if isinstance(y_dim, str):
        y_dim = [y_dim]
    if isinstance(x_dim, str):
        x_dim = [x_dim]

    rename_dict = {}
    for y in y_dim:
        if y in ds.dims:
            rename_dict[y] = "y"
            break
    for x in x_dim:
        if x in ds.dims:
            rename_dict[x] = "x"
            break
    for z in z_dim:
        if z in ds.dims:
            rename_dict[z] = "z"
            break

    if rename_dict:
        ds = ds.rename(rename_dict)

    coord_rename = {}
    # Case insensitive search for lat/lon if not found exactly
    # Try to find latitude/longitude
    actual_lat = None
    lat_names = [lat_name, "latitude", "lat", "LAT", "Latitude"]
    for ln in lat_names:
        if ln in ds.variables:
            actual_lat = ln
            break
    if actual_lat is None:
        for v in ds.variables:
            if v.lower() in ["latitude", "lat"]:
                actual_lat = v
                break

    actual_lon = None
    lon_names = [lon_name, "longitude", "lon", "LON", "Longitude"]
    for ln in lon_names:
        if ln in ds.variables:
            actual_lon = ln
            break
    if actual_lon is None:
        for v in ds.variables:
            if v.lower() in ["longitude", "lon"]:
                actual_lon = v
                break

    if actual_lat and actual_lat != "latitude":
        coord_rename[actual_lat] = "latitude"
    if actual_lon and actual_lon != "longitude":
        coord_rename[actual_lon] = "longitude"

    if coord_rename:
        ds = ds.rename(coord_rename)

    # Ensure they are coordinates
    to_set = []
    if "latitude" in ds.variables and "latitude" not in ds.coords:
        to_set.append("latitude")
    if "longitude" in ds.variables and "longitude" not in ds.coords:
        to_set.append("longitude")
    if to_set:
        ds = ds.set_coords(to_set)

    if "latitude" in ds.coords:
        ds["latitude"].attrs.update({"units": "degrees_north", "standard_name": "latitude"})
    if "longitude" in ds.coords:
        ds["longitude"].attrs.update({"units": "degrees_east", "standard_name": "longitude"})

    # Handle Time
    if time_name in ds.variables and "time" not in ds.variables:
        ds = ds.rename({time_name: "time"})
    elif "time" not in ds.variables:
        for v in ds.variables:
            if v.lower() == "time":
                ds = ds.rename({v: "time"})
                break

    if "time" in ds.variables and "time" not in ds.coords:
        # If it's a coordinate-like variable, set it
        if "time" in ds.dims or ds["time"].ndim == 1:
            ds = ds.set_coords("time")

    return ds

tai93_to_datetime(time_array)

Convert TAI93 time (seconds since 1993-01-01) to datetime64[ns].

Parameters:

Name Type Description Default
time_array DataArray

Input time array in seconds since 1993-01-01 00:00:00 UTC.

 required

Returns:

Type Description
DataArray

Time array in datetime64[ns].

Examples:

>>> ds["time"] = tai93_to_datetime(ds["Scan_Start_Time"])
Source code in monetio/readers/sat_utils.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def tai93_to_datetime(time_array: xr.DataArray) -> xr.DataArray:
    """
    Convert TAI93 time (seconds since 1993-01-01) to datetime64[ns].

    Parameters
    ----------
    time_array : xr.DataArray
        Input time array in seconds since 1993-01-01 00:00:00 UTC.

    Returns
    -------
    xr.DataArray
        Time array in datetime64[ns].

    Examples
    --------
    >>> ds["time"] = tai93_to_datetime(ds["Scan_Start_Time"])
    """

    def _convert(t):
        # pd.to_datetime expects 1D input
        return pd.to_datetime(t.ravel(), unit="s", origin="1993-01-01").values.reshape(t.shape)

    return apply_lazy_conversion(time_array, _convert, "datetime64[ns]")

update_history(ds, message)

Update the 'history' attribute of a dataset or dataframe backend-agnostic.

Parameters:

Name Type Description Default
ds xarray.Dataset, xarray.DataArray, pandas.DataFrame, or dask.DataFrame

Input object.

 required
message str

Message to add to history.

 required

Returns:

Type Description
object

The input object with updated history.
Source code in monetio/readers/sat_utils.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def update_history(ds: Union[xr.Dataset, xr.DataArray, pd.DataFrame, "dd.DataFrame"], message: str):
    """
    Update the 'history' attribute of a dataset or dataframe backend-agnostic.

    Parameters
    ----------
    ds : xarray.Dataset, xarray.DataArray, pandas.DataFrame, or dask.DataFrame
        Input object.
    message : str
        Message to add to history.

    Returns
    -------
    object
        The input object with updated history.
    """
    if not hasattr(ds, "attrs"):
        # For dask objects that don't have attrs yet or other edge cases
        return ds

    history = f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: {message}"
    try:
        if "history" in ds.attrs:
            ds.attrs["history"] = f"{ds.attrs['history']}\n{history}"
        else:
            ds.attrs["history"] = history
    except (AttributeError, TypeError):
        # Some dask-backed objects might have 'attrs' but it might be read-only or similar
        pass
    return ds