19147573697

Committed 06 Nov 2025 07:34PM UTC coverage: 20.335% (-0.2%) from 20.507%

Build # 19147573697

Build Type

Pull #258

github

Committed by

web-flow

Commit Message

Merge cbf57e240 into f3ea17ca6

Pull Request Pull Request #258: Bump the actions group in /.github/workflows with 6 updates + cruft update

Run Details

0 of 16 new or added lines in 7 files covered. (0.0%)

7 existing lines in 6 files now uncovered.

97 of 477 relevant lines covered (20.34%)

1.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

20.27

/src/xdatasets/spatial.py

import logging
import warnings

import pandas as pd
import xarray as xr
from clisops.core.subset import shape_bbox_indexer, subset_gridpoint
from tqdm.auto import tqdm

from .utils import HiddenPrints


try:
    import xagg as xa
except ImportError:
    warnings.warn("xagg is not installed. Please install it with `pip install xagg`", stacklevel=2)
    xa = None


def bbox_ds(ds_copy, geom):
    indexer = shape_bbox_indexer(ds_copy, geom)
    da = ds_copy.isel(indexer)
    da = da.chunk({"latitude": -1, "longitude": -1})
    return da


def clip_by_bbox(ds, space, dataset_name):
    msg = f"Spatial operations: processing bbox with {dataset_name}"
    logging.info(msg)
    indexer = shape_bbox_indexer(ds, space["geometry"])
    ds_copy = ds.isel(indexer).copy()
    return ds_copy


def create_weights_mask(da, poly):
    if xa is None:
        raise ImportError(
            "xagg is not installed. Please install it with `pip install xagg`",
        )

    weightmap = xa.pixel_overlaps(da, poly, subset_bbox=True)

    pixels = pd.DataFrame(
        index=weightmap.agg["pix_idxs"][0],
        data=list(map(list, weightmap.agg["coords"][0])),
        columns=["latitude", "longitude"],
    )

    weights = pd.DataFrame(
        index=weightmap.agg["pix_idxs"][0],
        data=weightmap.agg["rel_area"][0][0].tolist(),
        columns=["weights"],
    )

    df = pd.merge(pixels, weights, left_index=True, right_index=True)
    return df.set_index(["latitude", "longitude"]).to_xarray()


def aggregate(ds_in, ds_weights):
    return (ds_in * ds_weights.weights).sum(["latitude", "longitude"], min_count=1)


def clip_by_polygon(ds, space, dataset_name):
    # We are not using clisops for weighted averages because it is too unstable for now.
    # We use the xagg package instead.

    indexer = shape_bbox_indexer(ds, space["geometry"])
    ds_copy = ds.isel(indexer).copy()

    arrays = []
    pbar = tqdm(space["geometry"].iterrows(), position=0, leave=True)
    for idx, row in pbar:
        item = row[space["unique_id"]] if space["unique_id"] is not None and space["unique_id"] in row else idx
        pbar.set_description(
            f"Spatial operations: processing polygon {item} with {dataset_name}",
        )

        geom = space["geometry"].iloc[[idx]]
        da = bbox_ds(ds_copy, geom)

        # Average data array over shape
        # da = average_shape(da, shape=geom)
        with HiddenPrints():
            ds_weights = create_weights_mask(da.isel(time=0), geom)
        if space["averaging"] is True:
            da_tmp = aggregate(da, ds_weights)
            for var in da_tmp.variables:
                da_tmp[var].attrs = da[var].attrs
            da = da_tmp
        else:
            da = xr.merge([da, ds_weights])
            da = da.where(da.weights.notnull(), drop=True)
        da = da.expand_dims({"geom": geom.index.values})
        arrays.append(da)

    data = xr.concat(arrays, dim="geom")

    if space.get("unique_id") is not None:
        try:
            data = data.swap_dims({"geom": space["unique_id"]})
            data = data.drop_vars("geom")

            if space["unique_id"] not in data.coords:
                data = data.assign_coords(
                    {
                        space["unique_id"]: (
                            space["unique_id"],
                            space["geometry"][space["unique_id"]],
                        ),
                    },
                )
            data[space["unique_id"]].attrs["cf_role"] = "timeseries_id"
        except KeyError:  # noqa: S110
            pass
    return data


def clip_by_point(ds, space, dataset_name):
    # TODO : adapt logic for coordinate names
    msg = f"Spatial operations: processing points with {dataset_name}"
    logging.info(msg)

    lat, lon = zip(*space["geometry"].values(), strict=False)
    data = subset_gridpoint(
        ds.rename({"latitude": "lat", "longitude": "lon"}),
        lon=list(lon),
        lat=list(lat),
    )
    data = data.rename({"lat": "latitude", "lon": "longitude"})

    data = data.assign_coords({"site": ("site", list(space["geometry"].keys()))})

    data["site"].attrs["cf_role"] = "timeseries_id"

    return data

1	import logging	6✔
2	import warnings	6✔
3
4	import pandas as pd	6✔
5	import xarray as xr	6✔
6	from clisops.core.subset import shape_bbox_indexer, subset_gridpoint	6✔
7	from tqdm.auto import tqdm	6✔
8
9	from .utils import HiddenPrints	6✔
10
11
12	try:	6✔
13	import xagg as xa	6✔
14	except ImportError:	×
NEW 15	warnings.warn("xagg is not installed. Please install it with `pip install xagg`", stacklevel=2)	×
16	xa = None	×
17
18
19	def bbox_ds(ds_copy, geom):	6✔
20	indexer = shape_bbox_indexer(ds_copy, geom)	×
21	da = ds_copy.isel(indexer)	×
22	da = da.chunk({"latitude": -1, "longitude": -1})	×
23	return da	×
24
25
26	def clip_by_bbox(ds, space, dataset_name):	6✔
27	msg = f"Spatial operations: processing bbox with {dataset_name}"	×
28	logging.info(msg)	×
29	indexer = shape_bbox_indexer(ds, space["geometry"])	×
30	ds_copy = ds.isel(indexer).copy()	×
31	return ds_copy	×
32
33
34	def create_weights_mask(da, poly):	6✔
35	if xa is None:	×
36	raise ImportError(	×
37	"xagg is not installed. Please install it with `pip install xagg`",
38	)
39
40	weightmap = xa.pixel_overlaps(da, poly, subset_bbox=True)	×
41
42	pixels = pd.DataFrame(	×
43	index=weightmap.agg["pix_idxs"][0],
44	data=list(map(list, weightmap.agg["coords"][0])),
45	columns=["latitude", "longitude"],
46	)
47
48	weights = pd.DataFrame(	×
49	index=weightmap.agg["pix_idxs"][0],
50	data=weightmap.agg["rel_area"][0][0].tolist(),
51	columns=["weights"],
52	)
53
54	df = pd.merge(pixels, weights, left_index=True, right_index=True)	×
55	return df.set_index(["latitude", "longitude"]).to_xarray()	×
56
57
58	def aggregate(ds_in, ds_weights):	6✔
59	return (ds_in * ds_weights.weights).sum(["latitude", "longitude"], min_count=1)	×
60
61
62	def clip_by_polygon(ds, space, dataset_name):	6✔
63	# We are not using clisops for weighted averages because it is too unstable for now.
64	# We use the xagg package instead.
65
66	indexer = shape_bbox_indexer(ds, space["geometry"])	×
67	ds_copy = ds.isel(indexer).copy()	×
68
69	arrays = []	×
70	pbar = tqdm(space["geometry"].iterrows(), position=0, leave=True)	×
71	for idx, row in pbar:	×
NEW 72	item = row[space["unique_id"]] if space["unique_id"] is not None and space["unique_id"] in row else idx	×
UNCOV 73	pbar.set_description(	×
74	f"Spatial operations: processing polygon {item} with {dataset_name}",
75	)
76
77	geom = space["geometry"].iloc[[idx]]	×
78	da = bbox_ds(ds_copy, geom)	×
79
80	# Average data array over shape
81	# da = average_shape(da, shape=geom)
82	with HiddenPrints():	×
83	ds_weights = create_weights_mask(da.isel(time=0), geom)	×
84	if space["averaging"] is True:	×
85	da_tmp = aggregate(da, ds_weights)	×
86	for var in da_tmp.variables:	×
87	da_tmp[var].attrs = da[var].attrs	×
88	da = da_tmp	×
89	else:
90	da = xr.merge([da, ds_weights])	×
91	da = da.where(da.weights.notnull(), drop=True)	×
92	da = da.expand_dims({"geom": geom.index.values})	×
93	arrays.append(da)	×
94
95	data = xr.concat(arrays, dim="geom")	×
96
97	if space.get("unique_id") is not None:	×
98	try:	×
99	data = data.swap_dims({"geom": space["unique_id"]})	×
100	data = data.drop_vars("geom")	×
101
102	if space["unique_id"] not in data.coords:	×
103	data = data.assign_coords(	×
104	{
105	space["unique_id"]: (
106	space["unique_id"],
107	space["geometry"][space["unique_id"]],
108	),
109	},
110	)
111	data[space["unique_id"]].attrs["cf_role"] = "timeseries_id"	×
112	except KeyError: # noqa: S110	×
113	pass	×
114	return data	×
115
116
117	def clip_by_point(ds, space, dataset_name):	6✔
118	# TODO : adapt logic for coordinate names
119	msg = f"Spatial operations: processing points with {dataset_name}"	×
120	logging.info(msg)	×
121
NEW 122	lat, lon = zip(*space["geometry"].values(), strict=False)	×
123	data = subset_gridpoint(	×
124	ds.rename({"latitude": "lat", "longitude": "lon"}),
125	lon=list(lon),
126	lat=list(lat),
127	)
128	data = data.rename({"lat": "latitude", "lon": "longitude"})	×
129
130	data = data.assign_coords({"site": ("site", list(space["geometry"].keys()))})	×
131
132	data["site"].attrs["cf_role"] = "timeseries_id"	×
133
134	return data	×

hydrologie / xdatasets / 19147573697

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous