Skip to content

API reference#

earthcarekit.filter

Dataset filtering utilities.

Notes#

This module depends on other internal modules:

filter_frame #

filter_frame(
    ds: Dataset,
    frame_id: str | None = None,
    along_track_dim: str = ALONG_TRACK_DIM,
    lat_var: str = TRACK_LAT_VAR,
    add_trim_index_offset_var: bool = True,
    trim_index_offset_var: str = "trim_index_offset",
) -> Dataset

Trims the dataset to the region within the latitude frame bounds.

Parameters:

Name Type Description Default
ds Dataset

Input dataset to be trimmed.

 required
frame_id str | None

EarthCARE frame (single character between "A" and "H"). If given, speeds up trimming.

 None
along_track_dim str

Dimension along which to trim.

 ALONG_TRACK_DIM
lat_var str

Name of the latitude variable.

 TRACK_LAT_VAR
add_trim_index_offset_var bool

Whether the index offset between the original and trimmed dataset should be stored in the trimmed dataset (variable: "trim_index_offset").

 True
trim_index_offset_var str

Variable tracking index offsets from trimming/filtering.

 'trim_index_offset'

Returns:

Type Description
Dataset

Filtered dataset containing only samples within the EarthCARE frame.

Examples:

>>> import earthcarekit as eck
>>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True, trim_to_frame=False)
>>> ds_filtered = eck.filter_frame(ds)
>>> print("non-trimmed:", ds.along_track.shape)
>>> print("timmed:", ds_filtered.along_track.shape)
non-trimmed: (4992,)
timmed: (4964,)
Source code in earthcarekit/filter/_frame.py 
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
def filter_frame(
    ds: Dataset,
    frame_id: str | None = None,
    along_track_dim: str = ALONG_TRACK_DIM,
    lat_var: str = TRACK_LAT_VAR,
    add_trim_index_offset_var: bool = True,
    trim_index_offset_var: str = "trim_index_offset",
) -> Dataset:
    """Trims the dataset to the region within the latitude frame bounds.

    Args:
        ds: Input dataset to be trimmed.
        frame_id: EarthCARE frame (single character between "A" and "H"). If given, speeds up trimming.
        along_track_dim: Dimension along which to trim.
        lat_var: Name of the latitude variable.
        add_trim_index_offset_var: Whether the index offset between the original and trimmed dataset should be stored in the trimmed dataset (variable: "trim_index_offset").
        trim_index_offset_var: Variable tracking index offsets from trimming/filtering.

    Returns:
        Filtered dataset containing only samples within the EarthCARE frame.

    Examples:
        >>> import earthcarekit as eck
        >>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True, trim_to_frame=False)
        >>> ds_filtered = eck.filter_frame(ds)
        >>> print("non-trimmed:", ds.along_track.shape)
        >>> print("timmed:", ds_filtered.along_track.shape)
        non-trimmed: (4992,)
        timmed: (4964,)
    """
    slice_tuple = get_frame_index_range(
        frame_id=frame_id,
        ds=ds,
        lat_var=lat_var,
    )
    ds = ds.isel({along_track_dim: slice(*slice_tuple)})
    if add_trim_index_offset_var and slice_tuple[0] > 0:
        ds = insert_var(
            ds=ds,
            var=trim_index_offset_var,
            data=int(slice_tuple[0]),
            index=0,
            after_var="processing_start_time",
        )
        ds[trim_index_offset_var] = ds[trim_index_offset_var].assign_attrs(
            {
                "earthcarekit": "Added by earthcarekit: Used to calculate the index in the original, untrimmed dataset, i.e. by addition."
            }
        )
    return ds

filter_index #

filter_index(
    ds: Dataset,
    index: int | slice | tuple[_Start, _Stop] | NDArray | Sequence,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
) -> Dataset

Filters a dataset given an along-track index number, list/array or range/slice.

Parameters:

Name Type Description Default
ds Dataset

Input dataset with along-track dimension.

 required
index int | slice | tuple[_Start, _Stop] | NDArray | Sequence

Index, indices or slice to filter.

 required
along_track_dim str

Dimension along which to apply filtering.

 ALONG_TRACK_DIM
trim_index_offset_var str

Variable tracking index offsets from trimming/filtering.

 'trim_index_offset'
pad_idxs int

Number of additional samples added at both ends.

 0

Returns:

Type Description
Dataset

Filtered dataset.

Examples:

>>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
>>> ds_filtered = eck.filter_index(ds, 123)
>>> print(ds_filtered.sizes)
Frozen({'along_track': 1, 'vertical': 218})

>>> ds_filtered = eck.filter_index(ds, slice(0, 1000))
>>> print(ds_filtered.sizes)
Frozen({'along_track': 1000, 'vertical': 218})

A 2-element tuple is interpreted as (start, stop) and treated like a slice():

>>> ds_filtered = eck.filter_index(ds, (0, 1000))
>>> print(ds_filtered.sizes)
Frozen({'along_track': 1000, 'vertical': 218})

Any other n-element tuple is treated as a sequence of indices to select:

>>> ds_filtered = eck.filter_index(ds, (0, 1000, 2000))
>>> print(ds_filtered.sizes)
Frozen({'along_track': 3, 'vertical': 218})

Finally, any other sequence is treated as a sequence of indices to select:

>>> ds_filtered = eck.filter_index(ds, [0, 1000])
>>> print(ds_filtered.sizes)
Frozen({'along_track': 2, 'vertical': 218})
Source code in earthcarekit/filter/_filter_index.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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def filter_index(
    ds: Dataset,
    index: int | slice | tuple[_Start, _Stop] | NDArray | Sequence,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
) -> Dataset:
    """
    Filters a dataset given an along-track index number, list/array or range/slice.

    Args:
        ds: Input dataset with along-track dimension.
        index: Index, indices or slice to filter.
        along_track_dim: Dimension along which to apply filtering.
        trim_index_offset_var: Variable tracking index offsets from trimming/filtering.
        pad_idxs: Number of additional samples added at both ends.

    Returns:
        Filtered dataset.

    Examples:
        >>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
        >>> ds_filtered = eck.filter_index(ds, 123)
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 1, 'vertical': 218})

        >>> ds_filtered = eck.filter_index(ds, slice(0, 1000))
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 1000, 'vertical': 218})

        A 2-element tuple is interpreted as ``(start, stop)`` and treated like a [``slice()``](https://docs.python.org/3/library/functions.html#slice):

        >>> ds_filtered = eck.filter_index(ds, (0, 1000))
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 1000, 'vertical': 218})

        Any other n-element tuple is treated as a sequence of indices to select:

        >>> ds_filtered = eck.filter_index(ds, (0, 1000, 2000))
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 3, 'vertical': 218})

        Finally, any other sequence is treated as a sequence of indices to select:

        >>> ds_filtered = eck.filter_index(ds, [0, 1000])
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 2, 'vertical': 218})
    """
    if isinstance(index, tuple) and len(index) == 2:
        index = slice(int(index[0]), int(index[1]))

    if isinstance(index, (Sequence, np.ndarray)):
        if len(index) == 0:
            raise ValueError("index must be integer or non-empty array")
        elif len(index) == 1:
            index = int(index[0])

    if isinstance(index, (int, np.integer)):
        index = slice(index, index + 1)

    if isinstance(index, slice):
        index = slice(index.start - pad_idxs, index.stop + pad_idxs, index.step)
    else:
        index = flatten_array(index)

    ds_new = ds.copy().isel({along_track_dim: index})
    new_trim_index_offset: int | NDArray = 0

    if isinstance(index, slice):
        if isinstance(index.step, int) and index.step > 1:
            new_trim_index_offset = np.array(list(range(index.start, index.stop, index.step)))
        else:
            new_trim_index_offset = int(index.start)

    if isinstance(index, np.ndarray):
        if np.max(np.diff(index)) > 1:
            new_trim_index_offset = index
        else:
            new_trim_index_offset = int(index[0])

    ds_new = update_trim_index_offset(
        ds=ds_new,
        offset=int(np.atleast_1d(new_trim_index_offset)[0]),
        var=trim_index_offset_var,
    )

    return ds_new

filter_latitude #

filter_latitude(
    ds: Dataset,
    lat_range: NumberPairNoneLike,
    start_before_pole: bool = True,
    end_before_pole: bool = True,
    only_center: bool = False,
    lat_var: str = TRACK_LAT_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
    time_var: str = TIME_VAR,
) -> Dataset

Filters a dataset to include only points within a specified latitude range.

Parameters:

Name Type Description Default
ds Dataset

Input dataset with geolocation data.

 required
lat_range NumberPairNoneLike

A pair of latitude values (min_lat, max_lat) defining the selection range.

 required
start_before_pole bool

If True, selection starts before the pole when the track crosses one.

 True
end_before_pole bool

If True, selection ends before the pole when the track crosses one.

 True
only_center bool

If True, only the sample at the center index of selection is returned.

 False
lat_var str

Name of the latitude variable.

 TRACK_LAT_VAR
along_track_dim str

Dimension along which to apply filtering.

 ALONG_TRACK_DIM
trim_index_offset_var str

Variable tracking index offsets from trimming/filtering.

 'trim_index_offset'
pad_idxs int

Number of additional samples added at both ends.

 0
shift_idxs int

Offset number to shift selection of samples.

 0
pad_time TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None

Additional time padding; applied before index-based padding (pad_idxs, shift_idxs).

 None
time_var str

Name of the time variable in ds.

 TIME_VAR

Raises:

Type Description
ValueError

If selection is empty.

Returns:

Type Description
Dataset

Filtered dataset containing only points within the specified latitude range.

Examples:

>>> import numpy as np
>>> import earthcarekit as eck
>>> np.set_printoptions(precision=3)
>>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
>>> print(ds.latitude.values)
[-22.503 -22.512 -22.521 ... -67.482 -67.491 -67.499]

>>> ds_filtered = eck.filter_latitude(ds, (-40, -30))
>>> print(ds_filtered.latitude.values)
[-30.004 -30.013 -30.021 ... -39.981 -39.99  -39.998]
Source code in earthcarekit/filter/_filter_latitude.py 
 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
def filter_latitude(
    ds: xr.Dataset,
    lat_range: NumberPairNoneLike,
    start_before_pole: bool = True,
    end_before_pole: bool = True,
    only_center: bool = False,
    lat_var: str = TRACK_LAT_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
    time_var: str = TIME_VAR,
) -> xr.Dataset:
    """
    Filters a dataset to include only points within a specified latitude range.

    Args:
        ds: Input dataset with geolocation data.
        lat_range: A pair of latitude values (min_lat, max_lat) defining the selection range.
        start_before_pole: If True, selection starts before the pole when the track crosses one.
        end_before_pole: If True, selection ends before the pole when the track crosses one.
        only_center: If True, only the sample at the center index of selection is returned.
        lat_var: Name of the latitude variable.
        along_track_dim: Dimension along which to apply filtering.
        trim_index_offset_var: Variable tracking index offsets from trimming/filtering.
        pad_idxs: Number of additional samples added at both ends.
        shift_idxs: Offset number to shift selection of samples.
        pad_time: Additional time padding; applied before index-based padding (`pad_idxs`, `shift_idxs`).
        time_var: Name of the time variable in `ds`.

    Raises:
        ValueError: If selection is empty.

    Returns:
        Filtered dataset containing only points within the specified latitude range.

    Examples:
        >>> import numpy as np
        >>> import earthcarekit as eck
        >>> np.set_printoptions(precision=3)
        >>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
        >>> print(ds.latitude.values)
        [-22.503 -22.512 -22.521 ... -67.482 -67.491 -67.499]

        >>> ds_filtered = eck.filter_latitude(ds, (-40, -30))
        >>> print(ds_filtered.latitude.values)
        [-30.004 -30.013 -30.021 ... -39.981 -39.99  -39.998]
    """
    lats = ds[lat_var].values
    satellite_crosses_pole, is_first_increase, mask_before_pole, mask_after_pole = (
        _get_pole_crossing_masks(lats)
    )
    lat_range = validate_numeric_pair(lat_range, fallback=(lats[0], lats[-1]))

    lats_mask: NDArray[np.bool_] = (lats >= np.min(lat_range)) & (lats <= np.max(lat_range))

    if satellite_crosses_pole and start_before_pole and not end_before_pole:
        if is_first_increase:
            mask_from_start = lats >= lat_range[0]
            mask_from_end = lats >= lat_range[1]
        else:
            mask_from_start = lats <= lat_range[0]
            mask_from_end = lats <= lat_range[1]

        mask_from_start_before_pole = np.logical_and(mask_before_pole, mask_from_start)
        mask_from_end_after_pole = np.logical_and(mask_after_pole, mask_from_end)

        mask = np.logical_or(mask_from_start_before_pole, mask_from_end_after_pole)
    elif satellite_crosses_pole and start_before_pole and end_before_pole:
        mask = np.logical_and(lats_mask, mask_before_pole)
    elif satellite_crosses_pole and not start_before_pole:
        mask = np.logical_and(lats_mask, mask_after_pole)
    else:
        mask = lats_mask

    if only_center:
        mask_true_idxs = np.where(mask)[0]
        if len(mask_true_idxs) > 0:
            idx_center = mask_true_idxs[len(mask_true_idxs) // 2]
            mask[:] = False
            mask[idx_center] = True

    mask = _pad_mask(
        ds=ds,
        mask=mask,
        pad_idxs=pad_idxs,
        shift_idxs=shift_idxs,
        pad_time=pad_time,
        time_var=time_var,
    )

    if np.sum(mask) == 0:
        msg = f"No data falls into the given latitude range!\nIn the dataset latitude falls between {np.min(lats)} and {np.max(lats)}.\n"
        if satellite_crosses_pole:
            msg += "Note that the satellite crosses a pole (set `start_before_pole` and `end_before_pole`\nto clarify how the start and end of the range should be interpreted)."
        else:
            msg += "The satellite is not crossing a pole."
        raise ValueError(msg)

    da_mask: xr.DataArray = xr.DataArray(mask, dims=[along_track_dim], name=lat_var)

    ds_new: xr.Dataset = xr.Dataset(
        {
            var: (
                ds[var].copy().where(da_mask, drop=True)
                if along_track_dim in ds[var].dims
                else ds[var].copy()
            )
            for var in ds.data_vars
        }
    )
    ds_new.attrs = ds.attrs.copy()
    ds_new.encoding = ds.encoding.copy()

    ds_new = update_trim_index_offset(
        ds=ds_new,
        offset=int(np.argmax(mask)),
        var=trim_index_offset_var,
    )

    return ds_new

filter_radius #

filter_radius(
    ds: Dataset,
    radius_km: float = 100.0,
    center_lat: float | None = None,
    center_lon: float | None = None,
    site: SiteLike | None = None,
    lat_var: str = TRACK_LAT_VAR,
    lon_var: str = TRACK_LON_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    method: Literal["geodesic", "haversine"] = "geodesic",
    closest: bool = False,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
    time_var: str = TIME_VAR,
) -> Dataset

Filters a dataset to include only points within a specified radius of a geographic location.

Parameters:

Name Type Description Default
ds Dataset

Input dataset with geolocation data.

 required
radius_km float

Radius (in kilometers) around the center location.

 100.0
site SiteLike | None

Site object or name; center location retrieved if provided.

 None
center_lat float | None

Latitude of the center point; required if site is not set.

 None
center_lon float | None

Longitude of the center point; required if site is not set.

 None
lat_var str

Name of the latitude variable.

 TRACK_LAT_VAR
lon_var str

Name of the longitude variable.

 TRACK_LON_VAR
along_track_dim str

Dimension along which to apply filtering.

 ALONG_TRACK_DIM
method Literal['geodesic', 'haversine']

Distance calculation method.

 'geodesic'
closest bool

If True, returns only the single closest sample; otherwise returns all within radius.

 False
trim_index_offset_var str

Variable tracking index offsets from trimming/filtering.

 'trim_index_offset'
pad_idxs int

Number of additional samples added at both ends.

 0
shift_idxs int

Offset to shift the selected sample indices.

 0
pad_time TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None

Additional time padding applied before index-based padding; applied first.

 None
time_var str

Name of the time variable in ds.

 TIME_VAR

Returns:

Type Description
Dataset

Filtered dataset containing only points within the specified radius.

Raises:

Type Description
EmptyFilterResultError

If no data points are found within the radius.
ValueError

If the method is invalid.

Examples:

>>> import earthcarekit as eck
>>> ds = eck.ecload("ATL_EBD_2A", "01508B", download=True)
>>> print(ds.sizes)
Frozen({'along_track': 5143, 'vertical': 242, 'layer': 25, 'n_state': 351})

>>> ds_filtered = eck.filter_radius(ds, site="dushanbe")
>>> print(ds_filtered.sizes)
Frozen({'along_track': 197, 'vertical': 242, 'layer': 25, 'n_state': 351})

>>> ds_filtered = eck.filter_radius(ds, site="dushanbe", radius_km=200)
>>> print(ds_filtered.sizes)
Frozen({'along_track': 399, 'vertical': 242, 'layer': 25, 'n_state': 351})
Source code in earthcarekit/filter/_filter_radius.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
 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
def filter_radius(
    ds: xr.Dataset,
    radius_km: float = 100.0,
    center_lat: float | None = None,
    center_lon: float | None = None,
    site: SiteLike | None = None,
    lat_var: str = TRACK_LAT_VAR,
    lon_var: str = TRACK_LON_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    method: Literal["geodesic", "haversine"] = "geodesic",
    closest: bool = False,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
    time_var: str = TIME_VAR,
) -> xr.Dataset:
    """Filters a dataset to include only points within a specified radius of a geographic location.

    Args:
        ds: Input dataset with geolocation data.
        radius_km: Radius (in kilometers) around the center location.
        site: Site object or name; center location retrieved if provided.
        center_lat: Latitude of the center point; required if `site` is not set.
        center_lon: Longitude of the center point; required if `site` is not set.
        lat_var: Name of the latitude variable.
        lon_var: Name of the longitude variable.
        along_track_dim: Dimension along which to apply filtering.
        method: Distance calculation method.
        closest: If True, returns only the single closest sample; otherwise returns all within radius.
        trim_index_offset_var: Variable tracking index offsets from trimming/filtering.
        pad_idxs: Number of additional samples added at both ends.
        shift_idxs: Offset to shift the selected sample indices.
        pad_time: Additional time padding applied before index-based padding; applied first.
        time_var: Name of the time variable in `ds`.

    Returns:
        Filtered dataset containing only points within the specified radius.

    Raises:
        EmptyFilterResultError: If no data points are found within the radius.
        ValueError: If the `method` is invalid.

    Examples:
        >>> import earthcarekit as eck
        >>> ds = eck.ecload("ATL_EBD_2A", "01508B", download=True)
        >>> print(ds.sizes)
        Frozen({'along_track': 5143, 'vertical': 242, 'layer': 25, 'n_state': 351})

        >>> ds_filtered = eck.filter_radius(ds, site="dushanbe")
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 197, 'vertical': 242, 'layer': 25, 'n_state': 351})

        >>> ds_filtered = eck.filter_radius(ds, site="dushanbe", radius_km=200)
        >>> print(ds_filtered.sizes)
        Frozen({'along_track': 399, 'vertical': 242, 'layer': 25, 'n_state': 351})
    """
    _center_lat: float
    _center_lon: float

    if isinstance(site, str):
        site = get_site(site)

    if isinstance(site, Site):
        _center_lat = site.latitude
        _center_lon = site.longitude
    elif isinstance(center_lat, (int, float, np.integer, np.floating)) and isinstance(
        center_lon, (int, float, np.integer, np.floating)
    ):
        _center_lat = float(center_lat)
        _center_lon = float(center_lon)
    else:
        raise ValueError("Either 'site' or 'center_lat' and 'center_lon' must be given.")

    if method not in ["geodesic", "haversine"]:
        raise ValueError(r'Invalid method choosen. Available methods: {"geodesic", "haversine"}')

    satellite_coords = get_coords(ds, lat_var=lat_var, lon_var=lon_var)

    center_coords = (_center_lat, _center_lon)

    if method == "geodesic":
        distances = geodesic(center_coords, satellite_coords)
    else:
        distances = haversine(center_coords, satellite_coords)

    mask = np.array(distances < radius_km)

    if closest:
        closest_distance = np.min(distances)
        closest_filtered_index = int(np.argmin(np.abs(distances - closest_distance)))
        mask[:] = False
        mask[closest_filtered_index] = True

    mask = _pad_mask(
        ds=ds,
        mask=mask,
        pad_idxs=pad_idxs,
        shift_idxs=shift_idxs,
        pad_time=pad_time,
        time_var=time_var,
    )

    da_mask = xr.DataArray(data=mask, dims=[along_track_dim])
    if np.sum(da_mask.values) < 1:
        raise EmptyFilterResultError(
            f"Could not find valid overpass for given inputs. Data lies outside the given {radius_km} km radius around ({center_lat} degN {center_lon} degE).",
            min_distance=float(np.min(distances)),
        )

    ds_new: xr.Dataset = xr.Dataset(
        {
            var: (
                ds[var].copy().where(da_mask, drop=True)
                if along_track_dim in ds[var].dims
                else ds[var].copy()
            )
            for var in ds.data_vars
        }
    )
    ds_new.attrs = ds.attrs.copy()
    ds_new.encoding = ds.encoding.copy()

    ds_new = update_trim_index_offset(
        ds=ds_new,
        offset=int(np.argmax(mask)),
        var=trim_index_offset_var,
    )

    return ds_new

filter_time #

filter_time(
    ds: Dataset,
    time_range: TimeRangeLike | Iterable | None = None,
    timestamp: TimestampLike | None = None,
    only_center: bool = False,
    time_var: str = TIME_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
) -> Dataset

Filters an xarray Dataset to include only samples within a given time range.

Parameters:

Name Type Description Default
ds Dataset

Input dataset containing a time coordinate.

 required
time_range TimeRangeLike | Iterable | None

Start and end time of the range to filter.

 None
timestamp TimestampLike | None

Single timestamp; returns closest sample if provided.

 None
only_center bool

If True, returns only the center sample of the selection.

 False
time_var str

Name of the time variable in ds.

 TIME_VAR
along_track_dim str

Dimension name along which time is defined.

 ALONG_TRACK_DIM
trim_index_offset_var str

Variable tracking index offsets from trimming/filtering.

 'trim_index_offset'
pad_idxs int

Number of additional samples added at both ends.

 0
shift_idxs int

Offset to shift selected sample indices.

 0
pad_time TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None

Additional time padding applied before index-based padding.

 None

Returns:

Type Description
Dataset

Filtered dataset containing only samples within the specified time range.

Examples:

>>> import earthcarekit as eck
>>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
>>> print(ds.time.values[[0, -1]])
['2026-01-08T03:04:08.393852234' '2026-01-08T03:15:57.401298285']

>>> ds_filtered = eck.filter_time(ds, time_range=("2026-01-08 03:10", "2026-01-08 03:12"))
>>> print(ds_filtered.time.values[[0, -1]])
['2026-01-08T03:10:00.115605354' '2026-01-08T03:11:59.985651731']
Source code in earthcarekit/filter/_filter_time.py 
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
def filter_time(
    ds: xr.Dataset,
    time_range: TimeRangeLike | Iterable | None = None,
    timestamp: TimestampLike | None = None,
    only_center: bool = False,
    time_var: str = TIME_VAR,
    along_track_dim: str = ALONG_TRACK_DIM,
    trim_index_offset_var: str = "trim_index_offset",
    pad_idxs: int = 0,
    shift_idxs: int = 0,
    pad_time: TimedeltaLike | tuple[TimedeltaLike, TimedeltaLike] | None = None,
) -> xr.Dataset:
    """Filters an xarray Dataset to include only samples within a given time range.

    Args:
        ds: Input dataset containing a time coordinate.
        time_range: Start and end time of the range to filter.
        timestamp: Single timestamp; returns closest sample if provided.
        only_center: If True, returns only the center sample of the selection.
        time_var: Name of the time variable in `ds`.
        along_track_dim: Dimension name along which time is defined.
        trim_index_offset_var: Variable tracking index offsets from trimming/filtering.
        pad_idxs: Number of additional samples added at both ends.
        shift_idxs: Offset to shift selected sample indices.
        pad_time: Additional time padding applied before index-based padding.

    Returns:
        Filtered dataset containing only samples within the specified time range.

    Examples:
        >>> import earthcarekit as eck
        >>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True)
        >>> print(ds.time.values[[0, -1]])
        ['2026-01-08T03:04:08.393852234' '2026-01-08T03:15:57.401298285']

        >>> ds_filtered = eck.filter_time(ds, time_range=("2026-01-08 03:10", "2026-01-08 03:12"))
        >>> print(ds_filtered.time.values[[0, -1]])
        ['2026-01-08T03:10:00.115605354' '2026-01-08T03:11:59.985651731']
    """
    if time_range is not None and timestamp is not None:
        raise ValueError("Can not use both arguments time_range and timestamp at the same time.")

    mask = get_filter_time_mask(
        ds=ds,
        time_range=time_range,
        timestamp=timestamp,
        only_center=only_center,
        time_var=time_var,
        pad_idxs=pad_idxs,
        shift_idxs=shift_idxs,
        pad_time=pad_time,
    )

    if np.sum(mask) == 0:
        times = ds[time_var].values
        msg = (
            f"No data falls into the given time range!\n"
            f"In the dataset time ranges from {times[0]} to {times[-1]}.\n"
        )
        raise ValueError(msg)

    da_mask: xr.DataArray = xr.DataArray(mask, dims=[along_track_dim], name=time_var)

    ds_new: xr.Dataset = xr.Dataset(
        {
            var: (
                ds[var].copy().where(da_mask, drop=True)
                if along_track_dim in ds[var].dims
                else ds[var].copy()
            )
            for var in ds.data_vars
        }
    )
    ds_new.attrs = ds.attrs.copy()
    ds_new.encoding = ds.encoding.copy()

    ds_new = update_trim_index_offset(
        ds=ds_new,
        offset=int(np.argmax(mask)),
        var=trim_index_offset_var,
    )

    return ds_new

get_frame_index_range #

get_frame_index_range(
    latitude: ArrayLike | None = None,
    frame_id: str | None = None,
    ds: Dataset | None = None,
    lat_var: str = TRACK_LAT_VAR,
) -> tuple[int, int]

Generate an index range for trimming arrays or datasets to EarthCARE latitude frame bounds.

Parameters:

Name Type Description Default
latitude ArrayLike | None

Sequence of along-track latitude values.

 None
frame_id str | None

EarthCARE frame ID (letter between "A" and "H").

 None
ds Dataset | None

EarthCARE dataset containing along-track latitude values.

 None
lat_var str

Name of the latitude dataset variable.

 TRACK_LAT_VAR

Raises:

Type Description
ValueError

If inputs are missing (requires latitude and ds or frame_id).

Returns:

Type Description
tuple[int, int]

Index range of the EarthCARE frame (i.e., slice tuple).

Examples:

>>> import earthcarekit as eck
>>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True, trim_to_frame=False)
>>> slice_tuple = eck.filter.get_frame_index_range(ds=ds, frame_id="F")
>>> # Or: eck.filter.get_frame_index_range(latitude=ds.latitude.values, frame_id="F")
>>> ds_sliced = ds.isel({"along_track": slice(*slice_tuple)})
>>> print("non-trimmed:", ds.along_track.shape)
>>> print("timming slice:", slice_tuple)
>>> print("timmed:", ds_sliced.along_track.shape)
non-trimmed: (4992,)
timming slice: (14, 4978)
timmed: (4964,)
Source code in earthcarekit/filter/_frame.py 
 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 get_frame_index_range(
    latitude: ArrayLike | None = None,
    frame_id: str | None = None,
    ds: Dataset | None = None,
    lat_var: str = TRACK_LAT_VAR,
) -> tuple[int, int]:
    """Generate an index range for trimming arrays or datasets to EarthCARE latitude frame bounds.

    Args:
        latitude: Sequence of along-track latitude values.
        frame_id: EarthCARE frame ID (letter between "A" and "H").
        ds: EarthCARE dataset containing along-track latitude values.
        lat_var: Name of the latitude dataset variable.

    Raises:
        ValueError:
            If inputs are missing (requires `latitude` and `ds` or `frame_id`).

    Returns:
        Index range of the EarthCARE frame (i.e., slice tuple).

    Examples:
        >>> import earthcarekit as eck
        >>> ds = eck.ecload("CPR_FMR_2A", "09167F", download=True, trim_to_frame=False)
        >>> slice_tuple = eck.filter.get_frame_index_range(ds=ds, frame_id="F")
        >>> # Or: eck.filter.get_frame_index_range(latitude=ds.latitude.values, frame_id="F")
        >>> ds_sliced = ds.isel({"along_track": slice(*slice_tuple)})
        >>> print("non-trimmed:", ds.along_track.shape)
        >>> print("timming slice:", slice_tuple)
        >>> print("timmed:", ds_sliced.along_track.shape)
        non-trimmed: (4992,)
        timming slice: (14, 4978)
        timmed: (4964,)
    """
    if isinstance(ds, Dataset):
        lat = ds[lat_var].data
        if not isinstance(frame_id, str):
            frame_id = get_frame_id(ds)
    elif latitude is not None:
        lat = np.asarray(latitude)
    else:
        raise ValueError("Either ds or latitude array must be given")

    if not isinstance(frame_id, str):
        raise ValueError("Missing frame_id input")
    return get_frame_slice_tuple_check_gaps(lat, frame_id)

get_frame_slice_tuple #

get_frame_slice_tuple(latitude: ArrayLike, frame_id: str) -> tuple[int, int]

Return start and end index of an EarthCARE frame for a along-track latitude sequence.

This method assumes input latitudes from a continous sequence of satellite track coordinates spanning at most a single EarthCARE frame. The coordinates may extend slightly beyond the frame's limits (e.g., due to margins), but they must not span multiple frames or full orbits.

Parameters:

Name Type Description Default
latitude ArrayLike

EarthCARE dataset.

 required
frame_id str

EarthCARE frame ID letter (A-H).

 required

Raises:

Type Description
ValueError

When not able to retrieve frame ID from either the dataset encoding (i.e., ds.encoding["source"]) or a variable (i.e., "frame_id" or "frameID").

Returns:

Type Description
tuple[int, int]

Slice tuple matching the data within the EarthCARE frame.
Source code in earthcarekit/filter/_frame.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get_frame_slice_tuple(
    latitude: ArrayLike,
    frame_id: str,
) -> tuple[int, int]:
    """Return start and end index of an EarthCARE frame for a along-track latitude sequence.

    This method assumes input latitudes from a continous sequence of satellite track coordinates
    spanning at most a single EarthCARE frame. The coordinates may extend slightly beyond the
    frame's limits (e.g., due to margins), but they must not span multiple frames or full orbits.

    Args:
        latitude: EarthCARE dataset.
        frame_id: EarthCARE frame ID letter (A-H).

    Raises:
        ValueError: When not able to retrieve frame ID from either the dataset encoding (i.e., `ds.encoding["source"]`) or a variable (i.e., `"frame_id"` or `"frameID"`).

    Returns:
        Slice tuple matching the data within the EarthCARE frame.
    """
    start, stop = EC_LATITUDE_FRAME_BOUNDS[frame_id]
    idxs = argwhere_frame(latitude, start, stop)
    slice_tuple = int(idxs[0][0]), int(idxs[-1][0]) + 1
    return slice_tuple