17044350441

Committed 18 Aug 2025 03:00PM UTC coverage: 88.074%. First build

Build # 17044350441

Build Type

Pull #1015

github

Committed by

web-flow

Commit Message

Merge f38e4bbee into db8685e5e

Pull Request Pull Request #1015: workflow optimization (wip)

Coverage Stats

1666 of 2178 new or added lines in 51 files covered. (76.49%)

114929 of 130492 relevant lines covered (88.07%)

179856.12 hits per line

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

93.15

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

use std::sync::Arc;

use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedRasterOperator, InitializedSources, Operator,
    OperatorName, QueryContext, 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::{RasterQueryRectangle, SpatialResolution};
use geoengine_datatypes::raster::{
    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: Box<dyn RasterQueryProcessor<RasterType = 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: Box<dyn RasterQueryProcessor<RasterType = 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> RasterQueryProcessor for BandwiseExpressionProcessor<TO>
where
    TO: Pixel,
{
    type RasterType = TO;

    async fn raster_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 raster_result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }
}

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

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

geo-engine / geoengine / 17044350441

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