Skip to content

arcticdem

darts_preprocessing.engineering.arcticdem

Computation of ArcticDEM derived products.

logger module-attribute 

logger = logging.getLogger(
    __name__.replace("darts_", "darts.")
)

Distance dataclass 

Distance(pixel: int, meter: float)

Convenience class to represent a distance in pixels and meters.

meter instance-attribute 

meter: float

pixel instance-attribute 

pixel: int

__repr__ 

__repr__()
Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
29
30
def __repr__(self):  # noqa: D105
    return f"{self.pixel}px ({self.meter}m)"

parse classmethod 

parse(
    v: int | float | str, res: float
) -> darts_preprocessing.engineering.arcticdem.Distance

Parse a distance from a string or numeric value.

If the input is a string, it can be in the format of "10px" or "10m". If it is a numeric value, it is interpreted as meters and converted to pixels based on the resolution.

Parameters:

  • v (int | float | str) –

    The input distance value.

  • res (float) –

    The resolution in meters per pixel.

Raises:

  • ValueError –

    If the input distance is negative.

  • ValueError –

    If the input distance is not a valid string format.

  • TypeError –

    If the input distance is not a string, int, or float.

Returns:

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
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
@classmethod
def parse(cls, v: int | float | str, res: float) -> "Distance":
    """Parse a distance from a string or numeric value.

    If the input is a string, it can be in the format of "10px" or "10m".
    If it is a numeric value, it is interpreted as meters and converted to pixels based on the resolution.

    Args:
        v (int | float | str): The input distance value.
        res (float): The resolution in meters per pixel.

    Raises:
        ValueError: If the input distance is negative.
        ValueError: If the input distance is not a valid string format.
        TypeError: If the input distance is not a string, int, or float.

    Returns:
        Distance: The parsed distance in pixels and meters.

    """
    if isinstance(v, str):
        if v.endswith("px"):
            pixel = int(v[:-2])
            meter = pixel * res
        elif v.endswith("m"):
            meter = float(v[:-1])
            pixel = ceil(meter / res)
        else:
            raise ValueError(f"Invalid distance format: {v}")
    elif isinstance(v, (int, float)):
        if v < 0:
            raise ValueError("Distance cannot be negative")
        pixel = ceil(v / res)
        meter = pixel * res
    else:
        raise TypeError(f"Invalid type for distance: {type(v)}")
    return cls(pixel=pixel, meter=meter)

calculate_aspect 

calculate_aspect(
    arcticdem_ds: xarray.Dataset,
) -> xarray.Dataset

Calculate the aspect of the terrain surface from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated aspect added as a new variable 'aspect'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@stopwatch("Calculating aspect", printer=logger.debug)
def calculate_aspect(arcticdem_ds: xr.Dataset) -> xr.Dataset:
    """Calculate the aspect of the terrain surface from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.

    Returns:
        xr.Dataset: The input Dataset with the calculated aspect added as a new variable 'aspect'.

    """
    aspect_deg = aspect(arcticdem_ds.dem)
    aspect_deg.attrs = {
        "long_name": "Aspect",
        "units": "degrees",
        "description": "The compass direction that the slope faces, in degrees clockwise from north.",
        "source": "ArcticDEM",
    }
    arcticdem_ds["aspect"] = aspect_deg.compute()
    return arcticdem_ds

calculate_curvature 

calculate_curvature(
    arcticdem_ds: xarray.Dataset,
) -> xarray.Dataset

Calculate the curvature of the terrain surface from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated curvature added as a new variable 'curvature'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@stopwatch("Calculating curvature", printer=logger.debug)
def calculate_curvature(arcticdem_ds: xr.Dataset) -> xr.Dataset:
    """Calculate the curvature of the terrain surface from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.

    Returns:
        xr.Dataset: The input Dataset with the calculated curvature added as a new variable 'curvature'.

    """
    curvature_da = curvature(arcticdem_ds.dem)
    curvature_da.attrs = {
        "long_name": "Curvature",
        "units": "",
        "description": "The curvature of the terrain surface.",
        "source": "ArcticDEM",
    }
    arcticdem_ds["curvature"] = curvature_da.compute()
    return arcticdem_ds

calculate_dissection_index 

calculate_dissection_index(
    arcticdem_ds: xarray.Dataset, neighborhood_size: int
) -> xarray.Dataset

Calculate the Dissection Index (DI) from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

  • neighborhood_size (int) –

    The size of the neighborhood window (in meters) for the calculation.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated DI added as a new variable 'di'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@stopwatch("Calculating Dissection Index", printer=logger.debug)
def calculate_dissection_index(arcticdem_ds: xr.Dataset, neighborhood_size: int) -> xr.Dataset:
    """Calculate the Dissection Index (DI) from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.
        neighborhood_size (int): The size of the neighborhood window (in meters) for the calculation.

    Returns:
        xr.Dataset: The input Dataset with the calculated DI added as a new variable 'di'.

    """
    # Get neighborhood_size in pixels
    neighborhood_size: Distance = Distance.parse(neighborhood_size, arcticdem_ds.odc.geobox.resolution.x)

    di = dissection_index(arcticdem_ds.dem, window_size=neighborhood_size.pixel)

    di.attrs = {
        "long_name": "Dissection Index",
        "units": "",
        "description": (
            f"Dissection index calculated using a {neighborhood_size} neighborhood. "
            "Values range from 0 (smooth) to 1 (most rugged)."
        ),
        "source": "ArcticDEM",
    }

    arcticdem_ds["di"] = di.compute()
    return arcticdem_ds

calculate_hillshade 

calculate_hillshade(
    arcticdem_ds: xarray.Dataset,
    azimuth: int = 225,
    angle_altitude: int = 25,
) -> xarray.Dataset

Calculate the hillshade of the terrain surface from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

  • azimuth (int, default: 225 ) –

    The azimuth angle of the light source in degrees. Defaults to 225.

  • angle_altitude (int, default: 25 ) –

    The altitude angle of the light source in degrees. Defaults to 25.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated hillshade added as a new variable 'hillshade'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@stopwatch.f("Calculating hillshade", printer=logger.debug, print_kwargs=["azimuth", "angle_altitude"])
def calculate_hillshade(arcticdem_ds: xr.Dataset, azimuth: int = 225, angle_altitude: int = 25) -> xr.Dataset:
    """Calculate the hillshade of the terrain surface from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.
        azimuth (int, optional): The azimuth angle of the light source in degrees. Defaults to 225.
        angle_altitude (int, optional): The altitude angle of the light source in degrees. Defaults to 25.

    Returns:
        xr.Dataset: The input Dataset with the calculated hillshade added as a new variable 'hillshade'.

    """
    hillshade_da = hillshade(arcticdem_ds.dem, azimuth=azimuth, angle_altitude=angle_altitude)
    hillshade_da.attrs = {
        "long_name": "Hillshade",
        "units": "",
        "description": f"The hillshade based on azimuth {azimuth} and angle_altitude {angle_altitude}.",
        "source": "ArcticDEM",
    }
    arcticdem_ds["hillshade"] = hillshade_da.compute()
    return arcticdem_ds

calculate_slope 

calculate_slope(
    arcticdem_ds: xarray.Dataset,
) -> xarray.Dataset

Calculate the slope of the terrain surface from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated slope added as a new variable 'slope'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@stopwatch("Calculating slope", printer=logger.debug)
def calculate_slope(arcticdem_ds: xr.Dataset) -> xr.Dataset:
    """Calculate the slope of the terrain surface from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.

    Returns:
        xr.Dataset: The input Dataset with the calculated slope added as a new variable 'slope'.

    """
    slope_deg = slope(arcticdem_ds.dem)
    slope_deg.attrs = {
        "long_name": "Slope",
        "units": "degrees",
        "description": "The slope of the terrain surface in degrees.",
        "source": "ArcticDEM",
    }
    arcticdem_ds["slope"] = slope_deg.compute()
    return arcticdem_ds

calculate_terrain_ruggedness_index 

calculate_terrain_ruggedness_index(
    arcticdem_ds: xarray.Dataset, neighborhood_size: int
) -> xarray.Dataset

Calculate the Terrain Ruggedness Index (TRI) from an ArcticDEM Dataset.

Definition from ESRI: TRI expresses the amount of elevation difference between adjacent cells of a DEM. Using methodology developed by Riley et al (1999) and published in the paper “A Terrain ruggedness Index That Quantifies Topographic heterogeneity”, the tool measures the difference in elevation values from a center cell and eight cells directly surrounding it. Then, the eight elevation differences are squared and averaged. The square root of this average results is a TRI measurement for the center cell. This calculation is then conducted on every cell of the DEM.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

  • neighborhood_size (int) –

    The neighborhood_size in meters for the TRI calculation.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated TRI added as a new variable 'tri'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
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
263
264
265
@stopwatch("Calculating TRI", printer=logger.debug)
def calculate_terrain_ruggedness_index(arcticdem_ds: xr.Dataset, neighborhood_size: int) -> xr.Dataset:
    """Calculate the Terrain Ruggedness Index (TRI) from an ArcticDEM Dataset.

    Definition from ESRI:
    TRI expresses the amount of elevation difference between adjacent cells of a DEM.
    Using methodology developed by Riley et al (1999) and published in the paper
    “A Terrain ruggedness Index That Quantifies Topographic heterogeneity”,
    the tool measures the difference in elevation values from a center cell and eight cells directly surrounding it.
    Then, the eight elevation differences are squared and averaged.
    The square root of this average results is a TRI measurement for the center cell.
    This calculation is then conducted on every cell of the DEM.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.
        neighborhood_size (int): The neighborhood_size in meters for the TRI calculation.

    Returns:
        xr.Dataset: The input Dataset with the calculated TRI added as a new variable 'tri'.

    """
    cellsize_x, _cellsize_y = convolution.calc_cellsize(arcticdem_ds.dem)

    neighborhood_size: Distance = Distance.parse(neighborhood_size, cellsize_x)

    kernel = np.ones((neighborhood_size.pixel, neighborhood_size.pixel), dtype=float)
    kernel[neighborhood_size.pixel // 2, neighborhood_size.pixel // 2] = 0  # Set the center cell to 0
    kernel = convolution.custom_kernel(kernel)
    logger.debug(f"Calculating Terrain Ruggedness Index with square kernel of radius {neighborhood_size} cells.")

    # Change dtype of kernel to float32 since we don't need the precision and f32 is faster
    kernel = kernel.astype("float32")

    if has_cuda_and_cupy() and arcticdem_ds.cupy.is_cupy:
        kernel = cp.asarray(kernel)

    # Kernel compute of TRI as described here:
    # https://sites.utexas.edu/utarima/files/2024/02/terrain_roughness_index.pdf
    dem_squared = arcticdem_ds.dem**2
    focal_sum = convolution.convolution_2d(arcticdem_ds.dem, kernel)
    focal_sum_squared = convolution.convolution_2d(dem_squared, kernel)
    tri = np.sqrt((kernel.size - 1) * dem_squared - 2 * arcticdem_ds.dem * focal_sum + focal_sum_squared)

    tri.attrs = {
        "long_name": "Terrain Ruggedness Index",
        "units": "m",
        "description": (
            "The difference between the elevation of a cell and the mean elevation of the surrounding"
            f" cells within a square kernel of radius {neighborhood_size} cells."
        ),
        "source": "ArcticDEM",
    }

    arcticdem_ds["tri"] = tri.compute()
    return arcticdem_ds

calculate_topographic_position_index 

calculate_topographic_position_index(
    arcticdem_ds: xarray.Dataset,
    outer_radius: int,
    inner_radius: int,
) -> xarray.Dataset

Calculate the Topographic Position Index (TPI) from an ArcticDEM Dataset.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

  • outer_radius (int) –

    The outer radius of the annulus kernel in m.

  • inner_radius (int) –

    The inner radius of the annulus kernel in m.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated TPI added as a new variable 'tpi'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
 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
@stopwatch.f("Calculating TPI", printer=logger.debug, print_kwargs=["outer_radius", "inner_radius"])
def calculate_topographic_position_index(arcticdem_ds: xr.Dataset, outer_radius: int, inner_radius: int) -> xr.Dataset:
    """Calculate the Topographic Position Index (TPI) from an ArcticDEM Dataset.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.
        outer_radius (int, optional): The outer radius of the annulus kernel in m.
        inner_radius (int, optional): The inner radius of the annulus kernel in m.

    Returns:
        xr.Dataset: The input Dataset with the calculated TPI added as a new variable 'tpi'.

    """
    cellsize_x, cellsize_y = convolution.calc_cellsize(arcticdem_ds.dem)  # Should be equal to the resolution of the DEM
    # Use an annulus kernel if inner_radius is greater than 0
    outer_radius: Distance = Distance.parse(outer_radius, cellsize_x)
    if inner_radius > 0:
        inner_radius: Distance = Distance.parse(inner_radius, cellsize_x)
        kernel = convolution.annulus_kernel(cellsize_x, cellsize_y, outer_radius.meter, inner_radius.meter)
        attr_cell_description = (
            f"within a ring at a distance of {inner_radius}-{outer_radius} cells away from the focal cell."
        )
        logger.debug(
            f"Calculating Topographic Position Index with annulus kernel of {inner_radius}-{outer_radius} cells."
        )
    else:
        kernel = convolution.circle_kernel(cellsize_x, cellsize_y, outer_radius.meter)
        attr_cell_description = f"within a circle at a distance of {outer_radius} cells away from the focal cell."
        logger.debug(f"Calculating Topographic Position Index with circle kernel of {outer_radius} cells.")

    # Change dtype of kernel to float32 since we don't need the precision and f32 is faster
    kernel = kernel.astype("float32")

    if has_cuda_and_cupy() and arcticdem_ds.cupy.is_cupy:
        kernel = cp.asarray(kernel)

    tpi = arcticdem_ds.dem - convolution.convolution_2d(arcticdem_ds.dem, kernel) / kernel.sum()
    tpi.attrs = {
        "long_name": "Topographic Position Index",
        "units": "m",
        "description": "The difference between the elevation of a cell and the mean elevation of the surrounding"
        f"cells {attr_cell_description}",
        "source": "ArcticDEM",
    }

    arcticdem_ds["tpi"] = tpi.compute()

    return arcticdem_ds

calculate_vector_ruggedness_measure 

calculate_vector_ruggedness_measure(
    arcticdem_ds: xarray.Dataset, neighborhood_size: int
) -> xarray.Dataset

Calculate the Vector Ruggedness Measure (VRM) from an ArcticDEM Dataset.

Implementation of the vector ruggedness measure described in Sappington, J.M., K.M. Longshore, and D.B. Thomson. 2007. Quantifying Landscape Ruggedness for Animal Habitat Analysis: A case Study Using Bighorn Sheep in the Mojave Desert. Journal of Wildlife Management. 71(5): 1419-1426.

Parameters:

  • arcticdem_ds (xarray.Dataset) –

    The ArcticDEM Dataset containing the 'dem' variable.

  • neighborhood_size (int) –

    The size of the neighborhood window (in meters) for the calculation.

Returns:

  • xarray.Dataset –

    xr.Dataset: The input Dataset with the calculated VRM added as a new variable 'vrm'.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/arcticdem.py 
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
295
296
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
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
@stopwatch("Calculating Vector Ruggedness Measure", printer=logger.debug)
def calculate_vector_ruggedness_measure(arcticdem_ds: xr.Dataset, neighborhood_size: int) -> xr.Dataset:
    """Calculate the Vector Ruggedness Measure (VRM) from an ArcticDEM Dataset.

    Implementation of the vector ruggedness measure described in Sappington, J.M.,
    K.M. Longshore, and D.B. Thomson. 2007. Quantifying Landscape Ruggedness for
    Animal Habitat Analysis: A case Study Using Bighorn Sheep in the Mojave Desert.
    Journal of Wildlife Management. 71(5): 1419-1426.

    Args:
        arcticdem_ds (xr.Dataset): The ArcticDEM Dataset containing the 'dem' variable.
        neighborhood_size (int): The size of the neighborhood window (in meters) for the calculation.

    Returns:
        xr.Dataset: The input Dataset with the calculated VRM added as a new variable 'vrm'.

    """
    # Calculate slope and aspect
    slope_rad = arcticdem_ds.slope * (np.pi / 180)  # Convert to radians
    aspect_rad = arcticdem_ds.aspect * (np.pi / 180)  # Convert to radians

    # Calculate x, y, and z components of unit vectors
    xy = np.sin(slope_rad)
    z = np.cos(slope_rad)

    # Handle flat areas (where aspect = -1)
    if has_cuda_and_cupy() and arcticdem_ds.cupy.is_cupy:
        aspect_rad.variable._data = cp.where(aspect_rad.variable._data == -1, 0, aspect_rad.variable._data)
        # aspect_rad = aspect_rad.copy(data=aspect_rad_raw)
    else:
        aspect_rad = xr.where(aspect_rad == -1, 0, aspect_rad)
    x = np.sin(aspect_rad) * xy
    y = np.cos(aspect_rad) * xy

    # Get neighborhood_size in pixels
    neighborhood_size: Distance = Distance.parse(neighborhood_size, arcticdem_ds.odc.geobox.resolution.x)
    # Create convolution kernel for focal sum
    kernel = np.ones((neighborhood_size.pixel, neighborhood_size.pixel), dtype=float) / neighborhood_size.pixel**2
    kernel = convolution.custom_kernel(kernel)

    # Change dtype of kernel to float32 since we don't need the precision and f32 is faster
    kernel = kernel.astype("float32")

    if has_cuda_and_cupy() and arcticdem_ds.cupy.is_cupy:
        kernel = cp.asarray(kernel)

    logger.debug(f"Calculating Vector Ruggedness Measure with square kernel of size {neighborhood_size} cells.")

    # TODO: Write a custom kernel for this for speedup and smaller memory footprint
    # Calculate sums of x, y, and z components in the neighborhood
    x_sum = convolution.convolution_2d(x, kernel)
    y_sum = convolution.convolution_2d(y, kernel)
    z_sum = convolution.convolution_2d(z, kernel)

    # Calculate the resultant vector magnitude
    vrm = 1 - np.sqrt(x_sum**2 + y_sum**2 + z_sum**2)

    vrm.attrs = {
        "long_name": "Vector Ruggedness Measure",
        "units": "",
        "description": (
            f"Vector ruggedness measure calculated using a {neighborhood_size} neighborhood. "
            "Values range from 0 (smooth) to 1 (most rugged)."
        ),
        "source": "ArcticDEM",
    }

    arcticdem_ds["vrm"] = vrm.compute()
    return arcticdem_ds

dissection_index 

dissection_index(
    agg: xarray.DataArray,
    window_size: int = 3,
    name: str | None = "dissection_index",
) -> xarray.DataArray

Compute the dissection index of a 2D array.

Parameters:

  • agg (xarray.DataArray) –

    The input data array.

  • window_size (int, default: 3 ) –

    The size of the window to use for the computation. Defaults to 3.

  • name (str | None, default: 'dissection_index' ) –

    The name of the output data array. Defaults to "dissection_index".

Returns:

  • xarray.DataArray –

    xr.DataArray: The dissection index of the input data array.

Source code in darts-preprocessing/src/darts_preprocessing/engineering/dissection_index.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def dissection_index(agg: xr.DataArray, window_size: int = 3, name: str | None = "dissection_index") -> xr.DataArray:
    """Compute the dissection index of a 2D array.

    Args:
        agg (xr.DataArray): The input data array.
        window_size (int, optional): The size of the window to use for the computation. Defaults to 3.
        name (str | None, optional): The name of the output data array. Defaults to "dissection_index".

    Returns:
        xr.DataArray: The dissection index of the input data array.

    """
    mapper = ArrayTypeFunctionMapping(
        numpy_func=_run_numpy,
        dask_func=_run_dask_numpy,
        cupy_func=_run_cupy,
        dask_cupy_func=lambda *args: not_implemented_func(
            *args, messages="dissection_index() does not support dask with cupy backed DataArray"
        ),
    )

    out = mapper(agg)(agg.data, (window_size - 1) // 2)

    return xr.DataArray(out, name=name, coords=agg.coords, dims=agg.dims, attrs=agg.attrs)