19242935849

Committed 10 Nov 2025 07:03PM UTC coverage: 88.832%. First build

Build # 19242935849

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge 44ecb3aa8 into 113de40ca

Pull Request Pull Request #1083: feat: new gdal source workflow optimization

Coverage Stats

6213 of 7052 new or added lines in 71 files covered. (88.1%)

116189 of 130797 relevant lines covered (88.83%)

496372.7 hits per line

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

92.0

/operators/src/processing/bandwise_expression/mod.rs

use std::sync::Arc;

use crate::engine::{
    BoxRasterQueryProcessor, CanonicOperatorName, ExecutionContext, InitializedRasterOperator,
    InitializedSources, Operator, OperatorName, QueryContext, QueryProcessor, RasterOperator,
    RasterQueryProcessor, RasterResultDescriptor, ResultDescriptor, SingleRasterSource,
    TypedRasterQueryProcessor, WorkflowOperatorPath,
};

use crate::optimization::OptimizationError;
use crate::util::Result;
use async_trait::async_trait;
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use geoengine_datatypes::primitives::{BandSelection, RasterQueryRectangle, SpatialResolution};
use geoengine_datatypes::raster::{
    GridBoundingBox2D, GridOrEmpty2D, MapElementsParallel, Pixel, RasterDataType, RasterTile2D,
};
use geoengine_expression::{
    DataType, ExpressionAst, ExpressionParser, LinkedExpression, Parameter,
};
use serde::{Deserialize, Serialize};

use super::RasterExpressionError;
use super::expression::get_expression_dependencies;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BandwiseExpressionParams {
    pub expression: String,
    pub output_type: RasterDataType,
    pub map_no_data: bool,
    // TODO: new unit for each band?
}

/// This `QueryProcessor` performs a unary expression on all bands of its input raster series.
pub type BandwiseExpression = Operator<BandwiseExpressionParams, SingleRasterSource>;

impl OperatorName for BandwiseExpression {
    const TYPE_NAME: &'static str = "BandwiseExpression";
}

#[typetag::serde]
#[async_trait]
impl RasterOperator for BandwiseExpression {
    async fn _initialize(
        self: Box<Self>,
        path: WorkflowOperatorPath,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedRasterOperator>> {
        let name = CanonicOperatorName::from(&self);

        let source = self
            .sources
            .initialize_sources(path.clone(), context)
            .await?
            .raster;

        let in_descriptor = source.result_descriptor();

        // TODO: ensure all bands have same measurement unit?

        let result_descriptor = in_descriptor.map_data_type(|_| self.params.output_type);

        let parameters = vec![Parameter::Number("x".into())];

        let expression = ExpressionParser::new(&parameters, DataType::Number)
            .map_err(RasterExpressionError::from)?
            .parse(
                "expression", // TODO: what is the name used for?
                &self.params.expression,
            )
            .map_err(RasterExpressionError::from)?;

        Ok(Box::new(InitializedBandwiseExpression {
            name,
            path,
            params: self.params.clone(),
            result_descriptor,
            source,
            expression,
            map_no_data: self.params.map_no_data,
        }))
    }

    span_fn!(BandwiseExpression);
}

pub struct InitializedBandwiseExpression {
    name: CanonicOperatorName,
    params: BandwiseExpressionParams,
    path: WorkflowOperatorPath,
    result_descriptor: RasterResultDescriptor,
    source: Box<dyn InitializedRasterOperator>,
    expression: ExpressionAst,
    map_no_data: bool,
}

impl InitializedRasterOperator for InitializedBandwiseExpression {
    fn result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }

    fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {
        let typed_raster_processor = self.source.query_processor()?.into_f64();

        let output_type = self.result_descriptor().data_type;

        // TODO: spawn a blocking task for the compilation process
        let expression_dependencies = get_expression_dependencies()
            .map_err(|source| RasterExpressionError::Dependencies { source })?;

        let expression = LinkedExpression::new(
            self.expression.name(),
            &self.expression.code(),
            expression_dependencies,
        )
        .map_err(RasterExpressionError::from)?;

        Ok(call_generic_raster_processor!(
            output_type,
            BandwiseExpressionProcessor::new(
                typed_raster_processor,
                self.result_descriptor.clone(),
                expression,
                self.map_no_data
            )
            .boxed()
        ))
    }

    fn canonic_name(&self) -> CanonicOperatorName {
        self.name.clone()
    }

    fn name(&self) -> &'static str {
        BandwiseExpression::TYPE_NAME
    }

    fn path(&self) -> WorkflowOperatorPath {
        self.path.clone()
    }

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn RasterOperator>, OptimizationError> {
        Ok(BandwiseExpression {
            params: self.params.clone(),
            sources: SingleRasterSource {
                raster: self.source.optimize(target_resolution)?,
            },
        }
        .boxed())
    }
}

pub(crate) struct BandwiseExpressionProcessor<TO> {
    source: BoxRasterQueryProcessor<f64>,
    result_descriptor: RasterResultDescriptor,
    expression: Arc<LinkedExpression>,
    map_no_data: bool,
    phantom: std::marker::PhantomData<TO>,
}

impl<TO> BandwiseExpressionProcessor<TO>
where
    TO: Pixel,
{
    pub fn new(
        source: BoxRasterQueryProcessor<f64>,
        result_descriptor: RasterResultDescriptor,
        expression: LinkedExpression,
        map_no_data: bool,
    ) -> Self {
        Self {
            source,
            result_descriptor,
            expression: Arc::new(expression),
            map_no_data,
            phantom: Default::default(),
        }
    }

    #[inline]
    fn compute_expression(
        raster: RasterTile2D<f64>,
        expression: &LinkedExpression,
        map_no_data: bool,
    ) -> Result<GridOrEmpty2D<TO>> {
        let expression = unsafe {
            // we have to "trust" that the function has the signature we expect
            expression
                .function_1::<Option<f64>>()
                .map_err(RasterExpressionError::from)?
        };

        let map_fn = |in_value: Option<f64>| {
            // TODO: could be a |in_value: T1| if map no data is false!
            if !map_no_data && in_value.is_none() {
                return None;
            }

            let result = expression(in_value);

            result.map(TO::from_)
        };

        let res = raster.grid_array.map_elements_parallel(map_fn);

        Result::Ok(res)
    }
}

#[async_trait]
impl<TO> QueryProcessor for BandwiseExpressionProcessor<TO>
where
    TO: Pixel,
{
    type Output = RasterTile2D<TO>;
    type ResultDescription = RasterResultDescriptor;
    type Selection = BandSelection;
    type SpatialBounds = GridBoundingBox2D;

    async fn _query<'a>(
        &'a self,
        query: RasterQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<RasterTile2D<TO>>>> {
        let stream = self
            .source
            .raster_query(query, ctx)
            .await?
            .and_then(move |tile| async move {
                let expression = self.expression.clone();
                let map_no_data = self.map_no_data;

                let time = tile.time;
                let tile_position = tile.tile_position;
                let band = tile.band;
                let global_geo_transform = tile.global_geo_transform;
                let cache_hint = tile.cache_hint;

                let out = crate::util::spawn_blocking_with_thread_pool(
                    ctx.thread_pool().clone(),
                    move || Self::compute_expression(tile, &expression, map_no_data),
                )
                .await??;

                Ok(RasterTile2D::new(
                    time,
                    tile_position,
                    band,
                    global_geo_transform,
                    out,
                    cache_hint,
                ))
            });

        Ok(stream.boxed())
    }

    fn result_descriptor(&self) -> &Self::ResultDescription {
        &self.result_descriptor
    }
}

#[async_trait]
impl<TO> RasterQueryProcessor for BandwiseExpressionProcessor<TO>
where
    TO: Pixel,
{
    type RasterType = TO;

    async fn _time_query<'a>(
        &'a self,
        query: geoengine_datatypes::primitives::TimeInterval,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<geoengine_datatypes::primitives::TimeInterval>>> {
        self.source.time_query(query, ctx).await
    }
}

#[cfg(test)]
mod tests {
    use geoengine_datatypes::{
        primitives::{CacheHint, TimeInterval, TimeStep},
        raster::{
            Grid, GridBoundingBox2D, GridShape, MapElements, RenameBands,
            TilesEqualIgnoringCacheHint,
        },
        spatial_reference::SpatialReference,
        util::test::TestDefault,
    };

    use crate::{
        engine::{
            MockExecutionContext, MultipleRasterSources, RasterBandDescriptors,
            SpatialGridDescriptor, TimeDescriptor,
        },
        mock::{MockRasterSource, MockRasterSourceParams},
        processing::{RasterStacker, RasterStackerParams},
    };

    use super::*;

    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn it_computes_bandwise_expression() {
        let data: Vec<RasterTile2D<u8>> = vec![
            RasterTile2D {
                time: TimeInterval::new_unchecked(0, 5),
                tile_position: [-1, 0].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![0, 1, 2, 3]).unwrap().into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(0, 5),
                tile_position: [-1, 1].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![4, 5, 6, 7]).unwrap().into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(5, 10),
                tile_position: [-1, 0].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![8, 9, 10, 11]).unwrap().into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(5, 10),
                tile_position: [-1, 1].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![12, 13, 14, 15])
                    .unwrap()
                    .into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
        ];

        let data2: Vec<RasterTile2D<u8>> = vec![
            RasterTile2D {
                time: TimeInterval::new_unchecked(0, 5),
                tile_position: [-1, 0].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![16, 17, 18, 19])
                    .unwrap()
                    .into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(0, 5),
                tile_position: [-1, 1].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![20, 21, 22, 23])
                    .unwrap()
                    .into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(5, 10),
                tile_position: [-1, 0].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![24, 25, 26, 27])
                    .unwrap()
                    .into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
            RasterTile2D {
                time: TimeInterval::new_unchecked(5, 10),
                tile_position: [-1, 1].into(),
                band: 0,
                global_geo_transform: TestDefault::test_default(),
                grid_array: Grid::new([2, 2].into(), vec![28, 29, 30, 31])
                    .unwrap()
                    .into(),
                properties: Default::default(),
                cache_hint: CacheHint::default(),
            },
        ];

        let result_descriptor = RasterResultDescriptor {
            data_type: RasterDataType::U8,
            spatial_reference: SpatialReference::epsg_4326().into(),
            time: TimeDescriptor::new_regular_with_epoch(
                Some(
                    TimeInterval::new(
                        data.first().unwrap().time.start(),
                        data.last().unwrap().time.end(),
                    )
                    .unwrap(),
                ),
                TimeStep::millis(5),
            ),
            spatial_grid: SpatialGridDescriptor::source_from_parts(
                TestDefault::test_default(),
                GridBoundingBox2D::new_min_max(-2, -1, 0, 3).unwrap(),
            ),
            bands: RasterBandDescriptors::new_single_band(),
        };

        let mrs1 = MockRasterSource {
            params: MockRasterSourceParams {
                data: data.clone(),
                result_descriptor: result_descriptor.clone(),
            },
        }
        .boxed();

        let mrs2 = MockRasterSource {
            params: MockRasterSourceParams {
                data: data2.clone(),
                result_descriptor,
            },
        }
        .boxed();

        let stacker = RasterStacker {
            params: RasterStackerParams {
                rename_bands: RenameBands::Default,
            },
            sources: MultipleRasterSources {
                rasters: vec![mrs1, mrs2],
            },
        }
        .boxed();

        let expression = BandwiseExpression {
            params: BandwiseExpressionParams {
                expression: "x + 1".to_string(),
                output_type: RasterDataType::U8,
                map_no_data: false,
            },
            sources: SingleRasterSource { raster: stacker },
        }
        .boxed();

        let mut exe_ctx = MockExecutionContext::test_default();
        exe_ctx.tiling_specification.tile_size_in_pixels = GridShape {
            shape_array: [2, 2],
        };

        let query_rect = RasterQueryRectangle::new(
            GridBoundingBox2D::new_min_max(-2, -1, 0, 3).unwrap(),
            TimeInterval::new_unchecked(0, 10),
            [0, 1].try_into().unwrap(),
        );

        let query_ctx = exe_ctx.mock_query_context_test_default();

        let op = expression
            .initialize(WorkflowOperatorPath::initialize_root(), &exe_ctx)
            .await
            .unwrap();

        let qp = op.query_processor().unwrap().get_u8().unwrap();

        let result = qp
            .raster_query(query_rect, &query_ctx)
            .await
            .unwrap()
            .collect::<Vec<_>>()
            .await;
        let result = result.into_iter().collect::<Result<Vec<_>>>().unwrap();

        let expected: Vec<RasterTile2D<u8>> = data
            .into_iter()
            .zip(data2.into_iter().map(|mut tile| {
                tile.band = 1;
                tile
            }))
            .flat_map(|(a, b)| vec![a.clone(), b.clone()])
            .map(|mut tile| {
                tile.grid_array = tile.grid_array.map_elements(|in_value: u8| in_value + 1);
                tile
            })
            .collect();

        assert!(expected.tiles_equal_ignoring_cache_hint(&result));
    }
}

1	use std::sync::Arc;
2
3	use crate::engine::{
4	BoxRasterQueryProcessor, CanonicOperatorName, ExecutionContext, InitializedRasterOperator,
5	InitializedSources, Operator, OperatorName, QueryContext, QueryProcessor, RasterOperator,
6	RasterQueryProcessor, RasterResultDescriptor, ResultDescriptor, SingleRasterSource,
7	TypedRasterQueryProcessor, WorkflowOperatorPath,
8	};
9
10	use crate::optimization::OptimizationError;
11	use crate::util::Result;
12	use async_trait::async_trait;
13	use futures::stream::BoxStream;
14	use futures::{StreamExt, TryStreamExt};
15	use geoengine_datatypes::primitives::{BandSelection, RasterQueryRectangle, SpatialResolution};
16	use geoengine_datatypes::raster::{
17	GridBoundingBox2D, GridOrEmpty2D, MapElementsParallel, Pixel, RasterDataType, RasterTile2D,
18	};
19	use geoengine_expression::{
20	DataType, ExpressionAst, ExpressionParser, LinkedExpression, Parameter,
21	};
22	use serde::{Deserialize, Serialize};
23
24	use super::RasterExpressionError;
25	use super::expression::get_expression_dependencies;
26
27	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28	#[serde(rename_all = "camelCase")]
29	pub struct BandwiseExpressionParams {
30	pub expression: String,
31	pub output_type: RasterDataType,
32	pub map_no_data: bool,
33	// TODO: new unit for each band?
34	}
35
36	/// This `QueryProcessor` performs a unary expression on all bands of its input raster series.
37	pub type BandwiseExpression = Operator<BandwiseExpressionParams, SingleRasterSource>;
38
39	impl OperatorName for BandwiseExpression {
40	const TYPE_NAME: &'static str = "BandwiseExpression";
41	}
42
43	#[typetag::serde]	×
44	#[async_trait]
45	impl RasterOperator for BandwiseExpression {
46	async fn _initialize(
47	self: Box<Self>,
48	path: WorkflowOperatorPath,
49	context: &dyn ExecutionContext,
50	) -> Result<Box<dyn InitializedRasterOperator>> {	1✔
51	let name = CanonicOperatorName::from(&self);
52
53	let source = self
54	.sources
55	.initialize_sources(path.clone(), context)
56	.await?
57	.raster;
58
59	let in_descriptor = source.result_descriptor();
60
61	// TODO: ensure all bands have same measurement unit?
62
63	let result_descriptor = in_descriptor.map_data_type(\|_\| self.params.output_type);
64
65	let parameters = vec![Parameter::Number("x".into())];
66
67	let expression = ExpressionParser::new(&parameters, DataType::Number)
68	.map_err(RasterExpressionError::from)?
69	.parse(
70	"expression", // TODO: what is the name used for?
71	&self.params.expression,
72	)
73	.map_err(RasterExpressionError::from)?;
74
75	Ok(Box::new(InitializedBandwiseExpression {
76	name,
77	path,
78	params: self.params.clone(),
79	result_descriptor,
80	source,
81	expression,
82	map_no_data: self.params.map_no_data,
83	}))
84	}	1✔
85
86	span_fn!(BandwiseExpression);
87	}
88
89	pub struct InitializedBandwiseExpression {
90	name: CanonicOperatorName,
91	params: BandwiseExpressionParams,
92	path: WorkflowOperatorPath,
93	result_descriptor: RasterResultDescriptor,
94	source: Box<dyn InitializedRasterOperator>,
95	expression: ExpressionAst,
96	map_no_data: bool,
97	}
98
99	impl InitializedRasterOperator for InitializedBandwiseExpression {
100	fn result_descriptor(&self) -> &RasterResultDescriptor {	1✔
101	&self.result_descriptor	1✔
102	}	1✔
103
104	fn query_processor(&self) -> Result<TypedRasterQueryProcessor> {	1✔
105	let typed_raster_processor = self.source.query_processor()?.into_f64();	1✔
106
107	let output_type = self.result_descriptor().data_type;	1✔
108
109	// TODO: spawn a blocking task for the compilation process
110	let expression_dependencies = get_expression_dependencies()	1✔
111	.map_err(\|source\| RasterExpressionError::Dependencies { source })?;	1✔
112
113	let expression = LinkedExpression::new(	1✔
114	self.expression.name(),	1✔
115	&self.expression.code(),	1✔
116	expression_dependencies,	1✔
117	)
118	.map_err(RasterExpressionError::from)?;	1✔
119
120	Ok(call_generic_raster_processor!(	×
121	output_type,	1✔
122	BandwiseExpressionProcessor::new(	1✔
123	typed_raster_processor,	1✔
124	self.result_descriptor.clone(),	1✔
125	expression,	1✔
126	self.map_no_data	1✔
127	)	1✔
128	.boxed()	1✔
129	))
130	}	1✔
131
132	fn canonic_name(&self) -> CanonicOperatorName {	×
133	self.name.clone()	×
134	}	×
135
136	fn name(&self) -> &'static str {	×
137	BandwiseExpression::TYPE_NAME	×
138	}	×
139
140	fn path(&self) -> WorkflowOperatorPath {	×
141	self.path.clone()	×
142	}	×
143
NEW 144	fn optimize(	×
NEW 145	&self,	×
NEW 146	target_resolution: SpatialResolution,	×
NEW 147	) -> Result<Box<dyn RasterOperator>, OptimizationError> {	×
148	Ok(BandwiseExpression {
NEW 149	params: self.params.clone(),	×
150	sources: SingleRasterSource {
NEW 151	raster: self.source.optimize(target_resolution)?,	×
152	},
153	}
NEW 154	.boxed())	×
NEW 155	}	×
156	}
157
158	pub(crate) struct BandwiseExpressionProcessor<TO> {
159	source: BoxRasterQueryProcessor<f64>,
160	result_descriptor: RasterResultDescriptor,
161	expression: Arc<LinkedExpression>,
162	map_no_data: bool,
163	phantom: std::marker::PhantomData<TO>,
164	}
165
166	impl<TO> BandwiseExpressionProcessor<TO>
167	where
168	TO: Pixel,
169	{
170	pub fn new(	1✔
171	source: BoxRasterQueryProcessor<f64>,	1✔
172	result_descriptor: RasterResultDescriptor,	1✔
173	expression: LinkedExpression,	1✔
174	map_no_data: bool,	1✔
175	) -> Self {	1✔
176	Self {	1✔
177	source,	1✔
178	result_descriptor,	1✔
179	expression: Arc::new(expression),	1✔
180	map_no_data,	1✔
181	phantom: Default::default(),	1✔
182	}	1✔
183	}	1✔
184
185	#[inline]
186	fn compute_expression(	8✔
187	raster: RasterTile2D<f64>,	8✔
188	expression: &LinkedExpression,	8✔
189	map_no_data: bool,	8✔
190	) -> Result<GridOrEmpty2D<TO>> {	8✔
191	let expression = unsafe {	8✔
192	// we have to "trust" that the function has the signature we expect
193	expression	8✔
194	.function_1::<Option<f64>>()	8✔
195	.map_err(RasterExpressionError::from)?	8✔
196	};
197
198	let map_fn = \|in_value: Option<f64>\| {	32✔
199	// TODO: could be a \|in_value: T1\| if map no data is false!
200	if !map_no_data && in_value.is_none() {	32✔
201	return None;	×
202	}	32✔
203
204	let result = expression(in_value);	32✔
205
206	result.map(TO::from_)	32✔
207	};	32✔
208
209	let res = raster.grid_array.map_elements_parallel(map_fn);	8✔
210
211	Result::Ok(res)	8✔
212	}	8✔
213	}
214
215	#[async_trait]
216	impl<TO> QueryProcessor for BandwiseExpressionProcessor<TO>
217	where
218	TO: Pixel,
219	{
220	type Output = RasterTile2D<TO>;
221	type ResultDescription = RasterResultDescriptor;
222	type Selection = BandSelection;
223	type SpatialBounds = GridBoundingBox2D;
224
225	async fn _query<'a>(
226	&'a self,
227	query: RasterQueryRectangle,
228	ctx: &'a dyn QueryContext,
229	) -> Result<BoxStream<'a, Result<RasterTile2D<TO>>>> {	1✔
230	let stream = self
231	.source
232	.raster_query(query, ctx)
233	.await?
234	.and_then(move \|tile\| async move {	8✔
235	let expression = self.expression.clone();	8✔
236	let map_no_data = self.map_no_data;	8✔
237
238	let time = tile.time;	8✔
239	let tile_position = tile.tile_position;	8✔
240	let band = tile.band;	8✔
241	let global_geo_transform = tile.global_geo_transform;	8✔
242	let cache_hint = tile.cache_hint;	8✔
243
244	let out = crate::util::spawn_blocking_with_thread_pool(	8✔
245	ctx.thread_pool().clone(),	8✔
246	move \|\| Self::compute_expression(tile, &expression, map_no_data),	8✔
247	)
248	.await??;	8✔
249
250	Ok(RasterTile2D::new(	8✔
251	time,	8✔
252	tile_position,	8✔
253	band,	8✔
254	global_geo_transform,	8✔
255	out,	8✔
256	cache_hint,	8✔
257	))	8✔
258	});	16✔
259
260	Ok(stream.boxed())
261	}	1✔
262
263	fn result_descriptor(&self) -> &Self::ResultDescription {	2✔
264	&self.result_descriptor	2✔
265	}	2✔
266	}
267
268	#[async_trait]
269	impl<TO> RasterQueryProcessor for BandwiseExpressionProcessor<TO>
270	where
271	TO: Pixel,
272	{
273	type RasterType = TO;
274
275	async fn _time_query<'a>(
276	&'a self,
277	query: geoengine_datatypes::primitives::TimeInterval,
278	ctx: &'a dyn QueryContext,
279	) -> Result<BoxStream<'a, Result<geoengine_datatypes::primitives::TimeInterval>>> {	×
280	self.source.time_query(query, ctx).await
281	}	×
282	}
283
284	#[cfg(test)]
285	mod tests {
286	use geoengine_datatypes::{
287	primitives::{CacheHint, TimeInterval, TimeStep},
288	raster::{
289	Grid, GridBoundingBox2D, GridShape, MapElements, RenameBands,
290	TilesEqualIgnoringCacheHint,
291	},
292	spatial_reference::SpatialReference,
293	util::test::TestDefault,
294	};
295
296	use crate::{
297	engine::{
298	MockExecutionContext, MultipleRasterSources, RasterBandDescriptors,
299	SpatialGridDescriptor, TimeDescriptor,
300	},
301	mock::{MockRasterSource, MockRasterSourceParams},
302	processing::{RasterStacker, RasterStackerParams},
303	};
304
305	use super::*;
306
307	#[tokio::test]
308	#[allow(clippy::too_many_lines)]
309	async fn it_computes_bandwise_expression() {	1✔
310	let data: Vec<RasterTile2D<u8>> = vec![	1✔
311	RasterTile2D {	1✔
312	time: TimeInterval::new_unchecked(0, 5),	1✔
313	tile_position: [-1, 0].into(),	1✔
314	band: 0,	1✔
315	global_geo_transform: TestDefault::test_default(),	1✔
316	grid_array: Grid::new([2, 2].into(), vec![0, 1, 2, 3]).unwrap().into(),	1✔
317	properties: Default::default(),	1✔
318	cache_hint: CacheHint::default(),	1✔
319	},	1✔
320	RasterTile2D {	1✔
321	time: TimeInterval::new_unchecked(0, 5),	1✔
322	tile_position: [-1, 1].into(),	1✔
323	band: 0,	1✔
324	global_geo_transform: TestDefault::test_default(),	1✔
325	grid_array: Grid::new([2, 2].into(), vec![4, 5, 6, 7]).unwrap().into(),	1✔
326	properties: Default::default(),	1✔
327	cache_hint: CacheHint::default(),	1✔
328	},	1✔
329	RasterTile2D {	1✔
330	time: TimeInterval::new_unchecked(5, 10),	1✔
331	tile_position: [-1, 0].into(),	1✔
332	band: 0,	1✔
333	global_geo_transform: TestDefault::test_default(),	1✔
334	grid_array: Grid::new([2, 2].into(), vec![8, 9, 10, 11]).unwrap().into(),	1✔
335	properties: Default::default(),	1✔
336	cache_hint: CacheHint::default(),	1✔
337	},	1✔
338	RasterTile2D {	1✔
339	time: TimeInterval::new_unchecked(5, 10),	1✔
340	tile_position: [-1, 1].into(),	1✔
341	band: 0,	1✔
342	global_geo_transform: TestDefault::test_default(),	1✔
343	grid_array: Grid::new([2, 2].into(), vec![12, 13, 14, 15])	1✔
344	.unwrap()	1✔
345	.into(),	1✔
346	properties: Default::default(),	1✔
347	cache_hint: CacheHint::default(),	1✔
348	},	1✔
349	];
350
351	let data2: Vec<RasterTile2D<u8>> = vec![	1✔
352	RasterTile2D {	1✔
353	time: TimeInterval::new_unchecked(0, 5),	1✔
354	tile_position: [-1, 0].into(),	1✔
355	band: 0,	1✔
356	global_geo_transform: TestDefault::test_default(),	1✔
357	grid_array: Grid::new([2, 2].into(), vec![16, 17, 18, 19])	1✔
358	.unwrap()	1✔
359	.into(),	1✔
360	properties: Default::default(),	1✔
361	cache_hint: CacheHint::default(),	1✔
362	},	1✔
363	RasterTile2D {	1✔
364	time: TimeInterval::new_unchecked(0, 5),	1✔
365	tile_position: [-1, 1].into(),	1✔
366	band: 0,	1✔
367	global_geo_transform: TestDefault::test_default(),	1✔
368	grid_array: Grid::new([2, 2].into(), vec![20, 21, 22, 23])	1✔
369	.unwrap()	1✔
370	.into(),	1✔
371	properties: Default::default(),	1✔
372	cache_hint: CacheHint::default(),	1✔
373	},	1✔
374	RasterTile2D {	1✔
375	time: TimeInterval::new_unchecked(5, 10),	1✔
376	tile_position: [-1, 0].into(),	1✔
377	band: 0,	1✔
378	global_geo_transform: TestDefault::test_default(),	1✔
379	grid_array: Grid::new([2, 2].into(), vec![24, 25, 26, 27])	1✔
380	.unwrap()	1✔
381	.into(),	1✔
382	properties: Default::default(),	1✔
383	cache_hint: CacheHint::default(),	1✔
384	},	1✔
385	RasterTile2D {	1✔
386	time: TimeInterval::new_unchecked(5, 10),	1✔
387	tile_position: [-1, 1].into(),	1✔
388	band: 0,	1✔
389	global_geo_transform: TestDefault::test_default(),	1✔
390	grid_array: Grid::new([2, 2].into(), vec![28, 29, 30, 31])	1✔
391	.unwrap()	1✔
392	.into(),	1✔
393	properties: Default::default(),	1✔
394	cache_hint: CacheHint::default(),	1✔
395	},	1✔
396	];
397
398	let result_descriptor = RasterResultDescriptor {	1✔
399	data_type: RasterDataType::U8,	1✔
400	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
401	time: TimeDescriptor::new_regular_with_epoch(	1✔
402	Some(	1✔
403	TimeInterval::new(	1✔
404	data.first().unwrap().time.start(),	1✔
405	data.last().unwrap().time.end(),	1✔
406	)	1✔
407	.unwrap(),	1✔
408	),	1✔
409	TimeStep::millis(5),	1✔
410	),	1✔
411	spatial_grid: SpatialGridDescriptor::source_from_parts(	1✔
412	TestDefault::test_default(),	1✔
413	GridBoundingBox2D::new_min_max(-2, -1, 0, 3).unwrap(),	1✔
414	),	1✔
415	bands: RasterBandDescriptors::new_single_band(),	1✔
416	};	1✔
417
418	let mrs1 = MockRasterSource {	1✔
419	params: MockRasterSourceParams {	1✔
420	data: data.clone(),	1✔
421	result_descriptor: result_descriptor.clone(),	1✔
422	},	1✔
423	}	1✔
424	.boxed();	1✔
425
426	let mrs2 = MockRasterSource {	1✔
427	params: MockRasterSourceParams {	1✔
428	data: data2.clone(),	1✔
429	result_descriptor,	1✔
430	},	1✔
431	}	1✔
432	.boxed();	1✔
433
434	let stacker = RasterStacker {	1✔
435	params: RasterStackerParams {	1✔
436	rename_bands: RenameBands::Default,	1✔
437	},	1✔
438	sources: MultipleRasterSources {	1✔
439	rasters: vec![mrs1, mrs2],	1✔
440	},	1✔
441	}	1✔
442	.boxed();	1✔
443
444	let expression = BandwiseExpression {	1✔
445	params: BandwiseExpressionParams {	1✔
446	expression: "x + 1".to_string(),	1✔
447	output_type: RasterDataType::U8,	1✔
448	map_no_data: false,	1✔
449	},	1✔
450	sources: SingleRasterSource { raster: stacker },	1✔
451	}	1✔
452	.boxed();	1✔
453
454	let mut exe_ctx = MockExecutionContext::test_default();	1✔
455	exe_ctx.tiling_specification.tile_size_in_pixels = GridShape {	1✔
456	shape_array: [2, 2],	1✔
457	};	1✔
458
459	let query_rect = RasterQueryRectangle::new(	1✔
460	GridBoundingBox2D::new_min_max(-2, -1, 0, 3).unwrap(),	1✔
461	TimeInterval::new_unchecked(0, 10),	1✔
462	[0, 1].try_into().unwrap(),	1✔
463	);
464
465	let query_ctx = exe_ctx.mock_query_context_test_default();	1✔
466
467	let op = expression	1✔
468	.initialize(WorkflowOperatorPath::initialize_root(), &exe_ctx)	1✔
469	.await	1✔
470	.unwrap();	1✔
471
472	let qp = op.query_processor().unwrap().get_u8().unwrap();	1✔
473
474	let result = qp	1✔
475	.raster_query(query_rect, &query_ctx)	1✔
476	.await	1✔
477	.unwrap()	1✔
478	.collect::<Vec<_>>()	1✔
479	.await;	1✔
480	let result = result.into_iter().collect::<Result<Vec<_>>>().unwrap();	1✔
481
482	let expected: Vec<RasterTile2D<u8>> = data	1✔
483	.into_iter()	1✔
484	.zip(data2.into_iter().map(\|mut tile\| {	4✔
485	tile.band = 1;	4✔
486	tile	4✔
487	}))	4✔
488	.flat_map(\|(a, b)\| vec![a.clone(), b.clone()])	4✔
489	.map(\|mut tile\| {	8✔
490	tile.grid_array = tile.grid_array.map_elements(\|in_value: u8\| in_value + 1);	32✔
491	tile	8✔
492	})	8✔
493	.collect();	1✔
494
495	assert!(expected.tiles_equal_ignoring_cache_hint(&result));	1✔
496	}	1✔
497	}

geo-engine / geoengine / 19242935849

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