18890899227

Committed 28 Oct 2025 10:21PM UTC coverage: 88.317%. First build

Build # 18890899227

Build Type

Pull #1084

github

Committed by

web-flow

Commit Message

Merge 2f2fc88dc into 85068105d

Pull Request Pull Request #1084: feat: Add time_query() and TimeDescriptor

Coverage Stats

2409 of 2689 new or added lines in 73 files covered. (89.59%)

115141 of 130373 relevant lines covered (88.32%)

225661.33 hits per line

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

66.45

/operators/src/processing/temporal_raster_aggregation/first_last_subquery.rs

use crate::{
    adapters::{FoldTileAccu, FoldTileAccuMut, SubQueryTileAggregator},
    util::Result,
};
use async_trait::async_trait;
use futures::{Future, FutureExt, TryFuture, TryFutureExt, future::BoxFuture};
use geoengine_datatypes::{
    primitives::{CacheHint, RasterQueryRectangle, TimeInterval},
    raster::{EmptyGrid2D, Pixel, RasterTile2D, TileInformation},
};
use rayon::ThreadPool;
use std::{marker::PhantomData, sync::Arc};

/// Only outputs the first tile as accumulator.
pub fn first_tile_fold_fn<T>(
    acc: TemporalRasterAggregationTileAccu<T>,
    tile: RasterTile2D<T>,
) -> TemporalRasterAggregationTileAccu<T>
where
    T: Pixel,
{
    if acc.initial_state {
        let mut next_accu = tile;
        next_accu.time = acc.accu_tile.time;

        TemporalRasterAggregationTileAccu {
            accu_tile: next_accu,
            initial_state: false,
            pool: acc.pool,
        }
    } else {
        acc
    }
}

pub fn first_tile_fold_future<T>(
    accu: TemporalRasterAggregationTileAccu<T>,
    tile: RasterTile2D<T>,
) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>>
where
    T: Pixel,
{
    crate::util::spawn_blocking(|| first_tile_fold_fn(accu, tile)).then(move |x| async move {
        match x {
            Ok(r) => Ok(r),
            Err(e) => Err(e.into()),
        }
    })
}

/// Only outputs the last tile as accumulator.
#[allow(clippy::needless_pass_by_value)]
pub fn last_tile_fold_fn<T>(
    acc: TemporalRasterAggregationTileAccu<T>,
    tile: RasterTile2D<T>,
) -> TemporalRasterAggregationTileAccu<T>
where
    T: Pixel,
{
    let mut next_accu = tile;
    next_accu.time = acc.accu_tile.time;

    TemporalRasterAggregationTileAccu {
        accu_tile: next_accu,
        initial_state: false,
        pool: acc.pool,
    }
}

pub fn last_tile_fold_future<T>(
    accu: TemporalRasterAggregationTileAccu<T>,
    tile: RasterTile2D<T>,
) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>>
where
    T: Pixel,
{
    crate::util::spawn_blocking(|| last_tile_fold_fn(accu, tile)).then(move |x| async move {
        match x {
            Ok(r) => Ok(r),
            Err(e) => Err(e.into()),
        }
    })
}

#[derive(Debug, Clone)]
pub struct TemporalRasterAggregationTileAccu<T> {
    accu_tile: RasterTile2D<T>,
    initial_state: bool,
    pool: Arc<ThreadPool>,
}

#[async_trait]
impl<T: Pixel> FoldTileAccu for TemporalRasterAggregationTileAccu<T> {
    type RasterType = T;

    async fn into_tile(self) -> Result<RasterTile2D<Self::RasterType>> {
        Ok(self.accu_tile)
    }

    fn thread_pool(&self) -> &Arc<ThreadPool> {
        &self.pool
    }
}

impl<T: Pixel> FoldTileAccuMut for TemporalRasterAggregationTileAccu<T> {
    fn set_time(&mut self, time: TimeInterval) {
        self.accu_tile.time = time;
    }

    fn set_cache_hint(&mut self, new_cache_hint: CacheHint) {
        self.accu_tile.cache_hint = new_cache_hint;
    }
}

#[derive(Debug, Clone)]
pub struct TemporalRasterAggregationSubQuery<F, T: Pixel> {
    pub fold_fn: F,
    pub _phantom_pixel_type: PhantomData<T>,
}

impl<'a, T, FoldM, FoldF> SubQueryTileAggregator<'a, T>
    for TemporalRasterAggregationSubQuery<FoldM, T>
where
    T: Pixel,
    FoldM: Send
        + Sync
        + 'static
        + Clone
        + Fn(TemporalRasterAggregationTileAccu<T>, RasterTile2D<T>) -> FoldF,
    FoldF: Send + TryFuture<Ok = TemporalRasterAggregationTileAccu<T>, Error = crate::error::Error>,
{
    type TileAccu = TemporalRasterAggregationTileAccu<T>;
    type TileAccuFuture = BoxFuture<'a, Result<Self::TileAccu>>;

    type FoldFuture = FoldF;

    type FoldMethod = FoldM;

    fn new_fold_accu(
        &self,
        tile_info: TileInformation,
        query_rect: RasterQueryRectangle,
        pool: &Arc<ThreadPool>,
    ) -> Self::TileAccuFuture {
        build_temporal_accu(&query_rect, tile_info, pool.clone()).boxed()
    }

    fn tile_query_rectangle(
        &self,
        tile_info: TileInformation,
        _query_rect: RasterQueryRectangle,
        time: TimeInterval,
        band_idx: u32,
    ) -> Result<Option<RasterQueryRectangle>> {
        // The time is already snapped by the operator where the time stream is created.
        Ok(Some(RasterQueryRectangle::new(
            tile_info.global_pixel_bounds(),
            time,
            band_idx.into(),
        )))
    }

    fn fold_method(&self) -> Self::FoldMethod {
        self.fold_fn.clone()
    }
}

fn build_temporal_accu<T: Pixel>(
    query_rect: &RasterQueryRectangle,
    tile_info: TileInformation,
    pool: Arc<ThreadPool>,
) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>> + use<T> {
    let time_interval = query_rect.time_interval();
    crate::util::spawn_blocking(move || TemporalRasterAggregationTileAccu {
        accu_tile: RasterTile2D::new_with_tile_info(
            time_interval,
            tile_info,
            0,
            EmptyGrid2D::new(tile_info.tile_size_in_pixels).into(),
            CacheHint::max_duration(),
        ),

        initial_state: true,
        pool,
    })
    .map_err(From::from)
}

#[derive(Debug, Clone)]
pub struct TemporalRasterAggregationSubQueryNoDataOnly<F, T: Pixel> {
    pub fold_fn: F,
    pub _phantom_pixel_type: PhantomData<T>,
}

impl<'a, T, FoldM, FoldF> SubQueryTileAggregator<'a, T>
    for TemporalRasterAggregationSubQueryNoDataOnly<FoldM, T>
where
    T: Pixel,
    FoldM: Send
        + Sync
        + 'static
        + Clone
        + Fn(TemporalRasterAggregationTileAccu<T>, RasterTile2D<T>) -> FoldF,
    FoldF: Send + TryFuture<Ok = TemporalRasterAggregationTileAccu<T>, Error = crate::error::Error>,
{
    type TileAccu = TemporalRasterAggregationTileAccu<T>;
    type TileAccuFuture = BoxFuture<'a, Result<Self::TileAccu>>;
    type FoldFuture = FoldF;

    type FoldMethod = FoldM;

    fn new_fold_accu(
        &self,
        tile_info: TileInformation,
        query_rect: RasterQueryRectangle,
        pool: &Arc<ThreadPool>,
    ) -> Self::TileAccuFuture {
        build_temporal_no_data_accu(&query_rect, tile_info, pool.clone()).boxed()
    }

    fn tile_query_rectangle(
        &self,
        tile_info: TileInformation,
        _query_rect: RasterQueryRectangle,
        time: TimeInterval,
        band_idx: u32,
    ) -> Result<Option<RasterQueryRectangle>> {
        Ok(Some(RasterQueryRectangle::new(
            tile_info.global_pixel_bounds(),
            time, // The time is already snapped by the operator where the time stream is created.
            band_idx.into(),
        )))
    }

    fn fold_method(&self) -> Self::FoldMethod {
        self.fold_fn.clone()
    }
}

fn build_temporal_no_data_accu<T: Pixel>(
    query_rect: &RasterQueryRectangle,
    tile_info: TileInformation,
    pool: Arc<ThreadPool>,
) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>> + use<T> {
    let time_interval = query_rect.time_interval();
    crate::util::spawn_blocking(move || {
        let output_raster = EmptyGrid2D::new(tile_info.tile_size_in_pixels).into();

        TemporalRasterAggregationTileAccu {
            accu_tile: RasterTile2D::new_with_tile_info(
                time_interval,
                tile_info,
                0,
                output_raster,
                CacheHint::max_duration(),
            ),
            initial_state: true,
            pool,
        }
    })
    .map_err(From::from)
}

1	use crate::{
2	adapters::{FoldTileAccu, FoldTileAccuMut, SubQueryTileAggregator},
3	util::Result,
4	};
5	use async_trait::async_trait;
6	use futures::{Future, FutureExt, TryFuture, TryFutureExt, future::BoxFuture};
7	use geoengine_datatypes::{
8	primitives::{CacheHint, RasterQueryRectangle, TimeInterval},
9	raster::{EmptyGrid2D, Pixel, RasterTile2D, TileInformation},
10	};
11	use rayon::ThreadPool;
12	use std::{marker::PhantomData, sync::Arc};
13
14	/// Only outputs the first tile as accumulator.
15	pub fn first_tile_fold_fn<T>(	6✔
16	acc: TemporalRasterAggregationTileAccu<T>,	6✔
17	tile: RasterTile2D<T>,	6✔
18	) -> TemporalRasterAggregationTileAccu<T>	6✔
19	where	6✔
20	T: Pixel,	6✔
21	{
22	if acc.initial_state {	6✔
23	let mut next_accu = tile;	2✔
24	next_accu.time = acc.accu_tile.time;	2✔
25
26	TemporalRasterAggregationTileAccu {	2✔
27	accu_tile: next_accu,	2✔
28	initial_state: false,	2✔
29	pool: acc.pool,	2✔
30	}	2✔
31	} else {
32	acc	4✔
33	}
34	}	6✔
35
36	pub fn first_tile_fold_future<T>(	6✔
37	accu: TemporalRasterAggregationTileAccu<T>,	6✔
38	tile: RasterTile2D<T>,	6✔
39	) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>>	6✔
40	where	6✔
41	T: Pixel,	6✔
42	{
43	crate::util::spawn_blocking(\|\| first_tile_fold_fn(accu, tile)).then(move \|x\| async move {	6✔
44	match x {	6✔
45	Ok(r) => Ok(r),	6✔
46	Err(e) => Err(e.into()),	×
47	}
48	})	12✔
49	}	6✔
50
51	/// Only outputs the last tile as accumulator.
52	#[allow(clippy::needless_pass_by_value)]
53	pub fn last_tile_fold_fn<T>(	12✔
54	acc: TemporalRasterAggregationTileAccu<T>,	12✔
55	tile: RasterTile2D<T>,	12✔
56	) -> TemporalRasterAggregationTileAccu<T>	12✔
57	where	12✔
58	T: Pixel,	12✔
59	{
60	let mut next_accu = tile;	12✔
61	next_accu.time = acc.accu_tile.time;	12✔
62
63	TemporalRasterAggregationTileAccu {	12✔
64	accu_tile: next_accu,	12✔
65	initial_state: false,	12✔
66	pool: acc.pool,	12✔
67	}	12✔
68	}	12✔
69
70	pub fn last_tile_fold_future<T>(	12✔
71	accu: TemporalRasterAggregationTileAccu<T>,	12✔
72	tile: RasterTile2D<T>,	12✔
73	) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>>	12✔
74	where	12✔
75	T: Pixel,	12✔
76	{
77	crate::util::spawn_blocking(\|\| last_tile_fold_fn(accu, tile)).then(move \|x\| async move {	12✔
78	match x {	12✔
79	Ok(r) => Ok(r),	12✔
80	Err(e) => Err(e.into()),	×
81	}
82	})	24✔
83	}	12✔
84
85	#[derive(Debug, Clone)]
86	pub struct TemporalRasterAggregationTileAccu<T> {
87	accu_tile: RasterTile2D<T>,
88	initial_state: bool,
89	pool: Arc<ThreadPool>,
90	}
91
92	#[async_trait]
93	impl<T: Pixel> FoldTileAccu for TemporalRasterAggregationTileAccu<T> {
94	type RasterType = T;
95
96	async fn into_tile(self) -> Result<RasterTile2D<Self::RasterType>> {	12✔
97	Ok(self.accu_tile)	6✔
98	}	12✔
99
100	fn thread_pool(&self) -> &Arc<ThreadPool> {	×
101	&self.pool	×
102	}	×
103	}
104
105	impl<T: Pixel> FoldTileAccuMut for TemporalRasterAggregationTileAccu<T> {
106	fn set_time(&mut self, time: TimeInterval) {	×
107	self.accu_tile.time = time;	×
108	}	×
109
110	fn set_cache_hint(&mut self, new_cache_hint: CacheHint) {	×
111	self.accu_tile.cache_hint = new_cache_hint;	×
112	}	×
113	}
114
115	#[derive(Debug, Clone)]
116	pub struct TemporalRasterAggregationSubQuery<F, T: Pixel> {
117	pub fold_fn: F,
118	pub _phantom_pixel_type: PhantomData<T>,
119	}
120
121	impl<'a, T, FoldM, FoldF> SubQueryTileAggregator<'a, T>
122	for TemporalRasterAggregationSubQuery<FoldM, T>
123	where
124	T: Pixel,
125	FoldM: Send
126	+ Sync
127	+ 'static
128	+ Clone
129	+ Fn(TemporalRasterAggregationTileAccu<T>, RasterTile2D<T>) -> FoldF,
130	FoldF: Send + TryFuture<Ok = TemporalRasterAggregationTileAccu<T>, Error = crate::error::Error>,
131	{
132	type TileAccu = TemporalRasterAggregationTileAccu<T>;
133	type TileAccuFuture = BoxFuture<'a, Result<Self::TileAccu>>;
134
135	type FoldFuture = FoldF;
136
137	type FoldMethod = FoldM;
138
139	fn new_fold_accu(	×
140	&self,	×
141	tile_info: TileInformation,	×
142	query_rect: RasterQueryRectangle,	×
143	pool: &Arc<ThreadPool>,	×
144	) -> Self::TileAccuFuture {	×
145	build_temporal_accu(&query_rect, tile_info, pool.clone()).boxed()	×
146	}	×
147
148	fn tile_query_rectangle(	×
149	&self,	×
150	tile_info: TileInformation,	×
151	_query_rect: RasterQueryRectangle,	×
NEW 152	time: TimeInterval,	×
153	band_idx: u32,	×
154	) -> Result<Option<RasterQueryRectangle>> {	×
155	// The time is already snapped by the operator where the time stream is created.
156	Ok(Some(RasterQueryRectangle::new(	×
157	tile_info.global_pixel_bounds(),	×
NEW 158	time,	×
159	band_idx.into(),	×
160	)))	×
161	}	×
162
163	fn fold_method(&self) -> Self::FoldMethod {	×
164	self.fold_fn.clone()	×
165	}	×
166	}
167
168	fn build_temporal_accu<T: Pixel>(	×
169	query_rect: &RasterQueryRectangle,	×
170	tile_info: TileInformation,	×
171	pool: Arc<ThreadPool>,	×
172	) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>> + use<T> {	×
173	let time_interval = query_rect.time_interval();	×
174	crate::util::spawn_blocking(move \|\| TemporalRasterAggregationTileAccu {	×
175	accu_tile: RasterTile2D::new_with_tile_info(	×
176	time_interval,	×
177	tile_info,	×
178	0,
179	EmptyGrid2D::new(tile_info.tile_size_in_pixels).into(),	×
180	CacheHint::max_duration(),	×
181	),
182
183	initial_state: true,
184	pool,	×
185	})	×
186	.map_err(From::from)	×
187	}	×
188
189	#[derive(Debug, Clone)]
190	pub struct TemporalRasterAggregationSubQueryNoDataOnly<F, T: Pixel> {
191	pub fold_fn: F,
192	pub _phantom_pixel_type: PhantomData<T>,
193	}
194
195	impl<'a, T, FoldM, FoldF> SubQueryTileAggregator<'a, T>
196	for TemporalRasterAggregationSubQueryNoDataOnly<FoldM, T>
197	where
198	T: Pixel,
199	FoldM: Send
200	+ Sync
201	+ 'static
202	+ Clone
203	+ Fn(TemporalRasterAggregationTileAccu<T>, RasterTile2D<T>) -> FoldF,
204	FoldF: Send + TryFuture<Ok = TemporalRasterAggregationTileAccu<T>, Error = crate::error::Error>,
205	{
206	type TileAccu = TemporalRasterAggregationTileAccu<T>;
207	type TileAccuFuture = BoxFuture<'a, Result<Self::TileAccu>>;
208	type FoldFuture = FoldF;
209
210	type FoldMethod = FoldM;
211
212	fn new_fold_accu(	6✔
213	&self,	6✔
214	tile_info: TileInformation,	6✔
215	query_rect: RasterQueryRectangle,	6✔
216	pool: &Arc<ThreadPool>,	6✔
217	) -> Self::TileAccuFuture {	6✔
218	build_temporal_no_data_accu(&query_rect, tile_info, pool.clone()).boxed()	6✔
219	}	6✔
220
221	fn tile_query_rectangle(	6✔
222	&self,	6✔
223	tile_info: TileInformation,	6✔
224	_query_rect: RasterQueryRectangle,	6✔
225	time: TimeInterval,	6✔
226	band_idx: u32,	6✔
227	) -> Result<Option<RasterQueryRectangle>> {	6✔
228	Ok(Some(RasterQueryRectangle::new(	6✔
229	tile_info.global_pixel_bounds(),	6✔
230	time, // The time is already snapped by the operator where the time stream is created.	6✔
231	band_idx.into(),	6✔
232	)))	6✔
233	}	6✔
234
235	fn fold_method(&self) -> Self::FoldMethod {	6✔
236	self.fold_fn.clone()	6✔
237	}	6✔
238	}
239
240	fn build_temporal_no_data_accu<T: Pixel>(	6✔
241	query_rect: &RasterQueryRectangle,	6✔
242	tile_info: TileInformation,	6✔
243	pool: Arc<ThreadPool>,	6✔
244	) -> impl Future<Output = Result<TemporalRasterAggregationTileAccu<T>>> + use<T> {	6✔
245	let time_interval = query_rect.time_interval();	6✔
246	crate::util::spawn_blocking(move \|\| {	6✔
247	let output_raster = EmptyGrid2D::new(tile_info.tile_size_in_pixels).into();	6✔
248
249	TemporalRasterAggregationTileAccu {	6✔
250	accu_tile: RasterTile2D::new_with_tile_info(	6✔
251	time_interval,	6✔
252	tile_info,	6✔
253	0,	6✔
254	output_raster,	6✔
255	CacheHint::max_duration(),	6✔
256	),	6✔
257	initial_state: true,	6✔
258	pool,	6✔
259	}	6✔
260	})	6✔
261	.map_err(From::from)	6✔
262	}	6✔

geo-engine / geoengine / 18890899227

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